DPWorkerThread now sets up a coroutine stack.

[hom.git] / Source / DPMultiThreading.m

//
//  DPMultiThreading.m
//  HigherOrderMessaging
//
//  Created by Ofri Wolfus on 25/05/07.
//  Copyright 2007 Ofri Wolfus. All rights reserved.
//
//  Redistribution and use in source and binary forms, with or without modification,
//  are permitted provided that the following conditions are met:
//  
//  1. Redistributions of source code must retain the above copyright
//  notice, this list of conditions and the following disclaimer.
//  2. Redistributions in binary form must reproduce the above copyright
//  notice, this list of conditions and the following disclaimer in the
//  documentation and/or other materials provided with the distribution.
//  3. Neither the name of Ofri Wolfus nor the names of his contributors
//  may be used to endorse or promote products derived from this software
//  without specific prior written permission.
//
//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
//  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
//  WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
//  IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
//  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
//  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
//  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
//  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
//  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//

#include <libkern/OSAtomic.h>
#import "DPMultiThreading.h"
#import "DPCoroutine.h"


@interface NSInvocation (private)
// This is a private method that can be used to set
// a marg_list of an invocation.
// It's usually a bad idea to use private methods like this
// but there's no better alternative.
- (void)_setArgFrame:(void *)fp8;
@end

@implementation NSInvocation (DPMessageSupport)

+ (id)invocationWithMessage:(DPMessage *)msg receiver:(id)target {
        NSMethodSignature *sig = nil;
        NSInvocation *i = nil;
        
        // Let's be on the safe side
        NSAssert(msg != nil, @"Can't create an invocation from nil message!");
        NSAssert(target != nil, @"Can't create an invocation with no target."
                         " Isn't that what DPMessage is for?");
        
        sig = [target methodSignatureForSelector:[msg selector]];
        
        // Apparently +[NSInvocation invocationWithMethodSignature:] doesn't
        // assert for  nil signature, so we do it instead.
        if (!sig)
                [NSException raise:NSInternalInconsistencyException
                                        format:@"Can't get method signature for [%@ %@].",
                        NSStringFromClass([target class]), NSStringFromSelector([msg selector])];
        
        i = [NSInvocation invocationWithMethodSignature:sig];
        
        // No worries, NSInvocation takes care of copying the frame.
        [i _setArgFrame:[msg arguments]];
        
        // This must come after _setArgFrame: so that the invocation
        // can update its frame for the new target.
        [i setTarget:target];
        
        return i;
}

@end

@interface DPInvocationQueue (private)
- (void)_stopCurrentRunLoop;
- (BOOL)_theApocalypseArrived;
- (void)_startTheApocalypes;
@end

@interface NSThread (DPPrivate)
- (NSRunLoop *)_DPRunLoop;
@end

@implementation NSThread (DPWorkerThread)

+ (void)setupNewWorkThread:(DPConditionVariable *)cond {
        NSAutoreleasePool *pool = [[NSAutoreleasePool alloc] init];
        NSThread *thread = [self currentThread];
        DPInvocationQueue *queue = [DPInvocationQueue defaultQueue];
        DPCoroutineStack *stack = [[DPCoroutineStack alloc] init];
        
        // Set up a runloop
        NSRunLoop *rl = [NSRunLoop currentRunLoop];
        
        // Set up all the needed sources
        [rl addSource:queue forMode:NSDefaultRunLoopMode];
        [rl addSource:stack forMode:NSDefaultRunLoopMode];
        
        // Set the results pointer
        [cond signalValue:thread];
        
        // Set up an autorelease pool
        // NOTE: The DPSetUpAutoreleasePool() implementation included in
        // this file handles nil values but the one in UITestingKit 1.0b1
        // and 1.0rc1 will crash on nil.
        DPSetUpAutoreleasePool(NSDefaultRunLoopMode);
        
        // Keep a reference to our runloop
        [[thread threadDictionary] setObject:rl forKey:@"DPRunLoop"];
        
        // Run our runloop until someone stops it
        //CFRunLoopRun();
        while (![queue _theApocalypseArrived])
                CFRunLoopRunInMode(kCFRunLoopDefaultMode, 2.0, YES);
        
        // Clean up
        [stack release];
        [pool release];
}

+ (id)detachNewWorkerThread {
        DPConditionVariable *cond = [[[DPConditionVariable alloc] init] autorelease];
        
        // Launch the new thread
        [self detachNewThreadSelector:@selector(setupNewWorkThread:)
                                                 toTarget:self
                                           withObject:cond];
        
        // Wait for the thread to return us itself
        return [cond wait];
}

- (DPInvocationQueue *)defaultInvocationQueue {
        return [[self threadDictionary] objectForKey:@"DPInvocationQueue"];
}

- (void)setDefaultInvocationQueue:(DPInvocationQueue *)queue {
        NSMutableDictionary *dict = [self threadDictionary];
        
        @synchronized(dict) {
                [dict setObject:queue forKey:@"DPInvocationQueue"];
        }
}

- (NSRunLoop *)_DPRunLoop {
        return [[self threadDictionary] objectForKey:@"DPRunLoop"];
}

- (void)terminate {
        /*DPInvocationQueue *q = [self defaultInvocationQueue];
        
        if (q) {
                CFRunLoopRef rl = [self _DPRunLoop];
                [q appendInvocation:[NSInvocation invocationWithMessage:MSG(_stopCurrentRunLoop)
                                                                                                           receiver:q]];
                if (rl)
                        CFRunLoopWakeUp(rl);
        }*/
        [[self defaultInvocationQueue] _startTheApocalypes];
}

- (id)sendMessage:(DPMessage *)msg to:(id)obj {
        DPInvocationQueue *q = [self defaultInvocationQueue];
        NSRunLoop *rl = [self _DPRunLoop];
        id future = nil;
        
        // Make sure we got a queue
        if (!q)
                [NSException raise:NSInternalInconsistencyException
                                        format:@"Thread <%@> has no invocation queue.", self];
        
        // DPInvocationQueue returns a future for invocations with object
        // return or nil, so we simply return that.
        future = [q appendInvocation:[NSInvocation invocationWithMessage:msg
                                                                                                                        receiver:obj]];
        
        if (rl)
                // Wake up our runloop
                CFRunLoopWakeUp([rl getCFRunLoop]);
        
        return future;
}

#if !defined(MAC_OS_X_VERSION_10_5) || MAC_OS_X_VERSION_10_5 > MAC_OS_X_VERSION_MAX_ALLOWED
static NSThread *_mainThread = nil;

static __attribute__((constructor)) void _DPInitMainThread(void) {
        _mainThread = [[NSThread currentThread] retain];
}

+ (NSThread *)mainThread {
        return _mainThread;
}
#endif

// Set up the default invocation queue of the main thread
// and add it to the thread's runloop in the default mode.
// This allows us to easily do [[NSThread mainThread] sendMessage:msg to:obj]
// which is more flexible than -[obj preformSelectorOnMainThread:sel withObject:obj]
static __attribute__((constructor)) void _DPSetUpMainThreadInvocationQueue(void) {
        NSAutoreleasePool *pool = [[NSAutoreleasePool alloc] init];
        NSRunLoop *rl = [NSRunLoop currentRunLoop];
        
        [rl addSource:[DPInvocationQueue defaultQueue]
                  forMode:NSDefaultRunLoopMode];
        [[[NSThread currentThread] threadDictionary] setObject:rl
                                                                                                        forKey:@"DPRunLoop"];
        [pool release];
}

@end


/*
 * This is a base class for true proxies.
 * Unlike NSProxy, DPProxy implements only the basic (thread-safe)
 * memory management, and nothing more. This is in contrast to NSProxy
 * that under the hood implements tons of methods, making it hard to
 * create a proxy based on it.
 */
@interface DPProxy {
        Class           isa;
        u_int32_t       refCount;
}

+ (id)allocWithZone:(NSZone *)zone;
+ (id)alloc;

- (NSZone *)zone;

// These are thread safe
- (id)retain;
- (id)autorelease;
- (void)release;
- (u_int32_t)retainCount;

- (void)dealloc;

// Must be overridden by subclasses
- (id)forward:(SEL)sel :(marg_list)args;

@end

@implementation DPProxy

// Every base class must implement +initialize
+ (void)initialize {
}

+ (id)allocWithZone:(NSZone *)zone {
        DPProxy *p = (DPProxy *)NSAllocateObject(self, 0, zone);
        p->refCount = 1U;
        return p;
}

+ (id)alloc {
        return [self allocWithZone:NULL];
}

- (NSZone *)zone {
        return NSZoneFromPointer(self);
}

- (id)retain {
        if (!objc_collecting_enabled()) {
                OSAtomicIncrement32Barrier((int32_t *)&refCount);
        }
        
        return self;
}

- (id)autorelease {
        [NSAutoreleasePool addObject:self];
        return self;
}

- (void)release {
        if (!objc_collecting_enabled()) {
                if (OSAtomicDecrement32Barrier((int32_t *)&refCount) == 0U)
                        [self dealloc];
        }
}

- (void)dealloc {
        if (!objc_collecting_enabled())
                NSDeallocateObject((id)self);
}

- (u_int32_t)retainCount {
        OSMemoryBarrier();
        return refCount;
}

- (id)forward:(SEL)sel :(marg_list)args {
        //DP_SUBCLASS_MUST_IMPLEMENT;
        return self;
}

// Let's play nice with Cocoa, that assumes *everything*
// (yes, even classes!) is ref counted. Argh!
+ (id)retain {
        return self;
}

+ (id)autorelease {
        return self;
}

+ (void)release {
}

+ (id)forward:(SEL)sel :(marg_list)args {
        return self;
}

@end


/*
 * This DPProxy subclass implements the basic behaviour
 * a proxy needs.
 * Given a target, it'll simply forward *ANY* message to it
 * and return the target's result.
 * NOTE: DPFastProxy always returns the result of its target
 * even if the target returned itself (except for retain
 * and autorelease). Override -forward:: and implement this
 * behaviour as needed.
 */
@interface DPFastProxy : DPProxy {
        id      target;
        BOOL weakRef;
}

- (id)initWithTarget:(id)target weakRef:(BOOL)shouldRetainTarget;

@end

@implementation DPFastProxy

- (id)initWithTarget:(id)t weakRef:(BOOL)flag {
        weakRef = flag;
        target = weakRef ? t : [t retain];
        
        return self;
}

- (void)dealloc {
        if (!weakRef)
                [target release];
        
        [super dealloc];
}

- (id)forward:(SEL)sel :(marg_list)args {
        // dp_msgSendv() takes care for everything for us
        return dp_msgSendv(target, sel, args);
}

- (id)retain {
        [target retain];
        return [super retain];
}

- (id)autorelease {
        [target autorelease];
        return [super autorelease];
}

- (void)release {
        [target release];
        [super release];
}

@end


@interface DPConditionVariable (private)
- (int)_condition;
@end


@interface DPFuture : DPFastProxy {
        @private
        DPConditionVariable *_condition;
}

// For DPInvocationQueue's usage only
- (id)initWithCondition:(DPConditionVariable *)cond;

@end

@implementation DPFuture

static Class _DPFastProxyCls = Nil;

+ (void)initialize {
        _DPFastProxyCls = objc_getClass("DPFastProxy");
}

- (id)initWithCondition:(DPConditionVariable *)cond {
        // Guard against NULL pointers
        NSAssert(cond != nil, @"Can't init without a condition");
        
        _condition = [cond retain];
        // Set our target to nil so retain/release will just
        // message nil (which has no effect).
        target = nil;
        
        return self;
}

- (void)dealloc {
        [_condition release]; _condition = nil;
        [super dealloc];
}

- (id)forward:(SEL)sel :(marg_list)args {
        // We wait for the condition variable to become valid,
        // and then initialize with its value (which is our target).
        // NOTE: This is *NOT* thread safe as we assume a future lives
        // in a single thread.
        [self initWithTarget:[_condition wait] weakRef:NO];
        // We no longer need the condition variable
        [_condition release]; _condition = nil;
        // Let's skip this check and just mutate ourself to
        // a fast proxy.
        self->isa = _DPFastProxyCls;
        
        // Now we can safely forward the message.
        // Since we're actually a DPFastProxy instance at this point
        // it's [self forward::] and not [super forward::].
        return [self forward:sel :args];
}

BOOL DPObjectIsFuture(id obj) {
        if (obj->isa == objc_getClass("DPFuture"))
                return [((DPFuture *)obj)->_condition _condition] == -1;
        return NO;
}

@end


#ifndef __UITESTINGKIT__
@implementation DPRunLoopSource

static void _DPSchedule(void *info, CFRunLoopRef rl, CFStringRef mode) {
        [(DPRunLoopSource *)info scheduleWithRunLoop:rl forMode:(NSString *)mode];
}

static void _DPCancel(void *info, CFRunLoopRef rl, CFStringRef mode) {
        [(DPRunLoopSource *)info cancelMode:(NSString *)mode forRunLoop:rl];
}

static void _DPPerform(void *info) {
        [(DPRunLoopSource *)info fire];
}

- (id)initWithPriority:(unsigned)priority {
        if ((self = [super init])) {
                CFRunLoopSourceContext cont = {
                        0,                                      // version
                        self,                           // info
                        NULL,                           // retain
                        NULL,                           // release
                        /* the following 3 functions work fine with
                           ObjC objects, thanks to toll-free bridging */
                        CFCopyDescription,      // copy description
                        CFEqual,                        // equal
                        CFHash,                         // hash
                        _DPSchedule,            // schedule
                        _DPCancel,                      // cancel
                        _DPPerform                      // perform
                };
                
                _refCount = 1;
                _src = CFRunLoopSourceCreate(kCFAllocatorDefault, priority, &cont);
        }
        
        return self;
}

- (id)init {
        return [self initWithPriority:0];
}

- (void)dealloc {
        [self invalidate];
        CFRelease(_src); _src = NULL;
        [super dealloc];
}

// CF Runloop sources are thread safe, so we should be too
- (id)retain {
        OSAtomicIncrement32Barrier(&_refCount);
        return self;
}

- (void)release {
        if (OSAtomicDecrement32Barrier(&_refCount) == 0)
                [self dealloc];
}

- (unsigned)retainCount {
        OSMemoryBarrier();
        return (unsigned)_refCount;
}

- (void)fire {
}

- (void)signal {
        CFRunLoopSourceSignal(_src);
}

- (void)invalidate {
        CFRunLoopSourceInvalidate(_src);
}

- (BOOL)isValid {
        return CFRunLoopSourceIsValid(_src);
}

- (unsigned)priority {
        return CFRunLoopSourceGetOrder(_src);
}

- (void)scheduleWithRunLoop:(CFRunLoopRef)rl forMode:(NSString *)mode {
}

- (void)cancelMode:(NSString *)mode forRunLoop:(CFRunLoopRef)rl {
}

- (CFRunLoopSourceRef)getCFRunLoopSource {
        return _src;
}

@end


@implementation NSRunLoop (DPRunLoopSource)

- (void)addSource:(DPRunLoopSource *)src forMode:(NSString *)mode {
        CFRunLoopAddSource([self getCFRunLoop], [src getCFRunLoopSource], (CFStringRef)mode);
}

- (void)removeSource:(DPRunLoopSource *)src forMode:(NSString *)mode {
        CFRunLoopRemoveSource([self getCFRunLoop], [src getCFRunLoopSource], (CFStringRef)mode);
}

- (BOOL)containsSource:(DPRunLoopSource *)src inMode:(NSString *)mode {
        return (BOOL)CFRunLoopContainsSource([self getCFRunLoop], [src getCFRunLoopSource],
                                                                                 (CFStringRef)mode);
}

@end
#endif //__UITESTINGKIT__


@implementation DPQueue

- (id)init {
        if ((self = [super init])) {
                _refCount = 1;
                queue = [[NSMutableArray alloc] init];
                queueLock = OS_SPINLOCK_INIT;
        }
        
        return self;
}

- (void)dealloc {
        [queue release]; queue = nil;
        [super dealloc];
}

// We need thread-safe ref counting.
// This might actually be faster than NSObject's implementation.
- (id)retain {
        OSAtomicIncrement32Barrier(&_refCount);
        return self;
}

- (void)release {
        if (OSAtomicDecrement32Barrier(&_refCount) == 0)
                [self dealloc];
}

- (unsigned)retainCount {
        OSMemoryBarrier();
        return (unsigned)_refCount;
}

- (void)enqueu:(id)obj {
        OSSpinLockLock(&queueLock);
        [queue addObject:obj];
        OSSpinLockUnlock(&queueLock);
}

- (id)dequeue {
        id obj = nil;
        
        OSSpinLockLock(&queueLock);
        if ([queue count]) {
                obj = [[queue objectAtIndex:0] retain];
                [queue removeObjectAtIndex:0];
        }
        OSSpinLockUnlock(&queueLock);
        
        return [obj autorelease];
}

- (unsigned)count {
        unsigned count;
        
        OSSpinLockLock(&queueLock);
        count = [queue count];
        OSSpinLockUnlock(&queueLock);
        
        return count;
}

@end


@implementation DPInvocationQueue

+ (id)defaultQueue {
        NSMutableDictionary *dict = [[NSThread currentThread] threadDictionary];
        DPInvocationQueue *q = [dict objectForKey:@"DPInvocationQueue"];
        
        if (!q) {
                q = [[DPInvocationQueue alloc] init];
                @synchronized (dict) {
                        [dict setObject:q forKey:@"DPInvocationQueue"];
                }
                [q release];
        }
        
        return q;
}

- (id)init {
        if ((self = [super init])) {
                //queue = [[NSMutableArray allocWithZone:[self zone]] init];
                //queueLock = [[NSRecursiveLock allocWithZone:[self zone]] init];
                queue = [[DPQueue alloc] init];
                _theApocalypseArrived = 0;
        }
        
        return self;
}

- (void)dealloc {
        //[queueLock lock];
        [queue release]; queue = nil;
        //[queueLock unlock];
        //[queueLock release]; queueLock = nil;
        [super dealloc];
}

- (id)appendInvocation:(NSInvocation *)invocation {
        const char *retType = [[invocation methodSignature] methodReturnType];
        DPFuture *future = nil;
        DPConditionVariable *condition = nil;
        NSDictionary *dict;
        
        // For object values we'll create a condition variable and a future,
        // and keep them in our queue together with the invocation.
        if (strcmp(retType, @encode(id)) == 0 || strcmp(retType, @encode(Class)) == 0) {
                condition = [[[DPConditionVariable alloc] init] autorelease];
                future = [[(DPFuture *)[DPFuture alloc] initWithCondition:condition] autorelease];
        }
        
        dict = [NSDictionary dictionaryWithObjectsAndKeys:invocation, @"invocation",
                                // This method stops when it sees nil, so if we condition == nil
                                // we'll end up just with the invocation.
                                                                                  condition ?: nil, @"condition", nil];
        
        //[queueLock lock];
        //[queue addObject:dict];
        //[queueLock unlock];
        [queue enqueu:dict];
        
        [self signal];
        
        // We're done
        return future;
}

/*- (void)removeInvocation:(NSInvocation *)invocation {
        [queueLock lock];
        
        unsigned i, count = [queue count];
        for (i = 0; i < count; i++) {
                if ([[queue objectAtIndex:i] objectForKey:@"invocation"] == invocation) {
                        [queue removeObjectAtIndex:i];
                        break;
                }
        }
        
        [queueLock unlock];
}*/

- (void)pushAllInvocations {
        /*[queueLock lock];
        
        unsigned i, count = [queue count];
        // Loop through all invocations and invoke them
        for (i = 0U; i < count; i++) {
                NSDictionary *dict = [queue objectAtIndex:i];
                NSInvocation *invocation = [dict objectForKey:@"invocation"];
                id cond = [dict objectForKey:@"condition"];
                
                [invocation invoke];
                
                // If we have a memory block it means we're expecting an object reuturn.
                // In that case we'll copy the value from our invocation.
                if (cond) {
                        id val;
                        // Get the return value (which is an object (pointer)).
                        [invocation getReturnValue:&val];
                        [cond broadcastValue:val];
                }
        }
        
        // Clean up the queues
        [queue removeAllObjects];
        
        [queueLock unlock];*/
        
        while ([queue count]) {
                NSDictionary *dict = [queue dequeue];
                NSInvocation *invocation = [dict objectForKey:@"invocation"];
                id cond = [dict objectForKey:@"condition"];
                
                [invocation invoke];
                
                // If we have a memory block it means we're expecting an object reuturn.
                // In that case we'll copy the value from our invocation.
                if (cond) {
                        id val;
                        // Get the return value (which is an object (pointer)).
                        [invocation getReturnValue:&val];
                        [cond broadcastValue:val];
                }
        }
}

- (BOOL)process {
        NSDictionary *dict = [queue dequeue];
        // Lock
        /*[queueLock lock];
        // Stop now if the queue is empty
        if ([queue count] == 0) {
                [queueLock unlock];
                return NO;
        }
        // Get the first entry from the queue.
        // Must retain as removing releases the entry
        dict = [[queue objectAtIndex:0] retain];
        [queue removeObjectAtIndex:0];
        // We're good to do
        [queueLock unlock];*/
        
        if (!dict)
                return NO;
        
        // Signal so the runloop won't go to sleep
        [self signal];
        
        // Get the invocation and condition variable from the entry
        NSInvocation *invocation = [dict objectForKey:@"invocation"];
        id cond = [dict objectForKey:@"condition"];
        
        [invocation invoke];
        
        // If we have a condition variable it means somewhere
        // there's a future waiting on it (or not).
        if (cond) {
                id val;
                [invocation getReturnValue:&val];
                [cond broadcastValue:val];
        }
        
        // We're done
        //[dict release];
        return YES;
}

- (void)fire {
        [self process];
}

- (void)_stopCurrentRunLoop {
        CFRunLoopStop(CFRunLoopGetCurrent());
}

- (BOOL)_theApocalypseArrived {
        //OSMemoryBarrier();
        return _theApocalypseArrived != 0;
}

- (void)_startTheApocalypes {
        _theApocalypseArrived = YES;
        //OSAtomicCompareAndSwap32Barrier(0, 1, &_theApocalypseArrived);
}

@end


@implementation DPConditionVariable

- (id)init {
        if ((self = [super init])) {
                _refCount = 1;
                NSAssert(pthread_mutex_init(&_lock, NULL) == 0, @"Can't create pthread mutex");
                NSAssert(pthread_cond_init(&_condition, NULL) == 0, @"Can't create pthread condition variable");
                _value = NULL;
                _conditionValue = -1;
        }
        
        return self;
}

- (void)dealloc {
        NSAssert(pthread_cond_destroy(&_condition) == 0, @"Can't destroy pthread condition variable");
        NSAssert(pthread_mutex_destroy(&_lock) == 0, @"Can't destroy pthread mutex");
        [super dealloc];
}

// We need thread-safe ref counting.
// This might actually be faster than NSObject's implementation.
- (id)retain {
        OSAtomicIncrement32Barrier(&_refCount);
        return self;
}

- (void)release {
        if (OSAtomicDecrement32Barrier(&_refCount) == 0)
                [self dealloc];
}

- (unsigned)retainCount {
        OSMemoryBarrier();
        return (unsigned)_refCount;
}

// TODO: Implement double-checked locking?
// The needed memory barriers are already in place
// in the write methods.
- (void *)waitForCondition:(int)cond {
        void *val;
        
        pthread_mutex_lock(&_lock);
        
        while (_conditionValue != cond)
                pthread_cond_wait(&_condition, &_lock);
        
        val = (void *)_value;
        pthread_mutex_unlock(&_lock);
        
        return val;
}

- (void *)wait {
        return [self waitForCondition:1];
}

- (void)signalCondition:(int)cond withValue:(void *)val {
        pthread_mutex_lock(&_lock);
        _value = val;
        OSMemoryBarrier();
        _conditionValue = cond;
        pthread_mutex_unlock(&_lock);
        pthread_cond_signal(&_condition);
}

- (void)signalValue:(void *)val {
        [self signalCondition:1 withValue:val];
}

- (void)broadcastCondition:(int)cond withValue:(void *)val {
        pthread_mutex_lock(&_lock);
        _value = val;
        OSMemoryBarrier();
        _conditionValue = cond;
        pthread_mutex_unlock(&_lock);
        pthread_cond_broadcast(&_condition);
}

- (void)broadcastValue:(void *)cond {
        [self broadcastCondition:1 withValue:cond];
}

- (void)clear {
        pthread_mutex_lock(&_lock);
        _conditionValue = -1;
        OSMemoryBarrier();
        _value = NULL;
        pthread_mutex_unlock(&_lock);
}

- (int)_condition {
        OSMemoryBarrier();
        return _conditionValue;
}

@end

#include <sys/sysctl.h>

@implementation DPThreadPool

+ (int)activeCPUs {
        int num;
        unsigned long length = sizeof(num);
        
        if (sysctlbyname("hw.activecpu", &num, &length, NULL, 0))
                num = -1;       // An error had occured
        
        return num;
}

+ (id)poolWithNumberOfThreads:(unsigned)threadsCount {
        DPThreadPool *pool = [[self alloc] init];
        unsigned i;
        
        for (i = 0; i < threadsCount; i++)
                [pool spawnThread];
        
        return [pool autorelease];
}

+ (void)_stopRunLoop:(CFRunLoopRef)rl {
        CFRunLoopStop(rl);
}

- (id)init {
        if ((self = [super init])) {
                _refCount = 1;
                queue = [[DPInvocationQueue alloc] init];
                coroStack = [[DPCoroutineStack alloc] init];
                threads = [[NSMutableArray alloc] init];
        }
        
        return self;
}

// We need thread-safe ref counting.
// This might actually be faster than NSObject's implementation.
- (id)retain {
        OSAtomicIncrement32Barrier(&_refCount);
        return self;
}

- (void)release {
        if (OSAtomicDecrement32Barrier(&_refCount) == 0)
                [self dealloc];
}

- (unsigned)retainCount {
        OSMemoryBarrier();
        return (unsigned)_refCount;
}

- (void)dealloc {
        [queue _startTheApocalypes];
        [coroStack release]; coroStack = nil;
        [threads release]; threads = nil;
        [queue release]; queue = nil;
        [super dealloc];
}

// Since NSThread retains its target, we make this a class method
// in orde to avoid the retain loop. Retaining our class is meaningless.
+ (void)_setUpThread:(NSArray *)info {
        NSAutoreleasePool *pool = [[NSAutoreleasePool alloc] init];
        NSRunLoop *rl = [NSRunLoop currentRunLoop];
        NSThread *thread = [NSThread currentThread];
        DPInvocationQueue *queue = [[info objectAtIndex:0] retain];
        DPConditionVariable *cond = [[info objectAtIndex:1] retain];
        DPCoroutineStack *stack = [[info objectAtIndex:2] retain];
        
        // Add our queue as the default queue
        [thread setDefaultInvocationQueue:queue];
        [rl addSource:queue forMode:NSDefaultRunLoopMode];
        
        // Add the coroutine stack
        [DPCoroutineStack setCurrentStack:stack];
        [rl addSource:stack forMode:NSDefaultRunLoopMode];
        
        // Set up an autorelease pool
        // NOTE: The DPSetUpAutoreleasePool() implementation included in
        // this file handles nil values but the one in UITestingKit 1.0b1
        // and 1.0rc1 will crash on nil.
        DPSetUpAutoreleasePool(NSDefaultRunLoopMode);
        
        // Keep a reference to our runloop
        [[thread threadDictionary] setObject:rl forKey:@"DPRunLoop"];
        
        // We're good to go
        [cond signalValue:thread];
        
        // Run our runloop until someone stops it
        //CFRunLoopRun();
        while (![queue _theApocalypseArrived] || [stack hasWork])
                CFRunLoopRunInMode(kCFRunLoopDefaultMode, 2.0, true);
        
        [rl removeSource:stack forMode:NSDefaultRunLoopMode];
        [rl removeSource:queue forMode:NSDefaultRunLoopMode];
        
        //[stack empty];
        
        // Clean up
        [stack release];
        [cond release];
        [queue release];
        [info release];
        [pool release];
}

- (void)spawnThread {
        DPConditionVariable *cond = [[DPConditionVariable alloc] init];
        NSArray *info = [[NSArray alloc] initWithObjects:queue, cond, coroStack, nil];
        NSThread *thread;
        
        [NSThread detachNewThreadSelector:@selector(_setUpThread:)
                                                         toTarget:[self class]
                                                   withObject:info];
        
        thread = [cond wait];
        
        @synchronized (threads) {
                [threads addObject:thread];
        }
        [cond release];
}

- (void)terminateThread {
        [self sendMessage:MSG(_stopCurrentRunLoop)
                                   to:queue];
}

- (unsigned)numberOfThreads {
        return [threads count];
}

- (unsigned)activeThreads {
        unsigned i, count = [threads count], r = 0;
        
        for (i = 0; i < count; i++)
                r += !CFRunLoopIsWaiting([[[threads objectAtIndex:i] _DPRunLoop] getCFRunLoop]);
        
        return r;
}

- (id)sendMessage:(DPMessage *)msg to:(id)target {
        unsigned i, count = [threads count];
        id future = nil;
        
        // Add a termination invocation to the queue. The first thread that's idle
        // will receive it and terminate.
        future = [queue appendInvocation:[NSInvocation invocationWithMessage:msg
                                                                                                                                receiver:target]];
        
        // If there's a sleeping thread, wake it up so it can do some work.
        // There's no need to wake more than one thread for a single task.
        for (i = 0; i < count; i++) {
                NSRunLoop *rl = [[threads objectAtIndex:i] _DPRunLoop];
                
                if (rl && CFRunLoopIsWaiting([rl getCFRunLoop])) {
                        CFRunLoopWakeUp([rl getCFRunLoop]);
                        break;
                }
        }
        
        return future;
}

- (void)sendCoroutineMessage:(DPMessage *)msg to:(id)target {
        unsigned i, count = [threads count];
        
        // Add the coro to the stack
        DPCoroutine *coro = [[DPCoroutine alloc] initWithMessage:msg
                                                                                                          target:target];
        [coroStack addCoro:coro];
        [coro release];
        
        // If there's a sleeping thread, wake it up so it can do some work.
        // There's no need to wake more than one thread for a single task.
        for (i = 0; i < count; i++) {
                NSRunLoop *rl = [[threads objectAtIndex:i] _DPRunLoop];
                
                if (rl && CFRunLoopIsWaiting([rl getCFRunLoop])) {
                        CFRunLoopWakeUp([rl getCFRunLoop]);
                        break;
                }
        }
}

@end


@implementation DPThreadID

static pthread_key_t _DPThreadIDKey;

static void _DPThreadIDDestroy(void *ID) {
        [(DPThreadID *)ID release];
        pthread_setspecific(_DPThreadIDKey, NULL);
}

+ (void)initialize {
        pthread_key_create(&_DPThreadIDKey, _DPThreadIDDestroy);
}

+ (id)currentIdentifier {
        DPThreadID *ID = pthread_getspecific(_DPThreadIDKey);
        
        if (!ID)
                ID = [[self alloc] init];
        
        return ID;
}

- (id)init {
        DPThreadID *ID = pthread_getspecific(_DPThreadIDKey);
        
        // Prevent people from creating extra instances
        if (ID) {
                [self release];
                return [ID retain];
        }
        
        if ((self = [super init])) {
                threadId = pthread_self();
                pthread_setspecific(_DPThreadIDKey, ID);
        }
        
        return self;
}

- (pthread_t)getThreadID {
        return threadId;
}

extern unsigned DPIntHash(void *ptr);

- (unsigned)hash {
        return DPIntHash(self);
}

- (id)copyWithZone:(NSZone *)zone {
        return [self retain];
}

@end


@implementation NSObject (DPAsyncMessaging)

- (void)receiveMessage:(DPMessage *)msg afterDelay:(NSTimeInterval)sec {
        // If this isn't simple than what is?
        // Don't you love HOM? :)
        [NSTimer scheduledTimerWithTimeInterval:sec
                                                                 invocation:[NSInvocation invocationWithMessage:msg
                                                                                                                                           receiver:self]
                                                                        repeats:NO];
}

- (void)_DPFutureThread:(NSArray *)info {
        NSAutoreleasePool *pool = [[NSAutoreleasePool alloc] init];
        DPMessage *msg = [info objectAtIndex:0];
        DPConditionVariable *cond = [info objectAtIndex:1];
        
        [cond signalValue:[msg sendTo:self]];
        // They count on us to release the condition variable and the message
        [cond release];
        [msg release];
        [pool release];
}

- (id)future:(DPMessage *)msg {
        DPConditionVariable *cond = [[DPConditionVariable alloc] init];
        DPFuture *future = [[(DPFuture *)[DPFuture alloc] initWithCondition:cond] autorelease];
        
        [NSThread detachNewThreadSelector:@selector(_DPFutureThread:)
                                                         toTarget:self
                                                // Note that we must copy the message to ensure it won't go away.
                                                   withObject:[NSArray arrayWithObjects:[msg copy], cond, nil]];
        return future;
}

@end


#ifndef __UITESTINGKIT__
static void _DPMaintainAutoreleasePool(CFRunLoopObserverRef observer, CFRunLoopActivity activity, void *info) {
        NSAutoreleasePool **pool = info;
        [*pool release];
        *((NSAutoreleasePool **)info) = [[NSAutoreleasePool alloc] init];
}

static void _DPReleasePool(const void *info) {
        NSAutoreleasePool **pool = (void *)info;
        [*pool release];
        free((void *)info);
}

void DPSetUpAutoreleasePool(NSString *runLoopMode) {
        CFRunLoopRef rl = CFRunLoopGetCurrent();
        CFStringRef mode = runLoopMode ? CFRetain((CFStringRef)runLoopMode) : CFRunLoopCopyCurrentMode(rl);
        
        if (mode) {
                NSAutoreleasePool **p = malloc(sizeof(NSAutoreleasePool **));
                CFRunLoopObserverContext context = {
                        0,                              // version
                        p,                              // info
                        NULL,                   // retain
                        _DPReleasePool, // release
                        NULL                    // copy description
                };
                *p = [[NSAutoreleasePool alloc] init];
                CFRunLoopObserverRef observer = CFRunLoopObserverCreate(kCFAllocatorDefault,
                                                                                                                                kCFRunLoopEntry, true, 0,
                                                                                                                                _DPMaintainAutoreleasePool,
                                                                                                                                &context);
                
                CFRunLoopAddObserver(CFRunLoopGetCurrent(), observer, mode);
                CFRelease(mode);
        }
}
#endif //__UITESTINGKIT__