C# Threading and Queues

Question

This isn't about the different methods I could or should be using to utilize the queues in the best manner, rather something I have seen happening that makes no sense to me.

void Runner() {
    // member variable
    queue = Queue.Synchronized(new Queue());
    while (true) {
        if (0 < queue.Count) {
            queue.Dequeue();
        }
    }
}

This is run in a single thread:

var t = new Thread(Runner);
t.IsBackground = true;
t.Start();

Other events are "Enqueue"ing else where. What I've seen happen is over a period of time, the Dequeue will actually throw InvalidOperationException, queue empty. This should be impossible seeing as how the count guarantees there is something there, and I'm positive that nothing else is "Dequeue"ing.

The question(s):

1. Is it possible that the Enqueue actually increases the count before the item is fully on the queue (whatever that means...)?
2. Is it possible that the thread is somehow restarting (expiring, reseting...) at the Dequeue statement, but immediately after it already removed an item?

Edit (clarification):

These code pieces are part of a Wrapper class that implements the background helper thread. The Dequeue here is the only Dequeue, and all Enqueue/Dequeue are on the Synchronized member variable (queue).

Answer 1

Using Reflector, you can see that no, the count does not get increased until after the item is added.

As Ben points out, it does seem as you do have multiple people calling dequeue.

You say you are positive that nothing else is calling dequeue. Is that because you only have the one thread calling dequeue? Is dequeue called anywhere else at all?

EDIT:

I wrote a little sample code, but could not get the problem to reproduce. It just kept running and running without any exceptions.

How long was it running before you got errors? Maybe you can share a bit more of the code.

class Program
{
    static Queue q = Queue.Synchronized(new Queue());
    static bool running = true;

    static void Main()
    {
        Thread producer1 = new Thread(() =>
            {
                while (running)
                {
                    q.Enqueue(Guid.NewGuid());
                    Thread.Sleep(100);
                }
            });

        Thread producer2 = new Thread(() =>
        {
            while (running)
            {
                q.Enqueue(Guid.NewGuid());
                Thread.Sleep(25);
            }
        });

        Thread consumer = new Thread(() =>
            {
                while (running)
                {
                    if (q.Count > 0)
                    {
                        Guid g = (Guid)q.Dequeue();
                        Console.Write(g.ToString() + " ");
                    }
                    else
                    {
                        Console.Write(" . ");
                    }
                    Thread.Sleep(1);
                }
            });
        consumer.IsBackground = true;

        consumer.Start();
        producer1.Start();
        producer2.Start();

        Console.ReadLine();

        running = false;
    }
}

Answer 2

Here is what I think the problematic sequence is:

1. (0 < queue.Count) evaluates to true, the queue is not empty.
2. This thread gets preempted and another thread runs.
3. The other thread removes an item from the queue, emptying it.
4. This thread resumes execution, but is now within the if block, and attempts to dequeue an empty list.

However, you say nothing else is dequeuing...

Try outputting the count inside the if block. If you see the count jump numbers downwards, someone else is dequeuing.

Answer 3

Here's a possible answer from the MSDN page on this topic:

Enumerating through a collection is intrinsically not a thread-safe procedure. Even when a collection is synchronized, other threads can still modify the collection, which causes the enumerator to throw an exception. To guarantee thread safety during enumeration, you can either lock the collection during the entire enumeration or catch the exceptions resulting from changes made by other threads.

My guess is that you're correct - at some point, there's a race condition happening, and you end up dequeuing something that isn't there.

A Mutex or Monitor.Lock is probably appropriate here.

Good luck!

Answer 4

Are the other areas that are "Enqueuing" data also using the same synchronized queue object? In order for the Queue.Synchronized to be thread-safe, all Enqueue and Dequeue operations must use the same synchronized queue object.

From MSDN:

To guarantee the thread safety of the Queue, all operations must be done through this wrapper only.

Edited: If you are looping over many items that involve heavy computation or if you are using a long-term thread loop (communications, etc.), you should consider having a wait function such as System.Threading.Thread.Sleep, System.Threading.WaitHandle.WaitOne, System.Threading.WaitHandle.WaitAll, or System.Threading.WaitHandle.WaitAny in the loop, otherwise it might kill system performance.

Answer 5

question 1: If you're using a synchronized queue, then: no, you're safe! But you'll need to use the synchronized instance on both sides, the supplier and the feeder.

question 2: Terminating your worker thread when there is no work to do, is a simple job. However, you either way need a monitoring thread or have the queue start a background worker thread whenever the queue has something to do. The last one sounds more like the ActiveObject Pattern, than a simple queue (which's Single-Responsibily-Pattern says that it should only do queueing).

In addition, I'd go for a blocking queue instead of your code above. The way your code works requires CPU processing power even if there is no work to do. A blocking queue lets your worker thread sleep whenever there is nothing to do. You can have multiple sleeping threads running without using CPU processing power.

C# doesn't come with a blocking queue implementation, but there a many out there. See this example and this one.

Answer 6

Another option for making thread-safe use of queues is the ConcurrentQueue<T> class that has been introduced since 2009 (the year of this question). This may help avoid having to write your own synchronization code or at least help making it much simpler.

From .NET Framework 4.6 onward, ConcurrentQueue<T> also implements the interface IReadOnlyCollection<T>.