My requirements are:
- Create a .net 5 class that encapsulates a task runner (i.e. a wrapper around a Task object)
- The class should be thread-save
- Consumer of the class should be able to know if the task is running
- 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:
- Should I lock
this
under theDispose(bool disposing)
just before cancel the token? - (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? - Should I make the
Stop()
function awaitable and then await the the_runningTask
- I don't think is strictly required to lock the check to the `_disposed' variable?
Thanks for your comments!
await Task.Delay(Timeout.Infinite);
- there's no a normal way to handle that. The only way to stop any executing (even hung) code is running it on a separate OS-levelProcess
and.Kill()
the process if necessary. There's also deprecated killing a threadThread.Abort()
but it's not recommended and works not in all cases (no guarantee to kill the thread). 2)BackgroundTaskRunner
= naming issue.Task
isn't running,Thread
is running,Task
is waiting. MaybeExecuteAsync
? Don't forget aboutAsync
suffix for awaitable methods. \$\endgroup\$Task.Run
doesn't runs aTask
, it runs aThread
with code provided in lambda/delegate and creates aTask
that waits for theThread
exit. \$\endgroup\$lock(this)
: Please read this guidelines \$\endgroup\$