DPQueue is now a two-lock concurrent queue.

[hom.git] / Source / DPMultiThreading.h

//
//  DPMultiThreading.h
//  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.
//

#import <Foundation/Foundation.h>
#import <CoreFoundation/CoreFoundation.h>
#import <AvailabilityMacros.h>
#include <pthread.h>
#include <libkern/OSAtomic.h>   // For OSSpinLock

#if defined(__HOM_FRAMEWORK__)
#import <HigherOrderMessaging/DPObjCRuntime.h>
#elif defined(__UITESTINGKIT__)
#import <UITestingKit/DPObjCRuntime.h>
#import <UITestingKit/DPRunLoopSource.h>
#import <UITestingKit/DPUtilities.h>    // For DPSetUpAutoreleasePool()
#else
#import "DPObjCRuntime.h"       // For DP_EXTERN
#endif

#if defined(__HOM_FRAMEWORK__)
#import <HigherOrderMessaging/DPMessage.h>
#else
#import "DPMessage.h"
#endif


#ifndef __UITESTINGKIT__
/*!
 * @abstract A base class for runloop sources.
 * @discussion Subclass this class in order to create our own runloop source.
 * This class is thread-safe.
 */
@interface DPRunLoopSource : NSObject {
        @private
        int32_t _refCount;
        CFRunLoopSourceRef _src;
}

/*!
 * @abstract Initializes a new runloop source with a given priority.
 *
 * @param priority A priority index indicating the order in which run loop sources
 * are processed. When multiple run loop sources are firing in a single pass through
 * the run loop, the sources are processed in increasing order of this parameter.
 * If the run loop is set to process only one source per loop, only the highest
 * priority source, the one with the lowest order value, is processed.
 * Pass 0 unless there is a reason to do otherwise.
 */
- (id)initWithPriority:(unsigned)priority;

/*!
 * @abstract Called by the runloop in order for the source to do its work.
 * @discussion Override this method in your subclass to perform the actual work.
 * The default implementation of this method does nothing.
 */
- (void)fire;

/*!
 * @abstract Signals the receiver, marking it as ready to fire.
 * @discussion This method marks the receiver as ready to fire.
 * The receiver's -fire will then be invoked by one of the runloops the receiver
 * is added to, whichever happens to be available first.
 * After firing, the receiver must be signaled again in order to fire again.
 */
- (void)signal;


/*!
 * @abstract Invalidates the receiver, stopping it from ever firing again.
 * @discussion Once invalidated, the receiver will never fire again. This method
 * automatically removes the receiver from all run loop modes in which it was registered.
 */
- (void)invalidate;

/*!
 * @abstract Returns whether the receiver is valid and able to fire.
 */
- (BOOL)isValid;

/*!
 * @abstract Returns the ordering parameter for the receiver, which the run loop uses to
 * determine the order in which sources are processed when multiple sources are firing.
 */
- (unsigned)priority;


/*!
 * @abstract A scheduling callback for the receiver.
 * @discussion This method is invoked when the receiver is added to a run loop mode.
 * The default implementation of this method does nothing.
 */
- (void)scheduleWithRunLoop:(CFRunLoopRef)rl forMode:(NSString *)mode;

/*!
 * @abstract A cancel callback for the receiver.
 * @discussion This method is invoked when the receiver is removed from a run loop mode.
 * The default implementation of this method does nothing.
 */
- (void)cancelMode:(NSString *)mode forRunLoop:(CFRunLoopRef)rl;


/*!
 * @abstract Returns the underlying CFRunLoopSourceRef powering the receiver.
 */
- (CFRunLoopSourceRef)getCFRunLoopSource;

@end

/*!
 * @abstract Support for DPRunLoopSource.
 */
@interface NSRunLoop (DPRunLoopSource)

/*!
 * @abstract Registers <code>src</code> with the receiver for <code>mode</code>.
 */
- (void)addSource:(DPRunLoopSource *)src forMode:(NSString *)mode;

/*!
 * @abstract Unregisters <code>src</code> with the receiver for <code>mode</code>.
 */
- (void)removeSource:(DPRunLoopSource *)src forMode:(NSString *)mode;

/*!
 * @abstract Returns whether the receiver contains a given source.
 */
- (BOOL)containsSource:(DPRunLoopSource *)src inMode:(NSString *)mode;

@end
#endif //__UITESTINGKIT__


// Private
typedef struct {
        id value;
        void *next;
} _DPQueueNode;

/*!
 * @abstract A thread-safe queue.
 */
@interface DPQueue : NSObject {
        // Need our own ref-counting as Apple's implementation itsn't thread satfe
        int32_t _refCount;
        _DPQueueNode *head, *tail;
        OSSpinLock h_lock, t_lock;
        int32_t count;
}

/*!
 * @abstract Adds <code>obj</code> to the queue.
 */
- (void)enqueue:(id)obj;

/*!
 * @abstract Pops the topmost object in the queue and returns it.
 * @discussion Returns <code>nil</code> if the receiver is empty.
 */
- (id)dequeue;

/*!
 * @abstract Returns the number of objects in the receiver.
 */
- (unsigned)count;

@end


/*!
 * @abstract A queue of invocations that can be shared across
 * runloops to distribute work.
 *
 * @discussion An invocation queue is a runloop source, and therefor
 * can be shared across runloops/threads. Each runloop takes the topmost
 * invocation and invokes it, until the queue is empty.
 *
 * DPInvocation queue implements thread safe reference-counting so there's
 * no need to lock when retaining/releasing it.
 */
@interface DPInvocationQueue : DPRunLoopSource {
        DPQueue *queue;
        int32_t _theApocalypseArrived;
}

/*!
 * @abstract Returns the default queue in the current thread.
 * @discussion If there's no queue in the current thread one
 * is created.
 */
+ (id)defaultQueue;

/*!
 * @abstract Appends an invocation to the queue for later execution.
 *
 * @discussion If the invocation returns an object (either instance or class)
 * a future is returned. A future is a proxy to the result of the invocation
 * that does not yet exist.
 *
 * Upon the first message to the future it'll block the caller thread until
 * the result of the invocation is available. When the result is available,
 * the future will act as a simple proxy and just forward any message to its
 * target. No thread safety is implemented and the target of the proxy (the
 * result of the invocation) is assumed to be owned by the caller thread.
 *
 * For non-object return values nil is returned. If you have no use for the
 * proxy just ignore it, but if you do, you must retain it to keep it around.
 *
 * If you wish to get the real object behind the future, send it a
 * <code>self</code> message and use the result.
 *
 * Warning: Futures are not thread safe. They assume to be owned by excactly
 * one thread and to never be messaged from another thread. That being said,
 * after the first message to the future returns (meaning the future's target
 * is now available), the future can safely be shared across threads, assuming
 * its target is thread safe of course. Futures implement thread-safe reference
 * counting to allow sharing after a target is available.
 */
- (id)appendInvocation:(NSInvocation *)invocation;

/*!
 * @abstract Invokes all invocations of the receiver empties it.
 *
 * @discussion This method returns only after all invocations have
 * been invoked and the receiver is empty.
 * The queue is locked until all invocations are invoked, preventing
 * additions of invocations from other threads.
 */
- (void)pushAllInvocations;

@end

/*!
 * @abstract Returns whether a given object is a future or not.
 * @discussion Once a future gets its value, it is no longer
 * considered a future and this function returns <code>NO</code>.
 */
DP_EXTERN BOOL DPObjectIsFuture(id obj);


/*!
 * @abstract A condition variable that's shared across threads.
 * @discussion A condition variable is a variable that's shared
 * across threads and it's value is only available when a certain
 * condition is met.
 *
 * This is a wrapper around pthread's condition variable.
 */
@interface DPConditionVariable : NSObject {
        @private
        int32_t _refCount;
        pthread_mutex_t _lock;
        pthread_cond_t _condition;
        volatile void *_value;
        volatile int _conditionValue;
}

/*!
 * @abstract Waits for the default condition (1) and returns the value.
 */
- (void *)wait;

/*!
 * @abstract Waits until a certain condition is met, and returns the value.
 *
 * @param cond The condition to wait for. -1 is a reserved condition. Never use it yourself.
 * @result The value that was signaled/broadcasted.
 */
- (void *)waitForCondition:(int)cond;

/*!
 * @abstract Signals the default condition (1) with a given value.
 * @discussion If there are more than one thread waiting for this value
 * only one is unblocked.
 */
- (void)signalValue:(void *)val;

/*!
 * @abstract Signals a custom condition with a given value.
 * @discussion If there are more than one thread waiting for this value
 * only one is unblocked.
 *
 * @param cond The condition to signal. -1 is a reserved condition. Never use it yourself.
 * @param val The value to send with the signal.
 */
- (void)signalCondition:(int)cond withValue:(void *)val;

/*!
 * @abstract Broadcasts the default condition (1) with a given value.
 * @discussion A broadcast unblocks all threads waiting for the condition.
 */
- (void)broadcastValue:(void *)cond;

/*!
 * @abstract Broadcasts a custom condition with a given value.
 * @discussion A broadcast unblocks all threads waiting for the condition.
 *
 * @param cond The condition to signal. -1 is a reserved condition. Never use it yourself.
 * @param val The value to send with the signal.
 */
- (void)broadcastCondition:(int)cond withValue:(void *)val;

/*!
 * @abstract Resets the receiver to its initial condition and value.
 * @discussion If you'd like to reuse a condition variable, send it a
 * <code>clear</code> message before waiting for a new condition.
 */
- (void)clear;

@end

@class DPCoroutineStack, DPCoroutineScheduler, DPCoroutine;

/*!
 * @abstract A class representing a pool of threads that share work.
 *
 * @discussion A thread pool may have any number of threads and any
 * number of messages to send. The threads in the pool send the messages
 * whenever they can, and split the workload among themselves.
 *
 * Each thread pool has one shared invocation queue and one shared coroutine
 * stack. Both split the work among all active threads.
 */
@interface DPThreadPool : NSObject {
        @private
        int32_t _refCount;
        DPInvocationQueue *queue;
        DPCoroutineStack *coroStack;
        NSMutableArray *threads;
}

/*!
 * @abstract Returns the number of processors currently available for executing threads.
 * @discussion This number can change when power management modes are changed.
 * @result The number of active CPUs or -1 on error.
 */
+ (int)activeCPUs;

/*!
 * @abstract Creates and returns an autoreleased pool with a given
 * number of threads.
 */
+ (id)poolWithNumberOfThreads:(unsigned)threadsCount;

/*!
 * @abstract Spawns a new thread and adds it to the pool.
 */
- (void)spawnThread;

/*!
 * @abstract Terminates a thread from the pool.
 *
 * @discussion If there are sleeping threads one of them is terminated.
 * Otherwise, the first thread to become idle is terminated. Note that
 * any prioir messages added to the pool will be sent before the thread
 * is terminated.
 */
- (void)terminateThread;

/*!
 * @abstract Returns the total amount of threads in the pool.
 */
- (unsigned)numberOfThreads;

/*!
 * @abstract Returns the number of active threads in the pool.
 * @discussion An active thread is a thread that processes a messages.
 */
- (unsigned)activeThreads;

/*!
 * @abstract Adds a message to the pool's work queue.
 * @discussion The message is invoked whenever one of the threads in
 * the pool finds the time to do it.
 */
- (id)sendMessage:(DPMessage *)msg to:(id)target;

/*!
 * @abstract Creates a coroutine with the given message and target and adds
 * it to the pool's coroutine stack.
 */
- (void)sendCoroutineMessage:(DPMessage *)message to:(id)target;

@end


/*!
 * @abstract Adds an API for easy usage of worker threads.
 */
@interface NSThread (DPWorkerThread)

/*!
 * @abstract Detaches and returns a new worker thread.
 *
 * @discussion A worker thread is a thread that has a default
 * invocation queue set up together with an active runloop.
 *
 * This method does all the dirty work for you (including setting up an
 * autorelease pool), but you can set up a worker thread yourself buy calling
 * <code>[[NSRunLoop defaultRunLoop] addSource:[DPInvocationQueue defaultQueue]
                                                                           forMode:NSDefaultRunLoopMode];</code>
 * and running the runloop ins the default mode.
 */
+ (id)detachNewWorkerThread;

#if !defined(MAC_OS_X_VERSION_10_5) || MAC_OS_X_VERSION_10_5 > MAC_OS_X_VERSION_MAX_ALLOWED
/*!
 * @abstract Returns the main thread.
 */
+ (NSThread *)mainThread;
#endif

/*!
 * @abstract Returns the default invocaiton queue of the receiver, or <code>nil</code>
 * if none exists.
 */
- (DPInvocationQueue *)defaultInvocationQueue;

/*!
 * @abstract Sets the default invocation queue of the receiver.
 */
- (void)setDefaultInvocationQueue:(DPInvocationQueue *)queue;

/*!
 * @abstract Terminates a worker thread that was created using
 * +[NSThread detachNewWorkerThread].
 *
 * @discussion If the thread wasn't created with +[NSThread detachNewWorkerThread],
 * this method does nothing.
 */
- (void)terminate;

/*!
 * @abstract Sends a given message to an object inside the context of the
 * receiving thread.
 *
 * @discussion The object must respond to the passed message or an exception
 * will be thrown.
 *
 * @result A future if the message returns an object, <code>nil</code> otherwise.
 * See the documentation for -[DPInvocationQueue appendInvocation:] for more
 * information about futures.
 */
- (id)sendMessage:(DPMessage *)msg to:(id)obj;

@end


/*!
 * @abstract A unique thread ID.
 * @discussion Each thread has exactly one unique DPThreadID instance.
 * This instance can be used as a key in a dictionary and so on.
 */
@interface DPThreadID : NSObject {
        pthread_t threadId;
}

/*!
 * @abstract Returns the identifier of the current thread.
 */
+ (id)currentIdentifier;

/*!
 * @abstract Returns the pthread ID of the receiver.
 */
- (pthread_t)getThreadID;

@end

@interface NSObject (DPAsyncMessaging)

/*!
 * @abstract Makes the receiver receive <code>msg</code> after the given delay.
 */
- (void)receiveMessage:(DPMessage *)msg afterDelay:(NSTimeInterval)sec;

/*!
 * @abstract Spwns a new thread in which the passed message will be received.
 *
 * @discussion If the message doesn't return an object the behavior is undefined,
 * and will most likely cause a crash.
 *
 * @result A future to the message's result. Please refer to the
 * -[DPInvocationQueue appendInvocation:] documentation for more information
 * about futures.
 */
- (id)future:(DPMessage *)msg;

@end

@interface NSInvocation (DPMessageSupport)

+ (id)invocationWithMessage:(DPMessage *)msg receiver:(id)target;

@end

#ifndef __UITESTINGKIT__
/*!
 * @abstract Sets up an autorelease pool that's automatically released
 * and recreated at each entry of the current runloop.
 * @discussion The autorelease pool is added to the current runloop in
 * <code>runLoopMode</code>. If <code>runLoopMode</code> is <code>nil</code>,
 * the pool is added to the current mode.
 * Each call to this function creates and sets up a new autorelease pool.
 */
DP_EXTERN void DPSetUpAutoreleasePool(NSString *runLoopMode);
#endif