Usage of Task.WhenAll with infinite Tasks produced by BlockingCollection

Question

I am adding Background Tasks to a Blocking Collection (added in the Background).

I am waiting with Task.WhenAll on a Enumerable returned by GetConsumingEnumerable.

My question is: Is the overload of Task.WhenAll which receives an IEnumerable "prepared" to potentially receive an endless amount of tasks?

I am simply not sure if i can do it this way or if it was meant to be used this way?

private async Task RunAsync(TimeSpan delay, CancellationToken cancellationToken)
{
    using (BlockingCollection<Task> jobcollection = new BlockingCollection<Task>())
    {
        Task addingTask = Task.Run(async () =>
        {
            while (true)
            {
                DateTime utcNow = DateTime.UtcNow;
                var jobs = Repository.GetAllJobs();
                foreach (var job in GetRootJobsDue(jobs, utcNow))
                {
                    jobcollection.Add(Task.Run(() => RunJob(job, jobs, cancellationToken, utcNow), cancellationToken), cancellationToken);
                }

                await Task.Delay(delay, cancellationToken);
            }
        }, cancellationToken);

        await Task.WhenAll(jobcollection.GetConsumingEnumerable(cancellationToken));
    }
}

Answer 1

Since your goal is merely to wait until the cancellation token is cancelled, you should do that. For reasons others have explained, using WhenAll on an infinite sequence of tasks isn't the way to go about that. There are easier ways to get a task that will never complete.

await new TaskCompletionSource<bool>().Task
    .ContinueWith(t => { }, cancellationToken);

Answer 2

Task.WhenAll will not work with an infinite number of tasks. It will first (synchronously) wait for the enumerable to finish, and then (asynchronously) wait for them all to complete.

If you want to react to a sequence in an asynchronous way, then you need to use IObservable<Task> (Reactive Extensions). You can use a TPL Dataflow BufferBlock as a "queue" that can work with either synchronous or asynchronous code, and is easily convertible to IObservable<Task>.