SystemFiberScheduler.cs

Go to the documentation of this file.

/*

Author: Aaron Oneal, http://aarononeal.info

Copyright (c) 2012 Spicy Pixel, http://spicypixel.com

Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:

The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

*/
using System;
using System.Collections;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading;

namespace SpicyPixel.Threading
{       
    public class SystemFiberScheduler : FiberScheduler
    {
        public static SystemFiberScheduler StartNew()
        {
            return StartNew(null, CancellationToken.None, 0f);
        }
        
        public static SystemFiberScheduler StartNew(CancellationToken token, float updatesPerSecond = 0f)
        {
            return StartNew(null, token, updatesPerSecond);
        }
        
        public static SystemFiberScheduler StartNew(Fiber fiber)
        {
            return StartNew(fiber, CancellationToken.None, 0f);
        }
        
        public static SystemFiberScheduler StartNew(Fiber fiber, CancellationToken token, float updatesPerSecond = 0f)
        {
            SystemFiberScheduler backgroundScheduler = null;
            
            // Setup a thread to run the scheduler
            var wait = new ManualResetEvent(false);
            var thread = new Thread(() => {
                backgroundScheduler = (SystemFiberScheduler)FiberScheduler.Current;
                wait.Set();
                FiberScheduler.Current.Run(fiber, token, updatesPerSecond);
            });
            thread.Start();
            wait.WaitOne();
            
            return backgroundScheduler;
        }
        
        private const int MaxStackDepth = 10;
        
        [ThreadStatic]
        private static int stackDepthQueueFiber = 0;

        private ConcurrentQueue<Fiber> executingFibers = new ConcurrentQueue<Fiber>();
        
        private ConcurrentQueue<Tuple<Fiber, float>> sleepingFibers = new ConcurrentQueue<Tuple<Fiber, float>>();
                
        // A future queue may include waitingFibers (e.g. waiting on a signal or timeout)
        
        private float currentTime;
        
        private ManualResetEvent runWaitHandle = new ManualResetEvent(true);
            
        private ManualResetEvent disposeWaitHandle = new ManualResetEvent(false);
        
        private AutoResetEvent schedulerEventWaitHandle = new AutoResetEvent(false);
        
        protected ManualResetEvent RunWaitHandle
        {
            get { return runWaitHandle; }
        }
        
        protected WaitHandle DisposeWaitHandle
        {
            get { return disposeWaitHandle; }
        }
        
        protected WaitHandle SchedulerEventWaitHandle
        {
            get { return schedulerEventWaitHandle; }
        }
        
        protected int ExecutingFiberCount
        {
            get { return executingFibers.Count; }
        }
        
        protected int SleepingFiberCount
        {
            get { return sleepingFibers.Count; }
        }
        
        public SystemFiberScheduler ()
        {
        }
                
        protected sealed override void QueueFiber(Fiber fiber)
        {
            // Queueing can happen from completion callbacks
            // which may happen once the fiber has already
            // executed and changed state. It would be fine
            // if the queue did happen because non-running
            // fibers are skipped, but it's better to
            // shortcut here.
            if(fiber.Status != FiberStatus.WaitingToRun)
                return;
                
            // Entering queue fiber where recursion might matter
            Interlocked.Increment(ref stackDepthQueueFiber);
            
            try
            {               
                // Execute immediately to inline as much as possible
                //
                // Note: Some applications may want to always queue to control
                // performance more strictly by the run loop.
                if(AllowInlining && SchedulerThread == Thread.CurrentThread && stackDepthQueueFiber < MaxStackDepth)
                {
                    ExecuteFiberInternal(fiber);
                    return;
                }
                else 
                {
                    QueueFiberForExecution(fiber);
                    return;
                }
            }
            finally
            {
                // Exiting queue fiber
                Interlocked.Decrement(ref stackDepthQueueFiber);
            }
        }

        private void QueueFiberForExecution(Fiber fiber)
        {
            executingFibers.Enqueue(fiber);
            
            // Queueing a new execution fiber needs to trigger re-evaluation of the
            // next update time
            schedulerEventWaitHandle.Set();
        }
        
        private void QueueFiberForSleep(Fiber fiber, float timeToWake)
        {
            var tuple = new Tuple<Fiber, float>(fiber, timeToWake);
            sleepingFibers.Enqueue(tuple);
                
            // Fibers can only be queued for sleep when they return
            // a yield instruction. This can only happen when executing
            // on the main thread and therefore we will never be in
            // a wait loop with a need to signal the scheduler event handle.
        }
        
        protected void Update(float time)
        {
            currentTime = time;

            UpdateExecutingFibers();
            UpdateSleepingFibers();
        }
        
        private void UpdateExecutingFibers()
        {
            // Run executing fibers
            
            // Add null to know when to stop
            executingFibers.Enqueue(null);
            
            Fiber item;
            while(executingFibers.TryDequeue(out item))
            {
                // If we reached the marker for this update then stop
                if(item == null)
                    break;
                
                // Skip completed items
                if(item.IsCompleted)
                    continue;
                
                ExecuteFiberInternal(item);
            }
        }
        
        private void UpdateSleepingFibers()
        {
            // Wake sleeping fibers that it's time for
        
            // Add null to know when to stop
            sleepingFibers.Enqueue(null);
            
            Tuple<Fiber, float> item;
            while(sleepingFibers.TryDequeue(out item))
            {
                // If we reached the marker for this update then stop
                if(item == null)
                    break;
                
                Fiber fiber = item.Item1;
                
                // Skip completed items
                if(fiber.IsCompleted)
                    continue;
                
                // Run if time or cancelled otherwise re-enqueue
                if(item.Item2 <= currentTime || fiber.cancelToken.IsCancellationRequested)
                    ExecuteFiberInternal(item.Item1);
                else
                    sleepingFibers.Enqueue(item);
            }
        }
        
        protected sealed override void AbortRequested(Fiber fiber)
        {
            schedulerEventWaitHandle.Set();
        }
                        
        protected bool GetNextFiberWakeTime(out float fiberWakeTime)
        {
            fiberWakeTime = -1f;
            
            // Nothig to do if there are no sleeping fibers
            if(sleepingFibers.Count == 0)
                return false;

            // Find the earliest wake time
            foreach(var fiber in sleepingFibers)
            {
                if(fiber.Item1.cancelToken.IsCancellationRequested)
                {
                    fiberWakeTime = 0f; // wake immediately
                    break;
                }
                
                if(fiberWakeTime == -1f || fiber.Item2 < fiberWakeTime)
                    fiberWakeTime = fiber.Item2;
            }
            
            return true;
        }
        
        private IEnumerator CancelWhenComplete(Fiber waitOnFiber, CancellationTokenSource cancelSource)
        {
            yield return waitOnFiber;
            cancelSource.Cancel();
        }
                
        public override void Run(Fiber fiber, CancellationToken token, float updatesPerSecond)
        {
            long frequencyTicks = (long)(updatesPerSecond * (float)TimeSpan.TicksPerSecond); // min time between updates (duration)
            long startTicks = 0;        // start of update time (marker)
            long endTicks = 0;          // end of update time (marker)
            long sleepTicks;            // time to sleep (duration)
            long wakeTicks;             // ticks before wake (duration)
            int sleepMilliseconds;      // ms to sleep (duration)
            int wakeMilliseconds;       // ms before wake (duration)
            float wakeMarkerInSeconds;  // time of wake in seconds (marker)
            var mainFiberCompleteCancelSource = new CancellationTokenSource();
                        
            if(isDisposed)
                throw new ObjectDisposedException(GetType().FullName);
            
            // Run is not re-entrant, make sure we're not running
            if(!runWaitHandle.WaitOne(0))
                throw new InvalidOperationException("Run is already executing and is not re-entrant");
            
            // Verify arguments
            if(updatesPerSecond < 0f)
                throw new ArgumentOutOfRangeException("updatesPerSecond", "The updatesPerSecond must be >= 0");
            
            // Get a base time for better precision
            long baseTicks = DateTime.Now.Ticks;

            // Build wait list to terminate execution
            var waitHandleList = new List<WaitHandle>(4);
            waitHandleList.Add(schedulerEventWaitHandle);
            waitHandleList.Add(disposeWaitHandle);
            
            if(token.CanBeCanceled)
                waitHandleList.Add(token.WaitHandle);
                        
            try
            {
                if(fiber != null)
                {               
                    // Add the main fiber to the wait list so when it completes
                    // the wait handle falls through.
                    waitHandleList.Add(mainFiberCompleteCancelSource.Token.WaitHandle);

                    // Start the main fiber if it isn't running yet
                    if(fiber.Status == FiberStatus.Created)
                        fiber.Start(this);
                    
                    // Start another fiber that waits on the main fiber to complete.
                    // When it does, it raises a cancellation.
                    Fiber.Factory.StartNew(CancelWhenComplete(fiber, mainFiberCompleteCancelSource), this);
                }
                
                WaitHandle[] waitHandles = waitHandleList.ToArray();

                // FIXME: Unclear why below was included as the handle 
                // seems to be needed to wake sleeping fibers when abort is called.
                //waitHandleList.Remove(schedulerEventWaitHandle);

                WaitHandle[] sleepWaitHandles = waitHandleList.ToArray();
                
                runWaitHandle.Reset();
                
                while(true)
                {                   
                    // Stop executing if cancelled
                    if((token.CanBeCanceled && token.IsCancellationRequested) || mainFiberCompleteCancelSource.IsCancellationRequested || disposeWaitHandle.WaitOne(0))
                        return;
                    
                    // Snap current time
                    startTicks = DateTime.Now.Ticks;
    
                    // Update using this time marker (and convert ticks to s)
                    Update((float)((double)(startTicks - baseTicks) / (double)TimeSpan.TicksPerSecond));
    
                    // Only sleep to next frequency cycle if one was specified
                    if(updatesPerSecond > 0f)
                    {                       
                        // Snap end time
                        endTicks = DateTime.Now.Ticks;
                        
                        // Sleep at least until next update
                        sleepTicks = frequencyTicks - (endTicks - startTicks);
                        if(sleepTicks > 0)
                        {
                            sleepMilliseconds = (int)(sleepTicks / TimeSpan.TicksPerMillisecond);
                            
                            WaitHandle.WaitAny(sleepWaitHandles, sleepMilliseconds);
                            
                            // Stop executing if cancelled
                            if((token.CanBeCanceled && token.IsCancellationRequested) || mainFiberCompleteCancelSource.IsCancellationRequested || disposeWaitHandle.WaitOne(0))
                                return;
                        }
                    }
                                    
                    // Now keep sleeping until it's time to update
                    while(ExecutingFiberCount == 0)
                    {                       
                        // Assume we wait forever (e.g. until a signal)
                        wakeMilliseconds = -1;
    
                        // If there are sleeping fibers, then set a wake time
                        if(GetNextFiberWakeTime(out wakeMarkerInSeconds))
                        {
                            wakeTicks = baseTicks;
                            wakeTicks += (long)((double)wakeMarkerInSeconds * (double)TimeSpan.TicksPerSecond);                     
                            wakeTicks -= DateTime.Now.Ticks;
                                
                            // If there was a waiting fiber and it's already past time to awake then stop waiting
                            if(wakeTicks <= 0)
                                break;
                            
                            wakeMilliseconds = (int)(wakeTicks / TimeSpan.TicksPerMillisecond);
                        }

                        // FIXME: Sleeping tasks can be aborted and this should wake the scheduler.
                        // For some reason the schedulerEventWaitHandle which would do this was not
                        // in the wait list and removed above. Trying with it in the list again.
                                                
                        // There was no waiting fiber and we will wait for another signal,
                        // or there was a waiting fiber and we wait until that time.
                        WaitHandle.WaitAny(waitHandles, wakeMilliseconds);
                        
                        // Stop executing if cancelled
                        if((token.CanBeCanceled && token.IsCancellationRequested) || mainFiberCompleteCancelSource.IsCancellationRequested || disposeWaitHandle.WaitOne(0))
                            return;
                    }
                }   
            }
            finally
            {                   
                // Clear queues
                Fiber deqeueFiber;
                while(executingFibers.TryDequeue(out deqeueFiber));
                
                Tuple<Fiber, float> dequeueSleepingFiber;
                while(sleepingFibers.TryDequeue(out dequeueSleepingFiber));
                
                // Reset time
                currentTime = 0f;
                
                // Set for dispose
                runWaitHandle.Set();
            }
        }
                        
        private void ExecuteFiberInternal(Fiber fiber)
        {
            Fiber currentFiber = fiber;
            Fiber nextFiber;

            while(currentFiber != null)
            {
                // Execute the fiber
                var fiberInstruction = ExecuteFiber(currentFiber);
    
                // Nothing more to do if stopped
                if(currentFiber.IsCompleted)
                    return;
    
                // Handle special fiber instructions or queue for another update
                bool fiberQueued = false;
                OnFiberInstruction(currentFiber, fiberInstruction, out fiberQueued, out nextFiber);
                
                // If the fiber is still running but wasn't added to a special queue by
                // an instruction then it needs to be added to the execution queue
                // to run in the next Update().
                //
                // Check alive state again in case an instruction resulted
                // in an inline execution and altered state.
                if(!fiberQueued && !currentFiber.IsCompleted) {
                    // Send the fiber to the queue and don't execute inline
                    // since we're done this update
                    QueueFiberForExecution(currentFiber);
                }
                
                // Switch to the next fiber if an instruction says to do so
                currentFiber = nextFiber;
            }
        }
        
        private void OnFiberInstruction(Fiber fiber, FiberInstruction instruction, out bool fiberQueued, out Fiber nextFiber)
        {
            fiberQueued = false;
            nextFiber = null;
            
            YieldUntilComplete yieldUntilComplete = instruction as YieldUntilComplete;
            if(yieldUntilComplete != null)
            {
                // The additional complexity below is because this was going
                // to handle waiting for completions for fibers from other threads.
                // Currently fibers must belong to the same thread and this is enforced
                // by the instructions themselves for now.
                
                int completeOnce = 0;
                
                // FIXME: If we support multiple schedulers in the future
                // this callback could occur from another thread and
                // therefore after Dispose(). Would probably need a lock.
                
                yieldUntilComplete.Fiber.ContinueWith ((f) => {
                    var originalCompleteOnce = Interlocked.CompareExchange(ref completeOnce, 1, 0);
                    if(originalCompleteOnce != 0)
                        return;

                    QueueFiber(fiber); // optionally execute inline when the completion occurs
                });

                fiberQueued = true;
                return;
            }
            
            YieldForSeconds yieldForSeconds = instruction as YieldForSeconds;
            if(yieldForSeconds != null)
            {
                QueueFiberForSleep(fiber, currentTime + yieldForSeconds.Seconds);
                fiberQueued = true;
                return;
            }
            
            YieldToFiber yieldToFiber = instruction as YieldToFiber;
            if(yieldToFiber != null)
            {
                RemoveFiberFromQueues(yieldToFiber.Fiber);
                nextFiber = yieldToFiber.Fiber;
                fiberQueued = false;
                return;
            }
        }
        
        private void RemoveFiberFromQueues(Fiber fiber)
        {
            bool found = false;
            
            if(executingFibers.Count > 0)
            {
                Fiber markerItem = new Fiber(() => {});
                executingFibers.Enqueue(markerItem);
                
                Fiber item;
                while(executingFibers.TryDequeue(out item))
                {
                    if(item == markerItem)
                        break;
                    
                    if(item == fiber)
                        found = true;
                    else
                        executingFibers.Enqueue(item);
                }
                
                if(found)
                    return;
            }
            
            if(sleepingFibers.Count > 0)
            {
                Tuple<Fiber, float> markerTuple = new Tuple<Fiber, float>(null, 0f);
                sleepingFibers.Enqueue(markerTuple);
                
                Tuple<Fiber, float> itemTuple;
                while(sleepingFibers.TryDequeue(out itemTuple))
                {
                    if(itemTuple == markerTuple)
                        break;
                    
                    if(itemTuple != null && itemTuple.Item1 == fiber)
                        found = true;
                    else
                        sleepingFibers.Enqueue(itemTuple);
                }
            }
        }
        
        #region IDisposable implementation
        
        private bool isDisposed = false;
        
        protected override void Dispose(bool disposing) 
        {
            // Do nothing if already called
            if(isDisposed)
                return;
                        
            if(disposing) 
            {
                // Free other state (managed objects).
                disposeWaitHandle.Set();
                runWaitHandle.WaitOne();
            }
            
            // Free your own state (unmanaged objects).
            // Set large fields to null.
            
            // Mark disposed
            isDisposed = true;
            
            base.Dispose(disposing);
        }
        #endregion
    }
}