Stopping service with CancelationToken mechanism along with StopAsync method

Question

For each new message, the previous task (if available) should be stopped and a new one started.

SOLUTION

• The only solution I can see to my problems described below is to have CancellationTokenSource (CTS) in MessageHandler::Start and pass it to HandleAsync which will propogate it to all services. But, since _provider.StopAsync() must be called on cancelation, OnTaskStopping::_service.StopAsync() must be also called. Thus, I end up using CTS.Cancel and OnTaskStopping. Is this good approach to mix both CancellationToken and have Stop method?

PROBLEMS

• Let' assume the MessageHandler::Start::Task.Run hasn't started yet and we have a new incoming message and MessageHandler::Start::Task.WhenAll called first. That means, the OnTaskStopping() -> _service.StopAsync() -> _cts.Cancel() is called. When eventually MessageHandler::Start::Task.Run runs, it will create a new CancellationTokenSource and thus overwrite the _cts.Cancel(). Therefore, the foreach will not be canceled.
• If I move _cts = new CancellationTokenSource(); (denote NEW_CTS) to the end of MyService.StartAsync I might have a situation when _cts.Cancel() will be called right after NEW_CTS line. Meaning, when new MessageHandler::Start::Task.Run starts, the MyService.StartAsync will finish immediately since token is already canceled.

CODE

1    public abstract class MessageHandler
2    {
3        private readonly BlockingCollection<string> _pendingMessages;
4        public void Start()
5        {
6            _task = Task.Run(async () =>
7            {
8                Task handlerTask;
9                try
10                {
11                     // BlockingCollection is temporary. I'll switch to cahnnel to get async producer/consumer
12                     foreach (var msg in _pendingMessages.GetConsumingEnumerable(_cancellationTokenSource.Token))
13                     {
14                         try
15                         {
16                             // stop previous task
17                             if(handlerTask != null)
18                             {
19                                 await OnTaskStopping();
19.5                               await handlerTask;
20                             }
21    
22                             handlerTask = Task.Run(async () => await HandleAsync(msg));
23                         }
24                         catch (Exception ex)
25                         {
26                            ...
27                         }
28                     }
29                }
30                catch { } // OperationCanceledException
31            }
32        }
    
33        protected abstract Task HandleAsync(string msg);
34        protected abstract Task OnTaskStopping();
35    }
    
36    public class MyMessageHandler : MessageHandler
37    {
38        private readonly MyService _service;
    
39        public MyMessageHandler (MyService service)
40        {
41            _service = service;
42        }
    
43        protected override async Task HandleAsync(string msg)
44        {
45           ...
    
46            await _service.StartAsync(...);
47        }
    
48        protected override async Task OnTaskStopping()
49        {
50            await _service.StopAsync();
51        }
52    }
    
53    public class MyService
54    {
55        private CancellationTokenSource _cts;
56        private readonly IDevicesProvider _provider;
57       
58        public MyService()
59        { 
60           _cts = new CancellationTokenSource(); 
61        }
     
62        public async Task StartAsync(...)
63        {
64            _cts = new CancellationTokenSource();
    
65            foreach (var item in Items)
66            {
67                if(_cts.IsCancellationRequested)
68                   return;
                
69                ...
70            }
    
71            //_cts = new CancellationTokenSource(); 
72        }
    
73        public async Task<bool> StopAsync()
74        {
75            _cts.Cancel();
              
76            // THIS MUST HAPPEN
77            return await _provider.StopAsync();
78        }
79    } 

Answer 1

You can improve this in a couple of ways.

The first is not to access _cts constantly in StartAsync. Instead, read the token from _cts once into a local / parameter.

I also find it's worth saving the Task returned from StartAsync, so your StopAsync can wait for the cancellation to complete.

Something like:

public class MyService
{
    private CancellationTokenSource? _cts;
    private Task? _task;
    private readonly IDevicesProvider _provider;
 
    public async Task StartAsync(...)
    {
        await StopAsync(); // Or throw if we're already started?

        _cts = new CancellationTokenSource();
        _task = RunAsync(_cts.Token);
        await _task; 
    }

    private async Task RunAsync(CancellationToken token)
    {
        foreach (var item in Items)
        {
            if (token.IsCancellationRequested)
               return;
            
            ...
        }
    }

    public async Task<bool> StopAsync()
    {
        bool result = false;
        if (_cts != null)
        {
            _cts.Cancel();
            try
            {
                await _task;
            }
            catch (OperationCanceledException) { }

            try
            {
                // THIS MUST HAPPEN
                result = await _provider.StopAsync();
            }
            finally
            {
                _cts = null;
                _task = null;
            }
        }

        return result;
    }
}

Once you've got that far, it might make more sense to move the await _provider.StopAsync() into the RunAsync:

public class MyService
{
    private CancellationTokenSource? _cts;
    private Task<bool>? _task;
    private readonly IDevicesProvider _provider;
 
    public async Task StartAsync(...)
    {
        await StopAsync(); // Or throw if we're already started?

        _cts = new CancellationTokenSource();
        _task = RunAsync(_cts.Token);
        await _task; 
    }

    private async Task<bool> RunAsync(CancellationToken token)
    {
        try
        {
            foreach (var item in Items)
            {
                if (token.IsCancellationRequested)
                   return;
            
                ...
            }
        }
        catch (OperationCanceledException) { }
        finally
        {
            return await _provider.StopAsync();
        }
    }

    public async Task<bool> StopAsync()
    {
        bool result = false;
        if (_cts != null)
        {
            _cts.Cancel();
            try
            {
                result = await _task;
            }
            finally
            {
                _cts = null;
                _task = null;
            }
        }

        return result;
    }
}

Of course in both cases your StartAsync only completes after the task has fully completed (or been stopped) -- you can remove the await _task to avoid this.