External .NET10 SDK10.0.403, real Kestrel and HTTP driver. Experiments are limited to owned loopback processes. SQLite10.0.14 is a local-file classroom ledger, not a network production database.
Read or download the complete SDK project, then follow its setup on your computer. The separate C# companion practice below runs in SharpRune’s .NET 9 console environment.
A deadline applies to elapsed ownership, including waiting
A request can consume time before its useful operation begins: admission, queueing, gate waits and pool waits all belong to an elapsed budget when the caller expects one deadline. Limiting only the final dependency call can leave an accepted item waiting indefinitely elsewhere. The external endpoint links its bounded budget with RequestAborted and passes that lifetime into the real queued work. The portable companion uses fixed integer observations and pre-cancellation to explain the policy without pretending to own a request, run a scheduler or measure elapsed time inside the browser console.
Separate arithmetic expiry from cooperative execution
The console subtracts ninety from deadline one hundred and prints remaining ten. A second fixed observation at one hundred ten is expired. Its already canceled token then produces cancelled. In the external lesson, the startup settings explicitly select fifty milliseconds of controlled work. A request budget of five milliseconds yields504 while its caller remains connected, and stats eventually records cooperative cancellation. A separate unbudgeted request succeeds. Those are actual process observations with variable scheduling time, not a guarantee that a timeout response appears at exactly five milliseconds or that the worker stopped on the same instruction.
Give every waiting stage the same relevant lifetime
The accepted item carries its token into the gate wait, semaphore acquisition and processing delay. The worker also links its own shutdown lifetime, so process stop can end work it owns. A caller can stop awaiting a queued item before a worker dequeues it; the bounded queue retains that item until processing observes cancellation. This is why eventual cleanup is checked separately from the response. Do not treat cancellation as automatic removal from every data structure. Each owner still needs a bounded resource policy and an implementation that notices the intended cancellation boundary.
A budget timeout and a disconnected caller are different observations
The endpoint returns its classroom504 problem only when its internal budget expires while RequestAborted has not already been requested. An aborted caller takes another cancellation path and may not be available to receive a response at all. Neither case reverses a committed database operation or an accepted remote side effect. This testbed performs a controlled delay rather than such a mutation, keeping the demonstration precise. A production adapter should decide which phases can be abandoned, which need reconciliation and whether an independently owned operation is required after an explicit commit boundary.
Practice: expire while queued rather than while processing
Close the main gate and occupy the configured workers before submitting another request with a small budget. Predict the timeout result and the queue's eventual cancellation observation. Open the gate afterward so every owned item can reach a defined end. Compare that result with a request whose work begins immediately under the same budget. The hint is that a queued item has already consumed elapsed time even if it owns no pool permit. Keep total traffic bounded and record the response and later worker cleanup separately rather than forcing both into one timing assertion.
Practice: distinguish caller abort from the application deadline
Run one controlled request with a long enough application budget, then close its client connection before work completes. Compare that with a connected request whose short application budget expires. State which response, if any, can be observed by each client and which worker counter should eventually change. Use separate requests and a fresh fixture when exact counts matter. Do not call a missing response a successful rollback test. The exercise concerns two cancellation owners reaching cooperative work, while any future remote side effect would need its own reconciliation and idempotency evidence.
Account for abandoned work instead of just abandoned waits
A useful deadline review identifies the owning source, elapsed boundary, downstream token propagation, remaining resource ownership and failure category visible to the caller. The portable arithmetic checks only a small policy example; real HTTP and worker observations establish that the external implementation cooperates. Preserve those different claims when adding more stages. A timeout that merely returns quickly while unlimited work continues would move the capacity problem out of the caller's view. This bounded testbed makes accepted items and eventual cancellation visible so returning504 is not the only evidence used to judge the operation's lifecycle.
- Include queue and pool time in an operation deadline when that is the stated contract.
- Observe eventual owned-work cleanup separately from the time the caller stops waiting.
Deadlines, queued work and cancellation — real bounded loopback testbed
Set up the SDK project
- Save the complete Server and Driver directories plus global.json into one new owned folder. Install official .NET10 SDK10.0.403.
- Restore both projects explicitly: dotnet restore Server/Lab.Server.csproj --source https://api.nuget.org/v3/index.json ; then dotnet restore Driver/Lab.Driver.csproj --source https://api.nuget.org/v3/index.json
- In terminal1 start the first owned server: ASPNETCORE_ENVIRONMENT=Development dotnet run --project Server/Lab.Server.csproj --no-launch-profile --urls http://127.0.0.1:5081 --Lab:Replica=one --Lab:LedgerPath=./owned-ledger.db --Lab:Workers=2 --Lab:QueueCapacity=4 --Lab:PoolSize=2 --Lab:WorkMilliseconds=50
- For two-replica experiments use terminal2 with port5082, Lab:Replica=two and Lab:Partition=1. Resolve both ledger paths to the same owned absolute file only when the lesson explicitly tests shared local coordination.
- Run drivers only against your owned loopback ports. Maximum32 logical operations, maximum8 in flight and at most2 retry attempts keep each run below100 requests. Outputs include variable measured timings.
- Stop only the processes you started using Ctrl+C in their own terminals; clean up only your owned fixture directory.
- GET /work?budgetMs=5 while WorkMilliseconds=50. Expect504, then poll /stats for a canceled item; compare an unbudgeted normal /work request.
Download complete SDK project ↗
Project files
Server/Lab.Server.csproj
Download file ↗<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>
Server/Program.cs
Download file ↗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); }
}
Server/WorkLane.cs
Download file ↗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(); }
}
Server/LocalLedger.cs
Download file ↗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); }
}
Server/ClassroomCircuit.cs
Download file ↗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 });
}
}
}
Driver/Lab.Driver.csproj
Download file ↗<Project Sdk="Microsoft.NET.Sdk"><PropertyGroup><OutputType>Exe</OutputType><TargetFramework>net10.0</TargetFramework><Nullable>enable</Nullable><ImplicitUsings>enable</ImplicitUsings></PropertyGroup></Project>
Driver/Program.cs
Download file ↗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); }
Server/appsettings.json
Download file ↗{
"Lab": {
"Replica": "one",
"Workers": 2,
"QueueCapacity": 4,
"PoolSize": 2,
"WorkMilliseconds": 50
}
}
Expected result
Budgeted work is504 and cooperative work is eventually canceled; an ordinary unbudgeted request succeeds. Exact cancellation timing is variable.
Use the listed SDK and dependencies for this project. Open C# practice runs the separate portable exercise.
C# companion practice
using System;
using System.Linq;
using System.Collections.Generic;
using System.Globalization;
using System.Threading;
using System.Threading.Tasks;
int deadline = 100; Console.WriteLine("remaining:" + Math.Max(0, deadline - 90)); Console.WriteLine(110 >= deadline ? "expired" : "active");
using var owner = new CancellationTokenSource(); owner.Cancel(); try { owner.Token.ThrowIfCancellationRequested(); } catch (OperationCanceledException) { Console.WriteLine("cancelled"); }
C# practice output
remaining:10 expired cancelled
Try this
Include queue and pool time in the same owned budget; cancellation is not rollback.
Include queue and pool time in the same owned budget; cancellation is not rollback.