Is the SemaphoreSlim usage correct in the below code context? Will it limit the max running threads?

Question

If I pass the maxConcurrency as 10 will the code below limit the max parallel tasks to 10? How do I verify the number of tasks running at a time?

public BlockingCollection<Task> _workTaskQueue;

public void DequeueTask(int maxConcurrency)
{
    var tasks = new List<Task>();
    using (SemaphoreSlim concurrencySemaphore = new SemaphoreSlim(maxConcurrency))
    {
        foreach (var task in _workTaskQueue.GetConsumingEnumerable())
        {
            concurrencySemaphore.Wait();
            if (!(task.IsCanceled) && task.Status == TaskStatus.Created)
            {
                task.ContinueWith((t) => { concurrencySemaphore.Release(); });
                tasks.Add(task);
                task.Start();
            }
        }
    }
    Task.WaitAll(tasks.ToArray());
}

See also https://stackoverflow.com/questions/14075029/have-a-set-of-tasks-with-only-x-running-at-a-time — L.B, Aug 24 '17 at 16:50
Possible duplicate of [Have a set of Tasks with only X running at a time](https://stackoverflow.com/questions/14075029/have-a-set-of-tasks-with-only-x-running-at-a-time) — Cheesebaron, Aug 24 '17 at 21:16

score 1 · Accepted Answer · answered Aug 24 '17 at 17:51

Since you changed your logic a little bit by utilizing my other answer, I prepared a test code for you. (No need for var tasks = new List<Task>();)

Random rnd = new Random();
int maxConcurrency = 5;
var _workTaskQueue = new System.Collections.Concurrent.BlockingCollection<Task>();
for (int i = 0; i < 250; i++) 
{
    //Tasks running 250=500ms
    _workTaskQueue.Add(new Task(()=> { Task.Delay(250 + rnd.Next(250)).Wait(); }));
} 
_workTaskQueue.CompleteAdding();
int runningTaks = 0;

using (SemaphoreSlim concurrencySemaphore = new SemaphoreSlim(maxConcurrency))
{
    foreach (var task in _workTaskQueue.GetConsumingEnumerable())
    {
        Console.WriteLine("LoopStart: " + runningTaks);

        await concurrencySemaphore.WaitAsync();

        Console.WriteLine("GotASem  : " + runningTaks);

        task.Start(); 
        Interlocked.Increment(ref runningTaks);
        task.ContinueWith(t =>
            {
                Interlocked.Decrement(ref runningTaks);
                concurrencySemaphore.Release();
            });

        if (runningTaks > maxConcurrency) throw new Exception("ERROR");
        Console.WriteLine("LoopEnd  : " + runningTaks + Environment.NewLine);
    }

    Console.WriteLine("Finalizing: " + runningTaks);
    //Make sure all all tasks have ended.
    for (int i = 0; i < maxConcurrency; i++)
    {
        await concurrencySemaphore.WaitAsync();
    }

    Console.WriteLine("Finished: " + runningTaks);
}

score 0 · Answer 2 · answered Aug 24 '17 at 16:54

0

No. A call to wait (to enter into mutual exclusion block) is missing.

answered Aug 24 '17 at 16:54

Sunil Singhal

593
3
11

Sunil, I have modified the code to include the wait, will it work? How do I verify its working? – Nitheesh Aug 24 '17 at 17:27
Create more threads than shared concurrency, > 10 and induce sleep in a task for around a second or more and console printing. 10 Print statements will be seen and post that they will be delayed – Sunil Singhal Aug 25 '17 at 03:26

Is the SemaphoreSlim usage correct in the below code context? Will it limit the max running threads?

2 Answers2