4
\$\begingroup\$

Context and initial idea

I'm trying to teach myself the CQRS and Event Sourcing patterns by developing a simple ASP.NET Core application, loosely following Greg Young's example CQRS implementation on GitHub.

Once a command has been handled and the resulting events have been persisted to the database, I want to have any event handlers process the event on a background thread so that the thread that initiated the command can return without having to wait for the event handlers.

I've been taught that using Task.Run() for fire-and-forget-like background work like this is a bad idea for a multitude of reasons. I also want to be able to use scoped services from IServiceProvider (the default ASP.NET Core DI container) without having to worry about the services being disposed before the background work completes.

The ASP.NET Core documentation suggests a queue-based approach for background work which is based on the IHostedService abstraction and will run in the background for the entire duration of the web host and will even try to shut down any executing background work gracefully when the host stops. With some modifications, we could even have this service manage the service scopes of any enqueued background work (using IServiceProvider.CreateScope() and passing the scoped service provider to a Func<IServiceProvider, CancellationToken, Task> delegate) to be able to utilize the DI container to its full potential. Sounds great!

There is just one problem with this approach: it process the background work items in sequence, one at a time. If at all possible, I want to execute the background work in parallel.

Implementation

The implementation I've arrived at involves two parts: the interface IBackgroundTaskExecutor, which defines a simple Execute() method taking a delegate that defines the background work, and the class BackgroundTaskExecutingService which implements both the aforementioned interface and IHostedService and runs as a hosted background service alongside the web host.

BackgroundTaskExecutingService has the following responsibilities:

  • Process any incoming background work in parallel.
  • Log and handle any unhandled exceptions that are thrown by the background work.
  • Keep references to the executing Task instances until they complete.
  • Gracefully (attempt to) shut down all currently executing tasks when the web host stops.

(While not required by the contract that the ASP.NET Core web host specifies, the service can also be restarted after it has been stopped.)

All consumers have to do to dispatch background work is to retrieve a reference to the IBackgroundTaskExecutor service from the DI container and pass the background work they wish to process to Execute().

The implementation is designed as a reusable library and includes an IServiceCollection extension method for registering the implementation and any necessary dependencies.

Code

IBackgroundTaskExecutor.cs

using BackgroundTaskFactory = System.Func<
    System.IServiceProvider,
    System.Threading.CancellationToken,
    System.Threading.Tasks.Task>;

namespace BackgroundTasks
{
    public interface IBackgroundTaskExecutor
    {
        void Execute(BackgroundTaskFactory backgroundTaskFactory);
    }
}

BackgroundTaskExecutingService.cs

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

using BackgroundTaskFactory = System.Func<
    System.IServiceProvider,
    System.Threading.CancellationToken,
    System.Threading.Tasks.Task>;

namespace BackgroundTasks
{
    public sealed class BackgroundTaskExecutingService : IBackgroundTaskExecutor, IHostedService, IDisposable
    {
        private readonly IServiceProvider _services;
        private readonly ILogger _logger;

        private State _state;
        private ConcurrentDictionary<Task, byte> _executingTasks; // used as ConcurrentHashSet<Task>
        private CancellationTokenSource _stoppingCts;

        public BackgroundTaskExecutingService(
            IServiceProvider services,
            ILogger<BackgroundTaskExecutingService> logger)
        {
            if (services == null) throw new ArgumentNullException(nameof(services));
            if (logger == null) throw new ArgumentNullException(nameof(logger));

            _services = services;
            _logger = logger;
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            if (_state == State.Started) return Task.CompletedTask;

            _executingTasks = new ConcurrentDictionary<Task, byte>();
            _stoppingCts = new CancellationTokenSource();
            _state = State.Started;
            _logger.LogInformation("The background task executing service has started.");

            return Task.CompletedTask;
        }

        private void AssertStarted()
        {
            if (_state != State.Started)
            {
                throw new InvalidOperationException("The background task executing service is not started.");
            }
        }

        public void Execute(BackgroundTaskFactory backgroundTaskFactory)
        {
            if (backgroundTaskFactory == null) throw new ArgumentNullException(nameof(backgroundTaskFactory));

            AssertStarted();

            _ = ExecuteAsync(backgroundTaskFactory);
        }

        private async Task ExecuteAsync(BackgroundTaskFactory backgroundTaskFactory)
        {
            var task = ExecuteAndHandleExceptionAsync(backgroundTaskFactory);

            _executingTasks.TryAdd(task, default);
            await task;
            _executingTasks.TryRemove(task, out _);
        }

        private async Task ExecuteAndHandleExceptionAsync(BackgroundTaskFactory backgroundTaskFactory)
        {
            try
            {
                using (var scope = _services.CreateScope())
                {
                    await backgroundTaskFactory(scope.ServiceProvider, _stoppingCts.Token);
                }
            }
            catch (OperationCanceledException e) when (_stoppingCts.IsCancellationRequested)
            {
                _logger.LogInformation(e, "A background task was canceled.");
            }
            catch (Exception e)
            {
                _logger.LogError(e, "A background task threw an unhandled exception.");
            }
        }

        public async Task StopAsync(CancellationToken cancellationToken)
        {
            if (_state == State.Stopped) return;

            try
            {
                _logger.LogInformation("The background task executing service is stopping.");
                _stoppingCts.Cancel();
            }
            finally
            {
                // wait for any executing tasks to complete or the graceful shutdown process to be canceled
                await Task.WhenAny(
                    Task.WhenAll(_executingTasks.Keys),
                    Task.Delay(Timeout.Infinite, cancellationToken));

                _state = State.Stopped;
                _logger.LogInformation("The background task executing service has stopped.");
            }
        }

        public void Dispose()
        {
            _stoppingCts?.Cancel();
        }

        private enum State
        {
            Stopped,
            Started
        }
    }
}

BackgroundTasksServiceCollectionExtensions.cs

using System;
using BackgroundTasks;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;

namespace Microsoft.Extensions.DependencyInjection
{
    public static class BackgroundTasksServiceCollectionExtensions
    {
        public static IServiceCollection AddBackgroundTasks(this IServiceCollection services)
        {
            if (services == null) throw new ArgumentNullException(nameof(services));

            services.AddLogging();
            services.TryAddSingleton<BackgroundTaskExecutingService>();
            services.TryAddSingleton<IBackgroundTaskExecutor>(
                serviceProvider => serviceProvider.GetService<BackgroundTaskExecutingService>());
            services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService, BackgroundTaskExecutingService>(
                serviceProvider => serviceProvider.GetService<BackgroundTaskExecutingService>()));

            return services;
        }
    }
}

Example usage

This is an example of how the IBackgroundTaskExecutor is used within my web application to dispatch event messages to event handlers on a background thread:

private readonly IBackgroundTaskExecutor _backgroundTaskExecutor;

public void Dispatch<TEvent>(EventMetadata<TEvent> eventMetadata) where TEvent : Event
{
    _backgroundTaskExecutor.Execute((services, cancellationToken) =>
    {
        var eventHandlers = services.GetServices<IEventHandler<TEvent>>();

        return Task.WhenAll(eventHandlers.Select(eventHandler => eventHandler.HandleAsync(
            eventMetadata,
            cancellationToken)));
    });
}

Questions and concerns

  • Are there any immediate problems with this design? Am I being horribly naïve to use this as a solution to my problem?
  • This is for a relatively small application (a personal blog), but should I be concerned about things like thread starvation?
  • I'm not using ConfigureAwait(false) anywhere because it's not necessary for ASP.NET Core. Assuming I want to put this out as a reusable library targeting .NET Standard 2.0 (where it might be used by frameworks that have a SynchronizationContext)--on what lines would I need to use ConfigureAwait(false)?
  • Is there anything else that sticks out or looks bad?
\$\endgroup\$
  • \$\begingroup\$ Implementing Background Tasks in Web Applications is hard. Maybe you can check out Hangfire, a library for preforming Background Processing in Web Applications. \$\endgroup\$ – Volkmar Rigo Mar 31 at 10:26
  • \$\begingroup\$ @VolkmarRigo I considered using Hangfire, and if I needed support for retryable and/or scheduled tasks I probably would, but for now the problem seems simple enough that I want to try and solve it without tacking on an additional dependency. \$\endgroup\$ – qbicfeet Apr 1 at 11:07

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.