Bounded queues and backpressure

{
  "sdk": {
    "version": "10.0.403",
    "rollForward": "latestPatch"
  }
}

<Project Sdk="Microsoft.NET.Sdk.Web"><PropertyGroup><TargetFramework>net10.0</TargetFramework><Nullable>enable</Nullable><ImplicitUsings>enable</ImplicitUsings></PropertyGroup><ItemGroup><PackageReference Include="Microsoft.Data.Sqlite" Version="10.0.14" /></ItemGroup></Project>

using Microsoft.AspNetCore.Hosting.Server;
using Microsoft.AspNetCore.Hosting.Server.Features;
using Microsoft.Data.Sqlite;
var builder = WebApplication.CreateBuilder(args);
var settings = new LabSettings {
    Replica = builder.Configuration["Lab:Replica"] ?? "one",
    Workers = builder.Configuration.GetValue("Lab:Workers", 2),
    Partition = builder.Configuration.GetValue("Lab:Partition", 0),
    QueueCapacity = builder.Configuration.GetValue("Lab:QueueCapacity", 4),
    PoolSize = builder.Configuration.GetValue("Lab:PoolSize", 2),
    WorkMilliseconds = builder.Configuration.GetValue("Lab:WorkMilliseconds", 20),
    LedgerPath = builder.Configuration["Lab:LedgerPath"] ?? Path.Combine(Path.GetTempPath(), "sharprune-local-ledger.db")
};
settings.Validate(); builder.Services.AddSingleton(settings); builder.Services.AddSingleton<LabRuntime>(); builder.Services.AddSingleton<LocalIo>();
builder.Services.AddSingleton<IHostedService>(services => services.GetRequiredService<LabRuntime>());
builder.Services.AddSingleton<LocalLedger>(); var app = builder.Build();
app.Services.GetRequiredService<LocalLedger>().Initialize();
app.MapGet("/instance", (LabRuntime lab) => new { replica = settings.Replica, visits = lab.Visit() });
app.MapGet("/live", () => TypedResults.Text("live"));
app.MapGet("/io-delay", async (CancellationToken token) => { await Task.Delay(20, token); return TypedResults.Text("controlled loopback reply"); });
app.MapGet("/io", async (CancellationToken token, IServer server, LocalIo io) => {
    using var response = await io.Client.GetAsync(LocalAddress(server), token); response.EnsureSuccessStatusCode(); _ = await response.Content.ReadAsByteArrayAsync(token);
    return TypedResults.Ok(new { replica = settings.Replica, kind = "async-loopback-http" });
});
app.MapGet("/sync", (CancellationToken token, IServer server, LocalIo io) => {
    using var response = io.Client.GetAsync(LocalAddress(server), token).GetAwaiter().GetResult(); response.EnsureSuccessStatusCode(); _ = response.Content.ReadAsByteArrayAsync(token).GetAwaiter().GetResult();
    return TypedResults.Ok(new { replica = settings.Replica, kind = "deliberate-blocking-demo" });
});
app.MapGet("/cpu", () => { uint value = 1; for (int i = 0; i < 200000; i++) value = unchecked(value * 1664525 + 1013904223); return TypedResults.Ok(new { replica = settings.Replica, kind = "bounded-cpu", value }); });
app.MapGet("/threadpool", () => new { threads = ThreadPool.ThreadCount, pending = ThreadPool.PendingWorkItemCount, completed = ThreadPool.CompletedWorkItemCount });
app.MapGet("/ready", (LabRuntime lab) => lab.Accepting ? Results.Ok(new { ready = true }) : Results.StatusCode(503));
app.MapGet("/stats", (LabRuntime lab) => new { replica = settings.Replica, accepting = lab.Accepting, main = lab.Main.Snapshot(), batch = lab.Batch.Snapshot(), dependencyCalls = lab.Circuit.DependencyCalls });
app.MapGet("/work", (HttpContext context, LabRuntime lab, int? budgetMs) => Handle(context, lab.Main, lab, budgetMs));
app.MapGet("/batch", (HttpContext context, LabRuntime lab, int? budgetMs) => Handle(context, lab.Batch, lab, budgetMs));
app.MapGet("/unstable", (LabRuntime lab, string? key) => {
    if (key != "demo" && key != "always") return Results.BadRequest();
    int attempt = lab.Attempt(key); return key == "always" || attempt == 1 ? Results.StatusCode(503) : Results.Ok(new { replica = settings.Replica, attempt });
});
app.MapGet("/session/{name}", (LabRuntime lab, string name) => name == "demo" ? Results.Ok(new { replica = settings.Replica, value = lab.Session() }) : Results.BadRequest());
app.MapPost("/once/{key}", (string key, LedgerInput input, LocalLedger ledger) => {
    if (key != "demo" || input.Amount is < 1 or > 100) return Results.BadRequest();
    LedgerDecision result = ledger.Record(key, input.Amount); return result.Conflict ? Results.Conflict(new { error = "key amount conflict" }) : Results.Ok(result);
});
app.MapGet("/ledger", (LocalLedger ledger) => new { receipts = ledger.Count() });
app.MapGet("/guarded", (LabRuntime lab) => lab.Circuit.Invoke(lab.DependencyFailing));
app.MapGet("/partition/{key}", (string key, int? owner) => {
    if (key != "A" && key != "B") return Results.BadRequest();
    int partition = StablePartition.For(key, 2);
    return owner == partition && settings.Partition == partition ? Results.Ok(new { replica = settings.Replica, partition, key }) : Results.Conflict(new { expectedPartition = partition });
});
if (app.Environment.IsDevelopment()) {
    app.MapPost("/lab/gate/{lane}/{action}", (string lane, string action, LabRuntime lab) => {
        WorkLane? target = lane == "main" ? lab.Main : lane == "batch" ? lab.Batch : null;
        if (target is null || action != "open" && action != "close") return Results.BadRequest();
        if (action == "open") target.Gate.Open(); else target.Gate.Close(); return Results.NoContent();
    });
    app.MapPost("/lab/drain", (LabRuntime lab) => { lab.Drain(); return Results.NoContent(); });
    app.MapPost("/lab/dependency/{failing:bool}", (bool failing, LabRuntime lab) => { lab.DependencyFailing = failing; return Results.NoContent(); });
}
app.Run();
static async Task<IResult> Handle(HttpContext context, WorkLane lane, LabRuntime lab, int? budgetMs) {
    if (budgetMs is < 1 or > 1000) return Results.BadRequest();
    using var budget = CancellationTokenSource.CreateLinkedTokenSource(context.RequestAborted);
    if (budgetMs is int milliseconds) budget.CancelAfter(milliseconds);
    var item = new WorkItem(budget.Token);
    AdmissionDecision decision = lab.Admit(lane, item);
    if (decision == AdmissionDecision.Draining) return Results.StatusCode(503);
    if (decision == AdmissionDecision.Full) return Results.StatusCode(429);
    try { await item.Completion.Task.WaitAsync(budget.Token); return Results.Ok(new { replica = lab.Settings.Replica, completed = true }); }
    catch (OperationCanceledException) when (!context.RequestAborted.IsCancellationRequested) { return Results.Problem(statusCode: 504, title: "Classroom work budget expired"); }
}
static Uri LocalAddress(IServer server) {
    string address = server.Features.Get<IServerAddressesFeature>()!.Addresses.Single();
    return new UriBuilder("http", "127.0.0.1", new Uri(address).Port) { Path = "/io-delay" }.Uri;
}
public sealed class LocalIo : IDisposable { public HttpClient Client { get; } = new() { Timeout = TimeSpan.FromSeconds(5) }; public void Dispose() => Client.Dispose(); }
public sealed record LedgerInput(int Amount);
public sealed class LabSettings {
    public string Replica { get; init; } = "one"; public int Partition { get; init; } public int Workers { get; init; } public int QueueCapacity { get; init; }
    public int PoolSize { get; init; } public int WorkMilliseconds { get; init; } public string LedgerPath { get; init; } = "";
    public void Validate() {
        if (Partition is < 0 or > 1 || Replica.Length is < 1 or > 32 || Workers is < 1 or > 8 || QueueCapacity is < 1 or > 16 || PoolSize is < 1 or > 8 || WorkMilliseconds is < 1 or > 250 || string.IsNullOrWhiteSpace(LedgerPath)) throw new InvalidOperationException("Invalid bounded lab settings");
    }
}
public enum AdmissionDecision { Accepted, Full, Draining }
public sealed class LabRuntime : IHostedService, IDisposable {
    private readonly object admissionSync = new();
    private int accepting = 1, visits, sessions, demoAttempts, alwaysAttempts, failing = 1, disposed;
    public LabSettings Settings { get; } public WorkLane Main { get; } public WorkLane Batch { get; }
    public ClassroomCircuit Circuit { get; } = new();
    public LabRuntime(LabSettings settings) { Settings = settings; Main = new(settings.Workers, settings.QueueCapacity, settings.PoolSize, settings.WorkMilliseconds); Batch = new(1, 1, 1, settings.WorkMilliseconds); }
    public bool Accepting => Volatile.Read(ref accepting) == 1;
    public bool DependencyFailing { get => Volatile.Read(ref failing) == 1; set => Volatile.Write(ref failing, value ? 1 : 0); }
    public void Drain() { lock (admissionSync) Volatile.Write(ref accepting, 0); }
    public AdmissionDecision Admit(WorkLane lane, WorkItem item) {
        lock (admissionSync) {
            if (!Accepting) return AdmissionDecision.Draining;
            return lane.TrySubmit(item) ? AdmissionDecision.Accepted : AdmissionDecision.Full;
        }
    }
    public int Visit() => Interlocked.Increment(ref visits); public int Session() => Interlocked.Increment(ref sessions);
    public int Attempt(string key) => key == "demo" ? Interlocked.Increment(ref demoAttempts) : Interlocked.Increment(ref alwaysAttempts);
    public Task StartAsync(CancellationToken token) { Main.Start(); Batch.Start(); return Task.CompletedTask; }
    public Task StopAsync(CancellationToken token) => Task.WhenAll(Main.StopAsync(token), Batch.StopAsync(token));
    public void Dispose() { if (Interlocked.Exchange(ref disposed, 1) != 0) return; Main.Dispose(); Batch.Dispose(); }
}
public static class StablePartition {
    public static int For(string key, int count) { uint hash = 2166136261; foreach (byte b in System.Text.Encoding.UTF8.GetBytes(key)) hash = unchecked((hash ^ b) * 16777619); return (int)(hash % (uint)count); }
}

using System.Threading.Channels;
public sealed record WorkItem(CancellationToken Token) { public TaskCompletionSource<bool> Completion { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); }
public sealed class AsyncGate {
    private readonly object sync = new(); private TaskCompletionSource<bool> source = Completed();
    private static TaskCompletionSource<bool> Completed() { var value = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously); value.SetResult(true); return value; }
    public void Close() { lock (sync) { if (source.Task.IsCompleted) source = new(TaskCreationOptions.RunContinuationsAsynchronously); } }
    public void Open() { lock (sync) source.TrySetResult(true); }
    public Task WaitAsync(CancellationToken token) { lock (sync) return source.Task.WaitAsync(token); }
}
public sealed class WorkLane : IDisposable {
    private readonly Channel<WorkItem> queue; private readonly SemaphoreSlim pool; private readonly int workers, milliseconds;
    private readonly CancellationTokenSource stop = new(); private Task[] running = Array.Empty<Task>();
    private int active, inPool, maxActive, maxPool, accepted, rejected, completed, canceled, disposed;
    public AsyncGate Gate { get; } = new();
    public WorkLane(int workers, int capacity, int poolSize, int milliseconds) {
        this.workers = workers; this.milliseconds = milliseconds; pool = new(poolSize, poolSize);
        queue = Channel.CreateBounded<WorkItem>(new BoundedChannelOptions(capacity) { FullMode = BoundedChannelFullMode.Wait, SingleWriter = false, SingleReader = workers == 1 });
    }
    public void Start() { running = Enumerable.Range(0, workers).Select(_ => RunAsync()).ToArray(); }
    public bool TrySubmit(WorkItem item) { if (queue.Writer.TryWrite(item)) { Interlocked.Increment(ref accepted); return true; } Interlocked.Increment(ref rejected); return false; }
    private async Task RunAsync() {
        await foreach (WorkItem item in queue.Reader.ReadAllAsync(stop.Token)) {
            int now = Interlocked.Increment(ref active); Maximum(ref maxActive, now); bool entered = false;
            using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(item.Token, stop.Token);
            try {
                lifetime.Token.ThrowIfCancellationRequested(); await Gate.WaitAsync(lifetime.Token);
                await pool.WaitAsync(lifetime.Token); entered = true; Maximum(ref maxPool, Interlocked.Increment(ref inPool));
                await Task.Delay(milliseconds, lifetime.Token); Interlocked.Increment(ref completed); item.Completion.TrySetResult(true);
            } catch (OperationCanceledException) { Interlocked.Increment(ref canceled); item.Completion.TrySetCanceled(lifetime.Token); }
            finally { if (entered) { Interlocked.Decrement(ref inPool); pool.Release(); } Interlocked.Decrement(ref active); }
        }
    }
    private static void Maximum(ref int field, int value) { int old; do { old = Volatile.Read(ref field); if (old >= value) return; } while (Interlocked.CompareExchange(ref field, value, old) != old); }
    public object Snapshot() => new { active = Volatile.Read(ref active), queued = queue.Reader.Count, inPool = Volatile.Read(ref inPool), maxActive = Volatile.Read(ref maxActive), maxPool = Volatile.Read(ref maxPool), accepted = Volatile.Read(ref accepted), rejected = Volatile.Read(ref rejected), completed = Volatile.Read(ref completed), canceled = Volatile.Read(ref canceled) };
    public async Task StopAsync(CancellationToken token) {
        queue.Writer.TryComplete();
        try { await Task.WhenAll(running).WaitAsync(token); }
        catch (OperationCanceledException) when (token.IsCancellationRequested) { stop.Cancel(); Gate.Open(); try { await Task.WhenAll(running); } catch (OperationCanceledException) { } }
    }
    public void Dispose() { if (Interlocked.Exchange(ref disposed, 1) != 0) return; stop.Cancel(); pool.Dispose(); stop.Dispose(); }
}

using Microsoft.Data.Sqlite;
public sealed record LedgerDecision(bool Replay, bool Conflict, int Amount);
public sealed class LocalLedger {
    private readonly string connectionString;
    public LocalLedger(LabSettings settings) { connectionString = new SqliteConnectionStringBuilder { DataSource = settings.LedgerPath, DefaultTimeout = 2 }.ToString(); }
    private SqliteConnection Open() { var connection = new SqliteConnection(connectionString); connection.Open(); return connection; }
    public void Initialize() { using var connection = Open(); using var command = connection.CreateCommand(); command.CommandText = "CREATE TABLE IF NOT EXISTS receipts (request_key TEXT PRIMARY KEY, amount INTEGER NOT NULL)"; command.ExecuteNonQuery(); }
    public LedgerDecision Record(string key, int amount) {
        using var connection = Open(); using var transaction = connection.BeginTransaction();
        using var insert = connection.CreateCommand(); insert.Transaction = transaction;
        insert.CommandText = "INSERT INTO receipts(request_key, amount) VALUES ($key, $amount) ON CONFLICT(request_key) DO NOTHING";
        insert.Parameters.AddWithValue("$key", key); insert.Parameters.AddWithValue("$amount", amount); int changed = insert.ExecuteNonQuery();
        using var read = connection.CreateCommand(); read.Transaction = transaction; read.CommandText = "SELECT amount FROM receipts WHERE request_key = $key"; read.Parameters.AddWithValue("$key", key);
        int stored = Convert.ToInt32(read.ExecuteScalar(), System.Globalization.CultureInfo.InvariantCulture); transaction.Commit();
        return new LedgerDecision(changed == 0, stored != amount, stored);
    }
    public int Count() { using var connection = Open(); using var read = connection.CreateCommand(); read.CommandText = "SELECT COUNT(*) FROM receipts"; return Convert.ToInt32(read.ExecuteScalar(), System.Globalization.CultureInfo.InvariantCulture); }
}

public sealed class ClassroomCircuit {
    private readonly object sync = new(); private int failures, calls; private bool probe; private DateTimeOffset openUntil = DateTimeOffset.MinValue;
    public int DependencyCalls { get { lock (sync) return calls; } }
    public IResult Invoke(bool failing) {
        lock (sync) {
            DateTimeOffset now = DateTimeOffset.UtcNow;
            if (openUntil != DateTimeOffset.MinValue && now < openUntil) return Results.Json(new { state = "open", called = false }, statusCode: 503);
            if (probe) return Results.Json(new { state = "probe-busy", called = false }, statusCode: 503);
            bool halfOpen = openUntil != DateTimeOffset.MinValue; if (halfOpen) probe = true;
            calls++;
            // The dependency is an explicitly controlled in-process classroom outcome, not a real remote service.
            if (failing) {
                failures++; if (halfOpen || failures >= 2) openUntil = now.AddMilliseconds(250); probe = false;
                return Results.Json(new { state = openUntil == DateTimeOffset.MinValue ? "closed" : "open", called = true }, statusCode: 503);
            }
            failures = 0; openUntil = DateTimeOffset.MinValue; probe = false; return Results.Ok(new { state = "closed", called = true });
        }
    }
}

<Project Sdk="Microsoft.NET.Sdk"><PropertyGroup><OutputType>Exe</OutputType><TargetFramework>net10.0</TargetFramework><Nullable>enable</Nullable><ImplicitUsings>enable</ImplicitUsings></PropertyGroup></Project>

using System.Collections.Concurrent;
using System.Diagnostics;
using System.Globalization;
using System.Text.Json;
if (args.Length < 2) throw new ArgumentException("Usage: closed|open|retry|shard|hot|io|cpu|sync baseUrl[,baseUrl] [count<=32] [concurrency<=8] [intervalMs]");
string mode = args[0];
if (!new[] { "closed", "open", "retry", "shard", "hot", "io", "cpu", "sync" }.Contains(mode)) throw new ArgumentException("Unknown mode");
Uri[] targets = args[1].Split(',').Select(s => new Uri(s.TrimEnd('/') + "/")).ToArray();
if (targets.Length is < 1 or > 2 || targets.Any(u => !u.IsLoopback || u.Scheme != "http" || u.AbsolutePath != "/" || !string.IsNullOrEmpty(u.Query) || !string.IsNullOrEmpty(u.UserInfo))) throw new ArgumentException("Only one or two owned HTTP loopback roots are allowed");
int count = args.Length > 2 ? int.Parse(args[2], CultureInfo.InvariantCulture) : 16;
int concurrency = args.Length > 3 ? int.Parse(args[3], CultureInfo.InvariantCulture) : 4;
int interval = args.Length > 4 ? int.Parse(args[4], CultureInfo.InvariantCulture) : 2;
if (count is < 1 or > 32 || concurrency is < 1 or > 8 || interval is < 1 or > 100 || (mode == "shard" || mode == "hot") && targets.Length != 2) throw new ArgumentException("Bounded local settings required");
using var client = new HttpClient { Timeout = TimeSpan.FromSeconds(5) };
var latencies = new ConcurrentBag<double>(); var replicas = new ConcurrentDictionary<string, byte>(StringComparer.Ordinal);
int attempted = 0, successful = 0, rejected = 0, skipped = 0, logical = 0;
async Task Invoke(int index) {
    string key = mode == "hot" || index % 2 == 0 ? "A" : "B"; int partition = Partition(key);
    Uri target = targets[(mode == "shard" || mode == "hot") ? partition : index % targets.Length];
    string route = mode switch { "retry" => "unstable?key=demo", "shard" or "hot" => "partition/" + key + "?owner=" + partition, "io" => "io", "cpu" => "cpu", "sync" => "sync", _ => "work" };
    int maximumAttempts = mode == "retry" ? 2 : 1;
    for (int attempt = 1; attempt <= maximumAttempts; attempt++) {
        var begin = Stopwatch.GetTimestamp(); Interlocked.Increment(ref attempted);
        using var request = new HttpRequestMessage(HttpMethod.Get, new Uri(target, route));
        using var response = await client.SendAsync(request); string body = await response.Content.ReadAsStringAsync();
        latencies.Add(Stopwatch.GetElapsedTime(begin).TotalMilliseconds);
        if ((int)response.StatusCode == 503 && attempt < maximumAttempts) { await Task.Delay(10 + index % 7); continue; }
        if (response.IsSuccessStatusCode) {
            Interlocked.Increment(ref successful); using var doc = JsonDocument.Parse(body);
            if (doc.RootElement.TryGetProperty("replica", out var name)) replicas.TryAdd(name.GetString()!, 0);
        } else Interlocked.Increment(ref rejected);
        break;
    }
}
var elapsed = Stopwatch.StartNew();
if (mode == "open") {
    using var slots = new SemaphoreSlim(concurrency, concurrency); var running = new List<Task>();
    for (int i = 0; i < count; i++) {
        double wait = i * interval - elapsed.Elapsed.TotalMilliseconds; if (wait > 0) await Task.Delay(TimeSpan.FromMilliseconds(wait));
        Interlocked.Increment(ref logical);
        if (!slots.Wait(0)) { Interlocked.Increment(ref skipped); continue; }
        int index = i; running.Add(Release(index));
    }
    await Task.WhenAll(running);
    async Task Release(int index) { try { await Invoke(index); } finally { slots.Release(); } }
} else {
    int next = -1;
    await Task.WhenAll(Enumerable.Range(0, concurrency).Select(async _ => {
        while (true) { int index = Interlocked.Increment(ref next); if (index >= count) return; Interlocked.Increment(ref logical); await Invoke(index); }
    }));
}
elapsed.Stop(); double[] sorted = latencies.OrderBy(x => x).ToArray();
double Percentile(double probability) => sorted.Length == 0 ? 0 : sorted[(int)Math.Ceiling(probability * sorted.Length) - 1];
Console.WriteLine(JsonSerializer.Serialize(new { mode, scheduled = logical, attempted, successful, rejected, skipped, concurrency, elapsedMs = elapsed.Elapsed.TotalMilliseconds, p50Ms = Percentile(.5), p95Ms = Percentile(.95), p99Ms = Percentile(.99), replicas = replicas.Keys.OrderBy(x => x).ToArray() }));
static int Partition(string key) { uint hash = 2166136261; foreach (byte b in System.Text.Encoding.UTF8.GetBytes(key)) hash = unchecked((hash ^ b) * 16777619); return (int)(hash % 2); }

{
  "Lab": {
    "Replica": "one",
    "Workers": 2,
    "QueueCapacity": 3,
    "PoolSize": 2,
    "WorkMilliseconds": 20
  }
}