From 15bf6985ddefac29b7a715c3984b79c94f65b01b Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sun, 6 Sep 2026 11:51:59 +0800 Subject: [PATCH 1/5] feat: add concurrent load generator for coordinator benchmarks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New 'client bench' command: submits a batch of deterministic jobs from several client connections at once (one CoordinatorClient per generator thread — the client is a one-request-in-flight blocking protocol), waits for every job to reach a terminal state, and reports throughput, submit-acknowledgement latency, and submit-to-observed-terminal latency as nearest-rank p50/p95/p99. An optional untimed warm-up batch runs first so the timed phase is not measuring a stone-cold JIT. End-to-end numbers are an upper bound (quantized by the 20 ms status-poll interval), and generator and system share one machine; this is a local engineering benchmark for comparing the system against itself under different configurations, not a rigorous performance benchmark. --- .../achrafaittayeb/dtp/client/Bench.java | 301 ++++++++++++++++++ .../achrafaittayeb/dtp/client/ClientMain.java | 34 ++ 2 files changed, 335 insertions(+) create mode 100644 client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java new file mode 100644 index 0000000..989bc88 --- /dev/null +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java @@ -0,0 +1,301 @@ +package io.github.achrafaittayeb.dtp.client; + +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; +import io.github.achrafaittayeb.dtp.common.model.JobSnapshot; +import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskType; + +import java.io.IOException; +import java.io.PrintStream; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +/** + * Concurrent load generator for the coordinator: submits a batch of + * deterministic jobs from several client connections at once, waits for every + * job to reach a terminal state, and reports throughput plus latency + * percentiles. + * + *

This is a local engineering benchmark, not a rigorous performance + * benchmark: generator and system share one machine, latencies are measured + * by request/response round-trips and status polling (end-to-end numbers are + * an upper bound, quantized by the poll interval), and JIT warm-up is only + * approximated by an optional warm-up batch. Its purpose is comparing this + * system against itself under different configurations, not producing + * absolute numbers. + * + *

Each generator thread owns a private {@link CoordinatorClient} + * connection — the client is a blocking one-request-in-flight protocol and is + * not meant to be shared across threads. + */ +public final class Bench { + + /** One benchmark run's parameters. */ + public record Config( + String host, + int port, + int jobs, + int concurrency, + int warmupJobs, + TaskType taskType, + ObjectNode payload, + long waitTimeoutMillis) { + + public Config { + if (jobs < 1) { + throw new IllegalArgumentException("jobs must be >= 1"); + } + if (concurrency < 1 || concurrency > jobs) { + throw new IllegalArgumentException("concurrency must be in 1..jobs"); + } + if (warmupJobs < 0) { + throw new IllegalArgumentException("warmup must be >= 0"); + } + } + } + + /** Nearest-rank latency percentiles over one measured phase, in milliseconds. */ + public record LatencyStats(long count, double p50, double p95, double p99, double max) { + + static LatencyStats of(List nanos) { + if (nanos.isEmpty()) { + return new LatencyStats(0, 0, 0, 0, 0); + } + long[] sorted = nanos.stream().mapToLong(Long::longValue).sorted().toArray(); + return new LatencyStats(sorted.length, + millis(percentile(sorted, 50)), + millis(percentile(sorted, 95)), + millis(percentile(sorted, 99)), + millis(sorted[sorted.length - 1])); + } + + /** Nearest-rank percentile: the smallest value with at least p% of samples at or below it. */ + private static long percentile(long[] sorted, int p) { + int rank = (int) Math.ceil(p / 100.0 * sorted.length); + return sorted[Math.max(0, rank - 1)]; + } + + private static double millis(long nanos) { + return nanos / 1_000_000.0; + } + } + + /** Aggregated outcome of a benchmark run. */ + public record Report( + Config config, + int accepted, + int rejected, + int completed, + int failed, + long elapsedMillis, + LatencyStats submitAck, + LatencyStats endToEnd) { + + /** Completed jobs per second over the whole timed phase (submit of first to finish of last). */ + public double throughputPerSecond() { + return elapsedMillis == 0 ? 0 : completed * 1000.0 / elapsedMillis; + } + + public void print(PrintStream out) { + out.printf("Benchmark finished in %.1f s%n", elapsedMillis / 1000.0); + out.printf(" Jobs: %d requested, %d accepted, %d rejected%n", + config.jobs(), accepted, rejected); + out.printf(" Outcomes: %d completed, %d failed%n", completed, failed); + out.printf(" Throughput: %.1f completed jobs/s%n", throughputPerSecond()); + out.printf(" Submit ack: p50 %.1f ms p95 %.1f ms p99 %.1f ms max %.1f ms%n", + submitAck.p50(), submitAck.p95(), submitAck.p99(), submitAck.max()); + out.printf(" End-to-end: p50 %.1f ms p95 %.1f ms p99 %.1f ms max %.1f ms%n", + endToEnd.p50(), endToEnd.p95(), endToEnd.p99(), endToEnd.max()); + out.println(" (end-to-end = submit to observed-terminal via polling; upper bound)"); + } + } + + private final Config config; + + public Bench(Config config) { + this.config = config; + } + + /** + * Runs warm-up (untimed, sequential, one connection) and then the timed + * phase: {@code concurrency} threads submit their share of the jobs and + * poll them to terminal state. + */ + public Report run() throws Exception { + if (config.warmupJobs() > 0) { + warmUp(); + } + + List generators = new ArrayList<>(); + CountDownLatch startGate = new CountDownLatch(1); + int perThread = config.jobs() / config.concurrency(); + int remainder = config.jobs() % config.concurrency(); + for (int i = 0; i < config.concurrency(); i++) { + int share = perThread + (i < remainder ? 1 : 0); + generators.add(new GeneratorThread(i, share, startGate)); + } + for (GeneratorThread generator : generators) { + generator.thread.start(); + } + + long startedAt = System.nanoTime(); + startGate.countDown(); + for (GeneratorThread generator : generators) { + generator.thread.join(); + } + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAt); + + int accepted = 0; + int rejected = 0; + int completed = 0; + int failed = 0; + List submitAckNanos = new ArrayList<>(); + List endToEndNanos = new ArrayList<>(); + Exception firstFailure = null; + for (GeneratorThread generator : generators) { + accepted += generator.accepted; + rejected += generator.rejected; + completed += generator.completed; + failed += generator.failed; + submitAckNanos.addAll(generator.submitAckNanos); + endToEndNanos.addAll(generator.endToEndNanos); + if (firstFailure == null && generator.failure != null) { + firstFailure = generator.failure; + } + } + if (firstFailure != null) { + throw firstFailure; + } + return new Report(config, accepted, rejected, completed, failed, elapsedMillis, + LatencyStats.of(submitAckNanos), LatencyStats.of(endToEndNanos)); + } + + /** Submits and awaits a small untimed batch so JIT and connections are not stone cold. */ + private void warmUp() throws Exception { + try (CoordinatorClient client = new CoordinatorClient(config.host(), config.port())) { + List jobIds = new ArrayList<>(); + for (int i = 0; i < config.warmupJobs(); i++) { + jobIds.add(client.submit(config.taskType(), config.payload(), 0)); + } + for (String jobId : jobIds) { + client.awaitTerminal(jobId, Duration.ofMillis(20), + Duration.ofMillis(config.waitTimeoutMillis())); + } + } + } + + /** + * One generator: submits its share of jobs on a private connection, + * recording per-job submit-acknowledgement latency, then polls its own + * jobs until all are terminal, recording submit-to-observed-terminal + * latency. + */ + private final class GeneratorThread { + + final Thread thread; + final List submitAckNanos = new ArrayList<>(); + final List endToEndNanos = new ArrayList<>(); + int accepted; + int rejected; + int completed; + int failed; + Exception failure; + + private final int jobsToSubmit; + private final CountDownLatch startGate; + + GeneratorThread(int index, int jobsToSubmit, CountDownLatch startGate) { + this.jobsToSubmit = jobsToSubmit; + this.startGate = startGate; + this.thread = new Thread(this::generate, "bench-" + index); + } + + private void generate() { + try (CoordinatorClient client = new CoordinatorClient(config.host(), config.port())) { + startGate.await(); + Deque pending = submitAll(client); + awaitAll(client, pending); + } catch (Exception e) { + failure = e; + } + } + + private Deque submitAll(CoordinatorClient client) throws IOException { + Deque pending = new ArrayDeque<>(); + for (int i = 0; i < jobsToSubmit; i++) { + long before = System.nanoTime(); + String jobId = client.submit(config.taskType(), config.payload(), 0); + submitAckNanos.add(System.nanoTime() - before); + accepted++; + pending.add(new PendingJob(jobId, before)); + } + return pending; + } + + /** Round-robin polls this thread's jobs until every one is terminal. */ + private void awaitAll(CoordinatorClient client, Deque pending) + throws IOException, InterruptedException { + long deadline = System.nanoTime() + + TimeUnit.MILLISECONDS.toNanos(config.waitTimeoutMillis()); + while (!pending.isEmpty()) { + if (System.nanoTime() > deadline) { + throw new IOException(pending.size() + " jobs still not terminal after " + + config.waitTimeoutMillis() + " ms"); + } + int stillPending = pending.size(); + for (int i = 0; i < stillPending; i++) { + PendingJob job = pending.removeFirst(); + JobSnapshot snapshot = client.status(job.jobId()); + if (snapshot.state().isTerminal()) { + endToEndNanos.add(System.nanoTime() - job.submittedAtNanos()); + if (snapshot.state() == JobState.COMPLETED) { + completed++; + } else { + failed++; + } + } else { + pending.addLast(job); + } + } + if (!pending.isEmpty()) { + Thread.sleep(20); + } + } + } + } + + private record PendingJob(String jobId, long submittedAtNanos) { + } + + /** Builds the deterministic payload for a benchmark task type. */ + public static ObjectNode payloadFor(TaskType taskType, long sleepMillis, long primeLimit) { + ObjectNode payload = JsonNodeFactory.instance.objectNode(); + switch (taskType) { + case SHA256 -> payload.put("text", "distributed-task-processor-bench"); + case SLEEP -> payload.put("durationMillis", sleepMillis); + case PRIME_COUNT -> payload.put("limit", primeLimit); + default -> throw new IllegalArgumentException( + "Unsupported bench task type: " + taskType + + " (use one of: sha256, sleep, prime-count)"); + } + return payload; + } + + /** Maps the CLI task name to a benchable {@link TaskType}. */ + public static TaskType taskTypeFor(String name) { + return switch (name) { + case "sha256" -> TaskType.SHA256; + case "sleep" -> TaskType.SLEEP; + case "prime-count" -> TaskType.PRIME_COUNT; + default -> throw new IllegalArgumentException( + "Unsupported bench task type: " + name + + " (use one of: sha256, sleep, prime-count)"); + }; + } +} diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java index a57eadd..4b96b7b 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java @@ -27,6 +27,7 @@ * client wait <job-id> * client list * client workers + * client bench --jobs 500 --concurrency 4 * * * Global options: {@code --host} (default localhost), {@code --port} (default 7071), @@ -65,6 +66,13 @@ private static void run(String[] rawArgs) throws Exception { String host = options.get("host", "localhost"); int port = options.getInt("port", 7071); + // bench manages its own connections (one per generator thread), so it + // does not go through the single shared client below. + if (command.equals("bench")) { + bench(host, port, options); + return; + } + try (CoordinatorClient client = new CoordinatorClient(host, port)) { switch (command) { case "submit" -> submit(client, positionals, options); @@ -120,6 +128,29 @@ private static void submit(CoordinatorClient client, List positionals, A System.out.println(" State: QUEUED"); } + private static void bench(String host, int port, Args options) throws Exception { + TaskType taskType; + Bench.Config config; + try { + taskType = Bench.taskTypeFor(options.get("task", "sha256")); + config = new Bench.Config( + host, port, + options.getInt("jobs", 200), + options.getInt("concurrency", 4), + options.getInt("warmup", 25), + taskType, + Bench.payloadFor(taskType, + options.getLong("sleep-millis", 1_000), + options.getLong("prime-limit", 100_000)), + options.getLong("wait-timeout-millis", 10 * 60 * 1000)); + } catch (IllegalArgumentException badOption) { + throw new UsageException(badOption.getMessage()); + } + System.out.printf("Benchmarking %s:%d — %d %s jobs, %d connections, %d warm-up jobs%n", + host, port, config.jobs(), taskType, config.concurrency(), config.warmupJobs()); + new Bench(config).run().print(System.out); + } + private static void printJob(JobSnapshot job) { System.out.println("Job " + job.jobId()); System.out.println(" Type: " + job.taskType()); @@ -214,6 +245,9 @@ private static void printUsage() { cancel list workers + bench [--jobs ] [--concurrency ] [--warmup ] + [--task sha256|sleep|prime-count] [--sleep-millis ] + [--prime-limit ] [--wait-timeout-millis ] Global options: --host (default localhost), --port (default 7071) Submit options: --max-attempts (default: coordinator setting)"""); } From 55d6a7bfbe3af441d726b267ce4de14e1ed9cf9b Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sun, 6 Sep 2026 12:01:01 +0800 Subject: [PATCH 2/5] docs: record baseline throughput and overload behavior Baseline measurements from the new load generator, before admission control, on an Apple M1 Pro (3 workers x capacity 4 = 12 slots): - Durable write-through costs ~5.4x throughput on cheap SHA256 jobs (896 jobs/s file-backed vs 4843 jobs/s in-memory): synchronous per-transition SQLite writes on the single core thread are the bottleneck, not the task work. - Under sustained overload (240x 500ms sleep jobs, 12 slots), the coordinator accepts every job in ~4ms while end-to-end latency climbs to ~5s median / ~10s tail. Submit ack stays flat as the queue grows: the client gets no backpressure signal and the job set grows unbounded. This is the limitation the admission-control change addresses. Records methodology and honest benchmark limitations. --- docs/MEASURED_BEHAVIOR.md | 109 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 docs/MEASURED_BEHAVIOR.md diff --git a/docs/MEASURED_BEHAVIOR.md b/docs/MEASURED_BEHAVIOR.md new file mode 100644 index 0000000..d231cce --- /dev/null +++ b/docs/MEASURED_BEHAVIOR.md @@ -0,0 +1,109 @@ +# Measured Behavior + +This document records what the system actually does under load, measured with +the built-in `client bench` load generator. It is written in two parts: the +**baseline** (the unbounded system, before admission control) and the +**post-change** results (after `--max-active-jobs` was added). The engineering +point of the exercise is not a big throughput number — it is turning *undefined* +overload behavior into *defined* overload behavior, and being able to show the +before/after. + +## What the benchmark is, honestly + +`client bench` is a **local engineering benchmark**, not a rigorous performance +benchmark. Read the numbers as "this system compared against itself under +different configurations", never as absolute capacity claims. Specifically: + +- The load generator and the system under test share one machine, so they + compete for the same cores. +- Submit-acknowledgement latency is a genuine request/response round-trip. + End-to-end latency (submit → observed terminal) is measured by **polling job + status every 20 ms**, so it is an *upper bound* quantized by that interval, + not a precise completion timestamp. +- Warm-up is approximated by an untimed batch of jobs before the timed phase, + so the timed phase is not measuring a stone-cold JIT. It is not a rigorous + steady-state warm-up. +- Percentiles are nearest-rank over the run's samples. No attempt is made to + correct for coordinated omission. + +Every number below was produced on the machine described next. Nothing is +fabricated or extrapolated. + +## Environment + +| | | +|---|---| +| Machine | Apple M1 Pro, 8 cores, 16 GB RAM | +| OS | macOS 15.7.2 | +| JDK | 23 (Temurin), project compiled `--release 21` | +| Coordinator | 1 process, default timings | +| Workers | 3 processes, capacity 4 each → **12 concurrent execution slots** | +| Client | `client bench`, connections = `--concurrency` | +| Persistence | SQLite file (durable) and `:memory:` (non-durable), as noted per run | + +Reproduce with (three workers of `--capacity 4`, then): + +```bash +client bench --task sha256 --jobs 2000 --concurrency 4 --warmup 50 +client bench --task sleep --sleep-millis 500 --jobs 240 --concurrency 8 --warmup 5 +``` + +## Baseline: the unbounded system + +### 1. Cost of durable write-through (SHA256, 2000 jobs, concurrency 4) + +`sha256` is a cheap, CPU-bound, deterministic task, so this run is dominated by +coordinator bookkeeping and persistence rather than task execution. + +| Persistence | Throughput | Submit ack p50 / p99 | End-to-end p50 / p99 | +|---|---|---|---| +| SQLite file (durable) | **896 jobs/s** | 4.2 / 6.4 ms | 1110 / 2142 ms | +| `:memory:` (non-durable) | **4843 jobs/s** | 0.5 / 2.2 ms | 165 / 341 ms | + +**Finding.** Durable write-through costs roughly **5.4× throughput** here (896 +vs 4843 jobs/s). Every job state transition is written through to SQLite +synchronously on the single coordinator core thread (QUEUED → RUNNING → +COMPLETED is several writes), and those synchronous writes — not the SHA-256 +work — are the bottleneck for cheap tasks. This is the price of the durability +guarantee that makes coordinator restart recovery possible, and it is a +deliberate, documented trade-off (see +[DESIGN_DECISIONS.md](DESIGN_DECISIONS.md) ADR 7). The comparison is valid +because *only* the repository implementation changes between the two runs; +everything else is identical. + +### 2. Overload behavior (sleep 500 ms, 240 jobs, concurrency 8) + +With 12 execution slots and 500 ms tasks, the system's sustained service rate is +about `12 / 0.5 s = 24 jobs/s`. Submitting 240 such jobs deliberately exceeds +what the workers can service concurrently, so the excess must go *somewhere*. + +| | Value | +|---|---| +| Accepted | 240 / 240 (nothing rejected) | +| Throughput | 23.6 jobs/s (≈ the 24 jobs/s slot ceiling, as expected) | +| **Submit ack** p50 / p99 | **3.9 / 19.1 ms** (fast and flat) | +| **End-to-end** p50 / p99 / max | **5040 / 10028 / 10039 ms** | + +**Finding — this is the limitation the milestone fixes.** The coordinator +accepts *all* 240 jobs almost instantly (submit ack stays around 4 ms), while +end-to-end latency climbs to ~5 s at the median and ~10 s at the tail. The two +numbers diverge because the excess work is silently buffered in the coordinator: + +- **The client gets no backpressure signal.** A fast "accepted" is returned + regardless of how deep the queue already is, so a client (or a retry storm, or + a burst of clients) can pile on unboundedly. The in-memory `jobs` map and the + SQLite table both grow with every accepted job. +- **End-to-end latency grows with queue depth**, not with the task itself. A + 500 ms task takes 10 s to finish once it is 20 waves deep. This is Little's + law in miniature: with arrival rate above service rate, the queue — and + therefore the wait — grows without bound until submission stops. +- **Nothing pushes back and nothing sheds load.** The only reason the run ended + at all is that the benchmark submitted a finite batch. A truly open-loop + source would drive latency and memory up indefinitely. + +The pathology is not "the system is slow" — 24 jobs/s is exactly the honest slot +ceiling. The pathology is that **overload has no defined behavior**: the system +neither refuses work nor signals that it is saturated. That is what admission +control changes. + + From cff60910dabd32f5d45ea7810e7ff4a16f83a3ac Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sun, 6 Sep 2026 12:10:01 +0800 Subject: [PATCH 3/5] feat: bound active jobs with typed admission rejection Add --max-active-jobs (default 10000): the coordinator holds at most this many active (non-terminal) jobs before rejecting new submissions, so it sheds load instead of buffering unboundedly. Accounting is O(1). The core keeps a running activeJobCount in its single-writer state rather than scanning the jobs map on every submit (which would make admission most expensive exactly when it is most needed). It is incremented once when a job enters the active set (a new submission or a non-terminal job recovered from storage) and decremented once when a job leaves it. Every active -> terminal transition funnels through one persistRetired() helper, so the decrement lives in exactly one place and a guard makes underflow impossible. Retries stay active (RUNNING -> RETRY_WAIT -> QUEUED) and never touch the counter. Because all of this runs on the existing single core thread, the check and the counter need no extra locking. Rejections use a new typed SUBMIT_REJECTED protocol message carrying reason, activeCount, limit, and retryable=true - deliberately distinct from ErrorReply so a client can tell "valid request, back off and retry" apart from "malformed request". CoordinatorClient surfaces it as a dedicated SubmitRejectedException; the CLI prints it and exits non-zero, and the load generator counts it as a rejection. Overload semantics: only new submissions are admission-controlled; existing jobs are never discarded; a recovered coordinator may sit above a since-lowered limit and simply rejects new work until the count falls back below it. --- .../achrafaittayeb/dtp/client/Bench.java | 14 ++- .../achrafaittayeb/dtp/client/ClientMain.java | 12 ++- .../dtp/client/CoordinatorClient.java | 14 ++- .../dtp/client/SubmitRejectedException.java | 37 ++++++++ .../dtp/common/protocol/Message.java | 3 +- .../dtp/common/protocol/SubmitRejected.java | 25 +++++ .../dtp/coordinator/CoordinatorConfig.java | 13 +++ .../dtp/coordinator/CoordinatorCore.java | 92 +++++++++++++++++-- .../dtp/coordinator/CoordinatorMain.java | 1 + .../dtp/coordinator/net/ClientServer.java | 10 +- .../dtp/it/CoordinatorRestartIT.java | 3 +- .../github/achrafaittayeb/dtp/it/Testbed.java | 9 ++ 12 files changed, 215 insertions(+), 18 deletions(-) create mode 100644 client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRejectedException.java create mode 100644 common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/SubmitRejected.java diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java index 989bc88..6e59be7 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java @@ -230,10 +230,16 @@ private Deque submitAll(CoordinatorClient client) throws IOException Deque pending = new ArrayDeque<>(); for (int i = 0; i < jobsToSubmit; i++) { long before = System.nanoTime(); - String jobId = client.submit(config.taskType(), config.payload(), 0); - submitAckNanos.add(System.nanoTime() - before); - accepted++; - pending.add(new PendingJob(jobId, before)); + try { + String jobId = client.submit(config.taskType(), config.payload(), 0); + // Submit-ack latency is recorded for accepted jobs only, so the + // percentiles describe the cost of admitting real work. + submitAckNanos.add(System.nanoTime() - before); + accepted++; + pending.add(new PendingJob(jobId, before)); + } catch (SubmitRejectedException overloaded) { + rejected++; + } } return pending; } diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java index 4b96b7b..7451b8d 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/ClientMain.java @@ -121,7 +121,17 @@ private static void submit(CoordinatorClient client, List positionals, A default -> throw new UsageException("Unknown task type: " + positionals.getFirst()); }; - String jobId = client.submit(taskType, payload, options.getInt("max-attempts", 0)); + String jobId; + try { + jobId = client.submit(taskType, payload, options.getInt("max-attempts", 0)); + } catch (SubmitRejectedException rejected) { + System.err.println("Submission rejected (coordinator overloaded): " + rejected.getMessage()); + System.err.println(" Active jobs: " + rejected.activeCount() + + " (limit " + rejected.limit() + ")"); + System.err.println(" Retryable: back off and submit again once load subsides."); + System.exit(3); + return; + } System.out.println("Job submitted"); System.out.println(" ID: " + jobId); System.out.println(" Type: " + taskType); diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java index dbf6866..f0277d3 100644 --- a/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/CoordinatorClient.java @@ -17,6 +17,7 @@ import io.github.achrafaittayeb.dtp.common.protocol.ListWorkers; import io.github.achrafaittayeb.dtp.common.protocol.Message; import io.github.achrafaittayeb.dtp.common.protocol.SubmitJob; +import io.github.achrafaittayeb.dtp.common.protocol.SubmitRejected; import io.github.achrafaittayeb.dtp.common.protocol.WorkerListReply; import java.io.IOException; @@ -44,13 +45,24 @@ public CoordinatorClient(String host, int port) throws IOException { this.out = socket.getOutputStream(); } - /** Submits a job; returns its id. {@code maxAttempts <= 0} uses the server default. */ + /** + * Submits a job; returns its id. {@code maxAttempts <= 0} uses the server + * default. + * + * @throws SubmitRejectedException if the coordinator is at its active-job + * limit (a valid request refused under overload — distinct from a + * malformed-request error, and safe to retry after a back-off) + * @throws IOException on any other error reply or transport failure + */ public synchronized String submit(TaskType taskType, JsonNode payload, int maxAttempts) throws IOException { Message reply = exchange(new SubmitJob(taskType, payload, maxAttempts)); if (reply instanceof JobSubmitted submitted) { return submitted.jobId(); } + if (reply instanceof SubmitRejected rejected) { + throw new SubmitRejectedException(rejected); + } throw asError(reply); } diff --git a/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRejectedException.java b/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRejectedException.java new file mode 100644 index 0000000..a5169b1 --- /dev/null +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/SubmitRejectedException.java @@ -0,0 +1,37 @@ +package io.github.achrafaittayeb.dtp.client; + +import io.github.achrafaittayeb.dtp.common.protocol.SubmitRejected; + +import java.io.IOException; + +/** + * Thrown by {@link CoordinatorClient#submit} when the coordinator rejected a + * well-formed submission because it is at its active-job limit. It extends + * {@link IOException} so it flows through existing {@code throws IOException} + * signatures, but callers that care can catch it specifically to tell + * "overloaded, back off and retry" apart from a malformed-request error. + */ +public final class SubmitRejectedException extends IOException { + + private final int activeCount; + private final int limit; + + public SubmitRejectedException(SubmitRejected rejected) { + super(rejected.reason()); + this.activeCount = rejected.activeCount(); + this.limit = rejected.limit(); + } + + public int activeCount() { + return activeCount; + } + + public int limit() { + return limit; + } + + /** Overload rejection is always a transient, retryable condition. */ + public boolean retryable() { + return true; + } +} diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/Message.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/Message.java index 8c9332e..78c2d28 100644 --- a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/Message.java +++ b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/Message.java @@ -31,6 +31,7 @@ @JsonSubTypes.Type(value = CancelJob.class, name = "CANCEL_JOB"), // Coordinator -> Client @JsonSubTypes.Type(value = JobSubmitted.class, name = "JOB_SUBMITTED"), + @JsonSubTypes.Type(value = SubmitRejected.class, name = "SUBMIT_REJECTED"), @JsonSubTypes.Type(value = JobStatusReply.class, name = "JOB_STATUS"), @JsonSubTypes.Type(value = JobListReply.class, name = "JOB_LIST"), @JsonSubTypes.Type(value = WorkerListReply.class, name = "WORKER_LIST"), @@ -42,6 +43,6 @@ public sealed interface Message permits WorkerRegister, Heartbeat, TaskResult, WorkerRegistered, TaskAssign, TaskCancel, SubmitJob, GetJobStatus, ListJobs, ListWorkers, CancelJob, - JobSubmitted, JobStatusReply, JobListReply, WorkerListReply, JobCancelReply, + JobSubmitted, SubmitRejected, JobStatusReply, JobListReply, WorkerListReply, JobCancelReply, ErrorReply { } diff --git a/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/SubmitRejected.java b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/SubmitRejected.java new file mode 100644 index 0000000..26c1b38 --- /dev/null +++ b/common/src/main/java/io/github/achrafaittayeb/dtp/common/protocol/SubmitRejected.java @@ -0,0 +1,25 @@ +package io.github.achrafaittayeb.dtp.common.protocol; + +/** + * Reply to {@link SubmitJob} when the coordinator refuses a valid + * request because it is at its configured active-job limit (overload / load + * shedding). This is deliberately a distinct message from {@link ErrorReply}: + * an {@code ErrorReply} means the request was malformed or otherwise wrong and + * must not be resent as-is, whereas a {@code SubmitRejected} means the request + * was fine and the client should back off and retry later. + * + * @param reason human-readable explanation, safe to show to users + * @param activeCount number of active (non-terminal) jobs at the moment of rejection + * @param limit the configured {@code --max-active-jobs} ceiling + * @param retryable always {@code true} — overload is a transient condition + */ +public record SubmitRejected(String reason, int activeCount, int limit, boolean retryable) + implements Message { + + /** Builds an overload rejection; overload is always retryable. */ + public static SubmitRejected overloaded(int activeCount, int limit) { + return new SubmitRejected( + "coordinator at capacity: " + activeCount + " active jobs (limit " + limit + ")", + activeCount, limit, true); + } +} diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java index 215f915..47be81c 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfig.java @@ -20,6 +20,7 @@ public record CoordinatorConfig( int defaultMaxAttempts, long retryBaseDelayMillis, long retryMaxDelayMillis, + int maxActiveJobs, String databasePath) { public static final int DEFAULT_WORKER_PORT = 7070; @@ -33,6 +34,14 @@ public record CoordinatorConfig( public static final int DEFAULT_MAX_ATTEMPTS = 3; public static final long DEFAULT_RETRY_BASE_DELAY_MILLIS = 1_000; public static final long DEFAULT_RETRY_MAX_DELAY_MILLIS = 30_000; + /** + * Admission ceiling: the coordinator holds at most this many active + * (non-terminal) jobs before it rejects new submissions. Chosen to sit + * comfortably above any hand-driven or demo workload while still bounding + * coordinator memory and per-sweep work on a single node; tune it down to + * observe load shedding. + */ + public static final int DEFAULT_MAX_ACTIVE_JOBS = 10_000; public static final String DEFAULT_DATABASE_PATH = "data/coordinator.db"; /** Path {@code :memory:} selects the non-durable in-memory repository. */ @@ -49,6 +58,7 @@ public static CoordinatorConfig fromArgs(String[] args) { parsed.getInt("max-attempts", DEFAULT_MAX_ATTEMPTS), parsed.getLong("retry-base-delay-millis", DEFAULT_RETRY_BASE_DELAY_MILLIS), parsed.getLong("retry-max-delay-millis", DEFAULT_RETRY_MAX_DELAY_MILLIS), + parsed.getInt("max-active-jobs", DEFAULT_MAX_ACTIVE_JOBS), parsed.get("database", DEFAULT_DATABASE_PATH)); } @@ -63,5 +73,8 @@ public static CoordinatorConfig fromArgs(String[] args) { if (defaultMaxAttempts < 1) { throw new IllegalArgumentException("max-attempts must be >= 1"); } + if (maxActiveJobs < 1) { + throw new IllegalArgumentException("max-active-jobs must be >= 1"); + } } } diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java index 068c5a5..0f16a40 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorCore.java @@ -58,6 +58,22 @@ public final class CoordinatorCore implements AutoCloseable { private final Map jobs = new HashMap<>(); private final ScheduledExecutorService coreThread; + /** + * Number of active (non-terminal) jobs: QUEUED + RETRY_WAIT + RUNNING. Kept + * as a running counter — never recomputed by scanning {@link #jobs} — so the + * admission check in {@link #submitJob} stays O(1) even under overload, when + * scanning would be most expensive. Confined to the core thread, so the + * increment/decrement need no synchronization: the single-writer design that + * already prevents double-assignment also makes this counter race-free. + * + *

Invariant: it is incremented exactly once when a job enters the active + * set (a new submission, or a non-terminal job recovered from storage) and + * decremented exactly once when a job leaves it (any active → terminal + * transition, funnelled through {@link #persistRetired}). Retries stay + * active (RUNNING → RETRY_WAIT → QUEUED) and never touch it. + */ + private int activeJobCount; + public CoordinatorCore(CoordinatorConfig config, JobRepository repository) { this.config = config; this.repository = repository; @@ -115,22 +131,54 @@ public void onTaskResult(TaskResult result) { // Requests from client connections (block until the core thread answers) // ------------------------------------------------------------------ - public String submitJob(TaskType taskType, JsonNode payload, int requestedMaxAttempts) + /** Outcome of a submission: either accepted (with a job id) or rejected because the coordinator is at its active-job limit. */ + public record SubmitOutcome(boolean accepted, String jobId, int activeCount, int limit) { + + static SubmitOutcome accepted(String jobId, int activeCount, int limit) { + return new SubmitOutcome(true, jobId, activeCount, limit); + } + + static SubmitOutcome rejected(int activeCount, int limit) { + return new SubmitOutcome(false, null, activeCount, limit); + } + } + + /** + * Admission point for new work. A malformed payload is rejected up front + * (before the core thread) as {@link InvalidPayloadException}. A well-formed + * request is admitted only if the active-job count is below the configured + * limit; otherwise it is shed with a {@link SubmitOutcome#rejected} outcome — + * the coordinator refuses new work rather than buffering it unboundedly. + * Only fresh submissions pass through here; retries never do. + */ + public SubmitOutcome submitJob(TaskType taskType, JsonNode payload, int requestedMaxAttempts) throws InvalidPayloadException { TaskPayloads.validate(taskType, payload); int maxAttempts = requestedMaxAttempts > 0 ? requestedMaxAttempts : config.defaultMaxAttempts(); return askCore(() -> { + int limit = config.maxActiveJobs(); + if (activeJobCount >= limit) { + log.warn("Job submission rejected (overloaded): activeJobs={} limit={} type={}", + activeJobCount, limit, taskType); + return SubmitOutcome.rejected(activeJobCount, limit); + } Job job = Job.createQueued(taskType, payload, maxAttempts, config.taskTimeoutMillis(), now()); jobs.put(job.id(), job); + activeJobCount++; repository.save(job); - log.info("Job submitted: jobId={} type={} maxAttempts={}", - job.id(), taskType, maxAttempts); + log.info("Job submitted: jobId={} type={} maxAttempts={} activeJobs={}/{}", + job.id(), taskType, maxAttempts, activeJobCount, limit); scheduleQueuedJobs(); - return job.id(); + return SubmitOutcome.accepted(job.id(), activeJobCount, limit); }); } + /** Observability/test hook: current number of active (non-terminal) jobs, read on the core thread. */ + public int activeJobCount() { + return askCore(() -> activeJobCount); + } + public Optional getJob(String jobId) { return askCore(() -> Optional.ofNullable(jobs.get(jobId)).map(Job::snapshot)); } @@ -156,7 +204,7 @@ public CancelOutcome cancelJob(String jobId) { String attemptId = job.currentAttemptId(); String workerId = job.assignedWorkerId(); job.cancel(now()); - repository.save(job); + persistRetired(job); revokeAttemptOnWorker(jobId, attemptId, workerId, "cancelled by client"); log.info("Job cancelled by client: jobId={} previousWorker={}", jobId, workerId); return new CancelOutcome(true, true, job.snapshot()); @@ -284,12 +332,14 @@ private void retryOrFail(Job job, String attemptError) { job.scheduleRetry(attemptError, now + delay, now); log.info("Retry scheduled: jobId={} attempt={}/{} delayMillis={} cause=\"{}\"", job.id(), job.attempts(), job.maxAttempts(), delay, attemptError); + // Still active (RETRY_WAIT): the active-job count is unchanged. + repository.save(job); } else { job.failPermanently(attemptError + " (all " + job.maxAttempts() + " attempts used)", now); log.warn("Job failed permanently: jobId={} attempts={} error=\"{}\"", job.id(), job.attempts(), attemptError); + persistRetired(job); } - repository.save(job); } private void promoteDueRetries() { @@ -356,7 +406,7 @@ private void applyTaskResult(TaskResult result) { } if (result.success()) { job.complete(result.result(), now()); - repository.save(job); + persistRetired(job); log.info("Job completed: jobId={} workerId={} attempts={}", job.id(), result.workerId(), job.attempts()); } else { @@ -397,8 +447,17 @@ private void recoverFromRepository() { repository.save(job); } jobs.put(job.id(), job); + // Seed the active-job counter from each job's post-recovery state, so + // admission control resumes with an exact count. A restart may leave + // activeJobCount above a since-lowered limit; that is intended — the + // recovered jobs run, and only new submissions are rejected until the + // count drops back below the limit. + if (!job.state().isTerminal()) { + activeJobCount++; + } } - log.info("Recovery completed: {} jobs loaded, {} requeued", stored.size(), requeued); + log.info("Recovery completed: {} jobs loaded, {} requeued, {} active", + stored.size(), requeued, activeJobCount); } // ------------------------------------------------------------------ @@ -409,6 +468,23 @@ private long now() { return System.currentTimeMillis(); } + /** + * The single place a job leaves the active set. Every active → terminal + * transition (completion, permanent failure, cancellation) is persisted + * through here so {@link #activeJobCount} is decremented exactly once, in one + * spot, regardless of which path retired the job. The guard makes underflow + * impossible: the counter can never go negative even if a transition were + * ever double-applied. + */ + private void persistRetired(Job job) { + if (activeJobCount > 0) { + activeJobCount--; + } else { + log.error("activeJobCount underflow prevented while retiring job {}", job.id()); + } + repository.save(job); + } + private void runOnCore(Runnable event) { // Once shutdown begins, no further state transitions are applied: a // graceful stop leaves the same durable state as a crash, so recovery diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorMain.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorMain.java index 8934bac..e847bfc 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorMain.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorMain.java @@ -29,6 +29,7 @@ public static void main(String[] args) throws Exception { --max-attempts default 3 --retry-base-delay-millis default 1000 --retry-max-delay-millis default 30000 + --max-active-jobs default 10000 (admission limit) --database default data/coordinator.db"""); System.exit(2); return; diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/ClientServer.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/ClientServer.java index a8ff639..3bf2b0d 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/ClientServer.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/net/ClientServer.java @@ -13,6 +13,7 @@ import io.github.achrafaittayeb.dtp.common.protocol.ListWorkers; import io.github.achrafaittayeb.dtp.common.protocol.Message; import io.github.achrafaittayeb.dtp.common.protocol.SubmitJob; +import io.github.achrafaittayeb.dtp.common.protocol.SubmitRejected; import io.github.achrafaittayeb.dtp.common.protocol.WorkerListReply; import io.github.achrafaittayeb.dtp.common.task.InvalidPayloadException; import io.github.achrafaittayeb.dtp.coordinator.CoordinatorCore; @@ -105,8 +106,13 @@ private void handleConnection(Socket socket) { private Message handle(Message request) { try { return switch (request) { - case SubmitJob submit -> new JobSubmitted( - core.submitJob(submit.taskType(), submit.payload(), submit.maxAttempts())); + case SubmitJob submit -> { + CoordinatorCore.SubmitOutcome outcome = core.submitJob( + submit.taskType(), submit.payload(), submit.maxAttempts()); + yield outcome.accepted() + ? new JobSubmitted(outcome.jobId()) + : SubmitRejected.overloaded(outcome.activeCount(), outcome.limit()); + } case GetJobStatus status -> core.getJob(status.jobId()) .map(JobStatusReply::new) .orElseGet(() -> new ErrorReply("Unknown job: " + status.jobId())); diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/CoordinatorRestartIT.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/CoordinatorRestartIT.java index f02f016..2640934 100644 --- a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/CoordinatorRestartIT.java +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/CoordinatorRestartIT.java @@ -173,7 +173,8 @@ void workerAutomaticallyReconnectsToRestartedCoordinatorOnSamePorts() throws Exc new io.github.achrafaittayeb.dtp.coordinator.CoordinatorConfig( workerPort, clientPort, Testbed.HEARTBEAT_TIMEOUT_MILLIS, Testbed.SWEEP_INTERVAL_MILLIS, - Testbed.TASK_TIMEOUT_MILLIS, 3, 50, 200, database))) { + Testbed.TASK_TIMEOUT_MILLIS, 3, 50, 200, + Testbed.UNLIMITED_ACTIVE_JOBS, database))) { second.start(); try (CoordinatorClient client = new CoordinatorClient("localhost", clientPort)) { Testbed.waitUntil("worker re-registered after restart", Testbed.TERMINAL_TIMEOUT, diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/Testbed.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/Testbed.java index a6fafc6..cac3ca1 100644 --- a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/Testbed.java +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/Testbed.java @@ -30,18 +30,27 @@ private Testbed() { } /** Coordinator on ephemeral ports; {@code database} may be {@code :memory:} or a temp file. */ + /** Large enough that admission control never fires unless a test asks it to. */ + static final int UNLIMITED_ACTIVE_JOBS = 1_000_000; + static Coordinator startCoordinator(String database, int maxAttempts) throws IOException { return startCoordinator(database, maxAttempts, TASK_TIMEOUT_MILLIS); } static Coordinator startCoordinator(String database, int maxAttempts, long taskTimeoutMillis) throws IOException { + return startCoordinator(database, maxAttempts, taskTimeoutMillis, UNLIMITED_ACTIVE_JOBS); + } + + static Coordinator startCoordinator(String database, int maxAttempts, long taskTimeoutMillis, + int maxActiveJobs) throws IOException { Coordinator coordinator = new Coordinator(new CoordinatorConfig( 0, 0, HEARTBEAT_TIMEOUT_MILLIS, SWEEP_INTERVAL_MILLIS, taskTimeoutMillis, maxAttempts, 50, 200, + maxActiveJobs, database)); coordinator.start(); return coordinator; From 47327659b47c56c341e5f8bf7c429bd4faa7d880 Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sun, 6 Sep 2026 12:17:32 +0800 Subject: [PATCH 4/5] test: cover admission limits, concurrency boundary, and recovery Adds 14 tests for the active-job admission counter: - SubmitRejected wire round-trip and type discriminator (common). - max-active-jobs config parsing and rejection of non-positive values. - Integration (AdmissionControlIT), using no-worker coordinators to pin the active set and a ScriptedWorker for exact retry timing: * accepts exactly up to the limit, then rejects with a typed SubmitRejectedException carrying activeCount/limit/retryable, and the rejection persists nothing; * completion, permanent failure, and cancellation each free one slot; * a retry (RUNNING -> RETRY_WAIT -> RUNNING) neither frees nor consumes an extra slot, and the slot is freed exactly once on completion; * 8 clients racing 80 submissions at a limit of 20 admit exactly 20 - the single-writer core serialises admission without extra locking; * recovery reconstructs the count from non-terminal jobs only, ignoring terminal ones; * recovery above a since-lowered limit keeps the recovered jobs, rejects new work, and resumes accepting once the backlog drains below the limit. Exposes Coordinator.activeJobCount() so tests (and operators) can read the counter directly. No timing-based performance assertions. --- .../protocol/MessageSerializationTest.java | 17 ++ .../dtp/coordinator/Coordinator.java | 5 + .../coordinator/CoordinatorConfigTest.java | 36 +++ .../dtp/it/AdmissionControlIT.java | 287 ++++++++++++++++++ .../achrafaittayeb/dtp/it/ScriptedWorker.java | 4 + 5 files changed, 349 insertions(+) create mode 100644 coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfigTest.java create mode 100644 integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/AdmissionControlIT.java diff --git a/common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java b/common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java index f7b0c2a..683f8a4 100644 --- a/common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java +++ b/common/src/test/java/io/github/achrafaittayeb/dtp/common/protocol/MessageSerializationTest.java @@ -82,4 +82,21 @@ void rejectsNonJsonPayload() { byte[] garbage = {0x00, 0x01, 0x7F, (byte) 0xFF}; assertThatThrownBy(() -> MessageIO.decode(garbage)).isInstanceOf(ProtocolException.class); } + + @Test + void roundTripsSubmitRejected() throws IOException { + SubmitRejected rejected = SubmitRejected.overloaded(10, 10); + SubmitRejected decoded = (SubmitRejected) roundTrip(rejected); + assertThat(decoded).isEqualTo(rejected); + assertThat(decoded.activeCount()).isEqualTo(10); + assertThat(decoded.limit()).isEqualTo(10); + assertThat(decoded.retryable()).isTrue(); + } + + @Test + void submitRejectedCarriesTypeDiscriminator() throws IOException { + String json = new String(MessageIO.encode(SubmitRejected.overloaded(3, 3)), + StandardCharsets.UTF_8); + assertThat(json).contains("\"type\":\"SUBMIT_REJECTED\""); + } } diff --git a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/Coordinator.java b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/Coordinator.java index 2f92b3e..45eb3b6 100644 --- a/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/Coordinator.java +++ b/coordinator/src/main/java/io/github/achrafaittayeb/dtp/coordinator/Coordinator.java @@ -55,6 +55,11 @@ public int clientPort() { return clientServer.port(); } + /** Current number of active (non-terminal) jobs; the admission-control counter. */ + public int activeJobCount() { + return core.activeJobCount(); + } + @Override public void close() { // Stop the brain first: after this, disconnect events from closing diff --git a/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfigTest.java b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfigTest.java new file mode 100644 index 0000000..330731c --- /dev/null +++ b/coordinator/src/test/java/io/github/achrafaittayeb/dtp/coordinator/CoordinatorConfigTest.java @@ -0,0 +1,36 @@ +package io.github.achrafaittayeb.dtp.coordinator; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class CoordinatorConfigTest { + + @Test + void defaultsMaxActiveJobs() { + CoordinatorConfig config = CoordinatorConfig.fromArgs(new String[0]); + assertThat(config.maxActiveJobs()).isEqualTo(CoordinatorConfig.DEFAULT_MAX_ACTIVE_JOBS); + } + + @Test + void parsesMaxActiveJobsFlag() { + CoordinatorConfig config = CoordinatorConfig.fromArgs( + new String[]{"--max-active-jobs", "250"}); + assertThat(config.maxActiveJobs()).isEqualTo(250); + } + + @Test + void rejectsZeroMaxActiveJobs() { + assertThatThrownBy(() -> CoordinatorConfig.fromArgs(new String[]{"--max-active-jobs", "0"})) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("max-active-jobs"); + } + + @Test + void rejectsNegativeMaxActiveJobs() { + assertThatThrownBy(() -> CoordinatorConfig.fromArgs(new String[]{"--max-active-jobs", "-1"})) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("max-active-jobs"); + } +} diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/AdmissionControlIT.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/AdmissionControlIT.java new file mode 100644 index 0000000..9a9ad5a --- /dev/null +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/AdmissionControlIT.java @@ -0,0 +1,287 @@ +package io.github.achrafaittayeb.dtp.it; + +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import io.github.achrafaittayeb.dtp.client.CoordinatorClient; +import io.github.achrafaittayeb.dtp.client.SubmitRejectedException; +import io.github.achrafaittayeb.dtp.common.model.JobState; +import io.github.achrafaittayeb.dtp.common.model.TaskType; +import io.github.achrafaittayeb.dtp.common.protocol.TaskAssign; +import io.github.achrafaittayeb.dtp.coordinator.Coordinator; +import io.github.achrafaittayeb.dtp.coordinator.CoordinatorConfig; +import io.github.achrafaittayeb.dtp.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Scenario G: admission control. The coordinator holds at most + * {@code --max-active-jobs} active (non-terminal) jobs; further submissions are + * rejected with a typed {@code SUBMIT_REJECTED}, and the O(1) active-job counter + * stays exact across completion, failure, cancellation, retries, and recovery. + * + *

Most tests run with no workers, so submitted jobs sit in QUEUED + * (active) indefinitely — that pins the active set to exactly what was submitted + * and removes execution timing from the assertions. + */ +class AdmissionControlIT { + + @TempDir + Path tempDir; + + private static String submitSha(CoordinatorClient client) throws IOException { + return client.submit(TaskType.SHA256, + JsonNodeFactory.instance.objectNode().put("text", "x"), 0); + } + + /** Attempts one submission; returns true if it was rejected as overloaded. */ + private static boolean rejected(CoordinatorClient client) throws IOException { + try { + submitSha(client); + return false; + } catch (SubmitRejectedException overloaded) { + return true; + } + } + + @Test + void acceptsUpToLimitThenRejectsWithTypedReply() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 5); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + for (int i = 0; i < 5; i++) { + submitSha(client); // no worker → stays QUEUED (active) + } + assertThat(coordinator.activeJobCount()).isEqualTo(5); + + // The 6th submission is a valid request refused under overload. + assertThatThrownBy(() -> submitSha(client)) + .isInstanceOf(SubmitRejectedException.class) + .satisfies(thrown -> { + SubmitRejectedException e = (SubmitRejectedException) thrown; + assertThat(e.activeCount()).isEqualTo(5); + assertThat(e.limit()).isEqualTo(5); + assertThat(e.retryable()).isTrue(); + }); + + // Rejection must not have created or persisted anything. + assertThat(coordinator.activeJobCount()).isEqualTo(5); + assertThat(client.listJobs()).hasSize(5); + } + } + + @Test + void completionFreesCapacity() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 1); + Worker worker = Testbed.startWorker(coordinator, "worker-1", 2); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String first = submitSha(client); + assertThat(client.awaitTerminal(first, Testbed.POLL, Testbed.TERMINAL_TIMEOUT).state()) + .isEqualTo(JobState.COMPLETED); + assertThat(coordinator.activeJobCount()).isZero(); + + // Slot freed → a new submission is accepted. + String second = submitSha(client); + assertThat(client.awaitTerminal(second, Testbed.POLL, Testbed.TERMINAL_TIMEOUT).state()) + .isEqualTo(JobState.COMPLETED); + } + } + + @Test + void permanentFailureFreesCapacity() throws Exception { + // maxAttempts 1 so the first failure is terminal. + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 1, Testbed.TASK_TIMEOUT_MILLIS, 1); + Worker worker = Testbed.startWorker(coordinator, "worker-1", 2); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String failing = client.submit(TaskType.FAIL, + JsonNodeFactory.instance.objectNode().put("failUntilAttempt", 999), 0); + assertThat(client.awaitTerminal(failing, Testbed.POLL, Testbed.TERMINAL_TIMEOUT).state()) + .isEqualTo(JobState.FAILED); + assertThat(coordinator.activeJobCount()).isZero(); + + assertThat(rejected(client)).isFalse(); // capacity is free again + } + } + + @Test + void cancellationFreesCapacity() throws Exception { + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 1); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String queued = submitSha(client); // no worker → QUEUED (active) + assertThat(coordinator.activeJobCount()).isEqualTo(1); + assertThat(rejected(client)).isTrue(); // at limit + + client.cancel(queued); // synchronous → CANCELLED, counter freed + assertThat(coordinator.activeJobCount()).isZero(); + + assertThat(rejected(client)).isFalse(); // capacity available again + } + } + + @Test + void retryDoesNotFreeOrConsumeAnExtraSlot() throws Exception { + // A ScriptedWorker gives exact control over when the attempt fails and + // when the retry succeeds, so the counter can be sampled at known points. + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, 1); + ScriptedWorker worker = ScriptedWorker.register(coordinator.workerPort(), "w1", 2); + CoordinatorClient client = Testbed.connectClient(coordinator)) { + + String jobId = submitSha(client); + TaskAssign attempt1 = worker.awaitAssignment(); // RUNNING, 1 active + assertThat(coordinator.activeJobCount()).isEqualTo(1); + assertThat(rejected(client)).isTrue(); // single slot held while RUNNING + + worker.sendFailure(attempt1, "boom"); // RUNNING → RETRY_WAIT → requeued + + // awaitAssignment blocks until the coordinator has processed the + // failure and reassigned the retry, so no transient state needs to be + // caught. The active count is 1 in every non-terminal state, so the + // retry neither freed the slot (would drop to 0) nor consumed an extra + // one (would rise to 2). + TaskAssign attempt2 = worker.awaitAssignment(); + assertThat(attempt2.attemptNumber()).isEqualTo(2); + assertThat(coordinator.activeJobCount()).isEqualTo(1); + assertThat(rejected(client)).isTrue(); // still the same single slot + + worker.sendSuccess(attempt2, "ok"); // → COMPLETED + Testbed.waitUntil("job completed", Testbed.TERMINAL_TIMEOUT, + () -> uncheckedState(client, jobId) == JobState.COMPLETED); + assertThat(coordinator.activeJobCount()).isZero(); // freed exactly once + } + } + + @Test + void concurrentClientsAtBoundaryAdmitExactlyTheLimit() throws Exception { + int limit = 20; + int clients = 8; + int perClient = 10; // 80 attempts against 20 slots, no workers + AtomicInteger accepted = new AtomicInteger(); + AtomicInteger refused = new AtomicInteger(); + List unexpected = new CopyOnWriteArrayList<>(); + + try (Coordinator coordinator = Testbed.startCoordinator( + CoordinatorConfig.IN_MEMORY_DATABASE, 3, Testbed.TASK_TIMEOUT_MILLIS, limit)) { + + CountDownLatch startGate = new CountDownLatch(1); + List threads = new java.util.ArrayList<>(); + for (int c = 0; c < clients; c++) { + Thread thread = new Thread(() -> { + try (CoordinatorClient client = Testbed.connectClient(coordinator)) { + startGate.await(); + for (int i = 0; i < perClient; i++) { + if (rejected(client)) { + refused.incrementAndGet(); + } else { + accepted.incrementAndGet(); + } + } + } catch (Exception e) { + unexpected.add(e); + } + }, "submitter-" + c); + threads.add(thread); + thread.start(); + } + startGate.countDown(); + for (Thread thread : threads) { + thread.join(); + } + + assertThat(unexpected).isEmpty(); + assertThat(accepted).hasValue(limit); + assertThat(refused).hasValue(clients * perClient - limit); + assertThat(coordinator.activeJobCount()).isEqualTo(limit); + } + } + + @Test + void recoveryReconstructsActiveCountIgnoringTerminalJobs() throws Exception { + String database = tempDir.resolve("admission-recovery.db").toString(); + + // First lifetime: 2 jobs completed (terminal), 3 left QUEUED (active). + try (Coordinator first = Testbed.startCoordinator(database, 3, Testbed.TASK_TIMEOUT_MILLIS, 100)) { + try (Worker worker = Testbed.startWorker(first, "worker-1", 2); + CoordinatorClient client = Testbed.connectClient(first)) { + for (int i = 0; i < 2; i++) { + String id = submitSha(client); + assertThat(client.awaitTerminal(id, Testbed.POLL, Testbed.TERMINAL_TIMEOUT).state()) + .isEqualTo(JobState.COMPLETED); + } + } + // Worker is now gone; these three stay QUEUED (active). + try (CoordinatorClient client = Testbed.connectClient(first)) { + for (int i = 0; i < 3; i++) { + submitSha(client); + } + assertThat(first.activeJobCount()).isEqualTo(3); + } + first.close(); + } + + // Second lifetime on the same database: only the 3 non-terminal jobs count. + try (Coordinator second = Testbed.startCoordinator(database, 3, Testbed.TASK_TIMEOUT_MILLIS, 100)) { + assertThat(second.activeJobCount()).isEqualTo(3); + } + } + + @Test + void recoveryAboveLoweredLimitRejectsNewWorkUntilDrained() throws Exception { + String database = tempDir.resolve("admission-overage.db").toString(); + + // First lifetime: 5 active jobs under a generous limit, no workers. + try (Coordinator first = Testbed.startCoordinator(database, 3, Testbed.TASK_TIMEOUT_MILLIS, 5)) { + try (CoordinatorClient client = Testbed.connectClient(first)) { + for (int i = 0; i < 5; i++) { + submitSha(client); + } + assertThat(first.activeJobCount()).isEqualTo(5); + } + first.close(); + } + + // Restart with a lowered limit of 2: the 5 recovered jobs remain, but new + // submissions are rejected until enough recovered jobs drain below 2. + try (Coordinator second = Testbed.startCoordinator(database, 3, Testbed.TASK_TIMEOUT_MILLIS, 2)) { + assertThat(second.activeJobCount()).isEqualTo(5); // above the new limit + + try (CoordinatorClient client = Testbed.connectClient(second)) { + assertThat(client.listJobs()).hasSize(5); // recovered work kept + assertThat(rejected(client)).isTrue(); // new work refused (5 >= 2) + } + + // Attaching a worker drains the recovered jobs; once the active count + // drops below the limit, submissions are accepted again. + try (Worker worker = Testbed.startWorker(second, "worker-1", 4)) { + Testbed.waitUntil("recovered jobs drained below the lowered limit", + Testbed.TERMINAL_TIMEOUT, () -> second.activeJobCount() < 2); + try (CoordinatorClient client = Testbed.connectClient(second)) { + assertThat(rejected(client)).isFalse(); + } + } + } + } + + private static JobState uncheckedState(CoordinatorClient client, String jobId) { + try { + return client.status(jobId).state(); + } catch (IOException e) { + throw new RuntimeException(e); + } + } +} diff --git a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/ScriptedWorker.java b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/ScriptedWorker.java index 4392ead..b05b6f8 100644 --- a/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/ScriptedWorker.java +++ b/integration-tests/src/test/java/io/github/achrafaittayeb/dtp/it/ScriptedWorker.java @@ -123,6 +123,10 @@ void sendSuccess(TaskAssign assign, String result) throws IOException { sendResult(TaskResult.success(workerId, assign.jobId(), assign.attemptId(), result)); } + void sendFailure(TaskAssign assign, String error) throws IOException { + sendResult(TaskResult.failure(workerId, assign.jobId(), assign.attemptId(), error)); + } + void sendResult(TaskResult result) throws IOException { synchronized (out) { MessageIO.send(out, result); From 8b16d313881c47c67074723b94db0c21745bbfc6 Mon Sep 17 00:00:00 2001 From: Achraf Ait Tayeb <2023521460132@stu.scu.edu.cn> Date: Sun, 6 Sep 2026 12:25:17 +0800 Subject: [PATCH 5/5] docs: document overload control and measured behavior MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MEASURED_BEHAVIOR.md: post-change section. Same overload load, bounded to --max-active-jobs 24, drops end-to-end p99 from ~10s to ~1s while shedding 216 of 240 as typed rejections; throughput unchanged (~23 jobs/s). Healthy SHA256 runs confirm the O(1) check adds no measurable overhead below the limit (896->897 file, 4843->4751 memory). - README: bench command, admission-control / overload semantics with the measured before/after, an honest-semantics bullet, updated limitations, and a MEASURED_BEHAVIOR link. - PROTOCOL.md: SUBMIT_REJECTED message and why it is distinct from ERROR. - DESIGN_DECISIONS.md: ADR 14 — reject-new over buffer-forever / block / drop-accepted, typed rejection, and O(1) race-free accounting on the single-writer thread. - FAILURE_MODEL.md: overload as a defined, bounded condition. - ARCHITECTURE.md: admission control among the core-thread components. --- README.md | 78 ++++++++++++++++++++++++++++++++++++--- docs/ARCHITECTURE.md | 7 ++++ docs/DESIGN_DECISIONS.md | 47 +++++++++++++++++++++++ docs/FAILURE_MODEL.md | 24 ++++++++++++ docs/MEASURED_BEHAVIOR.md | 75 ++++++++++++++++++++++++++++++++++++- docs/PROTOCOL.md | 12 +++++- 6 files changed, 236 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index d06c72c..deff3ec 100644 --- a/README.md +++ b/README.md @@ -64,8 +64,8 @@ flowchart LR heartbeat thread, five built-in task types. - **client** — CLI (`submit`, `status`, `wait`, `list`, `workers`) and a small reusable client library. -- **integration-tests** — 18 end-to-end tests, including worker failure, - stale results, coordinator restart, and concurrency safety. +- **integration-tests** — end-to-end tests, including worker failure, + stale results, coordinator restart, admission control, and concurrency safety. ### Job lifecycle @@ -131,6 +131,7 @@ well under a minute. ./scripts/client.sh cancel ./scripts/client.sh list ./scripts/client.sh workers +./scripts/client.sh bench --jobs 2000 --concurrency 4 # load generator ``` Task types: `sleep --milliseconds N`, `word-count --text "..."`, @@ -151,6 +152,62 @@ longer sleep: ./scripts/client.sh submit sleep --milliseconds 20000 # exceeds the 3s deadline ``` +## Load generation and overload control + +The client includes a small concurrent load generator for measuring the system +against itself: + +```bash +./scripts/client.sh bench --jobs 2000 --concurrency 4 --task sha256 +``` + +It submits the requested jobs across `--concurrency` independent connections +(one blocking client per thread), waits for every job to reach a terminal state, +and reports throughput, submit-acknowledgement latency, and end-to-end +submit-to-terminal latency as p50/p95/p99, plus accepted/rejected counts. It is +a **local engineering benchmark** for comparing configurations, not a rigorous +performance benchmark — see [docs/MEASURED_BEHAVIOR.md](docs/MEASURED_BEHAVIOR.md) +for methodology, honest limitations, and the measured numbers below. + +**Admission control (backpressure).** The coordinator holds at most +`--max-active-jobs` active (non-terminal: `QUEUED` + `RETRY_WAIT` + `RUNNING`) +jobs, defaulting to 10 000. A submission that arrives at the limit is **rejected +with a typed `SUBMIT_REJECTED` reply** (distinct from a malformed-request error, +and marked retryable) rather than being buffered — the coordinator sheds excess +load instead of letting its queue and memory grow without bound. The active-job +count is an O(1) counter maintained inside the single-writer core, so the +admission check stays cheap even under heavy load. Overload semantics: + +- Only **new** submissions are admission-controlled; work already in the system + is never discarded. +- **Retries never re-enter admission control** — a retrying job is already + counted as active and stays counted across `RUNNING → RETRY_WAIT → RUNNING`. +- Completion, permanent failure, and cancellation each free exactly one slot. +- After a restart the coordinator may hold **more active jobs than a + since-lowered limit**; the recovered jobs run to completion normally, and only + *new* submissions are rejected until the count falls back below the limit. + +Watch it shed load with no workers running (jobs stay `QUEUED`, so they stay +active): + +```bash +./scripts/coordinator.sh --max-active-jobs 3 +./scripts/client.sh submit sleep --milliseconds 60000 # x4; the 4th is rejected +``` + +**Measured overload behavior** (Apple M1 Pro, 3 workers × capacity 4 = 12 slots, +240× 500 ms sleep jobs from 8 connections; full methodology in the doc): + +| | Unbounded (baseline) | `--max-active-jobs 24` | +|---|---|---| +| Accepted / rejected | 240 / 0 | 24 / 216 | +| End-to-end latency p99 | ~10.0 s | ~1.0 s | + +The change does not raise peak throughput (that was always the honest 12-slot +ceiling of ~24 jobs/s); it makes overload **defined** — admitted work has +bounded latency, and excess load is refused explicitly instead of silently +queued. + ## The failure-recovery demo ```bash @@ -238,7 +295,11 @@ containers. auto-reconnect), execution-deadline timeout on a healthy worker with reassignment, client cancellation of queued and running jobs, a 60-jobs/4-clients/5-workers scheduling race test asserting exactly one - assignment per job, and malformed-bytes robustness on both ports. + assignment per job, malformed-bytes robustness on both ports, and admission + control (accept-to-limit then typed rejection, slot accounting across + completion / failure / cancellation / retry, an 8-client race admitting + exactly the limit, and recovery of the active-job count including above a + lowered limit). ## Semantics, honestly stated @@ -260,6 +321,11 @@ containers. immediately and ask the worker to stop via thread interruption; a task that ignores interruption keeps running but can no longer affect job state. This is not guaranteed preemption, and it does not make execution exactly-once. +- **Overload is shed, not absorbed.** Beyond `--max-active-jobs` the + coordinator refuses new submissions with a typed, retryable `SUBMIT_REJECTED` + rather than queueing them. This bounds the coordinator's memory and the + latency of admitted work; it does *not* raise throughput, and it applies only + to new work — jobs already in the system always run to a terminal state. - **Single coordinator.** The coordinator is a single point of failure; durability (not availability) is what restart recovery provides. Multiple coordinators would require leader election/consensus — out of scope, by @@ -275,8 +341,9 @@ More detail in [docs/FAILURE_MODEL.md](docs/FAILURE_MODEL.md) and - One coordinator (see above); leader election is the natural next step. - Scheduling is FIFO / least-loaded; no priorities, deadlines, or fairness. - Results live in the job row; large results would need external storage. -- No backpressure or bounded submission queue yet; the job set grows - unboundedly. +- Admission is a single global active-job limit (`--max-active-jobs`); there is + no per-client fairness or priority in what gets shed, and the client does not + yet auto-retry rejections (it surfaces them for the caller to handle). - Cancellation is cooperative (thread interruption), not hard preemption: a task that ignores interruption keeps running until it finishes, though its lease is already revoked so its result is discarded. @@ -287,3 +354,4 @@ More detail in [docs/FAILURE_MODEL.md](docs/FAILURE_MODEL.md) and - [docs/PROTOCOL.md](docs/PROTOCOL.md) — framing, message schemas, error handling - [docs/FAILURE_MODEL.md](docs/FAILURE_MODEL.md) — failure scenarios and guarantees - [docs/DESIGN_DECISIONS.md](docs/DESIGN_DECISIONS.md) — ADR-style rationale +- [docs/MEASURED_BEHAVIOR.md](docs/MEASURED_BEHAVIOR.md) — load-generator methodology and measured throughput / overload behavior diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d703366..68da0a7 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -67,6 +67,13 @@ DESIGN_DECISIONS. - **Attempt leases** — `Job.assignTo` issues a fresh UUID per assignment; `applyTaskResult` accepts a result only if the job is RUNNING *and* the lease matches. Everything else is logged as stale and dropped. +- **Admission control** — an O(1) `activeJobCount` (non-terminal jobs) kept in + the core state. `submitJob` admits a new job only if the count is below + `max-active-jobs`, otherwise returns a typed `SUBMIT_REJECTED`; the count is + incremented on admission and decremented through the single `persistRetired` + chokepoint on any active→terminal transition. Because it lives on the + single-writer thread, the check-and-increment is atomic with no extra locking + — the same property that makes scheduling race-free. ### Persistence and recovery diff --git a/docs/DESIGN_DECISIONS.md b/docs/DESIGN_DECISIONS.md index 929b890..ce1200d 100644 --- a/docs/DESIGN_DECISIONS.md +++ b/docs/DESIGN_DECISIONS.md @@ -160,3 +160,50 @@ What this explicitly does **not** buy: it is not exactly-once (a timed-out task that actually finished still causes a retry — the at-least-once window is unchanged), and it is not a real-time guarantee (detection latency is bounded only by the sweep interval). + +## 14. Bounded admission (reject-new over buffer-forever) + +Before this decision the coordinator accepted every well-formed submission. The +[measured baseline](MEASURED_BEHAVIOR.md) showed the consequence: under a +sustained overload the submit acknowledgement stayed ~4 ms while end-to-end +latency climbed to ~10 s, because excess work was silently buffered in an +in-memory map and a growing SQLite table. The coordinator had no *defined* +overload behavior. `--max-active-jobs` gives it one: at the limit, new +submissions are rejected. + +- **Reject new, don't buffer forever.** An unbounded queue converts overload + into unbounded latency and unbounded memory (Little's law: with arrival rate + above service rate, the backlog grows without limit). Shedding the excess + keeps admitted work's latency bounded and the coordinator's footprint bounded. + The measured result is a ~10× lower overload tail on admitted jobs, with the + overflow surfaced as explicit rejections instead of hidden delay. +- **Why not block the client instead?** Blocking the submit call until a slot + frees just moves the unbounded queue into the clients' threads and hides the + saturation behind apparent slowness. An explicit, typed, retryable rejection + puts the backpressure decision where the client can see and act on it (retry + with back-off, shed, or route elsewhere). +- **Why not drop already-accepted work?** Admission bounds *intake*; it never + discards work the system already promised to run. Dropping accepted jobs would + break the durability and at-least-once guarantees the rest of the design is + built on. So only new submissions are gated, and retries — which are + already-accepted work — never re-enter admission control. +- **A typed message, not `ERROR`.** Overload rejection is retryable; a + malformed request is not. Collapsing them into one `ERROR` would force clients + to parse strings to tell "back off and retry" from "your request is broken". + `SUBMIT_REJECTED` carries `activeCount`/`limit`/`retryable` so the distinction + is structural. +- **O(1), and race-free for free.** The active-job count is a single integer + maintained in the core's state, incremented when a job enters the active set + and decremented through one `persistRetired` chokepoint when it leaves — never + recomputed by scanning the job map, which would make admission most expensive + exactly under the load it is meant to protect against. Because every state + mutation already runs on the single-writer core thread (decision 3), the + check-and-increment needs no lock and cannot race: the same design that + prevents double-assignment also makes "admit iff below the limit" atomic + without any new machinery. The 8-client boundary test admits *exactly* the + limit for this reason. + +What this explicitly does **not** buy: it is not fairness (a single global limit +with no per-client quota or priority in what gets shed) and it does not raise +throughput (the service rate is still the workers' slot ceiling) — the point is +*defined* overload behavior, not a faster system. diff --git a/docs/FAILURE_MODEL.md b/docs/FAILURE_MODEL.md index dee6ac3..1a3fbca 100644 --- a/docs/FAILURE_MODEL.md +++ b/docs/FAILURE_MODEL.md @@ -132,6 +132,30 @@ recovery discarded the old lease. While the coordinator is down the system is unavailable — see below. +## Overload (more work offered than the system can run) + +Overload is a *defined* condition, not an implicit degradation. The coordinator +holds at most `--max-active-jobs` active (non-terminal) jobs; a submission that +arrives at the limit is rejected with a typed, retryable `SUBMIT_REJECTED` +rather than being queued (see [DESIGN_DECISIONS.md](DESIGN_DECISIONS.md) 14 and +the measured before/after in [MEASURED_BEHAVIOR.md](MEASURED_BEHAVIOR.md)). + +- **Bounded, not unbounded.** Without a limit, offering work faster than the + workers can drain it grows the queue — and therefore both memory and the + latency of every admitted job — without bound. The limit caps the active set, + so admitted work has bounded latency and the coordinator has bounded memory. +- **Existing work is safe.** Only new submissions are shed. Jobs already + accepted always run to a terminal state, and retries never re-enter admission + control, so overload can never cause the system to drop or fail work it had + already taken on. +- **Recovery above a lowered limit.** If the coordinator restarts onto a + database holding more active jobs than a since-lowered limit, those jobs run + normally; only new submissions are rejected, until enough recovered jobs reach + a terminal state to bring the active count back below the limit. +- **Not a fairness or QoS mechanism.** The limit is a single global count with + no per-client quota and no priority in what gets shed. It bounds load; it does + not arbitrate between competing clients. + ## What is *not* tolerated - **Coordinator permanent loss**: single coordinator by design. The database diff --git a/docs/MEASURED_BEHAVIOR.md b/docs/MEASURED_BEHAVIOR.md index d231cce..7270348 100644 --- a/docs/MEASURED_BEHAVIOR.md +++ b/docs/MEASURED_BEHAVIOR.md @@ -106,4 +106,77 @@ ceiling. The pathology is that **overload has no defined behavior**: the system neither refuses work nor signals that it is saturated. That is what admission control changes. - +## Admission control: the change + +`--max-active-jobs N` caps the number of active (non-terminal) jobs the +coordinator will hold. A submission that arrives when the active-job count is +already at the limit is refused with a typed `SUBMIT_REJECTED` reply +(`retryable = true`) instead of being queued. The count is an O(1) running +counter kept inside the single-writer core — it is never recomputed by scanning +the job map, so the admission check does not itself get more expensive as load +rises. See [DESIGN_DECISIONS.md](DESIGN_DECISIONS.md) ADR 14 for the reject-new +rationale. + +## Post-change: the same overload, now bounded + +Same offered load as the baseline overload run — 240× 500 ms sleep jobs from 8 +connections, 12 execution slots — but with `--max-active-jobs 24` (twice the +slot count, so a full complement can run with an equal number queued): + +| | Baseline (unbounded) | Bounded (`--max-active-jobs 24`) | +|---|---|---| +| Accepted | 240 | **24** | +| Rejected | 0 | **216** | +| Completed throughput | 23.6 jobs/s | 22.8 jobs/s | +| Submit ack p50 / p99 | 3.9 / 19.1 ms | 7.7 / 13.1 ms | +| **End-to-end p50 / p99 / max** | **5040 / 10028 / 10039 ms** | **537 / 1033 / 1033 ms** | + +**Interpretation.** + +- **Overload latency is now bounded.** End-to-end p99 for admitted work fell + from ~10 s to ~1 s — roughly a 10× reduction — and, more importantly, it is + now *bounded by design*: with at most 24 active jobs over 12 slots, no job + waits behind more than about one extra wave, so the tail sits near + `(24 / 12) × 500 ms ≈ 1 s`, exactly as measured. In the unbounded run the tail + grew with the size of the submitted burst; here it cannot. +- **The system did not get faster, and that is the point.** Completed + throughput is essentially unchanged (22.8 vs 23.6 jobs/s) — it was always the + honest 12-slot ceiling. What changed is that excess load is now *shed + explicitly* (216 typed rejections a client can act on) instead of being + silently absorbed into an ever-growing queue. The intended result of this + milestone is **defined overload behavior**, not higher peak throughput. +- **Backpressure reaches the client.** In the baseline the client had no way to + know the system was saturated — every submit returned "accepted" in ~4 ms. + Now a saturated coordinator says so, and says it in a way the client can + distinguish from a malformed-request error and safely retry after a back-off. + +### Admission control is free when the limit is not hit + +Re-running the healthy SHA256 throughput scenarios with the default limit +(10 000, far above the 2000-job workload) shows the O(1) counter check adds no +measurable cost: + +| Persistence | Baseline throughput | Post-change throughput | +|---|---|---| +| SQLite file (durable) | 896 jobs/s | 897 jobs/s | +| `:memory:` (non-durable) | 4843 jobs/s | 4751 jobs/s | + +The differences are within run-to-run noise (the small in-memory dip is not +reproducible in direction across runs). Admission control costs one integer +comparison per submission on a thread that is already serializing all state +changes, so below the limit it is effectively invisible. + +## Benchmark limitations (recap) + +- Local, single-machine, generator and system co-resident — treat all absolute + numbers as this-machine-only. +- End-to-end latency is polled at 20 ms granularity, so it is an upper bound. +- Percentiles are nearest-rank with no coordinated-omission correction. +- Warm-up is approximate. Runs are short (hundreds to a few thousand jobs). +- The overload demonstration deliberately floods a small, fixed batch; a + true open-loop load source would make the *unbounded* case worse (unbounded + latency and memory), which only sharpens the contrast — it does not change the + bounded case. + +The value here is the **before/after contrast under identical offered load**, +produced by the same tool against the same system, not the raw magnitudes. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 2fe11f5..70d99ba 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -58,13 +58,23 @@ connection. | Request | Reply | Notes | |---|---|---| -| `SUBMIT_JOB` (`taskType`, `payload`, `maxAttempts`) | `JOB_SUBMITTED` (`jobId`) | `maxAttempts ≤ 0` → server default; payload validated before the job exists | +| `SUBMIT_JOB` (`taskType`, `payload`, `maxAttempts`) | `JOB_SUBMITTED` (`jobId`) or `SUBMIT_REJECTED` | `maxAttempts ≤ 0` → server default; payload validated before the job exists; rejected if the coordinator is at its active-job limit (see below) | | `GET_JOB_STATUS` (`jobId`) | `JOB_STATUS` (job snapshot) | unknown id → `ERROR` | | `LIST_JOBS` | `JOB_LIST` (snapshots, newest first) | | | `LIST_WORKERS` | `WORKER_LIST` (live workers) | | | `CANCEL_JOB` (`jobId`) | `JOB_CANCEL` (`cancelled`, job snapshot) | `cancelled=true` if this call moved it to CANCELLED; `false` if already terminal (snapshot shows real state); unknown id → `ERROR` | | any invalid | `ERROR` (`message`) | malformed frames get a best-effort `ERROR`, then close | +**`SUBMIT_REJECTED` (`reason`, `activeCount`, `limit`, `retryable`)** is the +reply to a *well-formed* `SUBMIT_JOB` that the coordinator refuses because it is +already holding `--max-active-jobs` active (non-terminal) jobs. It is a distinct +message from `ERROR` on purpose: `ERROR` means "your request was wrong, do not +resend it as-is", whereas `SUBMIT_REJECTED` means "your request was fine, the +coordinator is overloaded — back off and retry". `retryable` is always `true`; +`activeCount` and `limit` let the client report or adapt. Only new submissions +are gated this way; retries of already-accepted jobs never pass through +admission control. + A job snapshot contains: `jobId`, `taskType`, `state` (one of QUEUED, RUNNING, RETRY_WAIT, COMPLETED, FAILED, CANCELLED), `attempts`, `maxAttempts`, `workerId` (only while RUNNING), `result`, `error` (last attempt's error, kept