5
\$\begingroup\$

I am learning C# and in the process I am trying to implement a multi producer consumer thread safe class. Can you please code review and point me to any mistakes I have made and improvements I can make? I understand there would be libraries and other thread safe queue objects which would provide me the functionality which I am trying to implement, but the very reason I am doing this myself is to learn how threads and thread safe queues are coded and managed. My idea is to achieve maybe an under-optimized but correctly working code.

//simple thread safe many producer many consumer class
class ProducerConsumer<Task>
{
    public delegate void ConsumerCallback(Task t);

    protected bool keepRunning = false;
    protected Queue<Task> taskQueue = null;
    protected Thread[] consumerThreads = null;
    protected ConsumerCallback clientConsumerMethodCB = null;

    public ProducerConsumer(ConsumerCallback clientConsumerMethod, int consumerThreadsCount)
    {
        keepRunning = true;
        consumerThreads = new Thread[consumerThreadsCount];
        for (int i = 0; i < consumerThreadsCount; i++)
        {
            consumerThreads[i] = new Thread(ConsumerThreadFunction);
        }

        for (int i = 0; i < consumerThreadsCount; i++)
        {
            consumerThreads[i].Start();
        }

        taskQueue = new Queue<Task>();
        clientConsumerMethodCB = clientConsumerMethod;
    }

    public void Produce(Task t)
    {
        lock (taskQueue)
        {
            taskQueue.Enqueue(t);
        }

        Monitor.PulseAll(taskQueue);
    }

    protected void ConsumerThreadFunction()
    {
        Queue<Task> holdTaskQueueItem = new Queue<Task>();
        while (keepRunning)
        {
            while (taskQueue.Count <= 0)
            {
                Monitor.Wait(taskQueue);
            }

            lock (taskQueue)
            {
                if(taskQueue.Count > 0)
                {
                    holdTaskQueueItem.Enqueue(taskQueue.Dequeue());
                }
            }

            if(holdTaskQueueItem.Count > 0)
            {
                clientConsumerMethodCB?.Invoke(holdTaskQueueItem.Dequeue());
            }
        }
    }
}
\$\endgroup\$
8
  • \$\begingroup\$ Why not BlockingCollection? What is ConsumerCallback, why not Action<T>? Also consider TAP - Asynchronous Programming. Is there some usage example? \$\endgroup\$
    – aepot
    Oct 7, 2020 at 10:04
  • \$\begingroup\$ What is Task? Consider naming conflict with very popular System.Threading.Tasks.Task. \$\endgroup\$
    – aepot
    Oct 7, 2020 at 10:10
  • \$\begingroup\$ And finally, the cutting-edge-technology - System.Threading.Channels. \$\endgroup\$
    – aepot
    Oct 7, 2020 at 10:25
  • \$\begingroup\$ @aepot - thanks for your code review comments. I did not use any of the existing libraries or constructs for 2 reasons. I wanted a very simple solution and since I am new to C# wanted to implement simple producer consumer for understanding how such code is written and managed. I mentioned the same reason in my comment above the code. I will change the name Task to something else. \$\endgroup\$
    – anands
    Oct 7, 2020 at 17:14
  • \$\begingroup\$ Anyway, consider BlockingCollection, it's the easiest way to achieve the same thing but with less lines of code. I can make some example if you need. But can you fix the Task first e.g. with Job? You may add the fixed code to the question. \$\endgroup\$
    – aepot
    Oct 7, 2020 at 17:18

2 Answers 2

3
\$\begingroup\$

You have entered in the realm of concurrency. So there will be one or more functions which are competing to be executed. In other words there is no guarantee that your code will run in parallel.

If there is enough executor (CPU cores) then the runtime might schedule several threads to different cores. If there is not sufficient resource to be utilized then scheduler might use context switching.

Context switching will execute one portion of T1 and then it suspends T1 in order to avoid thread starvation (a single thread consumes all the available resources, while other threads remain idle). One portion of T2 is executed then it will be suspended. Now scheduler goes back to T1 to resume (hopefully finish) its current function.

These concepts are crucial to understand because these can cause strange behaviours for those that are used to program in a single threaded environment. Let me share with you a simple example:

Example

  1. Let's suppose there is one item in the queue
  2. Also let's assume that there are two consumers (T1 and T2) and they are both checking queue's length before they try to dequeue
  3. So T1 and T2 are at the same line of where they are checking the queue's length
  4. Scheduler suspends T1 after it successful checked the queue length
  5. T2 dequeues the item from the queue
  6. Scheduler resumes T1, which is in that belief that the queue has an item in it (which is not true, because T2 has already dequeued that)
  7. T1 tries to dequeue item from the queue but it will fail because it could not get data from an empty queue

The problem here is that the check and the dequeue functionalities are not treated as atomic. If they were treated as atomic then T1 could not be suspended (by the scheduler or by the GC) between the two queue method calls.


Back to your code, there are several ways to improve it. (I'll respect your intent to learn so I will not suggest alternatives that are already done this)

  1. Please do not use Task for the generic parameter name. It is a well-known structure in the .NET world. Please prefer either Job or Computation or what so ever.
  2. As I can see you would like to use the keepRunning for signaling (either for cancellation or for completion). Unfortunately it is declared as false then set to true in the initialization phase and then it is never modified.
  3. For locking the best practice is to introduce a private static readonly object lockObject = new object(); to avoid accidental exposure of your lock object. In your case the queue is protected, which means in a derived class it could be exposed publicly and a consumer of that derived class can add an extra lock for that. In that case the lock is out of your control.
  4. Please prefer TryDequeue over Count check and Dequeue.
  5. Please consider to use ConcurrentQueue or Synchronized Queue to avoid explicit locking.
    5.1) As always measure implementations with different workloads, ConcurrentQueue vs Queue performance
  6. According to my understanding holdTaskQueueItem will hold either one or zero item in it. A single Task or (Job/Computation) variable would be sufficient
  7. Whenever you allow to your consumer to pass any function (in your case ConcumerCallback) then be very-very careful. What if it runs for minutes? What if it deadlocks because it also contains a lock inside it?
\$\endgroup\$
3
  • \$\begingroup\$ Thanks for your code review comments. I have implemented some of them and have made changes to my code as well. Please check. \$\endgroup\$
    – anands
    Oct 10, 2020 at 20:56
  • 1
    \$\begingroup\$ I believe the problem you described in your sample pseudo code will not happen in code here, since only 1 thread will have access to taskQueue object in one time. In case the thread which owns the lock is suspended, other thread will not be able to access taskQueue as the lock would still be held by first thread. \$\endgroup\$
    – anands
    Oct 10, 2020 at 21:06
  • \$\begingroup\$ @anands Yes, you are right your code is not exposed to this problem because of explicit locking. I described that because I was not sure (based on the description) that it was a conscious decision or not. Most of the structures that are built to tackle SPMC problem try to use some sort of lock-free mechanism like CAS (compare-and-swap). In .NET world the Interlocked.CompareExchange is an example to provide low-level API for CAS. \$\endgroup\$ Oct 12, 2020 at 7:00
3
\$\begingroup\$

Not a review but my version of the same as usage example of Task, BlockingCollection and IDisposable implementation. It's thread-safe, thus you may create it and add jobs from multiple threads.

public class Consumer<T> : IDisposable
{
    private readonly int _maxConcurrency;
    private readonly Action<T> _action;
    private readonly BlockingCollection<T> _jobs;
    private readonly Task _task;

    public Consumer(int maxConcurrency, Action<T> action)
    {
        _maxConcurrency = maxConcurrency;
        _action = action;
        _jobs = new BlockingCollection<T>();
        _task = Task.Run(RunLoop);
    }

    public void Enqueue(T job) // BlockingCollection is FIFO collection
    {
        if (disposed) 
            throw new ObjectDisposedException("Can't add new job. Consumer was disposed.");
        _jobs.Add(job); 
    }

    private void RunLoop()
    {
        using (SemaphoreSlim semaphore = new SemaphoreSlim(_maxConcurrency))
        {
            List<Task> activeTasks = new List<Task>();
            foreach (T job in _jobs.GetConsumingEnumerable())
            {
                semaphore.Wait();
                lock (activeTasks)
                {
                    Task t = null;
                    activeTasks.Add(t = Task.Run(() => 
                    {
                        _action(job);
                        lock (activeTasks)
                        {
                            activeTasks.Remove(t);
                        }
                        semaphore.Release();
                    }));
                }
            }
            Task[] tasks;
            lock (activeTasks)
            {
                tasks = activeTasks.ToArray();
            }
            Task.WaitAll(tasks);
        }
    }

    bool disposed;

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

    protected virtual void Dispose(bool disposing)
    {
        if (!disposed)
        {
            _jobs.CompleteAdding();
            if (disposing) 
            {
                _task.GetAwaiter().GetResult();
            }
            disposed = true;
        }
    }

    ~Consumer() => Dispose(false);
}

Usage

public class Program
{
    static void Main()
    {
        using (Consumer<int> consumer = new Consumer<int>(2, x => { Thread.Sleep(100); Console.WriteLine(x + " End"); }))
        {
            for (int i = 0; i < 10; i++) // <= Producer
            {
                Console.WriteLine(i + " Start");
                consumer.Enqueue(i);
                Thread.Sleep(50);
            }
        } // <= here it will synchronously wait until all jobs is done.
        Console.WriteLine("Done.");
    }
}

Output

0 Start
1 Start
0 End
2 Start
1 End
3 Start
2 End
4 Start
3 End
5 Start
4 End
6 Start
5 End
7 Start
6 End
8 Start
7 End
9 Start
8 End
9 End
Done.

Why there's nothing about producers inside the class?

The application may run multiple threads with code that can add jobs to consume. So it supports any amount of producer threads. On other hand you may implement a producer class that will use the Consumer but here it's not a mandatory.

\$\endgroup\$
4
  • \$\begingroup\$ why the semaphore? \$\endgroup\$
    – esskar
    Nov 10, 2021 at 9:15
  • 1
    \$\begingroup\$ @esskar that was a year ago. Now I can do a better implementation. But why not? \$\endgroup\$
    – aepot
    Nov 10, 2021 at 9:28
  • \$\begingroup\$ The semaphore is not needed imho, as the RunLoop Method is just called once. \$\endgroup\$
    – esskar
    Nov 10, 2021 at 9:35
  • 1
    \$\begingroup\$ @esskar semaphore is for inner foreach loop. \$\endgroup\$
    – aepot
    Nov 10, 2021 at 9:36

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.