root/trunk/futurism/future.d

// future.d
//
// Futurism - A "Future" library for D.
//
// I wrote this after watching a talk by Herb Sutter about concurrent
// programming.  This library implements the "future" concept for D.
// The future concept is designed to promote scalable concurrency.
// The original invention of the concept is attributed to Henry Baker
// and Carl Hewitt in 1977.
//
// Author:   Kevin Bealer
// License:  Public Domain
// Modified: January 2007

module future;

version(Tango) {
    import tango.core.Thread;
    import tango.io.FileConduit;
    import tango.io.Conduit;
    import tango.stdc.stdlib;
    import tango.stdc.string;
    import tango.text.convert.Integer;
    import tango.text.Util;
    import tango.core.Version;
} else {
    import std.c.stdlib;
    import std.c.string;
    import std.conv;
    import std.file;
    import std.string;
    import std.thread;
    import std.traits;
    import std.cpuid;
    import std.stream;
}

template FutureTypeGroup(Delegate, Args...) {
    alias ReturnType!(Delegate)          Return;
    alias ParameterTypeTuple!(Delegate)  Params;
    alias Future!(Return, Params)        TFuture;
}

/*****
 * Build a future given a delegate and (optional) arguments.
 */

FutureTypeGroup!(Delegate, Args).TFuture
make_future(Delegate, Args...)(Delegate cmd, Args args)
{
    return new FutureTypeGroup!(Delegate, Args).TFuture(cmd, args);
}

FutureTypeGroup!(Delegate, Args).TFuture
make_future_priority(Delegate, Args...)(int p, Delegate cmd, Args args)
{
    return new FutureTypeGroup!(Delegate, Args).TFuture(p, cmd, args);
}

/*****
 * Scoped wrapper for ThreadPool.  If the number of threads is not
 * specified, this selects the number based on what it can find about
 * the hardware.  An environment variable, "FUTURISM_DEFAULT_THREADS",
 * allows users to adjust the default for compiled apps.
 */
scope class ThreadPoolScope {
    // Create a pool with N threads.
    this(int n = 0)
    {
        if (n < 1) {
            // Automatically generated values are 0-based.
            n = recommendThreads();
        } else {
            // A caller-specified value should be 1-based.
            n ++;
        }
        
        nthr_ = n;
        
        if (nthr_ > 1) {
            // Subtract one for the 'main' thread.
            pool_ = new ThreadPool(nthr_ - 1);
        }
    }
    
    ~this()
    {
        stop();
    }
    
    // Stop the thread pool.
    void stop()
    {
        if (pool_ !is null) {
            pool_.stop();
        }
    }
    
    // Create a thread pool, selecting the number of threads based on
    // an examination of the environment, hardware, and other factors.
    
    static int recommendThreads()
    {
        int N = getUserCount();
        
        if (! N)
            N = getHardwareCount();
        
        if (! N)
            N = getDefaultCount();
        
        assert(N);
        return N;
    }
    
    static int getUserCount()
    {
        int N = 0;
        
        char * cfg = getenv("FUTURISM_DEFAULT_THREADS");
        
        if (cfg) {
            char[] nstr = cfg[0..strlen(cfg)];
            N = toInt(nstr);
        }
        
        if (N < 1)
            N = 0;
        
        if (N > maxEnvarThreads)
            N = maxEnvarThreads;
        
        return N;
    }
    
    // Number of threads to use, including the main thread.  The
    // actual worker threads in the thread pool will be one less than
    // what is returned here.
    
    int threadCount()
    {
        return nthr_;
    }
    
private:
    // The rest of these are not public because they are heuristics
    // with little meaning outside this class.
    
    
    // The number of threads we might want per CPU; this is actually
    // something of a fudge factor.  This allows I/O bound threads to
    // run efficiently, lets a few futures wait on each other without
    // piling up too bad, and doesn't do much harm for CPU bound
    // systems.
    
    // My guess is that for futures we probably want a higher number
    // here than would be used for dedicated number cruncher threads
    // in a workhorse system.
    
    static int defaultOverlap()
    {
        return 4;
    }
    
    // Try to find some number representing the amoung of hardware
    // threading available here.
    
    static int getHardwareCount()
    {
        int N = 0;
        
        version(linux) {
            try {
                // Linux gives a seperate line for each CPU core, so
                // no additional std.cpuid is needed or useful here.
                
                version(Tango) {
                    // Use a FileConduit because 'files' in /proc do
                    // not report a correct length.
                    
                    auto f = new FileConduit("/proc/cpuinfo");
                    
                    char[] cpuinfo = "\n";
                    void[] tmp = new byte[4096];
                    int amt = 0;
                    
                    while(0 != (amt = f.fill(tmp))) {
                        cpuinfo ~= cast(char[]) tmp[0..amt];
                    }
                    
                    static if (Tango == 0.95) {
                        N = demarcate(cpuinfo, "\nprocessor").length-1;
                    } else {
                        N = split(cpuinfo, "\nprocessor").length-1;
                    }
                } else {
                    // std.file.read checks the file length first,
                    // so it is necessary to use a stream here.
                    
                    int new_N = 0;
                    
                    Stream file = new BufferedFile("/proc/cpuinfo");
                    foreach(ulong n, char[] line; file) {
                        if (find("\n" ~ line, "\nprocessor:") != -1) {
                            new_N ++;
                        }
                    }
                    
                    if (new_N) {
                        N = new_N;
                    }
                }
                
                N *= defaultOverlap;
            }
            catch(Exception e) {
                // Maybe the /proc fs is not enabled?
            }
        }
        
        if (N == 0) {
            // There is no good way to find number of CPUs, so use the
            // number of threads per CPU.  This only works for x86 and
            // is not yet supported in Tango.
            
            version(Tango) {
            } else {
                N = threadsPerCPU * defaultOverlap;
            }
        }
        
        return N;
    }
    
    // Just select a number.
    
    static int getDefaultCount()
    {
        // The number 2 here represents a "guess" of two CPUs or
        // cores, which will soon (its 2007 as I write this) be a
        // reasonable middle ground for consumer boxes.
        
        return 2 * defaultOverlap;
    }
    
    // Limit the range of the envar.
    
    static int maxEnvarThreads()
    {
        // Creating too many threads can cause the application to
        // crash.  This limit applies only to the envar, and is meant
        // to protect overly ambitious users from crashes.
        
        return 128;
    }
    
    // Number of threads we decided on.
    int        nthr_;
    
    // Pool, unless nthr_ is less than 2.
    ThreadPool pool_;
}

/*****
 * This class allows application writers to queue tasks that can be
 * done on concurrent worker threads.  Create an object of this type,
 * passing a delegate that represents some work that needs to be done
 * to the constructor.  A task object will be created and placed on a
 * global queue.
 *
 * Normally, a ThreadPool object (or more than one) is created with
 * several worker threads that can handle these queued requests whlie
 * the main thread does other things.
 *
 * When the results are needed, fetch them with the value() method.
 * If the results have not been computed by the time value() is
 * called, the thread asking for them will run the computation inside
 * the call to value().
 */

class Future(T, U...) {
    alias AsyncTask!(T, U) Task;
    
    // Construct a task at priority 0.
    this(T delegate(U) dg, U args)
    {
        task_ = new Task(0, dg, args);
    }
    
    // Construct a task, assigning a priority.
    this(int priority, T delegate(U) dg, U args)
    {
        task_ = new Task(priority, dg, args);
    }
    
    // Get the results of the calculation, waiting if necessary.
    T value()
    {
        return task_.getValue();
    }
    
    // Wait for the calculation to be finished.
    void wait()
    {
        task_.getValue();
    }
    
    // Test whether the calculation has finished.
    bool isDone()
    {
        return task_.isDone();
    }
    
 private:
    Task task_;
}


/*****
 * This class creates several worker threads.  If any asynchronous
 * jobs are queued by the Future class, these worker thread will try
 * to do the calculations concurrently with the main application and
 * each other.  Each thread pool must be stopped by calling stop()
 * before it is deleted or the application terminates.
 */
class ThreadPool {
    // Construct a thread pool with n threads.
    this(int n)
    {
        list_ = TaskList();
        
        for(int i = 0; i < n; i++) {
            version(Tango) {
                Thread t = new Thread(& workLoopTango);
            } else {
                Thread t = new Thread(& workLoopPhobos);
            }
            threads_ ~= t;
        }
        
        // Start all the threads.
        synchronized(this) {
            foreach(Thread t; threads_) {
                t.start();
            }
        }
        
        // Wait until all the threads are running before returning, so
        // that stop() cannot be called before all threads have
        // started.
        
        while(1) {
            synchronized(this) {
                volatile {
                    if (running_ == n) {
                        break;
                    }
                }
            }
            Thread.yield();
        }
    }
    
    // Destructor - you must call 'stop' before this runs.  We can't
    // call stop() here directly, because we don't know in what order
    // the destructors will be called for this and the Thread objects.
    
    ~this()
    {
        assert(stopped_);
    }
    
    // Call this to stop this thread pool -- it will wait for all
    // threads to terminate before returning.
    
    void stop()
    {
        bool done = false;
        
        // Signal all threads to resume and wait for them to finish
        // their workLoop() call.  If they are blocked in the task
        // list, this will unblock them.
        
        while(! done) {
            synchronized(this) {
                volatile {
                    stopping_ = true;
                    
                    if (running_ == 0) {
                        done = true;
                    }
                }
                
                version(Tango) {
                    // no resume needed.
                    Thread.yield();
                } else {
                    Thread.resumeAll();
                }
            }
            Thread.yield();
        }
        
        foreach(Thread t; threads_) {
            version(Tango) {
                t.join();
            } else {
                t.wait();
            }
        }
        
        stopped_ = true;
    }
    
    // Check whether a global stop has been requested.
    
    static bool checkGlobalStop()
    {
        volatile {
            return global_stop;
        }
    }
    
    // Request a global stop.  This prevents tasks from being queued
    // on thread pools, but is not a substitute for calling stop() on
    // each thread pool.
    
    static void stopAllTasks()
    {
        volatile {
            global_stop = true;
        }
    }
    
private:
    alias AsyncTaskBase Task;
    
    static bool global_stop = false;
    
    void workLoopTango()
    {
        void adjust_count(bool up)
        {
            synchronized(this) {
                volatile {
                    if (up) {
                        running_ ++;
                    } else {
                        running_ --;
                    }
                }
            }
        }
        
        // Keep track of number of threads still running in the while
        // loop below (and therefore possibly waiting in task_list.)
        
        adjust_count(true);
        scope(exit) adjust_count(false);
        
        bool do_break = false;
        
        while(1) {
            // first, check for and bail out if this thread pool is
            // closing shop.
            
            synchronized(this) {
                volatile {
                    if (stopping_ || checkGlobalStop()) {
                        do_break = true;
                    }
                }
            }
            
            if (do_break)
                break;

            // This method may call pause().
            Task task = list_.getTask();
            
            volatile {
                if ((! stopping_) && (task !is null)) {
                    task.checkTask(false);
                }
            }
        }
    }
    
    // Different signature for Phobos threads.
    int workLoopPhobos()
    {
        workLoopTango();
        
        return 0;
    }
    
    // All running tasks in the application.
    TaskList list_;
    
    // Threads in this thread pool.
    Thread[] threads_;
    
    // Count of running threads in workLoop()
    int      running_ = 0;
    
    // True when this thread pool is shutting down.
    bool     stopping_ = false;
    
    // True when shut-down has completed.
    bool     stopped_ = false;
}


// Asynchronous task.
//
// For a given data type T, this class represents a process that
// produces that data and might be run asynchronously.

private class AsyncTask(T, U...) : AsyncTaskBase {
    // Job to do.
    alias T delegate(U) Work;
    
    // Build a task item.
    this(int priority, Work w, U args)
    {
        super(priority);
        work_ = w;
        
        foreach(i,a; args) {
            args_[i] = a;
        }
        
        TaskList().addTask(this);
    }
    
    // do work and/or get result for consumer of this calculation.
    T getValue()
    {
        checkTask(true);

        if (error_ !is null) {
            throw error_;
        }

        static if(! is(T == void)) {
            return value_;
        }
    }

    // try to do work
    void work()
    {
        TaskList().removeTask(this);

        try {
            static if (is(T == void)) {
                work_(args_);
            } else {
                value_ = work_(args_);
            }
        }
        catch(Exception e) {
            error_ = e;
        }
    }

 private:
     Work      work_;
     U         args_;
     Exception error_;
     bool      doing_ = false;
     
     static if (! is(T == void)) {
         T      value_;
     }
}

// Base class for AsyncTask.

abstract private class AsyncTaskBase : IndexPQNode {
    this(int p)
    {
        synchronized {
            volatile {
                serial_ = next_serial_ ++;
            }
        }
        
        priority_ = p;
    }
    
    // Do task or wait for completion.  Value consumers should use
    // wait = true to sleep until results are available; thread pool
    // worker threads should use wait = false to avoid blocking on
    // already-queued work items.
    
    void checkTask(bool will_wait)
    {
        bool do_work = false;
        bool need_wait = false;
        
        synchronized(this) {
            volatile {
                if (state_ == State.start) {
                    state_ = State.run;
                    do_work = true;
                } else if (state_ == State.run) {
                    need_wait = true;
                } else {
                    assert(state_ == State.done);
                    return;
                }
            }
        }
        
        if (do_work) {
            work();
            haveResult();
        } else if (will_wait && need_wait) {
            needResult();
        }
    }
    
    int opCmp(Object obj)
    {
        AsyncTaskBase other = cast(AsyncTaskBase) obj;
        assert(other !is null);
        
        if (this is other) {
            return 0;
        }
        
        if (priority_ < other.priority_) {
            return -1;
        } else if (priority_ > other.priority_) {
            return 1;
        }
        
        if (serial_ < other.serial_) {
            return 1;
        }
        
        assert(serial_ > other.serial_);
        return -1;
    }
    
    bool isDone()
    {
        bool done = false;
        
        synchronized(this) {
            volatile {
                if (state_ == State.done) {
                    done = true;
                }
            }
        }
        
        return done;
    }
    
protected:
    abstract void work();
    
private:
    enum State {
        start, run, done
    }
    
    void haveResult()
    {
        // This is intended to change the state from start to done;
        // then it loops, calling resumeAll(), until there are no
        // waiters.
        
        bool set_done = false;
        
        while(1) {
            synchronized(this) {
                volatile {
                    if (set_done) {
                        assert(state_ == State.done);
                    } else {
                        assert(state_ == State.run);
                        state_ = State.done;
                        set_done = true;
                    }
                    
                    if (waiters_ == 0) {
                        return;
                    } else {
                        version(Tango) {
                            // no resume needed.
                        } else {
                            Thread.resumeAll();
                        }
                    }
                }
            }
            
            // To allow the waiters to wake up and leave.
            Thread.yield();
        }
    }
    
    void needResult()
    {
        version(Tango) {
            PseudoPause pauser;
            pauser.msg = "needResult";
        }
        
        synchronized(this) {
            volatile {
                if (state_ == State.done) {
                    return;
                } else {
                    assert(state_ == State.run);
                    waiters_ ++;
                }
            }
        }
        
        while(1) {
            synchronized(this) {
                volatile {
                    if (state_ == State.done) {
                        waiters_ --;
                        return;
                    }
                }
            }
            
            version(Tango) {
                // Simulate pause with yields and timed sleeps.
                pauser.pause();
            } else {
                Thread.getThis().pause();
            }
        }
    }
    
 private:
    State state_   = State.start;
    int   waiters_ = 0;
    
    long  serial_;
    int   priority_;
    
    static long next_serial_ = 1;
}

// Element of an index priority queue.
abstract private class IndexPQNode {
    abstract int opCmp(Object o);
    
    // position in the queue.
    int position()
    {
        return pq_index;
    }
    
private:
    long pq_index = -1;
}

// Index priority queue -- this is like any priority queue, but the
// elements know their own location, so that elements can be deleted
// in constant time.  This is a base class, because it doesn't know
// what type of node it is working with.

private class IndexPQ_Base {
    alias IndexPQNode Node;
    
    this()
    {
        nodes_.length = 1;
    }
    
    void add(Node n)
    {
        // Add the node
        int pos = nodes_.length;
        n.pq_index = pos;
        nodes_ ~= n;
        
        // Position it
        bubble_up_(pos);
    }
    
    void remove(Node n)
    in
    {
        assert(n !is null);
        assert(nodes_.length > 1);
        assert(n.pq_index != -1);
        assert(nodes_[n.pq_index] is n);
    }
    body
    {
        // Remove the indicated node, replacing it with the last node
        
        int pos = n.pq_index;
        
        if (pos != nodes_.length-1) {
            nodes_[pos] = nodes_[$-1];
            nodes_[pos].pq_index = pos;
        }
        
        nodes_[$-1] = null;
        nodes_.length = nodes_.length - 1;
        
        // Fix the position of the moved node.
        if (nodes_.length > 2 && nodes_.length != pos) {
            bubble_(pos);
        }
    }
    
    // Get the top node.
    Node top()
    in
    {
        assert(nodes_.length > 1);
    }
    out(result)
    {
        assert(result.pq_index == 1);
    }
    body
    {
        return nodes_[1];
    }
    
    // Return true if the queue is empty.
    int length()
    {
        assert(nodes_.length > 0);
        return nodes_.length - 1;
    }
    
private:
    void bubble_up_(int pos)
    in
    {
        assert(pos < nodes_.length && pos >= 1);
    }
    body
    {
        while(pos != 1 && higher_(pos, pos/2)) {
            swap_(pos, pos/2);
            pos = pos/2;
        }
        
        // Always try to bubble down after bubbling up
        bubble_down_(pos);
    }
    
    void bubble_down_(int pos)
    in
    {
        assert(pos < nodes_.length && pos >= 1);
    }
    body
    {
        while(pos*2 < nodes_.length) {
            int left = pos*2, right = pos*2 + 1;
        
            bool go_left = higher_(left, pos),
                go_right = higher_(right, pos);
        
            if (go_left && go_right) {
                if (higher_(left, right)) {
                    swap_(pos, left);
                    pos = left;
                } else {
                    swap_(pos, right);
                    pos = right;
                }
            } else if (go_left) {
                swap_(left, pos);
                pos = left;
            } else if (go_right) {
                swap_(right, pos);
                pos = right;
            } else {
                return;
            }
        }
    }
    
    void bubble_(int pos)
    in
    {
        assert(pos < nodes_.length && pos >= 1);
    }
    body
    {
        // when in doubt, bubble up.
        bubble_up_(pos);
    }
    
    bool higher_(int pos1, int pos2)
    {
        if (pos1 < 1 || pos1 >= nodes_.length ||
            pos2 < 1 || pos2 >= nodes_.length) {
            
            return false;
        }
        
        return (nodes_[pos1] > nodes_[pos2]);
    }
    
    void swap_(int pos1, int pos2)
    in
    {
        assert(pos1 >= 1); assert(pos1 < nodes_.length);
        assert(pos2 >= 1); assert(pos2 < nodes_.length);
    }
    body
    {
        Node n1 = nodes_[pos1];
        Node n2 = nodes_[pos2];
        
        n1.pq_index = pos2;
        n2.pq_index = pos1;
        
        nodes_[n1.pq_index] = n1;
        nodes_[n2.pq_index] = n2;
    }
    
    Node[] nodes_;
}

// Index priority queue, specialized by node type.

private class IndexPQ(T) : IndexPQ_Base {
    void add(T n)
    {
        super.add(n);
    }
    
    void remove(T n)
    {
        super.remove(n);
    }
    
    T top()
    {
        return cast(T) super.top();
    }
}

// All tasks in the system are queued here.

private class TaskList {
    alias AsyncTaskBase Task;
    
    // Get the singleton object.
    
    static TaskList opCall()
    {
        return instance_;
    }
    
    // This is called to add a new task to the queue.
    
    void addTask(Task task)
    {
        synchronized(this) {
            queue_.add(task);
            
            // Wake up threads waiting for work, if any.
            
            if (waiters_) {
                version(Tango) {
                    // no resume needed.
                } else {
                    Thread.resumeAll();
                }
            }
        }
    }
    
    // Remove a task because a thread is working on it.
    
    void removeTask(Task task)
    {
        synchronized(this) {
            removeTask_nl(task);
        }
    }
    
    // Get any new task to work on.
    
    Task getTask()
    {
        // If there are unassigned tasks, find the highest priority
        // one, move it to the assigned tasks queue, and return it to
        // the caller.
        
        synchronized(this) {
            volatile {
                if (queue_.length) {
                    Task t = getTask_nl();
                    return t;
                }
                waiters_ ++;
            }
        }
        
        // Otherwise, wait.
        
        version(Tango) {
            // done in caller
            PseudoPause p;
            p.sleep();
        } else {
            Thread.getThis().pause();
        }
        
        // Instead of checking the queue again, we return null.  This
        // allows the caller to check for other messages or conditions
        // before calling us again in a loop.
        
        synchronized(this) {
            volatile {
                waiters_ --;
            }
        }
        
        return null;
    }
    
private:
    // Constructor.
    this()
    {
        queue_ = new IndexPQ!(Task);
    }
    
    // Class constructor.
    static this()
    {
        instance_ = new TaskList();
    }
    
    // Remove a task from the unassigned tasks queue (if this is being
    // run "in-line" by a value consumer) or the assigned tasks list
    // (by a worker thread).
    
    void removeTask_nl(Task task)
    {
        for(int i = 0; i < assigned_.length; i++) {
            if (assigned_[i] is task) {
                assigned_[i] = assigned_[$-1];
                assigned_.length = assigned_.length - 1;
                
                return;
            }
        }
        
        queue_.remove(task);
    }
    
    // Move a task from the unassigned tasks queue to the assigned
    // tasks list.
    
    Task getTask_nl()
    {
        assert(queue_.length);
        Task t = queue_.top();
        queue_.remove(t);
        assigned_ ~= t;

        return t;
    }
    
    static TaskList instance_;
    
    IndexPQ!(Task) queue_;
    Task[] assigned_;
    
    // threads waiting here.
    int waiters_;
}

// The phobos version of Futurism uses pause and resumeAll to wait for
// state-change events.  Tango has a different tool set for this, but
// it is not available yet.  For now, I simulate pause events with
// yields and short sleeps, and do nothing for "resume".  The resume
// operations are not needed because the timed sleeps wake up.
//
// I expected the phobos version to run faster, because it wakes up
// waiters right away, whereas the tango version needs to wait for the
// sleep intervals to expire.  However, at the moment, the tango
// implementation seems to perform better (for some cases).
//
// Two possible explanations come to mind.  First, yield()ing several
// times before sleeping may be beneficial enough that the phobos
// version of this code should adopt a variant of it.  Secondly, the
// resumeAll() operation is inefficient because it causes unnecessary
// thread wakeups, and possibly also unnecessary signal handling in
// non-sleeping threads.  These together may explain the difference.

version(Tango) {
    struct PseudoPause {
        void pause()
        {
            if (serial1 == 0) {
                serial1 = serial ++;
            }
            
            if (count < yields) {
                Thread.yield();
            } else {
                sleep();
            }
            
            count ++;
        }
        
        void sleep()
        {
            Thread.sleep(Interval.second/10);
        }
        
        char[] msg;
        int count = 0;
        int yields = 10;
        int report = 1;
        static int serial = 100;
        int serial1 = 0;
    }
}