root/trunk/infrastructure/st/stackthread.d

/******************************************************
 * StackThreads are userland, cooperative, lightweight
 * threads. StackThreads are very efficient, requiring
 * much less time per context switch than real threads.
 * They also require far fewer resources than real
 * threads, which allows many more StackThreads to exist
 * simultaneously. In addition, StackThreads do not
 * require explicit synchronization since they are
 * non-preemptive.  There is no requirement that code
 * be reentrant.
 *
 * This module implements the stack thread system on top
 * of the context layer.
 *
 * Version: 0.3
 * Date: July 4, 2006
 * Authors:
 *  Mikola Lysenko, mclysenk@mtu.edu
 * License: Use/copy/modify freely, just give credit.
 * Copyright: Public domain.
 *
 * Bugs:
 *  Not thread safe.  May be changed in future versions,
 *  however this will require a radical refactoring.
 *
 * History:
 *  v0.7 - Switched timing resolution to milliseconds.
 *
 *  v0.6 - Removed timing functions from st_yield/st_throwYield
 *
 *  v0.5 - Addded st_throwYield and MAX/MIN_THREAD_PRIORITY
 *
 *  v0.4 - Unittests finished-ready for an initial release.
 *
 *  v0.3 - Changed name back to StackThread and added
 *      linux support.  Context switching is now handled
 *      in the stackcontext module, and much simpler to
 *      port.
 *
 *  v0.2 - Changed name to QThread, fixed many issues.
 *  
 *  v0.1 - Initial stack thread system. Very buggy.
 *
 ******************************************************/
module st.stackthread;

//Module imports
private import
    st.stackcontext,
    std.stdio,
    std.string;

/// The priority of a stack thread determines its order in
/// the scheduler.  Higher priority threads go first.
alias int priority_t;

/// The default priority for a stack thread is 0.
const priority_t DEFAULT_STACKTHREAD_PRIORITY = 0;

/// Maximum thread priority
const priority_t MAX_STACKTHREAD_PRIORITY = 0x7fffffff;

/// Minimum thread priority
const priority_t MIN_STACKTHREAD_PRIORITY = 0x80000000;

/// The state of a stack thread
enum THREAD_STATE
{
    READY,      /// Thread is ready to run
    RUNNING,    /// Thread is currently running
    DEAD,       /// Thread has terminated
    SUSPENDED,  /// Thread is suspended
}

/// The state of the scheduler
enum SCHEDULER_STATE
{
    READY,      /// Scheduler is ready to run a thread
    RUNNING,    /// Scheduler is running a timeslice
}

//Timeslices
private STPriorityQueue active_slice;
private STPriorityQueue next_slice;

//Scheduler state
private SCHEDULER_STATE sched_state;
    
//Start time of the time slice
private ulong sched_t0;

//Currently active stack thread
private StackThread sched_st;

version(Win32)
{
    private extern(Windows) int QueryPerformanceFrequency(ulong *);
    private ulong sched_perf_freq;
}


//Initialize the scheduler
static this()
{
    active_slice = new STPriorityQueue();
    next_slice = new STPriorityQueue();
    sched_state = SCHEDULER_STATE.READY;
    sched_t0 = -1;
    sched_st = null;
    
    version(Win32)
        QueryPerformanceFrequency(&sched_perf_freq);
}


/******************************************************
 * StackThreadExceptions are generated whenever the
 * stack threads are incorrectly invoked.  Trying to
 * run a time slice while a time slice is in progress
 * will result in a StackThreadException.
 ******************************************************/
class StackThreadException : Exception
{
    this(char[] msg)
    {
        super(msg);
    }
    
    this(StackThread st, char[] msg)
    {
        super(format("%s: %s", st.toString, msg));
    }
}



/******************************************************
 * StackThreads are much like regular threads except
 * they are cooperatively scheduled.  A user may switch
 * between StackThreads using st_yield.
 ******************************************************/
class StackThread
{
    /**
     * Creates a new stack thread and adds it to the
     * scheduler.
     *
     * Params:
     *  dg = The delegate we are invoking
     *  stack_size = The size of the stack for the stack
     *  thread.
     *  priority = The priority of the stack thread.
     */
    public this
    (
        void delegate() dg, 
        priority_t priority = DEFAULT_STACKTHREAD_PRIORITY,
        size_t stack_size = DEFAULT_STACK_SIZE
    )
    {
        this.m_delegate = dg;
        this.context = new StackContext(&m_proc, DEFAULT_STACK_SIZE);
        this.m_priority = priority;
        
        //Schedule the thread
        st_schedule(this);
        
        debug (StackThread) writefln("Created thread, %s", toString);
    }
    
    /**
     * Creates a new stack thread and adds it to the
     * scheduler, using a function pointer.
     *
     * Params:
     *  fn = The function pointer that the stack thread
     *  invokes.
     *  stack_size = The size of the stack for the stack
     *  thread.
     *  priority = The priority of the stack thread.
     */
    public this
    (
        void function() fn, 
        priority_t priority = DEFAULT_STACKTHREAD_PRIORITY,
        size_t stack_size = DEFAULT_STACK_SIZE
    )
    {
        this.m_delegate = &delegator;
        this.m_function = fn;
        this.context = new StackContext(&m_proc, DEFAULT_STACK_SIZE);
        this.m_priority = priority;
        
        //Schedule the thread
        st_schedule(this);
        
        debug (StackThread) writefln("Created thread, %s", toString);
    }
    
    /**
     * Converts the thread to a string.
     *
     * Returns: A string representing the stack thread.
     */
    public char[] toString()
    {
        debug(PQueue)
        {
            return format("ST[t:%8x,p:%8x,l:%8x,r:%8x]",
                cast(void*)this,
                cast(void*)parent,
                cast(void*)left,
                cast(void*)right);
        }
        else
        {
        static char[][] state_names =
        [
            "RDY",
            "RUN",
            "XXX",
            "PAU",
        ];
        
        //horrid hack for getting the address of a delegate
        union hack
        {
            struct dele
            {
                void * frame;
                void * fptr;
            }
            
            dele d;
            void delegate () dg;
        }
        hack h;
        if(m_function !is null)
            h.d.fptr = cast(void*) m_function;
        else if(m_delegate !is null)
            h.dg = m_delegate;
        else
            h.dg = &run;
        
        return format(
            "Thread[pr=%d,st=%s,fn=%8x]", 
            priority,
            state_names[cast(uint)state],
            h.d.fptr);
        }
    }
    
    invariant
    {
        assert(context);
        
        switch(state)
        {
            case THREAD_STATE.READY:
                assert(context.ready);
            break;
            
            case THREAD_STATE.RUNNING:
                assert(context.running);
            break;
            
            case THREAD_STATE.DEAD:
                assert(!context.running);
            break;
            
            case THREAD_STATE.SUSPENDED:
                assert(context.ready);
            break;

            default: assert(false);
        }
        
        if(left !is null)
        {
            assert(left.parent is this);
        }
        
        if(right !is null)
        {
            assert(right.parent is this);
        }
    }
    
    /**
     * Removes this stack thread from the scheduler. The
     * thread will not be run until it is added back to
     * the scheduler.
     */
    public final void pause()
    {
        debug (StackThread) writefln("Pausing %s", toString);
        
        switch(state)
        {
            case THREAD_STATE.READY:
                st_deschedule(this);
                state = THREAD_STATE.SUSPENDED;
            break;
            
            case THREAD_STATE.RUNNING:
                transition(THREAD_STATE.SUSPENDED);
            break;
            
            case THREAD_STATE.DEAD:
                throw new StackThreadException(this, "Cannot pause a dead thread");
            
            case THREAD_STATE.SUSPENDED:
                throw new StackThreadException(this, "Cannot pause a paused thread");

            default: assert(false);
        }
    }
    
    /**
     * Adds the stack thread back to the scheduler. It
     * will resume running with its priority & state
     * intact.
     */
    public final void resume()
    {
        debug (StackThread) writefln("Resuming %s", toString);
        
        //Can only resume paused threads
        if(state != THREAD_STATE.SUSPENDED)
        {
            throw new StackThreadException(this, "Thread is not suspended");
        }
        
        //Set state to ready and schedule
        state = THREAD_STATE.READY;
        st_schedule(this);
    }
    
    /**
     * Kills this stack thread in a violent manner.  The
     * thread does not get a chance to end itself or clean
     * anything up, it is descheduled and all GC references
     * are released.
     */
    public final void kill()
    {
        debug (StackThread) writefln("Killing %s", toString);
        
        switch(state)
        {
            case THREAD_STATE.READY:
                //Kill thread and remove from scheduler
                st_deschedule(this);
                state = THREAD_STATE.DEAD;
                context.kill();
            break;
            
            case THREAD_STATE.RUNNING:
                //Transition to dead
                transition(THREAD_STATE.DEAD);
            break;
            
            case THREAD_STATE.DEAD:
                throw new StackThreadException(this, "Cannot kill already dead threads");
            
            case THREAD_STATE.SUSPENDED:
                //We need to kill the stack, no need to touch scheduler
                state = THREAD_STATE.DEAD;
                context.kill();
            break;

            default: assert(false);
        }
    }
    
    /**
     * Waits to join with this thread.  If the given amount
     * of milliseconds expires before the thread is dead,
     * then we return automatically.
     *
     * Params:
     *  ms = The maximum amount of time the thread is 
     *  allowed to wait. The special value -1 implies that
     *  the join will wait indefinitely.
     *
     * Returns:
     *  The amount of millieconds the thread was actually
     *  waiting.
     */
    public final ulong join(ulong ms = -1)
    {
        debug (StackThread) writefln("Joining %s", toString);
        
        //Make sure we are in a timeslice
        if(sched_state != SCHEDULER_STATE.RUNNING)
        {
            throw new StackThreadException(this, "Cannot join unless a timeslice is currently in progress");
        }
        
        //And make sure we are joining with a valid thread
        switch(state)
        {
            case THREAD_STATE.READY:
                break;
            
            case THREAD_STATE.RUNNING:
                throw new StackThreadException(this, "A thread cannot join with itself!");
            
            case THREAD_STATE.DEAD:
                throw new StackThreadException(this, "Cannot join with a dead thread");
            
            case THREAD_STATE.SUSPENDED:
                throw new StackThreadException(this, "Cannot join with a paused thread");

            default: assert(false);
        }
        
        //Do busy waiting until the thread dies or the
        //timer runs out.
        ulong start_time = getSysMillis();
        ulong timeout = (ms == -1) ? ms : start_time + ms;
        
        while(
            state != THREAD_STATE.DEAD &&
            timeout > getSysMillis())
        {
            StackContext.yield();
        }
        
        return getSysMillis() - start_time;
    }
    
    /**
     * Restarts the thread's execution from the very
     * beginning.  Suspended and dead threads are not
     * resumed, but upon resuming, they will restart.
     */
    public final void restart()
    {
        debug (StackThread) writefln("Restarting %s", toString);
        
        //Each state needs to be handled carefully
        switch(state)
        {
            case THREAD_STATE.READY:
                //If we are ready,
                context.restart();
            break;
            
            case THREAD_STATE.RUNNING:
                //Reset the thread.
                transition(THREAD_STATE.READY);
            break;
            
            case THREAD_STATE.DEAD:
                //Dead threads become suspended
                context.restart();
                state = THREAD_STATE.SUSPENDED;
            break;
            
            case THREAD_STATE.SUSPENDED:
                //Suspended threads stay suspended
                context.restart();
            break;

            default: assert(false);
        }
    }
    
    /**
     * Grabs the thread's priority.  Intended for use
     * as a property.
     *
     * Returns: The stack thread's priority.
     */
    public final priority_t priority()
    {
        return m_priority;
    }
    
    /**
     * Sets the stack thread's priority.  Used to either
     * reschedule or reset the thread.  Changes do not
     * take effect until the next round of scheduling.
     *
     * Params:
     *  p = The new priority for the thread
     *
     * Returns:
     *  The new priority for the thread.
     */
    public final priority_t priority(priority_t p)
    {
        //Update priority
        if(sched_state == SCHEDULER_STATE.READY && 
            state == THREAD_STATE.READY)
        {
            next_slice.remove(this);
            m_priority = p;
            next_slice.add(this);
        }
        
        return m_priority = p;
    }
    
    /**
     * Returns: The state of this thread.
     */
    public final THREAD_STATE getState()
    {
        return state;
    }
    
    /**
     * Returns: True if the thread is ready to run.
     */
    public final bool ready()
    {
        return state == THREAD_STATE.READY;
    }
    
    /**
     * Returns: True if the thread is currently running.
     */
    public final bool running()
    {
        return state == THREAD_STATE.RUNNING;
    }
    
    /**
     * Returns: True if the thread is dead.
     */
    public final bool dead()
    {
        return state == THREAD_STATE.DEAD;
    }
    
    /**
     * Returns: True if the thread is not dead.
     */
    public final bool alive()
    {
        return state != THREAD_STATE.DEAD;
    }
    
    /**
     * Returns: True if the thread is paused.
     */
    public final bool paused()
    {
        return state == THREAD_STATE.SUSPENDED;
    }

    /**
     * Creates a stack thread without a function pointer
     * or delegate.  Used when a user overrides the stack
     * thread class.
     */
    protected this
    (
        priority_t priority = DEFAULT_STACKTHREAD_PRIORITY,
        size_t stack_size = DEFAULT_STACK_SIZE
    )
    {
        this.context = new StackContext(&m_proc, stack_size);
        this.m_priority = priority;
        
        //Schedule the thread
        st_schedule(this);
        
        debug (StackThread) writefln("Created thread, %s", toString);
    }
    
    /**
     * Run the stack thread.  This method may be overloaded
     * by classes which inherit from stack thread, as an
     * alternative to passing delegates.
     *
     * Throws: Anything.
     */
    protected void run()
    {
        m_delegate();
    }
    
    // Heap information
    private StackThread parent = null;
    private StackThread left = null;
    private StackThread right = null;

    // The thread's priority
    private priority_t m_priority;

    // The state of the thread
    private THREAD_STATE state;

    // The thread's context
    private StackContext context;

    //Delegate handler
    private void function() m_function;
    private void delegate() m_delegate;
    private void delegator() { m_function(); }
    
    //My procedure
    private final void m_proc()
    {
        try
        {
            debug (StackThread) writefln("Starting %s", toString);
            run;
        }
        catch(Object o)
        {
            debug (StackThread) writefln("Got a %s exception from %s", o.toString, toString);
            throw o;
        }
        finally
        {
            debug (StackThread) writefln("Finished %s", toString);
            state = THREAD_STATE.DEAD;
        }
    }

    /**
     * Used to change the state of a running thread
     * gracefully
     */
    private final void transition(THREAD_STATE next_state)
    {
        state = next_state;
        StackContext.yield();
    }
}



/******************************************************
 * The STPriorityQueue is used by the scheduler to
 * order the objects in the stack threads.  For the
 * moment, the implementation is binary heap, but future
 * versions might use a binomial heap for performance
 * improvements.
 ******************************************************/
private class STPriorityQueue
{
public:
    
    /**
     * Add a stack thread to the queue.
     *
     * Params:
     *  st = The thread we are adding.
     */
    void add(StackThread st)
    in
    {
        assert(st !is null);
        assert(st);
        assert(st.parent is null);
        assert(st.left is null);
        assert(st.right is null);
    }
    body
    {
        size++;
        
        //Handle trivial case
        if(head is null)
        {
            head = st;
            return;
        }
        
        //First, insert st
        StackThread tmp = head;
        int pos;
        for(pos = size; pos>3; pos>>>=1)
        {
            assert(tmp);
            tmp = (pos & 1) ? tmp.right : tmp.left;
        }
        
        assert(tmp !is null);
        assert(tmp);
        
        if(pos&1)
        {
            assert(tmp.left !is null);
            assert(tmp.right is null);
            tmp.right = st;
        }
        else
        {
            assert(tmp.left is null);
            assert(tmp.right is null);
            tmp.left = st;
        }
        st.parent = tmp;
        
        assert(tmp);
        assert(st);
        
        //Fixup the stack and we're good.
        bubble_up(st);
    }
    
    /**
     * Remove a stack thread.
     *
     * Params:
     *  st = The stack thread we are removing.
     */
    void remove(StackThread st)
    in
    {
        assert(st);
        assert(hasThread(st));
    }
    out
    {
        assert(st);
        assert(st.left is null);
        assert(st.right is null);
        assert(st.parent is null);
    }
    body
    {
        //Handle trivial case
        if(size == 1)
        {
            assert(st is head);
            
            --size;
            
            st.parent =
            st.left =
            st.right = 
            head = null;
            
            return;
        }
        
        //Cycle to the bottom of the heap
        StackThread tmp = head;
        int pos;
        for(pos = size; pos>3; pos>>>=1)
        {
            assert(tmp);
            tmp = (pos & 1) ? tmp.right : tmp.left;
        }
        tmp = (pos & 1) ? tmp.right : tmp.left;
        
        
        assert(tmp !is null);
        assert(tmp.left is null);
        assert(tmp.right is null);
        
        //Remove tmp
        if(tmp.parent.left is tmp)
        {
            tmp.parent.left = null;
        }
        else
        {
            assert(tmp.parent.right is tmp);
            tmp.parent.right = null;