How to avoid deadlock

Question

I have queue of MoveableItems in a class BackgroundTaskQueue (based on example from Microsoft here). Clients call BackgroundTaskQueue.QueueBackgroundWorkItem to add a MoveableItem to the queue.

An instance of BackgroundTaskQueue is passed to a class QueuedHostedService (which is also based on this). QueuedHostedService processes the queue on a separate thread to the enqueue action.

The BackgroundTaskQueue.DequeueAsync method blocks on a SemaphoreSlim (due to async). When a client calls BackgroundTaskQueue.QueueBackgroundWorkItem method it queues the MoveableItem and signals (releases) the semaphore. QueuedHostedService then processes the MoveableItem

Clients of would like to know all MoveableItem that are incomplete (defined as all items remaining in the queue and the currently processing item).

What suggested approaches exist to ensure Clients get an accurate audit of the incomplete items while Clients could be concurrently enqueuing items and asking for IncompleteItems, while Bar is concurrently dequeuing?

Problem I found is if I use another SemaphoreSlim in Bar as a mutex to ensure thread-safety of the Incomplete-Items list, then I get a deadlock race-condition - this happens when Bar is in the processing loop, its thread would claim the mutex-semaphore for protecting Incomplete-Items, then Bar calls Foo.Dequeue and blocks Foo's signalling-semaphore while it is waiting for a new item. At this point Bar has claimed the mutex and is blocked waiting for the enqueue signal. If a Client thread asks for the list of Incomplete Items it would attempt to claim Bar's mutex-semaphore and can't because it is held by Bar's queue processing thread.

public interface IBackgroundTaskQueue
{
    IList<MoveableItem> Items { get; }

    void QueueBackgroundWorkItem(MoveableItem moveableItem);

    Task<MoveableItem> DequeueAsync(
        CancellationToken cancellationToken);
}

public class BackgroundTaskQueue : IBackgroundTaskQueue
{
    private ConcurrentQueue<MoveableItem> _workItems =
        new ConcurrentQueue<MoveableItem>();
    private SemaphoreSlim _signal = new SemaphoreSlim(0);
    private readonly ILogger _log;

    public BackgroundTaskQueue(ILogger<BackgroundTaskQueue> logger)
    {
        _log = logger;
    }

    public void QueueBackgroundWorkItem(MoveableItem moveableItem)
    {
        if (moveableItem == null)
        {
            throw new ArgumentNullException(nameof(moveableItem));
        }

        _workItems.Enqueue(moveableItem);
        _signal.Release();
        _log.LogDebug($"{nameof(_signal)} release {nameof(QueueBackgroundWorkItem)}");
    }

    public async Task<MoveableItem> DequeueAsync(
        CancellationToken cancellationToken)
    {
        _log.LogDebug($"{nameof(_signal)} waiting async {nameof(QueueBackgroundWorkItem)}");

        await _signal.WaitAsync(cancellationToken);

        _log.LogDebug($"{nameof(_signal)} running {nameof(QueueBackgroundWorkItem)}");

        _workItems.TryDequeue(out var workItem);

        return workItem;
    }

    public bool IsEmpty()
    {
        return _workItems.IsEmpty;
    }

    public IList<MoveableItem> Items {  get => _workItems.ToList(); }
}


public class QueuedHostedService : BackgroundService
{
    private readonly ILogger _logger;
    private volatile MoveableItem _inProgressItem;
    private volatile IList<MoveableItem> _incompleteItems = new List<MoveableItem>();
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
    private readonly IBackgroundTaskQueue _taskQueue;

    public QueuedHostedService(IBackgroundTaskQueue taskQueue,
        ILogger<QueuedHostedService> logger)
    {
        _taskQueue = taskQueue;
        _logger = logger;
    }

    public override void Dispose()
    {
        _semaphore?.Dispose();
        base.Dispose();
    }

    public MoveableItem InProgressItem { get => _inProgressItem; private set => _inProgressItem = value; }

    public IList<MoveableItem> GetItems()
    {
        _logger.LogDebug($"{nameof(_semaphore)} waiting GetItems");
        _semaphore.Wait();
        _logger.LogDebug($"{nameof(_semaphore)} running GetItems");

        try
        {
            return _incompleteItems;
        }
        finally
        {
            _semaphore.Release(1);
            _logger.LogDebug($"{nameof(_semaphore)} release GetItems");
        }

    }

    protected async override Task ExecuteAsync(
        CancellationToken cancellationToken)
    {
        _logger.LogInformation("Queued Hosted Service is starting.");

        while (!cancellationToken.IsCancellationRequested)
        {
            MoveableItem moveableItem;

            moveableItem = await _taskQueue.DequeueAsync(cancellationToken);
            InProgressItem = moveableItem;

            _logger.LogDebug($"{nameof(_semaphore)} waiting ExecuteAsync StoreList");
            _semaphore.Wait();
            _logger.LogDebug($"{nameof(_semaphore)} running ExecuteAsync StoreList");

            try
            {
                var items = new List<MoveableItem>();
                if (InProgressItem != null)
                {
                    items.Add(InProgressItem);
                }
                items.AddRange(_taskQueue.Items);
                _incompleteItems = items;

            }
            finally
            {
                _semaphore.Release();
                _logger.LogDebug($"{nameof(_semaphore)} release ExecuteAsync StoreList");
            }

            try
            {
                // Do Stuff
            }
            catch (Exception ex)
            {
                _logger.LogError(ex,
                   $"Error occurred moviing {moveableItem}.");
            }
            finally
            {
                _logger.LogDebug($"{nameof(_semaphore)} waiting ExecuteAsync null");
                _semaphore.Wait();
                _logger.LogDebug($"{nameof(_semaphore)} running ExecuteAsync null");

                try
                {
                    InProgressItem = null;
                }
                finally
                {
                    _semaphore.Release();
                    _logger.LogDebug($"{nameof(_semaphore)} release ExecuteAsync null");
                }
            }
        }


        _logger.LogInformation("Queued Hosted Service is stopping.");
    }
}

Answer 1

Deadlocks occur when you have two threads that have a lock on resources the other thread needs. Since the threads are mutually waiting on the other thread to release the lock, neither will ever complete.

The best way to deal with deadlocks is to never lock in the first place. To prevent race conditions (the opposite of a deadlock), you can use collections that are designed for concurrent use like ConcurrentQueue or almost everything the System.Collections.Concurrent namespace.

If that's not possible, the following rules of thumb will at least minimize the likelihood of deadlocks if they don't completely eliminate them:

• Keep the period of the lock as short as possible
• Avoid acquiring multiple locks in the same thread
• Release locks before delegating tasks to background threads (including tasks)
• At the very least, locks should be released in the reverse order that they were acquired

Also know the impacts of the locking mechanisms you choose:

• lock keyword and code block: light weight lock across threads that is released as soon as the code exits the code block
• Semephore: light weight semaphore that locks across threads, same mechanics as a mutex though
• Mutex: very heavy weight mutex that protects resources across processes, uses the file system to manage the lock. Can be used to ensure only one instance of an application is ever running at a time.

The lock keyword is pretty straightforward and the easiest to get right. The Mutex has many different error states that you have to make doubly sure that you have the lock or not.

Answer 2

Solved deadlock by moving responsibility of tracking incomplete items into background task queue and removing the SemaphoreSlim that was acting as a mutex:

    public interface IBackgroundTaskQueue
    {
        IList<MoveableItem> IncompleteItems { get; }

        void QueueBackgroundWorkItem(MoveableItem moveableItem);

        Task<MoveableItem> DequeueAsync(
            CancellationToken cancellationToken);

        void CompleteItem();
    }

    public class BackgroundTaskQueue : IBackgroundTaskQueue
    {
        private ConcurrentQueue<MoveableItem> _workItems =
            new ConcurrentQueue<MoveableItem>();
        private ConcurrentQueue<MoveableItem> _incompleteItems =
            new ConcurrentQueue<MoveableItem>();
        private SemaphoreSlim _signal = new SemaphoreSlim(0);
        private readonly ILogger _log;

        public BackgroundTaskQueue(ILogger<BackgroundTaskQueue> logger)
        {
            _log = logger;
        }

        public void QueueBackgroundWorkItem(MoveableItem moveableItem)
        {
            if (moveableItem == null)
            {
                throw new ArgumentNullException(nameof(moveableItem));
            }

            _workItems.Enqueue(moveableItem);
            _incompleteItems.Enqueue(moveableItem);
            _signal.Release();
            _log.LogDebug($"{nameof(_signal)} release {nameof(QueueBackgroundWorkItem)}");
        }

        public async Task<MoveableItem> DequeueAsync(
            CancellationToken cancellationToken)
        {
            _log.LogDebug($"{nameof(_signal)} waiting async {nameof(QueueBackgroundWorkItem)}");

            await _signal.WaitAsync(cancellationToken);
            _log.LogDebug($"{nameof(_signal)} running {nameof(QueueBackgroundWorkItem)}");
            _workItems.TryDequeue(out var workItem);

            return workItem;
        }

        public void CompleteItem()
        {
            _incompleteItems.TryDequeue(out MoveableItem moveableItem);
        }

        public IList<MoveableItem> IncompleteItems {  get => _incompleteItems.ToList(); }
    }

    public class QueuedHostedService : BackgroundService
    {
        private readonly ILogger _logger;
        private readonly IBackgroundTaskQueue _taskQueue;

        public QueuedHostedService(IBackgroundTaskQueue taskQueue,
            ILogger<QueuedHostedService> logger)
        {
            _taskQueue = taskQueue;
            _logger = logger;
        }

        protected async override Task ExecuteAsync(
            CancellationToken cancellationToken)
        {
            _logger.LogInformation("Queued Hosted Service is starting.");

            while (!cancellationToken.IsCancellationRequested)
            {
                MoveableItem moveableItem;
                moveableItem = await _taskQueue.DequeueAsync(cancellationToken);

                try
                {
                    // Do stuff
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex,
                       $"Error occurred moving {moveableItem}.");
                }
                finally
                {
                    _taskQueue.CompleteItem();
                }
            }      

            _logger.LogInformation("Queued Hosted Service is stopping.");
        }
    }