OpenSim-Test/ThirdParty/SmartThreadPool/WorkItemsGroup.cs

using System;
using System.Threading;
using System.Runtime.CompilerServices;
using System.Diagnostics;

namespace Amib.Threading.Internal
{

    #region WorkItemsGroup class 

    /// <summary>
    /// Summary description for WorkItemsGroup.
    /// </summary>
    public class WorkItemsGroup : WorkItemsGroupBase
    {
        #region Private members

        private readonly object _lock = new object();

        /// <summary>
        /// A reference to the SmartThreadPool instance that created this 
        /// WorkItemsGroup.
        /// </summary>
        private readonly SmartThreadPool _stp;

        /// <summary>
        /// The OnIdle event
        /// </summary>
        private event WorkItemsGroupIdleHandler _onIdle;

        /// <summary>
        /// A flag to indicate if the Work Items Group is now suspended.
        /// </summary>
        private bool _isSuspended;

        /// <summary>
        /// Defines how many work items of this WorkItemsGroup can run at once.
        /// </summary>
        private int _concurrency;

        /// <summary>
        /// Priority queue to hold work items before they are passed 
        /// to the SmartThreadPool.
        /// </summary>
        private readonly PriorityQueue _workItemsQueue;

        /// <summary>
        /// Indicate how many work items are waiting in the SmartThreadPool
        /// queue.
        /// This value is used to apply the concurrency.
        /// </summary>
        private int _workItemsInStpQueue;

        /// <summary>
        /// Indicate how many work items are currently running in the SmartThreadPool.
        /// This value is used with the Cancel, to calculate if we can send new 
        /// work items to the STP.
        /// </summary>
        private int _workItemsExecutingInStp = 0;

        /// <summary>
        /// WorkItemsGroup start information
        /// </summary>
        private readonly WIGStartInfo _workItemsGroupStartInfo;

        /// <summary>
        /// Signaled when all of the WorkItemsGroup's work item completed.
        /// </summary>
        private readonly ManualResetEvent _isIdleWaitHandle = new ManualResetEvent(true);

        /// <summary>
        /// A common object for all the work items that this work items group
        /// generate so we can mark them to cancel in O(1)
        /// </summary>
        private CanceledWorkItemsGroup _canceledWorkItemsGroup = new CanceledWorkItemsGroup();

        #endregion 

        #region Construction

        public WorkItemsGroup(
            SmartThreadPool stp, 
            int concurrency, 
            WIGStartInfo wigStartInfo)
        {
            if (concurrency <= 0)
            {
                throw new ArgumentOutOfRangeException(
                    "concurrency",
                    concurrency,
                    "concurrency must be greater than zero");
            }
            _stp = stp;
            _concurrency = concurrency;
            _workItemsGroupStartInfo = new WIGStartInfo(wigStartInfo).AsReadOnly();
            _workItemsQueue = new PriorityQueue();
            Name = "WorkItemsGroup";

            // The _workItemsInStpQueue gets the number of currently executing work items,
            // because once a work item is executing, it cannot be cancelled.
            _workItemsInStpQueue = _workItemsExecutingInStp;

            _isSuspended = _workItemsGroupStartInfo.StartSuspended;
        }

        #endregion 

        #region WorkItemsGroupBase Overrides

        public override int Concurrency
        {
            get { return _concurrency; }
            set
            {
                Debug.Assert(value > 0);

                int diff = value - _concurrency;
                _concurrency = value;
                if (diff > 0)
                {
                    EnqueueToSTPNextNWorkItem(diff);
                }
            }
        }

        public override int WaitingCallbacks
        {
            get { return _workItemsQueue.Count; }
        }

        public override object[] GetStates()
        {
            lock (_lock)
            {
                object[] states = new object[_workItemsQueue.Count];
                int i = 0;
                foreach (WorkItem workItem in _workItemsQueue)
                {
                    states[i] = workItem.GetWorkItemResult().State;
                    ++i;
                }
                return states;
            }
        }

        /// <summary>
        /// WorkItemsGroup start information
        /// </summary>
        public override WIGStartInfo WIGStartInfo
        {
            get { return _workItemsGroupStartInfo; }
        }

        /// <summary>
        /// Start the Work Items Group if it was started suspended
        /// </summary>
        public override void Start()
        {
            // If the Work Items Group already started then quit
            if (!_isSuspended)
            {
                return;
            }
            _isSuspended = false;
            
            EnqueueToSTPNextNWorkItem(Math.Min(_workItemsQueue.Count, _concurrency));
        }

        public override void Cancel(bool abortExecution)
        {
            lock (_lock)
            {
                _canceledWorkItemsGroup.IsCanceled = true;
                _workItemsQueue.Clear();
                _workItemsInStpQueue = 0;
                _canceledWorkItemsGroup = new CanceledWorkItemsGroup();
            }

            if (abortExecution)
            {
                _stp.CancelAbortWorkItemsGroup(this);
            }
        }

        /// <summary>
        /// Wait for the thread pool to be idle
        /// </summary>
        public override bool WaitForIdle(int millisecondsTimeout)
        {
            SmartThreadPool.ValidateWorkItemsGroupWaitForIdle(this);
            return STPEventWaitHandle.WaitOne(_isIdleWaitHandle, millisecondsTimeout, false);
        }

        public override event WorkItemsGroupIdleHandler OnIdle
        {
            add { _onIdle += value; }
            remove { _onIdle -= value; }
        }

        #endregion 

        #region Private methods

        private void RegisterToWorkItemCompletion(IWorkItemResult wir)
        {
            IInternalWorkItemResult iwir = (IInternalWorkItemResult)wir;
            iwir.OnWorkItemStarted += OnWorkItemStartedCallback;
            iwir.OnWorkItemCompleted += OnWorkItemCompletedCallback;
        }

        public void OnSTPIsStarting()
        {
            if (_isSuspended)
            {
                return;
            }
            
            EnqueueToSTPNextNWorkItem(_concurrency);
        }

        public void EnqueueToSTPNextNWorkItem(int count)
        {
            for (int i = 0; i < count; ++i)
            {
                EnqueueToSTPNextWorkItem(null, false);
            }
        }

        private object FireOnIdle(object state)
        {
            FireOnIdleImpl(_onIdle);
            return null;
        }

        [MethodImpl(MethodImplOptions.NoInlining)]
        private void FireOnIdleImpl(WorkItemsGroupIdleHandler onIdle)
        {
            if(null == onIdle)
            {
                return;
            }

            Delegate[] delegates = onIdle.GetInvocationList();
            foreach(WorkItemsGroupIdleHandler eh in delegates)
            {
                try
                {
                    eh(this);
                }
                catch { }  // Suppress exceptions
            }
        }

        private void OnWorkItemStartedCallback(WorkItem workItem)
        {
            lock(_lock)
            {
                ++_workItemsExecutingInStp;
            }
        }

        private void OnWorkItemCompletedCallback(WorkItem workItem)
        {
            EnqueueToSTPNextWorkItem(null, true);
        }

        internal override void Enqueue(WorkItem workItem)
        {
            EnqueueToSTPNextWorkItem(workItem);
        }

        private void EnqueueToSTPNextWorkItem(WorkItem workItem)
        {
            EnqueueToSTPNextWorkItem(workItem, false);
        }

        private void EnqueueToSTPNextWorkItem(WorkItem workItem, bool decrementWorkItemsInStpQueue)
        {
            lock(_lock)
            {
                // Got here from OnWorkItemCompletedCallback()
                if (decrementWorkItemsInStpQueue)
                {
                    --_workItemsInStpQueue;

                    if(_workItemsInStpQueue < 0)
                    {
                        _workItemsInStpQueue = 0;
                    }

                    --_workItemsExecutingInStp;

                    if(_workItemsExecutingInStp < 0)
                    {
                        _workItemsExecutingInStp = 0;
                    }
                }

                // If the work item is not null then enqueue it
                if (null != workItem)
                {
                    workItem.CanceledWorkItemsGroup = _canceledWorkItemsGroup;

                    RegisterToWorkItemCompletion(workItem.GetWorkItemResult());
                    _workItemsQueue.Enqueue(workItem);
                    //_stp.IncrementWorkItemsCount();

                    if ((1 == _workItemsQueue.Count) && 
                        (0 == _workItemsInStpQueue))
                    {
                        _stp.RegisterWorkItemsGroup(this);
                        IsIdle = false;
                        _isIdleWaitHandle.Reset();
                    }
                }

                // If the work items queue of the group is empty than quit
                if (0 == _workItemsQueue.Count)
                {
                    if (0 == _workItemsInStpQueue)
                    {
                        _stp.UnregisterWorkItemsGroup(this);
                        IsIdle = true;
                        _isIdleWaitHandle.Set();
                        if (decrementWorkItemsInStpQueue && _onIdle != null && _onIdle.GetInvocationList().Length > 0)
                        {
                            _stp.QueueWorkItem(new WorkItemCallback(FireOnIdle));
                        }
                    }
                    return;
                }

                if (!_isSuspended)
                {
                    if (_workItemsInStpQueue < _concurrency)
                    {
                        WorkItem nextWorkItem = _workItemsQueue.Dequeue() as WorkItem;
                        try
                        {
                            _stp.Enqueue(nextWorkItem);
                        }
                        catch (ObjectDisposedException e)
                        {
                            e.GetHashCode();
                            // The STP has been shutdown
                        }

                        ++_workItemsInStpQueue;
                    }
                }
            }
        }

        #endregion
    }

    #endregion
}