How to make asynchronous methods that use a queue thread safe

Question

I have a service that ensures that exactly one popup is displayed at the same time. AddPopupAsync can be called concurrently, i.e. a popup is open while another 10 AddPopupAsync requests drop in. The code:

public async Task<int> AddPopupAsync(Message message)
{
    //[...]

    while (popupQueue.Count != 0)
    {
        await popupQueue.Peek();
    }

    return await Popup(interaction);
}

But I can spot two unwanted things that can happen because of the lack of thread safety:

1. If the queue is empty, Peek will throw an exception
2. If a thread A is preempted before the first statement in Popup, another thread B will not wait for the pending popup A since the queue is still empty.

Popup method works with a TaskCompletionSource and before calling its SetResult method, popupQueue.Dequeue() is called.

I think about using the atomic TryPeek from ConcurrentQueue in order to make #1 thread safe:

do
{
    Task<int> result;

    bool success = popupQueue.TryPeek(out result);

    if (!success) break;
    await result;
} 
while (true);

Then I read about a AsyncProducerConsumerCollection but I'm not sure if that is the most simple solution.

How can I ensure thread safety in a simple way? Thank you.

Answer 1

To simply add thread-safety you should use a ConcurrentQueue. It's a thread-safe implementation of a queue and you can use it just the same way you would use a queue without worrying about concurrency.

However if you want a queue you can await asynchronously (which means that you're not busy-waiting or blocking a thread while waiting) you should use TPL Dataflow's BufferBlock which is very similar to AsyncProducerConsumerCollection only already implemented for you by MS:

var buffer = new BufferBlock<int>();
await buffer.SendAsync(3); // Instead of enqueue.
int item = await buffer.ReceiveAsync(); // instead of dequeue.

ConcurrentQueue is fine for thread-safeness, but wasteful in high levels of contention. BufferBlock is not only thread-safe it also gives you asynchronous coordination (usually between a consumer and producer).