Create a cancelable Task background runner class

By Posted on

Problem

My requirements are:

  1. Create a .net 5 class that encapsulates a task runner (i.e. a wrapper around a Task object)
  2. The class should be thread-save
  3. Consumer of the class should be able to know if the task is running
  4. Consumer should be able to start/stop and start again the runner

I would love any expert advice regarding the class shown below. This is what I came up until now:

public class TaskWorker : IDisposable
{
    private CancellationTokenSource? _cancellationTokenSource;
    private bool _disposed;
    private bool _disposing;
    private readonly Func<CancellationToken, Task> _workerFunction;
    private readonly Action<Exception>? _onErrorCallback;
    private Task? _runningTask;
    private readonly object _syncLock = new();

    public TaskWorker(Func<CancellationToken, Task> workerFunction, Action<Exception>? onErrorCallback = null)
    {
        _workerFunction = workerFunction ?? throw new ArgumentNullException(nameof(workerFunction));
        _onErrorCallback = onErrorCallback;
    }

    public bool IsRunning { get; private set; }
    public bool IsStarted { get; private set; }

    public void Start()
    {
        if (_disposed)
        {
            throw new ObjectDisposedException(GetType().FullName);
        }

        lock (_syncLock)
        {
            if (_cancellationTokenSource != null)
                throw new InvalidOperationException();
            
            if (IsRunning)
                throw new InvalidOperationException();

            _cancellationTokenSource = new CancellationTokenSource();
            _runningTask = Task.Run(() => WorkerFuncCore(_cancellationTokenSource.Token), _cancellationTokenSource.Token);
            
            IsStarted = true;
        }
    }

    public void Stop()
    {
        if (_disposed)
        {
            throw new ObjectDisposedException(GetType().FullName);
        }

        lock (_syncLock)
        {
            if (_cancellationTokenSource == null)
                return;

            _cancellationTokenSource.Cancel();
            _cancellationTokenSource.Dispose();
            _cancellationTokenSource = null;

            _runningTask?.Wait();
            _runningTask = null;

            IsStarted = false;
        }
    }

    private async void WorkerFuncCore(CancellationToken cancellationToken)
    {
        try
        {
            IsRunning = true;
            await _workerFunction(cancellationToken);
        }
        catch (OperationCanceledException)
        {
            //do nothing
        }
        catch (Exception ex)
        {
            if (!_disposing && !_disposed)
            {
                _onErrorCallback?.Invoke(ex);
            }
        }

        IsRunning = false;
    }

    protected virtual void Dispose(bool disposing)
    {
        if (!_disposed)
        {
            if (disposing)
            {
                _disposing = true;
                try
                {
                    _cancellationTokenSource?.Cancel();
                    _cancellationTokenSource = null;
                }
                finally
                {
                    _disposing = false;
                }
            }

            _disposed = true;
        }
    }

    public void Dispose()
    {
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
}

This is a unit test:

    private AutoResetEvent _backgroundTaskRunnerStarted;

    [TestMethod]
    public void WorkerTask_Should_Run_A_Task_InBackground_And_Stops_Without_Errors()
    {
        var taskWorker = new TaskWorker(BackgroundTaskRunner, (ex) => Assert.Fail());

        taskWorker.Start();
        using AutoResetEvent backgroundTaskRunnerStarted = new AutoResetEvent(false);
        _backgroundTaskRunnerStarted = backgroundTaskRunnerStarted;
        _backgroundTaskRunnerStarted.WaitOne(1000).ShouldBeTrue();

        taskWorker.Stop();
        taskWorker.IsStarted.ShouldBeFalse();
        AssertExtensions.IsTrue(() => !taskWorker.IsRunning);
    }

    [TestMethod]
    public void WorkerTask_Should_Run_A_Blocking_Task_InBackground_And_Stops_Without_Errors()
    {
        var taskWorker = new TaskWorker(BackgroundTaskRunnerThatBlocksThread, (ex) => Assert.Fail());

        taskWorker.IsStarted.ShouldBeFalse();
        taskWorker.IsRunning.ShouldBeFalse();

        taskWorker.Start();
        taskWorker.IsStarted.ShouldBeTrue();

        using AutoResetEvent backgroundTaskRunnerStarted = new AutoResetEvent(false);
        _backgroundTaskRunnerStarted = backgroundTaskRunnerStarted;
        _backgroundTaskRunnerStarted.WaitOne(1000).ShouldBeTrue();
        taskWorker.IsRunning.ShouldBeTrue();

        taskWorker.Stop();
        taskWorker.IsStarted.ShouldBeFalse();
        AssertExtensions.IsTrue(() => !taskWorker.IsRunning);
    }

    [TestMethod]
    public void WorkerTask_Should_Run_A_Task_InBackground_And_Correctly_Handle_AnException()
    {
        Exception catchedExceptionInBackgroundTask = null;
        var taskWorker = new TaskWorker(BackgroundTaskRunnerThatThrowAnException, ex => catchedExceptionInBackgroundTask = ex);
        taskWorker.Start();

        AssertExtensions.IsTrue(() => catchedExceptionInBackgroundTask != null);
        AssertExtensions.IsTrue(() => !taskWorker.IsRunning);

        taskWorker.Stop();
        taskWorker.IsStarted.ShouldBeFalse();
    }

    private async Task BackgroundTaskRunner(CancellationToken cancellationToken)
    {
        _backgroundTaskRunnerStarted.Set();
        await Task.Delay(Timeout.Infinite, cancellationToken);
    }

    private Task BackgroundTaskRunnerThatBlocksThread(CancellationToken cancellationToken)
    {
        _backgroundTaskRunnerStarted.Set();
        while (!cancellationToken.IsCancellationRequested)
        {
            Thread.Sleep(10);
        }
        return Task.CompletedTask;
    }

    private async Task BackgroundTaskRunnerThatThrowAnException(CancellationToken cancellationToken)
    {
        _backgroundTaskRunnerStarted.Set();
        await Task.Delay(Timeout.Infinite, cancellationToken);
    }

public static class AssertExtensions
{
    public static void IsTrue(Func<bool> testFunc, int timeoutMilliseconds = 20000, Action beforeTimeoutExceptionAction = null)
    {
        while (!testFunc() && timeoutMilliseconds > 0)
        {
            timeoutMilliseconds -= 10;
            Thread.Sleep(10);
        }

        if (timeoutMilliseconds <= 0)
        {
            beforeTimeoutExceptionAction?.Invoke();
            throw new AssertFailedException();
        }
    }
 }

So far my concerns:

  1. Should I lock this under the Dispose(bool disposing) just before cancel the token?
  2. (Closed thanks to comment by Peter Csala) I’m correct to say that there is NO way to abort the execution of the function passed to the worker? in other words, if the user code doesn’t honor/check the passed cancellation token how I can abort it after a call to the Stop() function?
  3. Should I make the Stop() function awaitable and then await the the _runningTask
  4. I don’t think is strictly required to lock the check to the `_disposed’ variable?

Thanks for your comments!

Solution

Here are my observations/questions:

  • TaskWorker:
    • This name does not tell me anything
      • Is it a thing which works on a Task?
      • Is there a TaskManager or TaskDispatcher which distributes the tasks among workers?
    • So, please try to find better name which expresses your intent
  • IsStarted:
    • This property has been modified in Start and Stop
    • Let’s suppose for a second that it is set only in the Start method
      • In this case I would suggest to use HasStarted because the consumer of your class can retrieve this information at any time
    • To be honest with you I don’t understand why do set this property to false in Stop. If you stop the worker then what does IsStarted = false mean?
      • Has not been started yet???
  • TaskWorker‘s ctor: It accepts a cancellable task.
    • If it is already a cancellable task then why do we need this wrapper at all?
    • It works only with async methods. So, we can’t pass an async function (Task<TResult>).
  • Start‘s InvalidOperationException
    • There are two different cases when you throw this kind of exception
    • Please provide helpful error messages to be able to differentiate the two
  • Start‘s Task.Run
    • Task.Run accepts Action, Func<TResult>, Func<Task>, Func<Task<TResult>>
      • You have provided an Action which calls an async void. Why????
    • If the provided workerFunction is I/O bound then why do you move that operation on a seperate Thread?
  • Stop
    • _runningTask?.Wait(): This code will wait for the optional onErrorCallback to be finished. That means the Stop won’t finish until the user defined function completes.
      • Which might be okayish in some cases, but in most case this is undesirable.
  • WorkerFuncCore
    • Yet again this name does not provide any value for me
    • This method is not an async event handler, so I don’t see any valid reason why it is async void.
      • Please prefer async Task over async void.
    • I don’t get it why do you check dispose related variables inside the catch block.
      • If Dispose has been called then it will throw a OperationCanceledException because of this _cancellationTokenSource?.Cancel();
  • Dispose(bool disposing)
    • If your class provides Close or Stop or similar method then why don’t you use it inside the Dispose?
    • If they provide different functionality (like in your case) then how should I decide which way should I use it? (With or without using? Do I need to explicitly call Stop as well even if I use using?)

Leave a Reply

Your email address will not be published. Required fields are marked *