Moazzam’s Ramblings

Producer/consumer queues in C#

What is a producer/consumer queue

In a restaurant there are chefs in the kitchen who wait for orders from customers. As soon as a customer orders something, they (the chefs) make it and serve it to them (the customer). A producer/consumer queue is the same. There are workers who wait for a "job" to be entered into the queue and then process it when it is entered. The advantage of this method is that the workers consume negligible resources when they wait for a job to be added to the queue. So, "jobs" can be finished as soon as possible.

One of the advantages of this approach is that the class which adds the jobs to the queue (the customer) doesn't have to wait for the workers (the chefs) to finish what they are doing. It will keep inserting the jobs (orders) and the workers (chefs) can take their time processing it (cooking the order).

How do we accomplish this in C# you ask? Simple. The workers will check a job queue to see if anything has been inserted. If they find that there is a job waiting to be processed, then they will remove it from the queue and process it. This way, two workers won't end up working on the same job. If there are no jobs in the queue then the worker will go to sleep and wait for a signal to wake up when something is added to the queue. The "job manager" or the class that adds the jobs, will insert nulls if it is done adding jobs.

If that was hard to understand, let me explain it in another way. The restaurant will hire 20 chefs and 1 waiter. The customers will place their orders with the waiter. A chef will ask the waiter if he needs to cook anything for a customer. If the waiter says, "Yes, a customer ordered lamb curry" then the chef will start cooking lamb curry. If the orders of all customers have been cooked (or are being cooked by some other chef), then the chef will go to sleep and ask the waiter to wake him up if a customer orders something. However, the waiter will not wake up a specific chef. He will wake up all the chefs when an order comes in. The chef who wakes up first will start cooking and the rest of them will go back to sleep (and ask the waiter to wake them up if any other order comes in). At the end of the day, the waiter will give blank orders signaling to the chefs that they are done for the day.

Now that we know what a producer/consumer queue is, let's see how we can implement it in C#.

Frist we will take a look at the class which will "consume" the jobs - the resaurant containing the chefs and the waiter.

[csharp]
using System;
using System.Threading;
using System.Collections.Generic;

public class WorkerQueue : IDisposable
{
/* We will store our worker threads in this list. */
private List threads = new List();
/* The number of worker threads to use */
private int threadsToUse = 20;
/* synchronization lock */
private object locker = new object();
/* The queue in which jobs are stored */
private Queue jobs = new Queue();
/* wait handle - used to wake up sleeping threads and to make them wait (sleep) */
EventWaitHandle wh = new AutoResetEvent(false);

/* public variable for accessing threadsToUseProperty */
public int ThreadsToUse { set {threadsToUse=value;} get { return threadsToUse; } }

/* The constructor of this class */
public void WorkerQueue()
{
/* In the constructor, we will start worker threads, which will
* wait for a job to be entered into the queue.
*/
for (int i; i {
Thread t = new Thread(work);
threads.Add(t);
t.Start();
}
}

/* This is the method that other classes will use to add jobs to the queue
* All it does it, add the job to the queue and signal the worker threads
* that something has been added, so they can wake up and process it
* (if they are asleep). If the worker threads are already awake, then
* wh.Set() will have no effect (and won't error out)
*
* This is the waiter taking orders
*/
public void AddJob(string item)
{
lock (locker)
{
jobs.Enqueue(item);
}
wh.Set();
}

/* The method that will be used in worker threads. This will check our job queue
* to see if any jobs are available. If any job is available, then it will process that job
* and check for more jobs. If there are no jobs in the queue, then it will go to sleep
* (by using waitHandle.WaitOne method) until a new job is entered into the queue.
*
* In this method, we will end the worker queue by inserting nulls in the queue,
* so this worker thread will watch out for that and kill itself (by returning control)
* when it sees a null.
*
* This is our chef
*/
private void work()
{
while (true)
{
string item = null;
lock (locker)
{
/* chefs asking the waiter if anything needs to be cooked */
if (jobs.Count > 0)
{
item = jobs.Dequeue();
/* return if a null is found in the queue */
if (item == null) return;
}
}
if (item != null)
{
/* if a job was found then process it */
Console.WriteLine(item);
}
else
{
/* if a job was not found (meaning list is empty) then
* wait till something is added to it
*/
wh.WaitOne();
}
}
}

/* When the object of this class is about to be removed from memory
* we will signal all the worker threads to stop by inserting nulls in the queue
* and then we will wait for them to finish the jobs already in the queue
*
* Letting everyone know we are done for the day
*/
public void Dispose()
{
for (int i=0; i {
AddJob(null);
}
for (int i=0; i {
threads[i].Join();
}
}

}
[/csharp]

Now that we know what the restaurant looks like, let's take a look at the customer (the producer which will add the jobs)

[csharp]
using System;

public class Customer
{
public static void Main()
{
WorkerQueue q = new WorkerQueue();
q.AddJob("Lamb curry");
q.AddJob("Beef Steak");
}
}

[/csharp]

Share on Facebook