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
1
\$\begingroup\$

just come across this and I'm sure by now you've figured a lot of things out, but thought I'd add my 2 cents worth.

I wouldn't be worried about thread starvation, I would though be worried! I've played with Greg Young's m-r example a few times, from memory it processes events in-process which is a nice way to get things started. Ideally though if you don't want to do that the "best" way to go is to put the events on some kind of queue be it RabbitMQ, AWS SQS, Azure Service Bus or event MSMQ (but don't use MSMQ, ever!!!). Once your events are on a queue you can write another application/process that can read them and process at will.

Queue Benefits:

  • Resilience: This will make your system more resilient, if there's a crash during processing of an event it will just be put back on the queue and processed another time. Days even if you have a bug that needs fixing before you want the rest of the events to be processed (I've actually done this in the past, although I can definitely say it's not fun)
  • Scale: You can also scale with more cunning and precision. You can scale your event processing independently of your website and if you so wish you could have one process for each type of event and so you could scale per event type at the extreme end.

IHostedService

I'm working with IHostedService at the moment, I've written a console app that just runs one IHostedService (and reads events off a queue funnily enough). The problem that lead me to this page is that the hosted service runs in a background thread out of my control and I'm currently getting an exception thrown that no try/catch is picking up (I was taught to let exceptions bubble up and this exception is not now I'm having to put try/catch everywhere to find where the problem is). This exception is currently taking down the entire application, from what I've read this is a thing with IHostedServices. You don't want a background process to take down your website!

Single Responsibility

Not only is this the S in SOLID, but it's also a guiding principle in the microservices/distritubed systems world. Each system/service should do one thing and do it well.

Hosting

Last but not least where it's going to be hosted from is always going to be a factor IMO. I work a lot with Azure and now Kubernetes so while I build things that can be hosted anywhere there's always a lean towards leveraging the abilities of that hosting environment. For instance you could host in an Azure Web App, these have the concept of Web Jobs that are separate to the website, but run in the same App Service Plan. You could have your site write to an Azure Storage Queue and then a web job read the events from this (just an example)

ConfigureAwait In my opinion don't write your own framework, there are plenty of things out there that have had many eyes on them. If you're asking these sorts of questions use someone else's framework to start with and then once you've got familiar with it make the decision then about making your own. That's my opinion though. From what I've read on MS documention in the past, yes you should use ConfigureAwait(false) when creating a package.

Hope some of this helps. This is just from my experience and I'm sure other people will say other things :-)

\$\endgroup\$
  • \$\begingroup\$ I might be wrong, but in .NET Core I don't think ConfigureAwait(false) is needed since there is no synchronization context anymore. \$\endgroup\$ – dfhwze Aug 23 at 8:31
  • \$\begingroup\$ I can't confirm or deny this, I've not used it for a long time now. \$\endgroup\$ – matt_lethargic Aug 23 at 9:00

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.