Semaphore based concurrent work queue

Question

There's a need to have a mechanism in place that will process messages received from the network concurrently. However, only X number of messages can be allowed to be processed concurrently and there's a restriction: Similar messages must be processed sequentially. For simplicity's sake, similarity can be identified by the Tuple<T1, T2> or Integer TId generic parameter.

Consumption is straightforward: Throw any number of work items at SemaphoreWorkQueue<TId, TState> and it will do the processing. A single unit test is provided below for clarification.

Where can this be improved for performance and are there any pitfalls?

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace CodeReview.StackExchange.Com
{
    /// <summary>
    /// Executes actions concurrently limited by the semaphore count, but does not honor concurrency for actions with similar identifiers.
    /// </summary>
    public class SemaphoreWorkQueue<TId, TState> : IDisposable
    {
        private SemaphoreSlim _semaphore;
        private Task _workerTask;
        private ConcurrentDictionary<TId, TaskInfo> _runningTasks;
        private ConcurrentQueue<TaskInfo> _waitQueue;        
        private CancellationToken _cancellationToken;
        private volatile bool _disengaged;

        /// <summary>
        /// Initializes a new instance with an initial count of 10 concurrent actions and CancellationToken.None.
        /// </summary>
        public SemaphoreWorkQueue()
            : this(10, CancellationToken.None)
        {
        }

        /// <summary>
        /// Initializes a new instance with desired number of concurrent actions and CancellationToken.None.
        /// </summary>
        public SemaphoreWorkQueue(int concurrentTaskCount)
            : this(concurrentTaskCount, CancellationToken.None)
        {
        }

        /// <summary>
        /// Initializes a new instance with desired number of concurrent actions and cancellation token.
        /// </summary>
        public SemaphoreWorkQueue(int concurrentTaskCount, CancellationToken cancellationToken)
        {
            if (concurrentTaskCount < 1)
                throw new ArgumentException("Parameter concurrentTaskCount cannot be less then zero.", "concurrentTaskCount");

            _cancellationToken = cancellationToken;
            _semaphore = new SemaphoreSlim(concurrentTaskCount);
            _waitQueue = new ConcurrentQueue<TaskInfo>();
            _runningTasks = new ConcurrentDictionary<TId, TaskInfo>();            
            _workerTask = Task.Factory.StartNew(async () => await this.Engage(), cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
        }

        /// <summary>
        /// Queues an action with a specified identifier.
        /// </summary>
        public void EnqueueWork(TId id, Action<TState> action)
        {
            this.EnqueueWork(id, action, default(TState), _cancellationToken);
        }

        /// <summary>
        /// Queues an action with a specified identifier and state.
        /// </summary>
        public void EnqueueWork(TId id, Action<TState> action, TState state)
        {
            this.EnqueueWork(id, action, state, _cancellationToken);
        }

        /// <summary>
        /// Queues an action with a specified identifier and cancellation token.
        /// </summary>
        public void EnqueueWork(TId id, Action<TState> action, CancellationToken cancellationToken)
        {
            this.EnqueueWork(id, action, default(TState), cancellationToken);
        }

        /// <summary>
        /// Queues an action with a specified identifier, state and cancellation token.
        /// </summary>
        public void EnqueueWork(TId id, Action<TState> action, TState state, CancellationToken cancellationToken)
        {
            if (id == null)
                throw new ArgumentNullException("id", "Parameter id is required");

            if (action == null)
                throw new ArgumentNullException("action", "Parameter action is required");

            var task = new TaskInfo();
            task.Id = id;
            task.State = state;
            task.Action = action;
            task.CancellationToken = cancellationToken;

            _waitQueue.Enqueue(task);
        }

        private async Task<bool> Engage()
        {
            while (!_disengaged && !_cancellationToken.IsCancellationRequested)
            {
                TaskInfo waitingTask;
                if (_waitQueue.TryDequeue(out waitingTask))
                {
                    if (waitingTask.IsCancellationRequested)
                        break;

                    if (_runningTasks.ContainsKey(waitingTask.Id))
                    {
                        _waitQueue.Enqueue(waitingTask);
                    }
                    else
                    {
                        _runningTasks.TryAdd(waitingTask.Id, waitingTask);
                        await _semaphore.WaitAsync(); //decrease semaphore count

                        if (_disengaged)
                            break;

                        StartTask(waitingTask);
                    }
                }
                await Task.Delay(100);
            }
            return await Task.FromResult<bool>(_disengaged);
        }

        private void StartTask(TaskInfo taskInfo)
        {
            Task.Run(() => {
                try
                {
                    taskInfo.Action(taskInfo.State);
                }
                finally
                {
                    _runningTasks.TryRemove(taskInfo.Id, out taskInfo);
                    _semaphore.Release(); //increase semaphore count and remove task
                }
            });
        }

        private class TaskInfo
        {
            public TId Id;
            public TState State;
            public Action<TState> Action;
            public CancellationToken CancellationToken;

            public bool IsCancellationRequested
            {
                get { return CancellationToken.IsCancellationRequested; }
            }
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        private void Dispose(bool disposing)
        {
            if (disposing)
            {
                _disengaged = true;
                _semaphore.Dispose();
                _runningTasks.Clear();

                TaskInfo task;
                while (_waitQueue.TryDequeue(out task)) ;                
            }
        }
    }
}

And the unit test:

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;

namespace CodeReview.StackExchange.Com
{
    [TestClass]
    public class SemaphoreWorkQueueTests
    {
        [TestMethod]
        public void EnqueueWork_Can_Process_Tasks_Concurrently_No_Overlap()  
        {
            var actions = new List<Action<int>>();
            var rnd = new Random();
            var executedTasks = new ConcurrentBag<ActionInfo>();
            var maxTaskCount = 128;
            var concurrentTaskCount = 8;
            var delayMilliseconds = 1000;

            using (var semaphoreQueue = new SemaphoreWorkQueue<int, int>(concurrentTaskCount))
            {
                for (var i = 0; i < concurrentTaskCount; i++) //create X actions
                {
                    actions.Add(new Action<int>((int n) => {
                        var info = new ActionInfo();
                        info.Id = n;
                        info.StartTime = DateTime.Now.TimeOfDay;
                        Debug.WriteLine("I'm action #{0} @ {1} on Thread {2}", n, info.StartTime, Thread.CurrentThread.ManagedThreadId);                        
                        Task.Delay(delayMilliseconds).Wait();
                        info.EndTime = DateTime.Now.TimeOfDay;
                        executedTasks.Add(info);
                    }));
                }

                for (var i = 0; i < maxTaskCount; i++) //enqueue actions and start processing them randomly
                {
                    var next = rnd.Next(0, concurrentTaskCount);
                    semaphoreQueue.EnqueueWork(next, actions[next], next);
                }

                Task.Run(async () => { //wait for SemaphoreWorkQueue to process all tasks
                    while (executedTasks.Count < maxTaskCount)
                        await Task.Delay(delayMilliseconds);
                }).Wait(TimeSpan.FromMilliseconds(maxTaskCount * delayMilliseconds)); 
            }

            var groupById = executedTasks.GroupBy(x => x.Id); //group by similar identifiers and sort by StartTime

            foreach (var item in groupById)
            {
                if (item.Count() > 1)
                {
                    var group = item.ToList();
                    group.Sort((ActionInfo a, ActionInfo b) => { return a.StartTime.CompareTo(b.StartTime); });
                    for (var i = 1; i < group.Count; i++)
                    {
                        var first = group[i - 1];
                        var second = group[i];
                        /* sorted group will look like below
                         * 
                         * StartTime 11:00:00, EndTime 11:00:01 <--first
                         * StartTime 11:00:02, EndTime 11:00:03 <--second's StartTime should be greater than first's
                         * StartTime 11:00:04, EndTime 11:00:05
                         * StartTime 11:00:06, EndTime 11:00:07
                         */
                        if (second.StartTime < first.EndTime)
                            Assert.Fail("Task with same Id started while other task with same Id was already running.");
                    }
                }
            }
        }

        [DebuggerDisplay("{StartTime} {EndTime}")]
        private class ActionInfo
        {
            public int Id { get; set; }
            public TimeSpan StartTime { get; set; }
            public TimeSpan EndTime { get; set; }
        }
    }
}

Servy · Accepted Answer · 2016-11-02 20:01:47Z

There's one straight bug that I see in your implementation. _workerTask is not implemented properly. (It's also private and unused, so I'm not sure if you intended to just remove it entirely.) You have an async lambda in a call to StartNew. This is almost always wrong. You're scheduling a new thread pool thread that does nothing but start a new Task. The Task that StartNew returns will be completed when it finishes starting the inner task, not when the inner task finishes. You could use Task.Run, which will automatically unwrap the task for you, but honestly, you don't even need to do that. Just call Engage alone; it already produces a Task, and doesn't need to be run from another thread.

So the first change I'd make is to remove all notion of state for the work for a queue like this. There's just no reasons for such a queue to need to keep track of it. If someone wants to enqueue some work that requires state, they can trivially use an anonymous method to close over that state. That'll almost always be less work for them than using this queue's state variable anyway, so you're not really even providing a valuable service to consumers by providing it, and removing it will simplify the code in your class.

You also have the issue that your actual work loop is a busy loop. You check for work, and if there is none, you wait 100 milliseconds. Really you should set this up so that if new work comes in that is allowed to be performed it should be started immediately; this also avoids doing scheduling work constantly checking the queue when there is no work coming in. I'll get back to this later.

Next, there is no way for someone enqueuing some work to know when that work is done. They can know when all of the work is done, but never when any given piece is done. The EnqueueWork methods should probably all return a Task that represents the completion of that work item. (If you don't need this for your specific requirement at the moment, then that's fine, but I suspect if you end up planning to use a class like this much at all you'll end up wanting this feature.)

Along with this, you may want to allow your work items to support producing a value, rather than doing work that only ever causes side effects. If you design your multithreaded programs around individual pieces of work that produce a value, rather than as a collection of pieces of work that all mutate a common state, you dramatically simplify the code and remove huge swatches of very complex potential problems.

Next, consider what you should do if the work to be done isn't actually CPU bound work. In your specific case it may all be CPU bound work (although in your test case it isn't). You probably don't want your queue to accept a delegate and then call Task.Run on it. Instead your queue should probably accept a Func<Task>, so that the caller can provide any type of method that does work asynchronously, whether that be CPU bound work using Task.Run, IO work, or anything else. If you also want to provide overloads that don't accept a Task and that pass it to Task.Run, for convenience, then that's fine.

Rather than just having one class that is a queue that supports different categories, each of which need to run sequentially, but that can run with a fixed degree of parallelism with respect to each other, I'd break that up. Start out creating a queue that has no concept of categories, and simply runs the tasks sent to it with a fixed degree of parallelism. Once you have that, you can create a queue that does have the concept of categories and that leverages instances of your already existing queue as an implementation detail. This separates your different mechanisms into different classes, reducing the scope of each one, and making each one individually easier to work with.

So, that's a lot of words; let's get to some code. First off, a simple queue with N degrees of parallelism that supports the features mentioned above. You can actually leverage await to make this quite a lot simpler than your solution:

public sealed class TaskQueue : IDisposable
{
    private SemaphoreSlim semaphore;

    public TaskQueue() : this(degreesOfParallelism: 1)
    { }

    public TaskQueue(int degreesOfParallelism)
    {
        semaphore = new SemaphoreSlim(degreesOfParallelism);
    }

    public async Task<T> Enqueue<T>(Func<Task<T>> taskGenerator, CancellationToken token)
    {
        await semaphore.WaitAsync(token);
        try
        {
            return await taskGenerator();
        }
        finally
        {
            semaphore.Release();
        }
    }

    public async Task Enqueue(Func<Task> taskGenerator, CancellationToken token)
    {
        await semaphore.WaitAsync(token);
        try
        {
            await taskGenerator();
        }
        finally
        {
            semaphore.Release();
        }
    }

    public void Dispose()
    {
        semaphore.Dispose();
    }
}

Now for the category wrapper you can simply map each category to a SemaphoreSlim that you await to ensure that all operations within a category are serialized and then just use our existing TaskQueue to ensure that no more than N of these operations are running in parallel. Other than that it's mostly just various overloads to support different forms of input.

public sealed class CategorizedTaskQueue<TCategory> : IDisposable
{
    private ConcurrentDictionary<TCategory, SemaphoreSlim> categorySemaphores = new ConcurrentDictionary<TCategory, SemaphoreSlim>();
    private TaskQueue queue;

    public CategorizedTaskQueue() : this(degreesOfParallelism: 1)
    { }

    public CategorizedTaskQueue(int degreesOfParallelism)
    {
        queue = new TaskQueue(degreesOfParallelism);
    }

    public async Task<T> Enqueue<T>(TCategory category, Func<Task<T>> taskGenerator, CancellationToken token)
    {
        var myCategorySemaphore = categorySemaphores.GetOrAdd(category, _ => new SemaphoreSlim(1));
        await myCategorySemaphore.WaitAsync(token);
        try
        {
            return await queue.Enqueue(taskGenerator, token);
        }
        finally
        {
            myCategorySemaphore.Release();
        }
    }
    public Task<T> Enqueue<T>(TCategory category, Func<T> operation, CancellationToken token)
    {
        return Enqueue(category, () => Task.Run(operation, token), token);
    }
    public Task<T> Enqueue<T>(TCategory category, Func<T> operation)
    {
        return Enqueue(category, () => Task.Run(operation), CancellationToken.None);
    }

    public async Task Enqueue(TCategory category, Func<Task> taskGenerator, CancellationToken token)
    {
        var myCategorySemaphore = categorySemaphores.GetOrAdd(category, _ => new SemaphoreSlim(1));
        await myCategorySemaphore.WaitAsync(token);
        try
        {
            await queue.Enqueue(taskGenerator, token);
        }
        finally
        {
            myCategorySemaphore.Release();
        }
    }
    public Task Enqueue(TCategory category, Action operation, CancellationToken token)
    {
        return Enqueue(category, () => Task.Run(operation, token), token);
    }
    public Task Enqueue(TCategory category, Action operation)
    {
        return Enqueue(category, () => Task.Run(operation), CancellationToken.None);
    }

    public void Dispose()
    {
        queue.Dispose();
        foreach (var semaphore in categorySemaphores.Values)
            semaphore.Dispose();
    }
}

Nice review with a lot of knowledge. I need to read it several times before I can say I understand it. I think I will have to look at your answers on SO as well as I see you've quite a multithreading rep ;-) — t3chb0t
– t3chb0t, Commented Nov 3, 2016 at 8:33
Ohh, wow, thanks for great input. Will refactor and let you know. — Ostati
– Ostati, Commented Nov 3, 2016 at 18:34

Stack Exchange Network

Semaphore based concurrent work queue

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Semaphore based concurrent work queue

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions