Async Queue implementation without locking

Question

I wrote a quick implementation of an async queue that utilizes a backing ConcurrentQueue. It was pretty much based on an implementation given in this Stack Overflow answer. The only difference is that I got rid of the internal locks that implementation uses.

Since a ConcurrentQueue is already thread safe, I can't figure out why they chose to use a lock in their implementation. Normally I would ask them on Stack Overflow but the question is 4 years old and I would rather post my implementation here and have someone do a peer review to see if I have a potential threading issue.

/// <summary>
///     Asynchronous Queue.
/// </summary>
/// <typeparam name="T">
///     The queue's elements' type.
/// </typeparam>
public sealed class AsyncQueue<T> {
    /// <summary>
    ///     Items.
    /// </summary>
    private readonly ConcurrentQueue<T> _items;

    /// <summary>
    ///     Promises.
    /// </summary>
    private readonly ConcurrentQueue<TaskCompletionSource<T>> _promises;

    /// <summary>
    ///     Create an Asynchronous Queue.
    /// </summary>
    public AsyncQueue() {
        this._items = new ConcurrentQueue<T>();
        this._promises = new ConcurrentQueue<TaskCompletionSource<T>>();
    }

    /// <summary>
    ///     Dequeue an Item Asynchronously.
    /// </summary>
    /// <returns>
    ///     A task representing the asynchronous operation.
    /// </returns>
    public Task<T> DequeueAsync() {
        var dequeueTask = this.DequeueAsync(CancellationToken.None);
        return dequeueTask;
    }

    /// <summary>
    ///     Dequeue an Item Asynchronously.
    /// </summary>
    /// <param name="cancellationToken">
    ///     A cancellation token to cancel the asynchronous operation with.
    /// </param>
    /// <returns>
    ///     A task representing the asynchronous operation.
    /// </returns>
    public async Task<T> DequeueAsync(CancellationToken cancellationToken) {
        CancellationTokenRegistration? cancellationTokenRegistration = null;
        var promise = new TaskCompletionSource<T>();
        var itemFound = this._items.TryDequeue(out var item);
        if (!itemFound) {
            cancellationTokenRegistration = cancellationToken.Register(OnCancellationTokenCanceled, promise);
            this._promises.Enqueue(promise);
        }
        else {
            promise.TrySetResult(item);
        }

        try {
            item = await promise.Task;
            return item;
        }
        finally {
            cancellationTokenRegistration?.Dispose();
        }

        // <summary>
        //      On Cancellation Token Canceled.
        // </summary>
        void OnCancellationTokenCanceled(object cState) {
            var cPromise = (TaskCompletionSource<T>) cState;
            cPromise.TrySetCanceled();
        }
    }

    /// <summary>
    ///     Enqueue an Item.
    /// </summary>
    /// <param name="item">
    ///     An item to enqueue.
    /// </param>
    public void Enqueue(T item) {
        while (true) {
            var promiseFound = this._promises.TryDequeue(out var promise);
            if (!promiseFound) {
                this._items.Enqueue(item);
                break;
            }

            var promiseSet = promise.TrySetResult(item);
            if (promiseSet) {
                break;
            }
        }
    }
}

I wrote unit tests and it seems to work fine. I think there might be a rare race condition that happens in the following scenario, but I have not been able to trigger it and I want a second opinion:

1. Thread 1 enqueues an item
2. Thread 1 does not find an existing promise
3. Thread 1 is preempted
4. Thread 2 attempts to dequeue an item
5. Thread 2 does not find an existing item
6. Thread 2 is preempted
7. Thread 1 enqueues item
8. Thread 2 creates and enqueues a promise
9. Idle time until another enqueue or dequeue operation occurs

Answer 1

In short, your example race condition is a valid one.

If the code is truly "safe" then you should be able to add any number of Thread.Sleep() calls to the class without causing a change in behaviour. This Fiddle uses the AsyncQueue to enqueue/dequeue a single item from different background threads. Without the locks, it is possible simulate the race condition by adding a sleep to Enqueue(). This ultimately deadlocks the app. Add the locks back in, and the race condition is removed - try it yourself in the Fiddle.

public void Enqueue(T item)
{
    TaskCompletionSource<T> promise;
    do
    {
        if (_promisesQueue.TryDequeue(out promise) &&
            !promise.Task.IsCanceled &&
            promise.TrySetResult(item))
        {
            return;
        }
    }
    while (promise != null);

    // lock (_syncRoot)
    // {
        if (_promisesQueue.TryDequeue(out promise) &&
            !promise.Task.IsCanceled &&
            promise.TrySetResult(item))
        {
            return;
        }

        // DequeueAsync() adds a promise to _promisesQueue at this point
        Thread.Sleep(1000);

        _bufferQueue.Enqueue(item);
    // }
}

PS: Interesting question, but why not just use a BufferBlock<T>?!

Answer 2

The only difference is that I got rid of the internal locks that implementation uses.

Since a ConcurrentQueue is already thread safe, I can't figure out why they chose to use a lock in their implementation.

From my understanding of the meaning of "thread safety":

All public and protected members of ConcurrentQueue<T> are thread-safe and may be used concurrently from multiple threads.

A class being "thread safe" just means its members are, but only individually. So one method being called from many threads will not cause problems. Interacting with many members from many threads might not be "thread safe".

In some sense, the meaning of "thread safety" depends on your application.

If you look at the lock from the answer:

lock (_syncRoot)
{
    if (_promisesQueue.TryDequeue(out promise) &&
        !promise.Task.IsCanceled &&
        promise.TrySetResult(item))
    {
        return;
    }
    _bufferQueue.Enqueue(item);
} 

All those statements in the block are combined together to be accessed only by one thread, which is something the "thread safety" of the class alone cannot guarantee.

The lock allows you to group code into a thread-safety-wise "atomic" block.