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/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..6e59be7 --- /dev/null +++ b/client/src/main/java/io/github/achrafaittayeb/dtp/client/Bench.java @@ -0,0 +1,307 @@ +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(); + 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; + } + + /** 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..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 @@ -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); @@ -113,13 +121,46 @@ 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); 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 +255,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)"""); } 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/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/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/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/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 new file mode 100644 index 0000000..7270348 --- /dev/null +++ b/docs/MEASURED_BEHAVIOR.md @@ -0,0 +1,182 @@ +# 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. + +## 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 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/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/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); 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;