5
\$\begingroup\$

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();
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ One thing I am unclear on: is it necessary to use a volatile read when accessing the queue? I.e. previousQueue = Volatile.Read(queue)? The interaction between Interlocked and Volatile has always left me a bit confused, ever since (in c# 2.0) passing a volatile int to an Interlocked method would generate a compiler warning. \$\endgroup\$ – dbc Aug 13 '14 at 21:33
  • \$\begingroup\$ It is apparently not needed, according to this SO question. More specifically, it is mentioned by Anton: " you need volatile on your state variable only if you are going to write to it directly (...), reads of a variable updated with an interlocked (or volatile) operation will use the most recent value." \$\endgroup\$ – glopes Aug 13 '14 at 22:57
  • \$\begingroup\$ @But this answer suggests otherwise: stackoverflow.com/questions/6139699/…. Thinking further, if you get an out-of-date value from the cache the previousQueue != currentQueue check should catch the problem. \$\endgroup\$ – dbc Aug 13 '14 at 23:06
3
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ Maybe consumer/producer queue is not the best term to use for this case then... what I was after is to centralize execution of actions generated by multiple threads on a single thread. Something like windows forms message pumps, or the BeginInvoke pattern, but not tied in to the windows system. \$\endgroup\$ – glopes Aug 13 '14 at 12:37
  • 1
    \$\begingroup\$ @glopes Oh, I see. Well, you could use TPL Dataflow's ActionBlock, which does exactly what you need. You just post messages to a block, and you can configure the block to be consumed by a single thread or in parallel. \$\endgroup\$ – dcastro Aug 13 '14 at 12:45
3
\$\begingroup\$

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).

\$\endgroup\$
  • \$\begingroup\$ You make a good point about exception handling and thanks for the tip on using ImmutableQueue. I liked this approach in part due to the lack of dependencies. GetInvocationList() seems like a good compromise, though. Multicast delegate invocation lists basically implement immutable queues for free, no? Not all operations, I know, but enough for this case. \$\endgroup\$ – glopes Aug 13 '14 at 18:54
  • 1
    \$\begingroup\$ @glopes Yeah, they do. But I think that if you want a queue, it's clearer to use actual queue, not something that is meant for different purpose, but that also acts like a queue. \$\endgroup\$ – svick Aug 13 '14 at 19:00
  • \$\begingroup\$ Agreed, although for me needing no extra dependencies is still a plus. Also I think in general this usage is not so different from event callbacks. I can also imagine wanting to deal with exceptions from callbacks in the case of regular events (probably that's why GetInvocationList is there in the first place). But we can disagree on this one ;) thanks, anyway, great points! \$\endgroup\$ – glopes Aug 13 '14 at 19:07
  • \$\begingroup\$ On second thought, I agree that a proper immutable list implementation would be cleaner. Then you won't have to pay the unnecessary allocations of GetInvocationList() every time you process the queue. \$\endgroup\$ – glopes Aug 13 '14 at 19:43

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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