4
\$\begingroup\$

Problem space: I have the concept of a Job that needs to be processed. Jobs need to be processed sequentially, as one may depend on another. It's possible for a Job to be paused, in which case the next Job should not be started - the Job should either be explicitly continued, or cancelled, in order to get the next one to start.

If the background service doing these Jobs finishes, it should wait for an update to a RESTful resource before trying to start processing again. I don't want the service to busy-wait.

public class ExecutionBackgroundService : BackgroundService
{
    private readonly IJob _job;
    private readonly AutoResetEvent _isRunning = new AutoResetEvent(true);

    public ExecutionBackgroundService(IJob job)
    {
        _job = job;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Required for the method to be executed asynchronously, allowing startup to continue.
        await Task.Yield();
        while (!stoppingToken.IsCancellationRequested)
        {
            _isRunning.WaitOne(); // blocks the thread, waiting for the event to be set by something else calling the Resume method
            await _job.Process();
        }
    }

    /// <summary>
    /// Notifies the background service to resume processing.
    /// </summary>
    public void Resume()
    {
        _isRunning.Set();
    }
}

Jobs are represented by a DB entity. The IJob implementation looks something like this:

    public async Task Process()
    {
        using var scope = _serviceScopeFactory.CreateScope();
        var database = scope.ServiceProvider.GetService<MyDbContext>();
        var iterator = new DatabaseJobIterator(database);
        var worker = scope.ServiceProvider.GetService<IWorker>();
        while (iterator.HasNext())
        {
            await worker.DoLongRunningWork(iterator.Next);
        }
    }

Where the implementing class is singleton-scoped, and the DB and IWorker are both request-scoped (hence why I'm getting them using the scope factory).

DatabaseJobIterator is basically a simplified IEnumerator that just exposes two methods HasNext and Next, where HasNext will query the database for the oldest job, and assign it to Next. If the job is ready to be processed, it returns true. If the job is paused, it returns false.

An API controller that handles POSTs and PUTs to Jobs will then just call Resume on the background service, indicating that it might have new items to process if it isn't already running.

My question is whether I've used the AutoResetEvent correctly. Is there a more standard way to get the background service to wait without e.g. busy-waiting in a while (true) loop?

\$\endgroup\$
2
  • \$\begingroup\$ Are your API and your BackgroundService running in a same or in a separate process? In other words are we talking about cross-process synchronization or in-process? \$\endgroup\$ Nov 17, 2020 at 12:08
  • \$\begingroup\$ Just out of interested: Who would be calling Resume() and how? Do you pass around references to the ExecutionBackgroundService instance to other components? How would you get those references in the first place? \$\endgroup\$
    – Joerg
    Mar 29, 2023 at 9:11

1 Answer 1

3
\$\begingroup\$

I suggest SemaphoreSlim instead of AutoResetEvent here because it has asynchronous API and will not block the current thread, and it accepts CancellationToken. Task.Yield is good hack but it can be useless because await in case of not null SynchronizationContext will restore the context and WaitOne() can anyway block the current thread.

public class ExecutionBackgroundService : BackgroundService
{
    private readonly IJob _job;
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1); // 1 is concurrency degree

    public ExecutionBackgroundService(IJob job)
    {
        _job = job;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            while (true)
            {
                await _semaphore.WaitAsync(stoppingToken);
                await _job.Process();
            }
        }
        catch (OperationCanceledException ex)
        { } // occurs when WaitAsync receives canceling callback from the Token, you may log it if needed.
    }

    /// <summary>
    /// Notifies the background service to resume processing.
    /// </summary>
    public void Resume()
    {
        lock (_semaphore)
        {
            if (_semaphore.CurrentCount == 0) // is it waiting now?
                _semaphore.Release();
        }
    }
}

This way is better because now you may cancel the Token without calling the Resume().

Btw it's a programming pattern - Producer/Consumer. The best way for moving forward is using the specially designed for this purpose almost brand-new classes - System.Threading.Channels. Refer to this link.

\$\endgroup\$
2
  • 1
    \$\begingroup\$ The implementation of Resume is not thread-safe, two or more threads can enter the if-statement concurrently. \$\endgroup\$
    – Johnbot
    Nov 20, 2020 at 11:58
  • \$\begingroup\$ @Johnbot good catch, thanks! Fixed. \$\endgroup\$
    – aepot
    Nov 20, 2020 at 13:21

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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