Cancelling a long "string" of Task continuations properly? Forcing Task execution order and thread usage?

Question

I have a class that strings together small chunks of IO work as Task continuations. Each time a piece of work is received a new Task is created, it is then added as a continuation to the LastCreatedTask. I am trying to determine the proper way to cancel all of these Task properly?

Here is my setup currently

private Task LastCreatedTask { get; set; }
private CancellationTokenSource TaskQueueTokenSource { get; set; }

public void ScheduleChunk(IOWorkChunk inChunk, int procTimeout)
{
    Task ioChunkProcessTask = CreateProcessTask(inChunk, procTimeout);

    LastCreatedTask.ContinueWith(
        (t) => ioChunkProcessTask.Start(), 
        TaskContinuationOptions.ExecuteSynchronously);

    LastCreatedTask = ioChunkProcessTask;
}

private Task CreateProcessTask(IOWorkChunk inChunk, int procTimeout)
{
    // Create a TokenSource that will cancel after a given timeout period
    var ProcessTimeoutTokenSource = new CancellationTokenSource(
        TimeSpan.FromSeconds(procTimeout));

    // Create a TokenSource for the Task that is a 
    // link between the timeout and "main" token source
    var ProcessTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
        TaskQueueTokenSource.Token, 
        ProcessTimeoutTokenSource.Token);

    // Create a Task that will do the actual processing
    Task ioChunkProcessTask = new Task(() =>
    {
        if(!ProcessTokenSource.Token.IsCancellationRequested)
            inChunk.DoProcessing(ProcessTokenSource.Token);
    }, ProcessTokenSource.Token);

    return ioChunkProcessTask;
}

So in the function ScheduleChunk a "chunk" of IO work (and a timeout) are passed to CreateProcessTask which creates a Task that will do the actual processing of the IO work. A CancellationToken is passed to this Task which is made by chaining two CancellationTokenSources together.

The first is the "main" CancellationTokenSource; I want to be able to simply call Cancel on this source to cancel all of the Task that are chained. The second source is one that will automagically cancel after some given period of time (this stops long running/stalled IO chunks).

Finally once the constructed Task is returned to ScheduleChunk is it added as a continuation to the LastCreatedTask which is the last task the was added as a continuation. This in effect makes a chain of Task that run in order one after the other.

1. Is my method above the proper way to cancel the chain of Task? By calling Cancel on the TaskQueueTokenSource?

2. Is the use of the TaskContinuationOptions.ExecuteSynchronously along with continuations the proper way to ensure that these Task are executed in order one after the other?

3. Is there a way to specify that I would like the same thread from the underlying TPL ThreadPool to work on this chain of Task?

From what I can tell a new thread should not be created for each continuation point, though it is possible at some point that a new thread could pickup the chain.

Even though you've called Cancel on the cancellation token, you still will need to check if the token has been cancelled in all of the tasks.. — code4life
Oh yes, forgot to add that in there. I will make an edit shortly. — KDecker
Can you use await? Normally, all chains of tasks should be expressed as an async method. This makes all of this easy. — usr
Admitting my lack of knowledge, I am not sure how using async/await here would make this easier? I understand async/await to a point, just not how it would be applied. So to answer, no I don't know if I can or how I would? — KDecker

usr usr · Accepted Answer · 2016-02-11T20:08:29

The existing code does not work at all:

LastCreatedTask.ContinueWith((t) => ioChunkProcessTask)

This continuation merely returns a task whose result will be set to a constant object almost immediately. That's all it does.

Really, this code is structured awkwardly. This is much better:

async Task RunProcessing() {
 while (...) {
  await CreateProcessTask(...);
 }
}

await is made for sequencing async actions. await can replace ContinueWith most of the time.

The cancellation looks good. Maybe it can be simplified a bit but it's working fine.

Regarding 3, you should never need this. Thread affinity is a rare thing and to be avoided. There is no way to do this exactly as asked. Please elaborate what you want to achieve.

If you insist on using Task.Run, here's a sketch:

Task CreateProcessTask(IOWorkChunk inChunk, int procTimeout)
{
    // Create a TokenSource that will cancel after a given timeout period
    var ProcessTimeoutTokenSource = new CancellationTokenSource(
        TimeSpan.FromSeconds(procTimeout));

    // Create a TokenSource for the Task that is a 
    // link between the timeout and "main" token source
    var ProcessTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
        TaskQueueTokenSource.Token, 
        ProcessTimeoutTokenSource.Token);

    return Task.Run(() => {
        inChunk.DoProcessing(ProcessTokenSource.Token);
    }, ProcessTokenSource.Token);
}

Cancelling a long "string" of Task continuations properly? Forcing Task execution order and thread usage?

2 Answers