Serially process ConcurrentQueue and limit to one message processor. Correct pattern?

Question

I'm building a multithreaded app in .net.

I have a thread that listens to a connection (abstract, serial, tcp...).

When it receives a new message, it adds it to via AddMessage. Which then call startSpool. startSpool checks to see if the spool is already running and if it is, returns, otherwise, starts it in a new thread. The reason for this is, the messages HAVE to be processed serially, FIFO.

So, my questions are... Am I going about this the right way? Are there better, faster, cheaper patterns out there?

My apologies if there is a typo in my code, I was having problems copying and pasting.

    ConcurrentQueue<IMyMessage > messages = new ConcurrentQueue<IMyMessage>();

    const int maxSpoolInstances = 1;

    object lcurrentSpoolInstances;
    int currentSpoolInstances = 0;

    Thread spoolThread;

    public void AddMessage(IMyMessage message)
    {
        this.messages.Add(message);

        this.startSpool();
    }

    private void startSpool()
    {
        bool run = false;

        lock (lcurrentSpoolInstances)
        {
            if (currentSpoolInstances <= maxSpoolInstances)
            {
                this.currentSpoolInstances++;
                run = true;
            }
            else
            {
                return;
            }
        }

        if (run)
        {
            this.spoolThread = new Thread(new ThreadStart(spool));
            this.spoolThread.Start();
        }
    }

    private void spool()
    {
        Message.ITimingMessage message;

        while (this.messages.Count > 0)
        {
            // TODO: Is this below line necessary or does the TryDequeue cover this?
            message = null;

            this.messages.TryDequeue(out message);

            if (message != null)
            {
                // My long running thing that does something with this message.
            }
        }


        lock (lcurrentSpoolInstances)
        {
            this.currentSpoolInstances--;
        }
    }

Answer 1

This would be easier using BlockingCollection<T> instead of ConcurrentQueue<T>.

Something like this should work:

class MessageProcessor : IDisposable
{
    BlockingCollection<IMyMessage> messages = new BlockingCollection<IMyMessage>();

    public MessageProcessor()
    {
       // Move this to constructor to prevent race condition in existing code (you could start multiple threads...
       Task.Factory.StartNew(this.spool, TaskCreationOptions.LongRunning);
    }

    public void AddMessage(IMyMessage message)
    {
        this.messages.Add(message);
    }

    private void Spool()
    {
         foreach(IMyMessage message in this.messages.GetConsumingEnumerable())
         {
               // long running thing that does something with this message.
         }
    }

    public void FinishProcessing()
    {
         // This will tell the spooling you're done adding, so it shuts down
         this.messages.CompleteAdding();
    }

    void IDisposable.Dispose()
    {
        this.FinishProcessing();
    }
}

---

Edit: If you wanted to support multiple consumers, you could handle that via a separate constructor. I'd refactor this to:

    public MessageProcessor(int numberOfConsumers = 1)
    {
        for (int i=0;i<numberOfConsumers;++i)
            StartConsumer();
    }

    private void StartConsumer()
    {
       // Move this to constructor to prevent race condition in existing code (you could start multiple threads...
       Task.Factory.StartNew(this.spool, TaskCreationOptions.LongRunning);
    }

This would allow you to start any number of consumers. Note that this breaks the rule of having it be strictly FIFO - the processing will potentially process "numberOfConsumer" elements in blocks with this change.

Multiple producers are already supported. The above is thread safe, so any number of threads can call Add(message) in parallel, with no changes.

Answer 2

I think that Reed's answer is the best way to go, but for the sake of academics, here is an example using the concurrent queue -- you had some races in the code that you posted (depending upon how you handle incrementing currnetSpoolInstances)

The changes I made (below) were:

• Switched to a Task instead of a Thread (uses thread pool instead of incurring the cost of creating a new thread)
• added the code to increment/decrement your spool instance count
• changed the "if currentSpoolInstances <= max ... to just < to avoid having one too many workers (probably just a typo)
• changed the way that empty queues were handled to avoid a race: I think you had a race, where your while loop could have tested false, (you thread begins to exit), but at that moment, a new item is added (so your spool thread is exiting, but your spool count > 0, so your queue stalls).

private ConcurrentQueue<IMyMessage> messages = new ConcurrentQueue<IMyMessage>();

const int maxSpoolInstances = 1;
object lcurrentSpoolInstances = new object();
int currentSpoolInstances = 0;

public void AddMessage(IMyMessage message)
{
    this.messages.Enqueue(message);
    this.startSpool();
}

private void startSpool()
{
    lock (lcurrentSpoolInstances)
    {
        if (currentSpoolInstances < maxSpoolInstances)
        {
            this.currentSpoolInstances++;
            Task.Factory.StartNew(spool, TaskCreationOptions.LongRunning);
        }
    }
}

private void spool()
{
    IMyMessage message;
    while (true)
    {
// you do not need to null message because it is an "out" parameter, had it been a "ref" parameter, you would want to null it.

        if(this.messages.TryDequeue(out message))
        {
            // My long running thing that does something with this message. 
        }
        else
        {
            lock (lcurrentSpoolInstances)
            {
                if (this.messages.IsEmpty)
                {
                    this.currentSpoolInstances--;
                    return;
                }
            }
        }
    }
}