Async Queue implementation without locking 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.

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

    /// &lt;summary&gt;
    ///     Promises.
    /// &lt;/summary&gt;
    private readonly ConcurrentQueue&lt;TaskCompletionSource&lt;T&gt;&gt; _promises;

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

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

    /// &lt;summary&gt;
    ///     Dequeue an Item Asynchronously.
    /// &lt;/summary&gt;
    /// &lt;param name="cancellationToken"&gt;
    ///     A cancellation token to cancel the asynchronous operation with.
    /// &lt;/param&gt;
    /// &lt;returns&gt;
    ///     A task representing the asynchronous operation.
    /// &lt;/returns&gt;
    public async Task&lt;T&gt; DequeueAsync(CancellationToken cancellationToken) {
        CancellationTokenRegistration? cancellationTokenRegistration = null;
        var promise = new TaskCompletionSource&lt;T&gt;();
        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();
        }

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

    /// &lt;summary&gt;
    ///     Enqueue an Item.
    /// &lt;/summary&gt;
    /// &lt;param name="item"&gt;
    ///     An item to enqueue.
    /// &lt;/param&gt;
    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:


Thread 1 enqueues an item
Thread 1 does not find an existing promise
Thread 1 is preempted
Thread 2 attempts to dequeue an item
Thread 2 does not find an existing item
Thread 2 is preempted
Thread 1 enqueues item
Thread 2 creates and enqueues a promise
Idle time until another enqueue or dequeue operation occurs
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&lt;T&gt; 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) &amp;&amp;
        !promise.Task.IsCanceled &amp;&amp;
        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.