Lock-free multiple producer single consumer message queue

Question

For a while now I've been after a lock-free, simple and scalable implementation of a multiple producer, single consumer queue for delegates in C#. I think I finally have it. I've run basic tests on it showing it works, and the design is so simple that I've managed to convince myself it is rock-solid.

This relies on a compare-and-swap approach to update the queue, similar to the new lock-free pattern used to generate event field accessors in C# 4.0 (see here), combined with an Interlocked.Exchange to read-and-set the queue to null.

Essentially, this derived from the realization that message-queues are really one-shot multicast delegates that reset their invocation lists after message execution!

However, parallel code is very hard to get right so I would like confirmation that this pattern is indeed correct. I've found my intuitions can be surprisingly misleading when it comes to parallelism, and there's always some crazy edge-case driving me off...

So, to the question: Can anyone confirm to me that the below message queue design pattern is thread-safe?

public class MessageQueue
{
    Action queue;

    public void Enqueue(Action message)
    {
        Action currentQueue;
        var previousQueue = queue;
        do
        {
            currentQueue = previousQueue;
            var newQueue = currentQueue + message;
            previousQueue = Interlocked.CompareExchange(ref queue, newQueue, currentQueue);
        }
        while (previousQueue != currentQueue);
    }

    public void Process()
    {
        var current = Interlocked.Exchange(ref queue, null);
        if (current != null)
        {
            current();
        }
    }
}

Answer 1

As far as thread safety goes, your code is fine. But there's a couple of things I want to point out:

The meaning of your variables (previousQueue, newQueue and currentQueue) isn't very clear, at least not to me. And when writing multi-threaded code, readability becomes extremely important.

Also, for CAS-loops, I always find a "while(true) - break" loop a lot easier on the eyes, but that's my personal opinion.

Here's my suggestion for improving readability:

while(true)
{
    var expectedOldQueue = queue;
    var newQueue = expectedOldQueue + message;
    var actualOldQueue = Interlocked.CompareExchange(ref queue, newQueue, expectedOldQueue);

    if(expectedOldQueue == actualOldQueue)
        break;
}

---

Also, I'm a bit concerned about the design you chose to achieve a multiple producer/single consumer queue, or maybe I'm just not understanding something...

If I understood correctly, producers will enqueue actions, instead of items that need to be consumed, and producers will simply call Process to trigger those actions, correct?

So, instead of this:

//producer
queue.EnqueueItem(item);

//consumer
var item = queue.Dequeue();
Console.WriteLine(item);

You're proposing this:

//producer
queue.Enqueue(() => Console.WriteLine(item));

//consumer
queue.Process();

If so, this worries me because it goes against the main vein of a consumer/producer architecture, where the consumers are detached from the producers, and have no idea how items will be consumed.

With your proposal, producers will be in charge of producing work items and define how they are going to be consumed.

Answer 2

I think that combining delegates is a weird way to achieve this. The disadvantage of this approach is that it makes proper exception handling difficult. With your code, whenever an enqueued action throws an exception, Process() will throw that exception and all the actions that are after it in the same delegate won't be executed.

If this is not what you want, changing it while still using combing delegates will be awkward (you would have to use GetInvocationList().

I think a cleaner solution would be to use a thread-safe collection, like ConcurrentQueue, or even ActionBlock from TPL Dataflow, that will also handle scheduling consuming of the items.

If you want to keep using CompareExchange(), you could also use ImmutableQueue.

---

How is Process() going to be called? I think that with this design, it should be the work of the message queue to schedule executing the delegates, not of the user of your class.

One way to do that would be to use the ActionBlock mentioned above, but there are also other options (like using a BlockingCollection for storing of the delegates and then have a processing Task, that blocks on the collection when it's empty).