C# Aggregate of Tasks

Question

Let's say I have a number of tasks, and I want to execute them one by one with a delay in between. The idea I had is to fold them with Aggregate into one task by combining with ContinueWith and inserting Task.Delay() in between each pair:

var tasks = new[] {1, 2, 3, 4, 5}.Select(async x => Console.WriteLine(x));
var superTask =
    tasks.Aggregate(Task.CompletedTask, async (task1, task2) =>
        await (await task1.ContinueWith(_ => Task.Delay(1000))).ContinueWith(_ => task2));

await superTask;

Question is why it doesn't work?

Answer 1

There's a couple problems. First of all, you aren't awaiting the delay, so it may as well not even exist. Second, Tasks will automatically start immediately when invoked that way-- you need to use the constructor method instead (see this question).

This bit of code will do what you want.

var tasks = new[] { 1, 2, 3, 4, 5 }
    .Select
    (
        x => new Task( () => Console.WriteLine(x) )
    );
var superTask =
    tasks.Aggregate
    (
        Task.CompletedTask,
        async (task1, task2) =>
        {
            await task1;
            await Task.Delay(1000);
            task2.Start();
        }
    );
await superTask;

All this being said, I highly doubt this is the right approach. Just write a simple loop to iterate over the array and work on them one at a time; async isn't even needed.

Answer 2

You wrote:

I want to execute them one by one with a delay in between.

There are several problems with your code.

In your example you created an enumerable without enumerating over it. Therefore your tasks have not started yet. They will be started as soon as you start enumerating. So as soon as you use foreach, or ToList(), or low-level enumeration: GetEnumerator() and MoveNext().

var tasks = new[] {1, 2, 3, 4, 5}.Select(async x => Console.WriteLine(x))
     .ToList();

Every task starts running as soon as it is created, so now they are running simultaneously.

Your aggregate function also enumerates every element, one by one, and performs your ContinueWith. The enumeration already starts the task.

Take a look at the source code of Enumerable.Aggregate

public static TSource Aggregate<TSource>(
    this IEnumerable<TSource> source,
    Func<TSource, TSource, TSource> func)
    {
        ... // some test for null
        using (IEnumerator<TSource> e = source.GetEnumerator())
        {
            if (!e.MoveNext()) throw Error.NoElements(); // there must be at least one element

            TSource result = e.Current;
            while (e.MoveNext()) result = func(result, e.Current);
            return result;
        }
    }

So what it does:

• The first MoveNext() executes your Select statement once. The first Task is Created and is scheduled to run as soon as possible, probably immediately.
• e.Current is used to save this running task variable Result.
• The 2nd MoveNext() executes the Select statement again. A second Task is created and scheduled to run as soon as possible, possible immediately.
• e.Current contains your 2nd task. the It is used to perform func(result, e.Current)

This func will do the following:

<first task>.ContinueWith(delay task).ContinueWith(<2ndTask>).

The result is put in variable result, MoveNext is done, etc.

Remember: 1st task and 2nd task are already running!

So in fact your agregate is something like:

Task result = empty task
for every task
{
     Create it and schedule it to start running as soon as possible
     Result = Result.ContinueWith(delayTask).ContinueWith(already running created task)
}

Now what happens if you take a running task, and ContinueWith another already running task?

If you take a look at Task.ContinueWith, it says that the continuationAction parameter of ContinueWith means:

An action to run when the task completes.

What happens if you run a task that is already running? I don't know, some test code will give you the answer. My best guess is that it won't do anything. It certainly won't stop the already running task.

This is not what you want!

It seems to me, that this is not what you want. You want to specify some actions to perform sequentially with a delay time in between. Something like this:

Create a Task for Action 1, and wait until finished
Delay
Create a Task for Action 2, and wait until finished
Delay
...

So what you want is a function with input a sequence of Actions and a DelayTime. Every Action will be started, awaited until completed, after which the DelayTime is awaited.

You could do this using an Aggregate where the input is a sequence of Actions. The result will be horrible. Difficult to read, test and maintain. A procedure will be much simpler to understand.

To make it LINQ compatible, I'll create an extension method. See extension methods demystified

static Task PerformWithDelay(this IEnumerable<Action> actionsToPerform, TimeSpan delayTime)
{
    var actionEnumerator = actionsToPerform.GetEnumerator();

    // do nothing if no actions to perform
    if (!actionEnumerator.MoveNext())
        return Task.CompletedTask;
    else
    {    // Current points to the first action
         Task result = actionEnumerator.Current;

         // enumerate over all other actions:
         while (actionEnumerator.MoveNext())
         {
             // there is a next action. ContinueWith delay and next task
             result.ContinueWith(Task.Delay(delayTime)
                   .ContinueWith(Task.Run( () => actionEnumerator.Current);
         } 
     }
}

Well, if you really, really want to use an aggregate to impress your collegues:

Task resultTask = actionsToPerform.Aggregate<Action, Task> (
    action =>
    {
         previousResultTask.ContinueWith(Task.Delay(delayTime))
                           .ContinueWith(Task.Run( () => action));
     });

So we seed the aggregation with the first item of your sequence, for every action in your sequence of Actions: ContinueWith a new Task.Delay, and ContinueWith a new Task that executes the action.

Problem: exception if your input is empty.