Benzene

Map-Reduce & High-Volume Compute

Status: DRAFT v0.1 — part of the Patterns "at enterprise scale" set.

Some enterprise workloads are not "handle a request" but "calculate over a very large dataset, fast" — revalue a book of a million positions at end of day, run a risk simulation across thousands of scenarios, reprice a portfolio when the curve moves, score a batch of transactions. The shape that fits is map-reduce / scatter-gather: split the work into independent shards, run them in parallel across many workers (the map), and combine the results (the reduce).

Benzene has no dedicated map-reduce primitive — and that is fine, because it composes cleanly from parts Benzene does give you. This document is honest about which parts are built-in and which you assemble, and works an end-of-day portfolio risk calculation through it.

What's built-in, what you compose

Be clear-eyed about this up front — it is the difference between using the framework and fighting it:

Piece Status
Scatter — dispatch N units of work concurrently Composed — Task.WhenAll over SendAsync (to core services / Lambdas), or concurrent Lambda-to-Lambda invokes. Ordinary async fan-out; the sender is stateless per call.
Bounded parallelism — a cap on how many run at once Built-in helper — BoundedFanOut.WhenAllAsync(source, body, maxDegreeOfParallelism) (semaphore-gated Task.WhenAll, results in source order); ConcurrentRequests + BoundedConcurrentDispatcher on self-hosted workers.
One message → many transports Built-in — UseParallel((..),(..)) on an outbound route, all-must-succeed. Fan-out publish, not scatter-gather of distinct work.
A fixed, heterogeneous parallel step set with rollback Built-in — a saga stage runs its steps concurrently (Task.WhenAll) and compensates on failure.
Reduce — aggregate the workers' results Composed — entirely app-level. There is no built-in response aggregator; you write the fold.

So: the map is a fan-out you compose; the reduce is a fold you write. Benzene supplies the transport, the topic addressing, the bounded-concurrency helper, and (for a fixed step set) the saga — it does not supply a scatter-gather-with-reduction API, and you should not go looking for one.

The shape

A coordinator service owns the map-reduce; the workers are ordinary Benzene services (or Lambdas) that each compute one shard:

                         ┌──────────────────────────────────────────┐
        trigger  ──────► │                COORDINATOR                │
    (schedule / API)     │  1. split dataset into shards             │
                         │  2. scatter: fan out a job per shard      │
                         │  3. gather + reduce the partial results   │
                         └──────────────────────────────────────────┘
                            │        │        │        │        │
                    SendAsync("valuation:shard", …) × N  (bounded fan-out)
                            ▼        ▼        ▼        ▼        ▼
                        ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐
                        │worker│ │worker│ │worker│ │worker│ │worker│  each: compute one shard,
                        └──────┘ └──────┘ └──────┘ └──────┘ └──────┘  return a partial result

Two-level scatter for very large fan-out

At a million positions, one coordinator issuing a million calls is itself a bottleneck. Shard hierarchically: the coordinator scatters to a modest number of partition workers (say, one per book or per asset class), and each partition worker scatters again over its slice and reduces locally, returning a partial to the coordinator's final reduce. Fan-out becomes a tree; each level stays a bounded fan-out. This is the serverless equivalent of a map-reduce shuffle, assembled from the same SendAsync + fold at each level.

How you build it with Benzene

Scatter with a concurrency cap

(informative, .NET) The map is a bounded parallel fan-out — BoundedFanOut.WhenAllAsync keeps the in-flight count under a ceiling and returns results in source order, so the reduce is deterministic:

// shards: the dataset split into independent units of work
var partials = await BoundedFanOut.WhenAllAsync(
    shards,
    shard => _sender.SendAsync<ValueShard, ShardResult>("valuation:shard", shard),
    maxDegreeOfParallelism: 64);

var total = partials
    .Where(p => p.IsSuccessful)
    .Aggregate(RiskVector.Zero, (acc, p) => acc + p.Payload);   // the reduce — your fold

On AWS the worker call resolves to a Lambda-to-Lambda invoke (service communication): cheap, fast, and burst-parallel — a thousand shards become a thousand concurrent Lambdas, each billed only for its own runtime. For a genuinely uniform "same calculation over a partitioned collection", that burst model is Benzene's sweet spot.

The worker

A worker is a normal handler — one shard in, one partial out:

[Message("valuation:shard")]
public class ValueShard : IMessageHandler<ValueShard, ShardResult>
{
    public Task<IBenzeneResult<ShardResult>> HandleAsync(ValueShard shard)
        => Task.FromResult(BenzeneResult.Ok(Revalue(shard.Positions, shard.MarketData)));
}

Handling partial failure

Fan-out at scale will have stragglers and failures. Two honest options, and the choice is a business decision:

Never silently drop a failed shard into the reduce — a partial total presented as complete is the worst outcome. If coverage was reduced, the result must say so.

Worked example: end-of-day portfolio risk

The system: each evening, revalue and risk-assess a book of ~1M positions against the day's closing curves, produce the firm's risk numbers, and store them for reporting and regulatory submission.

  1. Trigger. A schedule fires the risk coordinator (a Benzene service behind a timer trigger).
  2. Split. The coordinator partitions the book by portfolio into a few hundred shards (a shard sized so a worker finishes in seconds).
  3. Scatter. It fans out SendAsync("risk:shard", shard) under a BoundedFanOut cap of, say, 128 — a few hundred Lambda workers run concurrently, each revaluing its portfolio against the curves.
  4. Workers compute. Each risk:shard handler revalues its positions and returns a partial risk vector (P&L, greeks, VaR contributions). Workers are stateless and idempotent, so a retried shard is safe.
  5. Reduce. The coordinator folds the partial vectors into the firm-level risk — deterministically, because BoundedFanOut returns results in shard order.
  6. Persist + publish. The final numbers are written (they become the authoritative end-of-day figures) and a risk:completed event is emitted — from which a reporting read model projects the regulatory views and an event-sourced ledger records the run for audit.

The whole thing is minutes of wall-clock for a job that is hours if run serially — because the map is a burst of hundreds of independent Lambdas — and every part of it is an ordinary Benzene handler plus a fold you own.

When to use it

Checklist

Map-reduce is well-formed when:

See also: service communication (the Lambda-to-Lambda calls the scatter rides on) and stream processing (for ordered, rolling computations rather than batch fan-out).