Source/DPCoroutine.m

   1 //
   2 //  DPCoroutine.m
   3 //  HigherOrderMessaging
   4 //
   5 //  Created by Ofri Wolfus on 01/08/07.
   6 //  Copyright 2007 Ofri Wolfus. All rights reserved.
   7 //
   8
   9 #import "DPCoroutine.h"
  10 #include <libkern/OSAtomic.h>
  11
  12
  13 @interface DPCoroutineStack (private)
  14
  15 - (DPCoroutine *)mainCoro;
  16 - (DPCoroutine *)newCoroutineWithMessage:(DPMessage *)msg target:(id)t;
  17 - (BOOL)nextCoro;
  18 - (void)_setMain;
  19
  20 @end
  21
  22 @interface DPCoroutine (private)
  23
  24 + (void)setCurrentCoroutine:(id)coro;
  25
  26 - (void)setMain;
  27 - (BOOL)isMain;
  28
  29 - (BOOL)continue;
  30 - (void)return;
  31 - (void)resetWithMessage:(DPMessage *)msg;
  32 - (void)setTarget:(id)target;
  33
  34 - (BOOL)isActive;
  35
  36 - (void)start;
  37 - (Coro *)_coro;
  38 - (void)setParent:(DPCoroutine *)co;
  39
  40 @end
  41
  42 @implementation DPCoroutine
  43
  44 #define DPCURRENT_COROUTINE_KEY @"DPCurrentCoroutine"
  45
  46 + (id)currentCoroutine {
  47         return [[[NSThread currentThread] threadDictionary] objectForKey:DPCURRENT_COROUTINE_KEY];
  48 }
  49
  50 + (void)setCurrentCoroutine:(id)coro {
  51         if (coro)
  52                 [[[NSThread currentThread] threadDictionary] setObject:coro forKey:DPCURRENT_COROUTINE_KEY];
  53         else
  54                 [[[NSThread currentThread] threadDictionary] removeObjectForKey:DPCURRENT_COROUTINE_KEY];
  55 }
  56
  57 - (id)init {
  58         if ((self = [super init])) {
  59                 _refCount = 1;
  60                 _coro = Coro_new();
  61                 isActive = NO;
  62                 message = nil;
  63         }
  64
  65         return self;
  66 }
  67
  68 - (id)initWithMessage:(DPMessage *)msg target:(id)t {
  69         // Can't initialize without a message or a targe
  70         if (!msg || !t) {
  71                 [self release];
  72                 return nil;
  73         }
  74
  75         if ((self = [self init])) {
  76                 message = [msg copy];
  77                 target = [t retain];
  78         }
  79
  80         return self;
  81 }
  82
  83 - (void)dealloc {
  84         [target release]; target = nil;
  85         [message release]; message = nil;
  86         Coro_free(_coro);
  87         [super dealloc];
  88 }
  89
  90 - (id)retain {
  91         OSAtomicIncrement32Barrier(&_refCount);
  92         return self;
  93 }
  94
  95 - (void)release {
  96         if (OSAtomicDecrement32Barrier(&_refCount) == 0)
  97                 [self dealloc];
  98 }
  99
 100 - (unsigned)retainCount {
 101         OSMemoryBarrier();
 102         return (unsigned)_refCount;
 103 }
 104
 105 - (void)setMain {
 106         if (!_coro->isMain) {
 107                 Coro_initializeMainCoro(_coro);
 108                 isActive = YES;
 109         }
 110 }
 111
 112 - (BOOL)isMain {
 113         return _coro->isMain;
 114 }
 115
 116 - (void)resetWithMessage:(DPMessage *)msg {
 117         //Coro_clear(_coro);
 118         [message release];
 119         message = [msg retain];
 120 }
 121
 122 - (void)setTarget:(id)t {
 123         if (target != t) {
 124                 [target release];
 125                 target = [t retain];
 126         }
 127 }
 128
 129 - (void)yield {
 130         //Coro_yield();
 131         if (__builtin_expect(Coro_stackSpaceAlmostGone(_coro), 0)) {
 132                 NSLog(@"Coroutine %@'s stack is almost full. Allocating some more in hope not to crash.", self);
 133                 Coro_setStackSize_(_coro, Coro_stackSize(_coro) * 2);
 134         }
 135
 136         Coro_switchTo_(_coro, [parent _coro]);
 137 }
 138
 139 - (void)return {
 140         isActive = NO;
 141         //Coro_end();
 142         [self yield];
 143 }
 144
 145 - (BOOL)isActive {
 146         return isActive;
 147 }
 148
 149 static void _DPCoroEntry(DPCoroutine *coro) {
 150         coro->isActive = YES;
 151         [coro->message sendTo:coro->target];
 152         [coro return];
 153 }
 154
 155 - (Coro *)_coro {
 156         return _coro;
 157 }
 158
 159 - (void)start {
 160         NSMutableDictionary *dict = [[NSThread currentThread] threadDictionary];
 161         [dict setObject:self forKey:DPCURRENT_COROUTINE_KEY];
 162         Coro_startCoro_([parent _coro], _coro, self, (void *)_DPCoroEntry);
 163         [dict setObject:parent forKey:DPCURRENT_COROUTINE_KEY];
 164 }
 165
 166 - (BOOL)continue {
 167         if (isActive) {
 168                 NSMutableDictionary *dict = [[NSThread currentThread] threadDictionary];
 169
 170                 [dict setObject:self forKey:DPCURRENT_COROUTINE_KEY];
 171                 Coro_switchTo_([parent _coro], _coro);
 172                 [dict setObject:parent forKey:DPCURRENT_COROUTINE_KEY];
 173                 return isActive;
 174         }
 175
 176         return NO;
 177 }
 178
 179 - (void)setParent:(DPCoroutine *)co {
 180         parent = co;
 181 }
 182
 183 @end
 184
 185 @implementation DPCoroutineStack
 186
 187 static DPCoroutineStack *_mainStack = nil;
 188
 189 static void __attribute__((constructor)) _DPSetupMainCoroStack(void) {
 190         _mainStack = [[DPCoroutineStack alloc] init];
 191         [DPCoroutineStack setCurrentStack:_mainStack];
 192         [[NSRunLoop currentRunLoop] addSource:_mainStack
 193                                                                   forMode:NSDefaultRunLoopMode];
 194 }
 195
 196 + (id)mainStack {
 197         return _mainStack;
 198 }
 199
 200 + (id)currentStack {
 201         return [[[NSThread currentThread] threadDictionary] objectForKey:@"DPCoroutineStack"];
 202 }
 203
 204 + (void)setCurrentStack:(id)stack {
 205         if (stack)
 206                 [[[NSThread currentThread] threadDictionary] setObject:stack
 207                                                                                                                 forKey:@"DPCoroutineStack"];
 208         else
 209                 [[[NSThread currentThread] threadDictionary] removeObjectForKey:@"DPCoroutineStack"];
 210 }
 211
 212 - (id)init {
 213         if ((self = [super init])) {
 214                 mainCoros = [[NSMutableDictionary alloc] init];
 215                 mainCorosLock = OS_SPINLOCK_INIT;
 216                 stack = [[DPQueue alloc] init];
 217                 unusedCoroutines = [[DPQueue alloc] init];
 218                 activeRoutines = 0;
 219         }
 220
 221         return self;
 222 }
 223
 224 - (void)dealloc {
 225         [unusedCoroutines release]; unusedCoroutines = nil;
 226         [stack release]; stack = nil;
 227         [mainCoros release]; mainCoros = nil;
 228         [super dealloc];
 229 }
 230
 231 - (DPCoroutine *)newCoroutineWithMessage:(DPMessage *)msg target:(id)target {
 232         DPCoroutine *coro = [unusedCoroutines dequeue];
 233
 234         if (coro) {
 235                 [coro resetWithMessage:msg];
 236                 [coro setTarget:target];
 237         } else {
 238                 coro = [DPCoroutine new];
 239         }
 240
 241         return [coro autorelease];
 242 }
 243
 244 - (void)addCoro:(DPCoroutine *)coro {
 245         [stack enqueu:coro];
 246         [coro setParent:[self mainCoro]];
 247         [self signal];
 248 }
 249
 250 - (BOOL)nextCoro {
 251         DPCoroutine *coro;
 252
 253         OSAtomicIncrement32Barrier(&activeRoutines);
 254         coro = [stack dequeue];
 255
 256         if (coro) {
 257                 BOOL shouldContinue;
 258
 259                 [coro retain];
 260                 [self retain];
 261                 [self signal];
 262
 263                 if ([coro isActive]) {
 264                         shouldContinue = [coro continue];
 265                 } else {
 266                         [coro start];
 267                         shouldContinue = [coro isActive];
 268                 }
 269
 270                 // There's no active coroutine
 271                 [DPCoroutine setCurrentCoroutine:nil];
 272
 273                 // Clean up
 274                 shouldContinue ? [stack enqueu:coro] : [unusedCoroutines enqueu:coro];
 275                 [coro release];
 276                 [self release];
 277         }
 278
 279         OSAtomicDecrement32Barrier(&activeRoutines);
 280
 281         return coro != nil;
 282 }
 283
 284 - (DPCoroutine *)mainCoro {
 285         DPThreadID *trID = [DPThreadID currentIdentifier];
 286         OSSpinLockLock(&mainCorosLock);
 287         DPCoroutine *coro = [mainCoros objectForKey:trID];
 288
 289         if (!coro) {
 290                 coro = [[DPCoroutine alloc] init];
 291                 [coro setMain];
 292                 [mainCoros setObject:coro forKey:trID];
 293                 [coro release];
 294         }
 295         OSSpinLockUnlock(&mainCorosLock);
 296         return coro;
 297 }
 298
 299 - (void)empty {
 300         while ([self hasWork])
 301                 [self nextCoro];
 302 }
 303
 304 - (void)fire {
 305         [self nextCoro];
 306 }
 307
 308 - (BOOL)hasWork {
 309         return [stack count] + activeRoutines > 0;
 310 }
 311
 312 @end
 313
 314 @implementation NSObject (DPCoroutine)
 315
 316 - (void)coroutine:(DPMessage *)msg {
 317         DPCoroutine *coro = [[DPCoroutine alloc] initWithMessage:msg
 318                                                                                                           target:self];
 319         [[DPCoroutineStack currentStack] addCoro:coro];
 320         [coro release];
 321 }
 322
 323 @end