3
\$\begingroup\$

I am attempting to construct a background service for an IIS-hosted ASP.NET Core project that basically queues tasks and runs them, and I decided to go with the IHostedService interface in Core 2.1 because of this tutorial. The tutorial does describe the case of running tasks one-by-one and in the order received, very well, and it does suffice for my purposes, but I was wondering whether it would be possible to run tasks in parallel via the same implementation of a BackgroundService, in such a way that one may control the maximum number of tasks running concurrently.

I found this question on SO, but was unable to get the proposed solution to work for the following task creation loop:

for (var i = 0; i < 20; i++) {
   var j = i;
   await _queue.QueueBackgroundWorkItem(async token =>
   {
       int s = j % 2 == 0 ? 5 : 1;
       Thread.Sleep(s * 1000);
       Debug.WriteLine("Processed " + j + " in " + s + " seconds");
   }
}

where _queue is the IBackgroundTaskQueue defined in the tutorial. Indeed, for the SemaphoreSlim in the solution, but with the initialCount setting set to 4, I still receive the numbers 0, 1, 2, 3, etc. in order.

I then tried to implement my own version of the QueueBackgroundWorkItem and DequeueAsync methods as an implemenation of IBackgroundTaskQueue called NewMultiThreadBackgroundTaskQueue:

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace QueueSystem.Queue
{
    public class NewMultiThreadBackgroundTaskQueue : IBackgroundTaskQueue
    {
        private readonly ConcurrentQueue<Func<CancellationToken, Task>> _workItems =
            new ConcurrentQueue<Func<CancellationToken, Task>>();

        private readonly List<Task> _runItems = new List<Task>();

        private readonly SemaphoreSlim _signal;
        private const int InitialCount = 4;

        public NewMultiThreadBackgroundTaskQueue()
        {
            _signal = new SemaphoreSlim(InitialCount);
        }

        public Task QueueBackgroundWorkItem(Func<CancellationToken, Task> workItem)
        {
            if (workItem == null)
            {
                throw new ArgumentNullException(nameof(workItem));
            }

            _workItems.Enqueue(async token =>
            {
                await _signal.WaitAsync(token);
                try
                {
                    await workItem(token);
                }
                finally
                {
                    _signal.Release();
                }
            });
            return Task.CompletedTask;
        }

        public async Task<Func<CancellationToken, Task>> DequeueAsync(CancellationToken cancellationToken)
        {
            while (!_workItems.IsEmpty && _runItems.Count < InitialCount)
            {
                _workItems.TryDequeue(out var workItem);
                if (workItem != null)
                {
                    _runItems.Add(Task.Run(async () => await workItem(cancellationToken), cancellationToken));
                }
            }

            if (!_runItems.Any()) return async token => await Task.CompletedTask;

            var completed = await Task.WhenAny(_runItems);
            _runItems.Remove(completed);
            return async token => await completed;
        }
    }
}

I use the DequeueAsync method in a QueueHostedService just as in the mentioned tutorial (i.e., nothing has changed):

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using QueueSystem.Queue;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace QueueSystem.Services
{
    public class QueueHostedService : BackgroundService
    {
        private readonly ILogger _logger;

        public QueueHostedService(IBackgroundTaskQueue taskQueue,
            ILoggerFactory loggerFactory)
        {
            TaskQueue = taskQueue;
            _logger = loggerFactory.CreateLogger<QueueHostedService>();
        }

        public IBackgroundTaskQueue TaskQueue { get; }

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


            while (!cancellationToken.IsCancellationRequested)
            {
                var workItem = await TaskQueue.DequeueAsync(cancellationToken);

                try
                { 
                    await workItem(cancellationToken);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex,
                        "Error occurred executing {WorkItem}.", nameof(workItem));
                }
            }

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

}

The DequeueAsync method does work as intended in the above background service, but I fear that I'm not using the CancellationToken correctly and my usage of Task.CompletedTask is a bit on the frivolous side.

EDIT:

Following this comment, I have changed the content of the QueueBackgroundWorkItem and DequeueAsync methods:

public void QueueBackgroundWorkItem(Func<CancellationToken, Task> workItem)
{
    if (workItem == null)
    {
       throw new ArgumentNullException(nameof(workItem));
    }
    _workItems.Enqueue(workItem);
}

public async Task<Func<CancellationToken, Task>> DequeueAsync(CancellationToken cancellationToken)
{
    if (_workItems.IsEmpty) return async token => await Task.CompletedTask;

    var removed = _runItems.RemoveAll(x => x.IsCompleted);
    while (!_workItems.IsEmpty && _runItems.Count < InitialCount)
    {
        _workItems.TryDequeue(out var workItem);
        if (workItem != null)
        {
            _runItems.Add(Task.Run(() => workItem(cancellationToken), cancellationToken));
        }
    }
    return async token => await Task.WhenAny(_runItems);
}

This still works (it dequeues and runs at most InitialCount tasks in parallel), but I'm still curious if there are any obvious improvements that I'm missing.

\$\endgroup\$
5
  • 2
    \$\begingroup\$ Hi! The code is working as intended and tasks are executed in parallel, I'm sorry if that wasn't clear. \$\endgroup\$
    – Bryder
    Dec 7 '19 at 16:47
  • \$\begingroup\$ For anyone in the Close Queue, the author believes the code is working as intended, do not close for that reason. \$\endgroup\$
    – pacmaninbw
    Dec 7 '19 at 18:02
  • 1
    \$\begingroup\$ _workItems.Enqueue(async token => { await _signal.WaitAsync(token); try { await workItem(token); } finally { _signal.Release(); } });why? why this is this being done... not even sure what its about. surely it should take a task and thats it? the rest is handled by the dequeue, could you explain this \$\endgroup\$
    – Seabizkit
    Jan 23 at 12:25
  • \$\begingroup\$ Thanks for your comment! It might work without the SemaphoreSlim, but I don't know for sure. I guess it probably will due to the SemaphoreSlim-free handling of the dequeueing of tasks (which means that I see what you see). I'll get back to you. :) \$\endgroup\$
    – Bryder
    Jan 24 at 14:44
  • \$\begingroup\$ You are correct - SemaphoreSlim is irrelevant. Also, I figured it would be nicer to check if the ConcurrentQueue is empty in the beginning of the DequeueAsync method and return a completed task if it is - then remove all completed tasks at the start of the while loop and return the Task.WhenAny(_runItems) after the while loop. That looks much better. Thanks for the feedback! \$\endgroup\$
    – Bryder
    Feb 4 at 7:45

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.