Do the same work for many items at once (Parallel, SemaphoreSlim)

Parallel.For(from, to, i => …) · Parallel.ForEach(items, item => …) · await Parallel.ForEachAsync(items, async (item, ct) => …) · new ParallelOptions { MaxDegreeOfParallelism = n } · new SemaphoreSlim(n)

Parallel.For, Parallel.ForEach and Parallel.ForEachAsync run one body per item, each as a task beside the others, and wait for all of them; failures come back together as one AggregateException. A SemaphoreSlim limits how many run a section at once — await gate.WaitAsync() takes a slot, gate.Release() gives it back.

Summary#

Parallel.ForEach(items, item => Work(item)) runs Work for every item, several at the same time, and returns when all of them have finished. Use it for work that waits on someone else — an outbound call per item — so a hundred calls take as long as the slowest few, not as long as all of them in a row.

await Parallel.ForEachAsync(urls, async (url, ct) => {
  await gate.WaitAsync();
  try { await Fetch(url); } finally { gate.Release(); }
});

Signature#

Parallel.For(from, to, i => …)                         // i from `from` up to, not including, `to`
Parallel.ForEach(items, item => …)                     // every item of a list or array
await Parallel.ForEachAsync(items, async (item, ct) => …)   // a Task; ct stops when the run does
Parallel.ForEach(items, options, item => …)            // …each takes options before the body:
new ParallelOptions { MaxDegreeOfParallelism = 4, CancellationToken = cts.Token }
await Parallel.ForEachAsync(items, token, async (item, ct) => …)   // ForEachAsync also takes a bare token

new SemaphoreSlim(initialCount) · new SemaphoreSlim(initialCount, maxCount)
await gate.WaitAsync()   · gate.Wait()                 // take a slot (waiting for one if none is free)
gate.Release() · gate.Release(n)                       // give it back; answers the count before
gate.CurrentCount · gate.Dispose()

Description#

How many bodies run at once?#

As many as the machine has cores, as .NET's default does — or MaxDegreeOfParallelism in a ParallelOptions. When only one section of each body needs the limit — an API that allows three calls at a time — share a SemaphoreSlim(3) between the bodies and take a slot around the call. A ParallelOptions' CancellationToken stops the loop: no body starts once it is cancelled, and the loop raises OperationCanceledException.

What may a body do?#

Each body is a task, so it follows the rules a task does ([[function-async-await#writes]]): it reads what its caller can read, makes outbound calls, and returns or throws — but it may not create, change or delete rows in its caller's unit of work, because the bodies would race. Gather what should be written and write it after the loop. A body may be an expression or a block, and it reads the values around it as they were when the loop started.

What happens when a body fails?#

Every body still runs, and then the failures come back together as one AggregateException whose message names each one in item order — One or more errors occurred. (…) (…) — as C#'s does.

And in the browser?#

The bodies run one after another, where the loop is — a schedule C# also allows — with the same answers and the same AggregateException. A SemaphoreSlim counts the same way.

Examples#

app Prices {
  model "model/**/*.osy";
  use Osysharp.Http;
}

async Task<int> Price(string sku) {
  var r = Http.Get("https://prices.example/" + sku);
  return int.Parse(r.Body);
}

async Task<int> Refresh(List<string> skus) {
  var gate = new SemaphoreSlim(3);
  await Parallel.ForEachAsync(skus, async (sku, ct) => {
    await gate.WaitAsync();
    try { await Price(sku); } finally { gate.Release(); }
  });
  return skus.Count;
}

async Task<int> RefreshGently(List<string> skus, CancellationToken stop) {
  var options = new ParallelOptions { MaxDegreeOfParallelism = 2, CancellationToken = stop };
  try { Parallel.ForEach(skus, options, sku => Price(sku)); }
  catch (OperationCanceledException) { return 0; }
  return skus.Count;
}

int Square(int i) { return i * i; }

int Squares(int n) {
  Parallel.For(0, n, i => Square(i));
  return n;
}