using System;

using System.Collections.Concurrent;

using System.Threading;

using System.Threading.Tasks;

​

namespace AsyncQueue

{

    public class Program

    {

        public static void Main(string[] args)

        {

            var asyncQueue = new AsyncQueue<string>();

​

            try

            {

                using (var cts = new CancellationTokenSource(3000))

                {

                    Task.WaitAll(

                        Task.Run(() => asyncQueue.Enqueue("foo")),

                        Task.Run(() => asyncQueue.DequeueAsync(cts.Token)));

​

                    Console.WriteLine("Did not deadlock!");

                }

            }

            catch (AggregateException ex) when (ex.InnerException is OperationCanceledException)

            {

                Console.WriteLine("Deadlocked! Add locks in order to resolve.");

            }

        }

    }

​

    public class AsyncQueue<T>

    {

        private readonly ConcurrentQueue<T> _bufferQueue;

        private readonly ConcurrentQueue<TaskCompletionSource<T>> _promisesQueue;

        private readonly object _syncRoot = new object();

​

        public AsyncQueue()

        {

            _bufferQueue = new ConcurrentQueue<T>();

            _promisesQueue = new ConcurrentQueue<TaskCompletionSource<T>>();

        }

​

        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;

                }

​

                // Without a lock, DequeueAsync() could add a promise to _promisesQueue at this point

                Thread.Sleep(1000);

​

                _bufferQueue.Enqueue(item);

//            }

        }

​

        public Task<T> DequeueAsync(CancellationToken cancellationToken)

        {

            if (!_bufferQueue.TryDequeue(out var item))

            {

//                lock (_syncRoot)

//                {

                    if (!_bufferQueue.TryDequeue(out item))

                    {

                        var promise = new TaskCompletionSource<T>();

                        cancellationToken.Register(() => promise.TrySetCanceled());

​

                        // Without a lock, Enqueue() could add an item to the _bufferQueue at this point

​

                        _promisesQueue.Enqueue(promise);

​

                        return promise.Task;

                    }

//                }

            }

​

            return Task.FromResult(item);

        }

    }

}

​