Production engineering workshop · 45 MIN

C# · Bound asynchronous work and propagate cancellation

Bound concurrency independently from the size of the input.

Parallel.ForEachAsync limits simultaneous delegate executions. CancellationToken must be passed into the delegate’s asynchronous operation; cancellation is cooperative. Store each result at its input index to preserve order without a shared append operation. Exceptions cause the returned task to fail; decide explicitly whether partial results are useful. This implementation expects a materialized array, so it does not claim constant-memory streaming.

First reproduce the normal path. Then force a failure at each boundary and inspect what remains true.

Read the example

using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public static class Pipeline {
  public static async Task<int[]> MapAsync(
      int[] input, int concurrency,
      Func<int,CancellationToken,Task<int>> transform,
      CancellationToken ct) {
    if (concurrency < 1) throw new ArgumentOutOfRangeException(nameof(concurrency));
    var result = new int[input.Length];
    await Parallel.ForEachAsync(Enumerable.Range(0,input.Length),
      new ParallelOptions { MaxDegreeOfParallelism=concurrency, CancellationToken=ct },
      async (index, token) => { result[index] = await transform(input[index], token); });
    return result;
  }
}
// Example inside async Main:
// var result = await Pipeline.MapAsync(new[]{1,2,3}, 2,
//   async (n, ct) => { await Task.Delay(10,ct); return n*n; }, CancellationToken.None);

Review the behavior in the stated project environment.

Your challenge

In a .NET console project, implement the function below and test an empty array, cancellation during Task.Delay, and a delegate that throws. Replace the sample transform with an HttpClient operation and propagate the supplied token. Use an injected managed HttpClient in a server project.

Solution cost: O(n) bookkeeping plus bounded asynchronous work. time · O(n + c) output and concurrency state. space

Common trap

Task.WhenAll over all input items does not itself bound the number of operations started.

using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public static class Pipeline {
  public static async Task<int[]> MapAsync(
      int[] input, int concurrency,
      Func<int,CancellationToken,Task<int>> transform,
      CancellationToken ct) {
    if (concurrency < 1) throw new ArgumentOutOfRangeException(nameof(concurrency));
    var result = new int[input.Length];
    await Parallel.ForEachAsync(Enumerable.Range(0,input.Length),
      new ParallelOptions { MaxDegreeOfParallelism=concurrency, CancellationToken=ct },
      async (index, token) => { result[index] = await transform(input[index], token); });
    return result;
  }
}
// Example inside async Main:
// var result = await Pipeline.MapAsync(new[]{1,2,3}, 2,
//   async (n, ct) => { await Task.Delay(10,ct); return n*n; }, CancellationToken.None);

Further reading: Official documentation