Threadsafe FIFO Queue/Buffer

Question

I need to implement a sort of task buffer. Basic requirements are:

• Process tasks in a single background thread
• Receive tasks from multiple threads
• Process ALL received tasks i.e. make sure buffer is drained of buffered tasks after a stop signal is received
• Order of tasks received per thread must be maintained

I was thinking of implementing it using a Queue like below. Would appreciate feedback on the implementation. Are there any other brighter ideas to implement such a thing?

public class TestBuffer
{
    private readonly object queueLock = new object();
    private Queue<Task> queue = new Queue<Task>();
    private bool running = false;

    public TestBuffer()
    {
    }

    public void start()
    {
        Thread t = new Thread(new ThreadStart(run));
        t.Start();
    }

    private void run()
    {
        running = true;

        bool run = true;
        while(run)
        {
            Task task = null;
            // Lock queue before doing anything
            lock (queueLock)
            {
                // If the queue is currently empty and it is still running
                // we need to wait until we're told something changed
                if (queue.Count == 0 && running)
                {
                    Monitor.Wait(queueLock);
                }

                // Check there is something in the queue
                // Note - there might not be anything in the queue if we were waiting for something to change and the queue was stopped
                if (queue.Count > 0)
                {
                    task = queue.Dequeue();
                }
            }

            // If something was dequeued, handle it
            if (task != null)
            {
                handle(task);
            }

            // Lock the queue again and check whether we need to run again
            // Note - Make sure we drain the queue even if we are told to stop before it is emtpy
            lock (queueLock)
            {
                run = queue.Count > 0 || running;
            }
        }
    }

    public void enqueue(Task toEnqueue)
    {
        lock (queueLock)
        {
            queue.Enqueue(toEnqueue);
            Monitor.PulseAll(queueLock);
        }
    }

    public void stop()
    {
        lock (queueLock)
        {
            running = false;
            Monitor.PulseAll(queueLock);
        }
    }

    public void handle(Task dequeued)
    {
        dequeued.execute();
    }
}

Solution 1

You can actually handle this with the out-of-the-box BlockingCollection.

It is designed to have 1 or more producers, and 1 or more consumers. In your case, you would have multiple producers and one consumer.

When you receive a stop signal, have that signal handler

• Signal producer threads to stop
• Call CompleteAdding on the BlockingCollection instance

The consumer thread will continue to run until all queued items are removed and processed, then it will encounter the condition that the BlockingCollection is complete. When the thread encounters that condition, it just exits.

Solution 2

You should think about ConcurrentQueue, which is FIFO, in fact. If not suitable, try some of its relatives in Thread-Safe Collections. By using these you can avoid some risks.

Solution 3

I suggest you take a look at TPL DataFlow. BufferBlock is what you're looking for, but it offers so much more.

Comments

[user1703401]

possible duplicate of Queues And Wait Handles in C#

[bmm6o]

Do you want to allow enqueueing after stop() has been called?

[user1300560]

Thanks for the suggestion. Unfortunately I am limited to .Net 3.5, so that rules out BlockingCollection. Sorry I should have mentioned that in the requirements ;)

[user1300560]

Thanks for the suggestion. Unfortunately I am limited to .Net 3.5, so that rules these out. Sorry I should have mentioned that in the requirements ;)

[Theodor Zoulias]

According to the guidelines the Enqueue method should be named EnqueueAsync.

[Alexander Danilov]

@TheodorZoulias why Task.Run and TaskFactory.StartNew don't have async then? Same with Task.ContinueWith. Should be RunAsync, StartNewAsync, ContinueWithAsync.

[Alexander Danilov]

@TheodorZoulias moreover, actual enqueue happens synchronously, so after method finishes action is already enqueued.

[Theodor Zoulias]

If Enqueue runs synchronously, then returning a Task makes no sense. What this Task is supposed to represent? The asynchronous completion of what?

[Alexander Danilov]

Enqueue to the task queue is one thing, task completion is another. Enqueue is synchronous, but completion is not. You can await completion if you want. Same as TaskFactory.StartNew - task creation and starting is synchronous, but completion is not. And it also returns Task.

[Theodor Zoulias]

Your claim that TaskFactory.StartNew is synchronous sounds outrageous to me. This method returns a Task, so it can only be synchronous if the returned task is completed. Could you also give an example of an asynchronous method, so that I can understand what is the meaning of the word "asynchronous" for you?

[Alexander Danilov]

Two absolutely equal methods that return Tasks: getContactsListAsync, startContactsListRequest. First is async, second is sync, because of different naming. In both methods request is sent immediately (synchronously), but completion is awaited asynchronously.

[Theodor Zoulias]

These methods are not provided by the .NET Framework. Could you give an example of a built-in asynchronous method? Or there aren't any?

[Theodor Zoulias]

What about the built-in method HttpClient.SendAsync. Is it synchronous or asynchronous, and why?

[Alexander Danilov]

Try to answer my questions about TaskFactory etc first pls.

[Theodor Zoulias]

Alexander you have asked no questions, and you are avoiding to answer my questions. You have made some egregious claims that TaskFactory.StartNew is not asynchronous, and that asynchrony has something to do with the naming of a method. If you don't feel comfortable about clarifying your ideas, then your ideas stand no chance of being propagated. Simple.

[Alexander Danilov]

@TheodorZoulias question in my 1 comment, no answer and you talk about avoiding answers. I've told enough about my ideas, and seems that you've heard them right. This is my last answer to you.

[Theodor Zoulias]

Your question is why Microsoft didn't follow its own guidelines when naming the method Task.ContinueWith. I don't know the answer to that. If you want an official answer ask Microsoft, not me. I didn't oppose your judgement to ignore the guidelines because of that. Instead I asked you to explain your naming patterns, and your definition of asynchrony. If you have developed a robust and coherent system/theory I could become a follower.