Load, Stress, Soak, and Capacity Testing

public sealed record Sample(bool Successful, double Milliseconds);
public sealed record LoadReport(int Successful, int Failed, int PeakInFlight,
    double ElapsedMs, double[] LatenciesMs);

public static class Load
{
    public static async Task<LoadReport> Run(int requests, int concurrency,
        Func<int, System.Threading.CancellationToken, Task> operation,
        System.Threading.CancellationToken cancellation = default)
    {
        if (requests < 1 || requests > 1000 || concurrency < 1 || concurrency > 32)
            throw new ArgumentOutOfRangeException("Use 1–1000 requests and 1–32 workers.");
        using var slots = new System.Threading.SemaphoreSlim(concurrency);
        object guard = new object();
        int active = 0, peak = 0;
        var total = System.Diagnostics.Stopwatch.StartNew();
        Task<Sample>[] tasks = Enumerable.Range(0, requests).Select(async id =>
        {
            var timer = System.Diagnostics.Stopwatch.StartNew();
            await slots.WaitAsync(cancellation);
            lock (guard) { active++; peak = Math.Max(peak, active); }
            bool succeeded = true;
            try { await operation(id, cancellation); }
            catch (Exception error) when (error is not OperationCanceledException)
            { succeeded = false; }
            finally
            {
                lock (guard) { active--; }
                slots.Release();
            }
            return new Sample(succeeded, timer.Elapsed.TotalMilliseconds);
        }).ToArray();
        Sample[] samples = await Task.WhenAll(tasks);
        total.Stop();
        int successful = samples.Count(sample => sample.Successful);
        return new LoadReport(successful, requests - successful, peak,
            total.Elapsed.TotalMilliseconds, samples.Select(sample => sample.Milliseconds).ToArray());
    }
}

using var stop = new System.Threading.CancellationTokenSource(TimeSpan.FromSeconds(5));
LoadReport report = await Load.Run(20, 3,
    (id, token) => Task.Delay(2, token), stop.Token);
Console.WriteLine($"Completed: {report.Successful}; failed: {report.Failed}");
Console.WriteLine($"Peak in flight: {report.PeakInFlight}; elapsed: {report.ElapsedMs:F2} ms");

Twenty operations complete when the local delay succeeds. At most three execute at once. The timeout is propagated to the wait and the operation; cancellation is not counted as an ordinary completed failure. WhenAll observes completion of the child tasks before returning or throwing.

The Python example uses TaskGroup, available in Python 3.11 and later.

import asyncio
from time import perf_counter


async def run_load(requests, concurrency, operation):
    if not 1 <= requests <= 1000 or not 1 <= concurrency <= 32:
        raise ValueError("Use 1–1000 requests and 1–32 workers.")
    slots = asyncio.Semaphore(concurrency)
    active = 0
    peak = 0
    started = perf_counter()

    async def one(item_id):
        nonlocal active, peak
        item_started = perf_counter()
        async with slots:
            active += 1
            peak = max(peak, active)
            succeeded = True
            try:
                await operation(item_id)
            except Exception:
                succeeded = False
            finally:
                active -= 1
        return succeeded, (perf_counter() - item_started) * 1000

    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(one(i)) for i in range(requests)]
    samples = [task.result() for task in tasks]
    successful = sum(1 for succeeded, _ in samples if succeeded)
    return {
        "successful": successful,
        "failed": requests - successful,
        "peak_in_flight": peak,
        "elapsed_ms": (perf_counter() - started) * 1000,
        "latencies_ms": [elapsed for _, elapsed in samples],
    }

async def demonstration():
    async def operation(item_id):
        await asyncio.sleep(0.002)
    async with asyncio.timeout(5):
        report = await run_load(20, 3, operation)
    print(report["successful"], report["failed"])
    print(report["peak_in_flight"], report["elapsed_ms"])

asyncio.run(demonstration())

The counters run on one event loop with no suspension between incrementing and updating the peak. They are not shared-thread counters. CancelledError is not swallowed by the ordinary exception handler, and the task group manages the child tasks’ lifetime.