Thread-Safe Task Queue Implementation

Question

Purpose:

This class exists to ensure actions created from other threads are executed in a synchronous manner. Additionally, it provides functionality to add actions that run after a delay.

Code: (Note that the comments may be inconsistent and inaccurate in places.)

class ActionQueue
{
    // We don't lock on the actual objects, just in case (for whatever bad reason) we ever pass references or expose them publicly.
    private readonly object _actionsLock = new object();
    private readonly object _scheduledActionsLock = new object();

    private List<Action> actions;
    private List<ScheduledAction> scheduledActions;
    private ManualResetEvent hasActions;

    public ActionQueue()
    {
        this.actions = new List<Action>();
        this.scheduledActions = new List<ScheduledAction>();
        this.hasActions = new ManualResetEvent(true);
    }

    /// <summary>
    /// Adds an action to the queue.
    /// </summary>
    /// <param name="action">The action to be executed.</param>
    /// <param name="immediate">If true, this action will be pushed to the front of the queue.</param>
    public void Enqueue(System.Action action, bool immediate = false)
    {
        lock (this._actionsLock)
        {
            if (immediate)
            {
                this.actions.Insert(0, new Tasking.Action(action));
            }
            else
            {
                this.actions.Add(new Tasking.Action(action));
            }

            this.hasActions.Set(); // The actions queue was previously empty, so we must signal that it is no longer.
        }
    }

    /// <summary>
    /// Schedule an action to be added to the task queue after a specified delay.
    /// </summary>
    /// <param name="delay">The delay after which this action will be added to the queue.</param>
    /// <param name="action">The action to will be executed.</param>
    /// <param name="immediate">Optional. If true, this action will be added to the front of the queue instead of the back.</param>
    /// <returns></returns>
    public ScheduledAction Schedule(int delay, System.Action action, bool immediate = false)
    {
        ScheduledAction scheduledAction = new ScheduledAction(delay, action, immediate);

        lock (this._scheduledActionsLock)
        {
            if (this.scheduledActions.Count == 0)
            {
                this.scheduledActions.Add(scheduledAction);
            }
            else
            {
                foreach (ScheduledAction scheduled in this.scheduledActions)
                {
                    // 
                    if (scheduled.TimeUntilReady > scheduledAction.TimeUntilReady)
                    {
                        this.scheduledActions.Insert(this.scheduledActions.IndexOf(scheduled), scheduledAction);

                        break; // Break out of the loop. Also because we don't want shouted at for modifying the collection. Which would be kinda redundant anyway.
                    }
                }
            }

            this.hasActions.Set();
        }

        return scheduledAction;
    }

    /// <summary>
    /// Processes the queue and executes all ready actions sequentially.
    /// </summary>
    public void Process()
    {
        int waitFor = 0;
        bool timedWait = false;
        // There's potential for deadlock here if you're not careful. This should be the only point in which both of these objs are simultaneously locked.
        lock (this._actionsLock) lock (this._scheduledActionsLock)
            {
                if (this.actions.Count == 0 && this.scheduledActions.Count == 0)
                {
                    // We have no actions at all so we wait until we have some.
                    this.hasActions.Reset();
                }
                else if (this.actions.Count == 0 && this.scheduledActions.Count > 0)
                {
                    if(!this.scheduledActions[0].Ready)
                    {
                        timedWait = true;
                        waitFor = this.scheduledActions[0].TimeUntilReady;
                        this.hasActions.Reset();
                    }
                }
            }

        // Wait outside to prevent deadlock.
        if (timedWait) 
        {
            this.hasActions.WaitOne(waitFor);
        }
        else
        {
            this.hasActions.WaitOne();
        }

        lock (this._scheduledActionsLock)
        {
            // Iterate through scheduled actions and shift ready ones into the action queue.
            if (this.scheduledActions.Count > 0)
            {
                if (this.scheduledActions[0].Ready)
                {
                    // Go through ready ones.
                    foreach (ScheduledAction action in this.scheduledActions.ToList())
                    {
                        if (action.Ready)
                        {
                            lock (this._actionsLock) // Hmm. Could be moved to somewhere more sensible?
                            {
                                if (action.IsImmedate)
                                {
                                    this.actions.Insert(0, action);
                                }
                                else
                                {
                                    this.actions.Add(action);
                                }

                                this.scheduledActions.Remove(action);
                            }
                        }
                    }
                }
            }
        }

        Action toExecute = null;

        lock (this._actionsLock)
        {
            if (this.actions.Count > 0)
            {
                toExecute = this.actions[0];
                this.actions.RemoveAt(0);
            }
            else
            {
                lock (this._scheduledActionsLock) // I do not like this extra lock...
                {
                    if (this.scheduledActions.Count == 0) // Update: Sync issue resolved. See above lock. -_-
                    {
                        this.hasActions.Reset(); // Reset so we block on next iteration. We have nothing to do.
                    }
                }
            }
        }

        if (toExecute != null) // Run this outside of the lock to prevent other threads blocking.
        {
            toExecute.Run();
        }
    }

    /// <summary>
    /// Starts a loop that continually processes tasks. Testing only!
    /// </summary>
    public void Run()
    {
        while (true)
        {
            this.Process();
        }
    }
}

Things I'd like advice on changing:

• Naming of classes - Within my project, this code lies in the namespace Tasking. My Tasking.Action conflicts with System.Action. While this isn't really a problem problem, it doesn't feel sane.

Potential concerns:

• Deadlocks - I'm fairly new to concurrent programming, so I'm probably not aware of any obvious pitfalls that will trip me up.
• Performance - This code will be used in a real-time game server. I've already taken measures to ensure that there are no unnecessary delays within lock blocks (therefore preventing the blocking of threads that would be adding actions to the queue), but am unsure if anything else could be a potential problem.

Answer 1

The .NET framework provides you with tons of concurrent solutions that help you reduce the amount of locks and this way the chance of deadlocks. I've revised your code and found the already present implementations in the framework suitable for your needs.

I sincerely hope you don't mind that I have reworked your solution to be lock-free (and also solved your issue of naming the classes). You can find the reworked class below my explanation.

First of all, since we are working in a multi-threaded environment, we need an intermediate solution to synchronize the workflow of the multiple threads. There can be multiple worker threads, that pass the actions to the one processor thread.

Our intermediate layer consists of two ConcurrentQueue<> and a ConcurrentBag<>. The reason for the two ConcurrentQueue<> variables is that I have separated the primary actions from the other ones. Instead of inserting the action to the front of the queue, I insert it to a dedicated queue. When the dedicated priority queue has an action waiting to be processed, it is going to interrupt the progression of the other queue (can be also interrupted in the middle of the cycle without any problems). The ConcurrentBag<> is responsible for the scheduled actions. Their order is not necessary to be kept, since they won't processed until the timeout expires. We check in each cycle and update the time left. When it has timed out, we pass it to the appropriate queue.

(code untested)

public class ScheduledSynchronizedAction
{
    public Action Action { get; set; }
    public int DelayMilliseconds { get; set; }
    public bool IsPriority { get; set; }
}

public class SynchronizedActionQueue
{
    private ConcurrentQueue<Action> SynchronizedActions { get; set; }
    private ConcurrentQueue<Action> PrioritySynchronizedActions { get; set; }
    private ConcurrentBag<ScheduledSynchronizedAction> ScheduledSynchronizedActions { get; set; }

    public SynchronizedActionQueue()
    {
        SynchronizedActions = new ConcurrentQueue<Action>();
        PrioritySynchronizedActions = new ConcurrentQueue<Action>();
        ScheduledSynchronizedActions = new ConcurrentBag<ScheduledSynchronizedAction>();
    }

    public void Enqueue(Action Action, bool IsPriority = false)
    {
        if (IsPriority)
            PrioritySynchronizedActions.Enqueue(Action);
        else
            SynchronizedActions.Enqueue(Action);
    }

    public void Schedule(Action Action, int DelayMilliseconds, bool IsPriority = false) => ScheduledSynchronizedActions.Add(new ScheduledSynchronizedAction { Action = Action, DelayMilliseconds = DelayMilliseconds, IsPriority = IsPriority });

    public void Update(int Diff)
    {
        Action DummyAction;
        ScheduledSynchronizedAction DummyScheduledAction;
        while (!SynchronizedActions.IsEmpty || !PrioritySynchronizedActions.IsEmpty)
        {
            while (!PrioritySynchronizedActions.IsEmpty)
                if (PrioritySynchronizedActions.TryDequeue(out DummyAction))
                    DummyAction.Invoke();

            if (SynchronizedActions.TryDequeue(out DummyAction))
                DummyAction.Invoke();
        }        

        Queue<ScheduledSynchronizedAction> TempScheduledSynchronizedActions = new Queue<ScheduledSynchronizedAction>();
        while (!ScheduledSynchronizedActions.IsEmpty)
        {
            if (ScheduledSynchronizedActions.TryTake(out DummyScheduledAction))
            {
                DummyScheduledAction.DelayMilliseconds -= Diff;

                if (DummyScheduledAction.DelayMilliseconds <= 0)
                    Enqueue(DummyScheduledAction.Action, DummyScheduledAction.IsPriority);
                else
                    TempScheduledSynchronizedActions.Enqueue(DummyScheduledAction);
            }
        }

        while (TempScheduledSynchronizedActions.Count > 0)
            ScheduledSynchronizedActions.Add(TempScheduledSynchronizedActions.Dequeue());
    }

    public void Run()
    {
        while (true)
        {
            Update(50);
            Thread.Sleep(50); // todo: implement this properly
        }
    }
}