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.");
}
}