diff --git a/.agent/rules/distributed-compilation.md b/.agent/rules/distributed-compilation.md index fa9054746..f2425ea84 100644 --- a/.agent/rules/distributed-compilation.md +++ b/.agent/rules/distributed-compilation.md @@ -2178,7 +2178,8 @@ symptom was that they were slow. and a learner granted none, so only a broken client sends one there. - **A node running NO consensus follows no leader**, and answers as the `pace-kv` daemon does: the capacity verbs (`Register`, `Heartbeat`, `Withdraw`, `Lease`, `Release`, - `NodeAnnounce`) `DispatchNotPermitted`, the cluster verbs `NoCluster` — the daemon reaching + `NodeAnnounce`, and the admission verbs `Admit`, `Settle`, `MachineBudget`) + `DispatchNotPermitted`, the cluster verbs `NoCluster` — the daemon reaching `NodeAnnounce`'s code through `RefusalFor`'s fallback. The codes agree on every scheduling verb, and `NodeFrameSurface_test` pins that agreement on the wire; the words are each binary's own. @@ -2373,6 +2374,13 @@ Six more about what the tier IS and who gets to see it: and that is correct rather than the same defect: a node offering no slots has a capacity of zero, which is a reading rather than an unmade measurement. The two fields answer different questions and the same word would be wrong in one of them. + **"No worker" is not "no machine" for the fleet's compile queue (#1728, R6)**: every + node, a `--slots=0` one included, states its machine's compile budget + (`MachineBudgetOf`: `--machine-slots`, `--machine-memory-budget`) on the presence loop's + round, which runs everywhere. It is a budget record and not a worker registration -- such + a node still registers nothing and is never sent a dispatched compile -- and its flags are + deliberately NOT worker-column rows, or the generated rule above would refuse them on + exactly the node that needs them. **The worker is `WorkerTier`, null on such a node** (#1387): no idle pool thread, no capacity sized to zero, no validator. It owns the `SchedulerLink`, so a worker cannot exist without one and a start that reaches it with no `--scheduler` is refused by @@ -3774,9 +3782,11 @@ cluster's pre-shared key instead; #178 PR 6 made it prove WHICH machine it is, b its own identity key, and sealed every frame after the proof. Each rule below is what some plausible simpler design gets wrong. -**No address admits another machine, and the verbs a machine JOINS with need a proof.** They -- `Register`, `NodeAnnounce`, `Heartbeat`, `Withdraw` -- decide where the fleet's work -is sent, and they are `IdentityRequirement::ProvenNodeOnly`, a COLUMN of `OpTable` pinned by -`JoiningVerbsNeedAnIdentity`, refused `NodeIdentityRequired` and counted +**No address admits another machine, and the verbs a machine JOINS with need a proof.** They -- +`Register`, `NodeAnnounce`, `Heartbeat`, `Withdraw` -- decide where the fleet's work is sent, and +`MachineBudget` decides how much work a machine is ADMITTED (the memory budget the leader charges +it against, #1728); all five are `IdentityRequirement::ProvenNodeOnly`, a COLUMN of `OpTable` +pinned by `JoiningVerbsNeedAnIdentity`, refused `NodeIdentityRequired` and counted (`SchedulerRequestsRefusedNodeIdentityRequired`). Loopback is NOT exempt: a process on the scheduler's own host could otherwise register an endpoint of its choosing and be leased the fleet's jobs. Every node naming `--scheduler` therefore needs a key to prove -- and every node @@ -4034,6 +4044,95 @@ here asserts first, with the addition beside it only to stop a change that satis half from passing. The in-process fleet asserts it from addresses nobody listed (`FleetTickets_test`, `FrameEndpoint_test`'s relabelled-peer cases). +## The compile queue (#1728) + + + +A build with more compiler processes than the machine has memory for was killed by the kernel +rather than slowed. The leader now holds a queue in front of every machine's compile budget. +Each rule below is a shape a plausible simpler design gets wrong, and none of them fails a build. + +- **A miss is ADMITTED (queued) at the leader, never refused for capacity.** A refusal sends the + client to a local compile, which is exactly the load the queue exists to bound. Waiting is the + product; the leader pulses (`Status::Progress`, `Admit` being the second verb allowed it) so a + long queue reads as alive against the client's sliding idle bound. A request larger than a whole + machine's budget waits for an idle machine and runs ALONE; it is never starved and never refused. + A `NoRoom` Local grant is an ordinary queue decision, not a refusal, and is not counted as one. +- **The connection is the lease.** A grant is charged to the connection that asked, carries no + token to expire, and is released by `SETTLE` on that connection or by its close. A close is an + abandonment, counted apart (`AdmissionAbandoned`); a `kill -9` of ninja must give the memory back + at once with no timer. The only frame a held connection may send next is its `SETTLE`, or, after + `Settle{WorkerRefused}`, a fresh `ADMIT` that binds to the SAME ticket (Ruling 9). +- **A worker refusal keeps the ticket's place and count.** The leader re-parks the same ticket at + the FRONT; after `MaxWorkerRefusals` the grant is Local. Abandoning it instead would lose the + front position and let a flapping worker starve the head of the queue. +- **A worker FAILURE settles `NotStored`, never abandon**: an + abandon raises the ESTIMATE of a translation unit that merely met a bad worker, and feeding that to the + estimator poisons every later admission of it. +- **Deduplication is woken by STORE, never by a release.** Only a `storable` compile that shares a + cache with its peers waits on another's key, and it is answered `Stored` when the object is + STORED. A release wakes nobody: a compile that failed stored nothing, and the waiters must then + compile. +- **Every node states its machine budget, `--slots=0` included, and that is not a worker + registration.** `MachineBudget` rides the presence loop's round, which runs everywhere; its flags + (`--machine-slots`, `--machine-memory-budget`) are deliberately not worker-column rows. +- **A failover admits nothing on a machine before that machine's budget report arrives**, and the + ledger charges `max(charged, held)`: the new leader knows only what the machine says it holds, + and an admission before that report double-books memory a surviving compile still has. +- **Every non-leader forwards `Admit`/`Settle` inline to the leader, never to itself**: a voting + follower as well as a learner. A voter that answered `NotLeader` would leave a launcher with no + queue on every node but one. The relay holds one upstream connection per client connection. +- **The grant says what it granted**: `workerKey`, `lifetime` and `dialHint` as `LeaseGrant` does + -- the hint from the same `DialHintFor`, so a worker reachable only where it was last seen is + dialled there whichever path granted it -- and a `reason` on a Local grant of a dispatchable + request, so a fleet that never distributes still says why (`DeclineDispatch`) instead of reading + green. +- **An admission's worker lease is filed under its TICKET, never under the cache key** (Ruling 17). + Deduplication is the QUEUE's, and only where a waiter can fetch the result; a lease table keyed by + the cache key refused the second of two machines that missed one key `AlreadyInFlight`, which its + launcher read as "compile here" -- an uncharged compile, the load the queue exists to bound. The + token still signs the OBJECT key, as a `Lease` grant's does: the table's key is bookkeeping the + worker never sees. +- **A launcher asks the queue at its OWN node, `PACE_ADDR`, whatever `PACE_SCHEDULER` names** (Ruling + 16, `Cc::AdmissionEndpointOf`). The leader charges a compile to the machine that ASKED, and names + it only by a key the connection PROVED or by loopback: the launcher's node proves its machine's key + when it relays, while a launcher dialling a remote scheduler directly proves nothing and would be + charged to no machine. `PACE_SCHEDULER` keeps one meaning, where a `LEASE` goes when no queue + answers, and a plan is refused for a remembered-unreachable endpoint only when EVERY place it could + be spent was unreached (`Cc::PlanUnreachableAt`). +- **A launcher fails OPEN only where no queue ANSWERED.** An unreachable, refusing or unreadable + node costs the build nothing, and a launcher whose FETCH already found the node unreachable does + not dial it again: on Windows a refused loopback connect can take the whole connect budget. The + one compile with no such evidence is a passthrough whose command line did not parse -- it makes + no FETCH -- so with the node stopped and the reachability memo empty it pays one connect, up to a + second on Windows, and records it; a stopped node costs a build at most one such connect per + fifteen-second memo window, never one per compile. Where the queue DID answer and the launcher + gives up on it anyway -- `MaxWorkerRefusals` worker refusals in a row -- the compile is asked for + again `LocalOnly`, excluding that worker, and runs HERE charged, never in the open. +- **A held admission connection carries TCP keepalive at BOTH ends** (Ruling 18, #1736): the + launcher's dial of its node (`Cc::DefaultAdmitBudget`) and the node's relay to the leader + (`Node::ForwardDialBudget`) arm it per dial, as `wire-and-protocol.md` requires, and the leader arms + it on the ACCEPTED connection once that connection holds a grant, through core-cpp's + `ISocket::armKeepAlive` (v0.11.0). The intervals are core-cpp's `KeepAliveSettings {}`, which every + `KeepAlive::Yes` dial applies -- a dial takes none of its own -- and ONE constant names them for all + three, `Pace::AdmitKeepAliveProbe` (`Pace/Protocol/AdmitKeepAlive.hpp`), pinned by a `static_assert`: + a peer whose host vanished is noticed in `idle + count * interval` = **16 s** on Linux and macOS and + **30 s** on Windows, where the OS fixes the count at 10. That bounds the relay noticing a vanished + LEADER (failing its launcher open, giving back its own held charge), a launcher noticing a vanished + node, and the leader noticing a vanished caller -- whose Worker grant would otherwise hold another + machine's budget until `MaxCompileLeaseLifetime` (one hour) sweeps it, which stays as the backstop. + - **What arms is a COLUMN of what is held**, `IConnectionHold::KeepAlive()`, pure so a new hold + answers rather than inherits a silence; `FrameServer::ArmHeldKeepAlive` asks it on the loop thread + (which `armKeepAlive` requires) once a reply that left a hold is WRITTEN, and a queued request + that holds nothing is not armed. Once per CONNECTION (`ConnectionScope::NoteKeepAliveArmed`), so a + requeue on the same socket is not a second request. A failure to arm is a `Warn`, never a + refusal: the sweep is still there. + - **Every pace socket that wraps an `ISocket` forwards `armKeepAlive`** (`SealedFrameSocket`, + `TotalDeadlineSocket`, and the test fake `RelabelledPeerSocket`), or the call lands on the + interface default, `Unsupported` -- `SetReceiveDeadline`'s defect class, one rule higher. The two + production wrappers each have a case asserting the request reaches the socket under them; the fake's + is the proven-leader forwarding case, whose ledger sits UNDER it. + ## Open work @@ -4082,3 +4181,7 @@ half from passing. The in-process fleet asserts it from addresses nobody listed it can be spawned and once for its banner, and the first is in a serial loop in front of the pool built to hide exactly that. `CompilerBanner` knows both facts and reports neither, so two callers reconstruct what it discarded. +- **[#1730](https://github.com/LASTRADA-Software/pace/issues/1730)** — one cache key is not + deduplicated across machines that share a cache, so the same compile can be admitted twice. +- **[#1731](https://github.com/LASTRADA-Software/pace/issues/1731)** — admission test gaps + deferred by the #1728 reviews. diff --git a/.agent/rules/testing.md b/.agent/rules/testing.md index 01f2b0950..ec3ce4d11 100644 --- a/.agent/rules/testing.md +++ b/.agent/rules/testing.md @@ -3094,3 +3094,10 @@ passed. does not judge whether a phrase is the right tripwire. Marked `untriaged:` and not `none:`, and `ctest -R rulebook-tripwires` prints the count against this issue on every run. + +- **[#1735](https://github.com/LASTRADA-Software/pace/issues/1735)** — a test that uses a link must + verify a link can be followed here, and SKIP otherwise (Windows error 448). +- **[#1733](https://github.com/LASTRADA-Software/pace/issues/1733)** — the FileTrust test hard-codes + the English LOCAL SERVICE account name. +- **[#1734](https://github.com/LASTRADA-Software/pace/issues/1734)** — `python3` resolves to the + Windows Store stub on this host. diff --git a/.agent/rules/wire-and-protocol.md b/.agent/rules/wire-and-protocol.md index 271ec43f6..2f85eae2f 100644 --- a/.agent/rules/wire-and-protocol.md +++ b/.agent/rules/wire-and-protocol.md @@ -52,7 +52,8 @@ Every rule below has already been a bug. replay, an oversized or unframed header) closes UNANSWERED, because there the frame is not known to be the peer's at all. - **A reply is one frame per request, and `Status::Progress` is the first of two exceptions -- - bounded to exactly one verb, carrying nothing, and paid for with a version step.** The + bounded to exactly the verbs a table names (`Compile`, and `Admit` since #1728), carrying + nothing, and paid for with a version step.** The second is `Status::Push`, below, bounded the same way. A dispatched compile is bounded by how long a COMPILER runs, so its client's total deadline is minutes by construction and was therefore also how long a worker whose @@ -72,7 +73,9 @@ Every rule below has already been a bug. "The framing already lets a receiver step over what it does not know" is the tempting reason not to bump, and it is a fact about REQUESTS. - **Which verbs may be answered with it is a column of `OpTable`**, and - `ProgressIsCompileOnly()` is `static_assert`ed. Every other verb on this wire is + `ProgressVerbsAre({Compile, Admit})` is `static_assert`ed (`Admit` joined at version 16, + #1728: a queue waits as long as it is long, and a pulse is what tells that from a dead leader). + Every other verb on this wire is answered from a table in microseconds, so a pulse on one would be a frame no client could observe and a second reply shape every reader would have to handle for nothing -- and a mask widened by hand would give it to them by omission. @@ -853,7 +856,9 @@ Every rule below has already been a bug. link is torn down, fleet-wide, for a change nobody asked for. Same rule as SIGPIPE above, reached from the option side rather than the signal side: per socket, never process-wide. `DialOptions::keepAlive` says so at the call, `Detail::ArmKeepAlive` - applies it, and `SocketAddress_test` asserts the socket is **still not probing** + applies it, and an ACCEPTED socket has the second door: `ISocket::armKeepAlive`, per socket, + called on the loop thread by the server once it decides the connection is long-lived (#1736); + `SocketAddress_test` asserts the socket is **still not probing** after `ApplyHotSocketOptions` — the negative half, which is the assertion that fails when somebody later folds the two together for tidiness ([#247](https://github.com/LASTRADA-Software/pace/issues/247)). Four things diff --git a/AGENT.md b/AGENT.md index cb9d04e41..13c6f5429 100644 --- a/AGENT.md +++ b/AGENT.md @@ -421,6 +421,12 @@ launcher's cache key is made of. Before `apps/pace-cc/`, `CompileCache/`. - A cordon is the worker PROCESS's state, never replicated and never persisted, so a restart lifts it. Stop, cordon and full are ONE locked decision with the slot take (`SlotAdmission`), and a compile's slot rides its reply (`IReplyHold`) until the endpoint has WRITTEN it. +- A miss is ADMITTED (queued) at the fleet's leader, never refused for capacity (#1728): the + connection is the lease (a close abandons, `SETTLE` releases), a worker refusal keeps the ticket's + place, dedup is woken by STORE, and a worker failure settles `NotStored`. +- Every node sends its `MachineBudget`, `--slots=0` included, and that is not a worker registration; + a failover admits nothing on a machine before its report, charging `max(charged, held)`. +- Every non-leader, voter or learner, forwards `Admit`/`Settle` inline to the leader, never to itself. - A heartbeat age is a duration on a report, never a `TimePoint` on `WorkerInfo` — a raw instant breaks every `ManualClock` test. - A CoW store file is claimed exclusively at `Open` and a second opener is refused by name diff --git a/CMakeLists.txt b/CMakeLists.txt index bd9dc7eea..56bf48b41 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -533,8 +533,8 @@ endif() CPMAddPackage( NAME core-cpp GITHUB_REPOSITORY contour-terminal/core-cpp - GIT_TAG v0.6.0 - VERSION 0.6.0 + GIT_TAG v0.11.0 + VERSION 0.11.0 SYSTEM YES EXCLUDE_FROM_ALL YES OPTIONS diff --git a/docs/getting-started/distributed-compilation.md b/docs/getting-started/distributed-compilation.md index 5545f276d..dad0c03d5 100644 --- a/docs/getting-started/distributed-compilation.md +++ b/docs/getting-started/distributed-compilation.md @@ -585,6 +585,22 @@ start it with `--admin-listen` and these count the outcomes. | `pace_dispatch_leases_field_too_long_total` | A client asked for a lease with a key, toolchain fingerprint or toolchain label longer than a scheduler records, and was refused before any of it was kept. Each is a few dozen bytes from a real client, and `pace-cc` sends none longer, so any rise names a client that does. | | `pace_dispatch_worker_registrations_field_too_long_total` | A worker registered with a fingerprint, endpoint, version, toolchain label, display name or codec list longer than a scheduler records, and was refused before any of it was kept. Each is kept in the worker's entry, and this project's nodes send none that long -- a compiler whose name would overrun its ceiling registers without the name -- so any rise names a peer that does. | | `pace_dispatch_node_announcements_field_too_long_total` | A machine announced itself with an endpoint or version longer than a scheduler records, and was refused before any of it was kept. A condition field has a ceiling too, but an overlong one is refused earlier, when the frame is decoded, as a malformed frame no counter moves for -- so this series never sees it from the network. This project's nodes send none that long, so any rise names a peer that does. | +| `pace_admission_queued_total` | Compile requests the leader queued for admission. Every request enters here, whether it is granted at once or waits. | +| `pace_admission_granted_worker_total` | Admissions granted on a worker, charged the compile's estimated peak memory against that machine's budget. | +| `pace_admission_granted_local_total` | Admissions granted to run on the machine that asked (a compile that cannot be dispatched, or a toolchain no worker serves). | +| `pace_admission_deduplicated_total` | Requests answered *stored*: another compile of the same key stored its object while they waited, so they fetch it instead of compiling. Duplicate-work suppression doing its job. | +| `pace_admission_ran_alone_total` | Admissions whose estimate exceeded a whole machine's budget, so the compile waited for an idle machine and ran alone. A steady rise means a budget below what this site's largest translation units need. | +| `pace_admission_abandoned_total` | A queued or granted request whose connection closed (a `kill -9` of `pace-cc` or ninja), or whose next frame was not a settle. The charge is given back at once; a measured peak is not learned from it. | +| `pace_admission_worker_refused_total` | A worker declined a compile the leader had admitted: a cordon, a stop, or a stale view before the machine's next report. The worker's charge and lease go back, the same request returns to the head of the queue with the refusal counted, and the launcher waits on it again over the same connection; after three refusals it compiles on its own machine. | +| `pace_admission_wait_milliseconds_total` | Time admitted requests spent queued, a `_sum`: each grant adds the time since its request was last parked -- its enqueue, or the worker refusal that sent it back -- so a refused request's first wait is not summed twice. | +| `pace_admission_wait_count_total` | The `_count` of that pair: one per grant, a regrant after a worker refusal included. Divide the sum by it for the mean wait. | +| `pace_admission_forwarded_total` | Queue requests (an admit, a settle) this node relayed to the leader for a launcher on this machine and got the leader's answer to. Written by a node while it does not lead -- a learner, or a voter following another -- and present, reading zero, on the leader: its launchers are answered by its own queue, and it relays again the moment another node takes over. Absent only on a node that runs no consensus. | +| `pace_admission_forward_failed_total` | Queue requests this node could not relay: the leader could not be reached or proved no identity, went silent or was lost before it answered, or moved more often than a request follows it (twice). The launcher was refused `EndpointBusy` and compiles without the queue, so a steady rise means this machine's builds are running unprotected from the OOM kills the queue prevents. | +| `pace_admission_forward_abandoned_total` | Relayed queue requests whose launcher left before the leader's answer reached it. The relay's connection to the leader closes with it, and the leader releases the request and counts it in `pace_admission_abandoned_total`. | +| `pace_admission_queue_depth` | A gauge: requests waiting for admission right now. Absent on a node that schedules nothing. | +| `pace_admission_charged_bytes{machine}` | A gauge: memory the leader has charged to admitted compiles on each machine, the machine named by its identity key in hex. A machine with nothing charged has no sample. | +| `pace_admission_budget_bytes{machine}` | A gauge: the memory budget each machine reported -- what it lets compiles hold at once, and what the leader admits against. A machine that has not reported one has no sample. | +| `pace_admission_in_use_bytes{machine}` | A gauge: what the leader counts as in use on each reporting machine when it admits -- the larger of what it charged and what the machine reports it holds. It reads above `charged_bytes` after a failover, while compiles the old leader admitted still run; a machine stays full while this is near its budget. | The first three refusals are different operator problems and are deliberately counted apart: summing them hides a misconfiguration behind a busy fleet, and diff --git a/docs/operations/deployment.md b/docs/operations/deployment.md index 940cf049d..e8cd1ed3f 100644 --- a/docs/operations/deployment.md +++ b/docs/operations/deployment.md @@ -648,6 +648,18 @@ is where they are explained one by one. | `pace_dispatch_leases_released_late_total` | A compile outran the lease timeout and reported back too late. Read as a fraction of `released_total`; a steady fraction means the lease bound is too short for this site's slowest translation unit. | | `pace_dispatch_leases_malformed_total` | A client asked for a lease naming its key, toolchain or toolchain label in bytes that are not UTF-8, and was refused. | | `pace_dispatch_leases_field_too_long_total` | A client asked for a lease with a key, toolchain fingerprint or toolchain label longer than a scheduler records, and was refused. | +| `pace_admission_queued_total` | Compile requests the leader queued for admission. | +| `pace_admission_granted_worker_total` | Admissions granted on a worker, charged the compile's estimated memory against that machine's budget. | +| `pace_admission_granted_local_total` | Admissions granted to run on the machine that asked. | +| `pace_admission_deduplicated_total` | Requests answered *stored*: another compile of the same key stored its object while they waited. | +| `pace_admission_ran_alone_total` | Admissions whose estimate exceeded a whole machine's budget, so the compile ran alone. | +| `pace_admission_abandoned_total` | Queued or granted requests whose connection closed, or whose next frame was not a settle. | +| `pace_admission_worker_refused_total` | Admitted compiles a worker declined; the same request goes back to the head of the queue, and runs locally after three refusals. | +| `pace_admission_wait_milliseconds_total` | Milliseconds admitted requests spent queued, each grant counting from its request's last park: a `_sum`, whose count is `pace_admission_wait_count_total`. | +| `pace_admission_wait_count_total` | Grants the wait sum covers, one per grant: the `_count` of that pair. | +| `pace_admission_forwarded_total` | Queue requests this node, which does not lead, relayed to the leader for a launcher on this machine. | +| `pace_admission_forward_failed_total` | Queue requests this node could not relay to the leader; the launcher compiled without the queue. | +| `pace_admission_forward_abandoned_total` | Relayed queue requests whose launcher left before the leader answered. | | `pace_dispatch_worker_registrations_total` | Workers registering. A steady rise means heartbeats are not arriving. | | `pace_dispatch_worker_registrations_malformed_total` | A peer named its toolchain, endpoint or version in bytes that are not UTF-8 and was refused. | | `pace_dispatch_worker_registrations_field_too_long_total` | A peer registered with a fingerprint, endpoint, version, toolchain label, display name or codec list longer than a scheduler records, and was refused. | diff --git a/docs/operations/upgrading-a-fleet.md b/docs/operations/upgrading-a-fleet.md index ba14852d1..5a549ac91 100644 --- a/docs/operations/upgrading-a-fleet.md +++ b/docs/operations/upgrading-a-fleet.md @@ -33,12 +33,16 @@ What makes it a *fleet* problem rather than a daemon problem is that a fleet has than one process. A cache daemon is upgraded, restarted, and done; twenty nodes cannot be. -## This release: 0xFC version 15, discovery version 3, consensus version 5 +## This release: 0xFC version 16, discovery version 3, consensus version 5 Every one of the three wires changed its grammar, so every node, every `pace-kv` serving `0xFC`, every `pace-cc` and every `pacectl` moves together (the supported procedure below). +- **`0xFC` 16** (on top of 15's changes below): + - New verbs ADMIT, SETTLE and MACHINE-BUDGET put a queue in front of the fleet's capacity; ADMIT may + be answered with progress pulses, a reply status an older client cannot read. + - A version-15 peer is refused `unsupported-version`, naming the range. - **`0xFC` 15:** - AUTH carries a credential kind, so a launcher presents a machine ticket. - LEASE carries the workers the client could not reach (at most 16) and the client's diff --git a/docs/protocols/compile-cache.md b/docs/protocols/compile-cache.md index e1c45144d..eada652b6 100644 --- a/docs/protocols/compile-cache.md +++ b/docs/protocols/compile-cache.md @@ -51,7 +51,7 @@ daemon, the launcher, and the test client cannot disagree about the layout. | `0x00` | Miss | FETCH found nothing. Payload empty. A legitimate negative, not an error. | | `0x01` | Ok | Command succeeded; payload is the result, if any. | | `0x02` | Error | Command refused; payload is `[u8 errorCode][message]`. | -| `0x03` | Progress | The server is still working on this request. Payload empty. **Not** an outcome: zero or more precede exactly one of the three above, and only on COMPILE. | +| `0x03` | Progress | The server is still working on this request. Payload empty. **Not** an outcome: zero or more precede exactly one of the three above, and only on COMPILE and ADMIT. | | `0x04` | Push | One frame of a subscription's stream. **Not** an outcome either, and only on SUBSCRIBE. | Which statuses an op may be answered with is table data, not convention: @@ -66,6 +66,9 @@ Which statuses an op may be answered with is table data, not convention: | LEASE | Ok, Error | | RELEASE | Ok, Error | | COMPILE | Ok, Progress, Error | +| ADMIT | Ok, Progress, Error | +| SETTLE | Ok, Error | +| MACHINE-BUDGET | Ok, Error | A miss and a refusal being distinct is the point. When both were the byte `0x00`, a client the daemon could not serve saw an endlessly cold cache and no @@ -117,7 +120,7 @@ diagnostic — the build merely got slower, forever, with nothing to show for it | `0x28` | *(reserved)* | Never sent. Was `roster-expired`, named for the certified roster's lapse ([#178](https://github.com/LASTRADA-Software/pace/issues/178)); what it still meant is `grant-unverifiable`. The number is burnt. | | `0x29` | node-key-unknown | A node proof verified, under a key this cluster does not hold for the id it named: a machine nobody admitted, or one presenting a key other than the one admitted under its id. The remedy is an admission, not a new key. | | `0x2a` | node-key-revoked | A node proof verified under a key the cluster has **revoked** — the forgotten machine itself. The connection is kept and marked, and every later verb on it is refused as the forgotten machine's, from any address, this machine's included. | -| `0x2b` | node-identity-required | A verb only a machine that **proved** its identity may send — REGISTER, NODE-ANNOUNCE, HEARTBEAT, WITHDRAW — arrived on a connection that has not, loopback included. A machine ticket admits a client and never satisfies these: a ticketed connection is not sealed. | +| `0x2b` | node-identity-required | A verb only a machine that **proved** its identity may send — REGISTER, NODE-ANNOUNCE, HEARTBEAT, WITHDRAW, MACHINE-BUDGET — arrived on a connection that has not, loopback included. A machine ticket admits a client and never satisfies these: a ticketed connection is not sealed. | | `0x2c` | enrollment-host-full | The request came from an address that already has as many enrollment requests waiting as one host may hold, counted by the address each request first came from, so it was not recorded. Its own code rather than enrollment-full, because one address asking a lot and many machines waiting are different problems; a joiner treats both as a wait. | | `0x2d` | ticket-refused | An AUTH presenting a machine ticket this node did not accept: a signature that does not verify, a machine the roster does not hold, a revoked key, an audience that is not this node, or an expiry passed. One code for all of them. The message names the reason when the caller could have checked it itself -- malformed bytes, a wrong audience, an expiry -- or when it is the node's own state; a forgery, a machine the roster does not hold and a revoked key all read `not admitted by this node`, because AUTH answers any address and those three would tell a stranger which ids the roster holds. The node still counts each of them apart. Not `unauthenticated`, which says a credential is still owed; this one says a credential was presented and judged. | | `0x2e` | identified-caller-required | An operator's **control** verb -- `CLUSTER-ADMIT` in every form, `CLUSTER-FORGET`, `CLUSTER-SET`, `ENROLL-CONTROL` -- from a caller admitted by `--fleet-open` alone. `--fleet-open` admits a caller to what the fleet serves and never to what decides the fleet; a caller on this machine, one that proved a node key, or one presenting a verified machine ticket is past this refusal, and is then asked for an operator's standing (`operator-standing-required`). Not `not-a-member`: the caller is admitted, and the remedy is to identify itself. | @@ -505,6 +508,73 @@ wait past its own grant by roughly a round trip ([#1122](https://github.com/LASTRADA-Software/pace/issues/1122)), which was harmless only because the worker declines an outrun job first. +### ADMIT, SETTLE and MACHINE-BUDGET + +Version 16 puts a queue in front of the fleet's capacity. Three verbs, all of the scheduler +family and all `RequiresAuth`; a cache listener refuses them `dispatch-not-permitted` for the +reason it refuses `LEASE`, and a node that runs no consensus the same. + +``` +ADMIT [0xFC][ver][0x22][u32 len] payload: [kind][fingerprint][cacheKey][sourcePath] + [treeLabel][storable][codecs][exclusions] + [toolchainLabel] + reply Ok: [kind][endpoint][leaseToken][codecs] + [u64 chargedBytes][u64 queuedMs] + [u32 lifetime][workerKey][reason][dialHint] +SETTLE [0xFC][ver][0x23][u32 len] payload: [outcome][peakBytes] + reply Ok: empty +MACHINE-BUDGET [0xFC][ver][0x24][u32 len] payload: [u64 memoryBudget][u32 slots][u32 heldSlots] + [u64 heldBytes][peaks] + reply Ok: empty +``` + +`ADMIT` asks the fleet's leader to admit one compile: it waits for the memory the compile is +estimated to need and is then told where it runs. Because a queue is as long as it is, the reply +may be preceded by any number of `Progress` pulses -- the second verb, after `COMPILE`, allowed +one -- which is what lets a client tell a queued request from a dead leader against its sliding +idle bound. `kind` is one byte, `0` dispatchable or `1` local-only (it must run on the machine +that asked); `storable` is `0` or `1`; `codecs` and `exclusions` are the lists `LEASE` carries. +`cacheKey` is at most 64 bytes, `fingerprint` at most 64, `sourcePath` (root-relative) at most 512 +and `treeLabel` at most 256; a field over its cap is refused `malformed-frame`, never truncated. +`toolchainLabel` is what a person calls the toolchain (`cl 19.44.35207`), as `LEASE` carries it: +display only, empty when the client did not say, and -- like `treeLabel` -- refused unless it is +UTF-8, because the leader keeps it and the fleet page renders it. + +The grant is exactly ten fields. Its `kind` is `0` a worker (`endpoint` and `leaseToken` name +it), `1` the machine that asked, or `2` stored -- another compile of the same key finished and +stored the object, so the client fetches it. `chargedBytes` is the memory estimate charged for the +admission and `queuedMs` how long the request waited since it was enqueued. The last four are a +worker grant's, empty or zero otherwise, and are what a `LEASE` grant carries for the same reasons: +`lifetime` is how long the token lives from when the client receives it, in milliseconds; +`workerKey` the named worker's identity key (empty, or exactly a key's width), which the compile +reply must be sealed under; and `dialHint` where the worker was last seen, at most as long as an +endpoint, dialled before `endpoint` and never signed -- the token signs `endpoint`, so a stale hint +is refused `lease-endpoint-mismatch` by whatever answers there. `reason` is one byte saying why a +dispatchable request was granted the machine that asked: `0` it asked to (every other grant +says this), `1` no worker serves its toolchain, `2` every worker that does was excluded, `3` none +had room. Every byte is pinned: an `AdmitKind`, `GrantKind`, `LocalGrantReason` or `SettleOutcome` +past its last value is refused. + +Two machines that missed the same key are each granted a worker: the leader files a worker +grant's lease under the admission, never under the cache key, so no admission is refused +`already-in-flight`. A second request for a key already being compiled waits for that compile only +where its result can be fetched -- a storable compile asked for by the same machine -- and is then +answered `2` stored. + +`SETTLE` tells the leader how an admitted compile ended, **on the connection that admitted it** -- +the charge is scoped to the connection, so it carries no lease token, and the leader releases +whatever a connection still holds when it closes. `outcome` is `0` stored, `1` not stored, `2` +failed or `3` the worker refused the job. `peakBytes` is the measured peak as a big-endian `u64`, +or an **empty** field when none was measured: absent is not zero. + +`MACHINE-BUDGET` is a node reporting its own ledger -- the memory it will let compiles hold, its +slots and what is held now, with a bounded sample of measured peaks. It is a machine speaking for +itself, so it requires a proven node identity exactly as `REGISTER` does. `peaks` is one nested +field list of at most 64 entries, each four fields: cache key, source path, compiler identity and +the peak as a `u64`. + +All three are capped at the control ceiling, 64 KiB. + ### Bulk fields carry a codec envelope A preprocessed translation unit and an object file are both large, so those @@ -518,7 +588,7 @@ its cache because two peers were compiled with different codec sets. ### Control verbs have a lower ceiling -`REGISTER`, `HEARTBEAT`, `LEASE` and `RELEASE` are capped at 64 KiB rather than the +`REGISTER`, `HEARTBEAT`, `LEASE`, `RELEASE`, `ADMIT`, `SETTLE` and `MACHINE-BUDGET` are capped at 64 KiB rather than the session cap. That listener is meant to be reachable by a whole fleet, and a scheduler that can be made to allocate 256 MiB per frame by anything that authenticated once is a scheduler that stops scheduling. `COMPILE` is the deliberate exception, @@ -527,7 +597,7 @@ since it carries a whole translation unit. ## Node identity and sealed frames A machine that **joins the fleet** — registers a worker, announces itself, heartbeats, -withdraws — proves which machine it is before it sends any of those verbs, and every frame +withdraws, reports its memory budget — proves which machine it is before it sends any of those verbs, and every frame after that proof is sealed ([#178](https://github.com/LASTRADA-Software/pace/issues/178)). A client asking for a lease or a cache entry runs no handshake and pays no round trip for any of this. A remote one proves its machine with a machine ticket instead (`AUTH`, below); @@ -691,7 +761,7 @@ through the daemon's connection logger, so a rejection is visible to the operator as well as to the client. An `unsupported-version` message names the offered version *and* the supported -range (`unsupported wire version 14; this server speaks 15..15`). A rejection that +range (`unsupported wire version 14; this server speaks 16..16`). A rejection that does not say what would have worked cannot be acted on, and this is the only message an operator with a mismatched install will ever see. diff --git a/docs/tools/pace-cc.md b/docs/tools/pace-cc.md index b18432633..b9bb6dd8f 100644 --- a/docs/tools/pace-cc.md +++ b/docs/tools/pace-cc.md @@ -691,11 +691,14 @@ Writes use an atomic append (`FILE_APPEND_DATA` / `O_APPEND`) so the hundreds of concurrent compilers in one build interleave whole lines instead of shredding each other's. Recording failures are swallowed: statistics never break a build. -A line is tab-separated and names its own format first: `v2`, then fifteen columns in +A line is tab-separated and names its own format first: `v3`, then seventeen columns in this order — `outcome`, `prefetch-group`, `value-bytes`, `elapsed-ms`, `source`, `detail`, `preprocess-ms`, `cache-ms`, `direct-ms`, `direct-hit`, `timestamp` (seconds -since the Unix epoch), `dispatch`, `dispatch-detail`, `dispatch-specifics` and -`exit-code`, the code the compile exited with. A line with no version was written by +since the Unix epoch), `dispatch`, `dispatch-detail`, `dispatch-specifics`, +`exit-code` (the code the compile exited with), `queued-ms` (milliseconds waited for a +compile slot) and `peak-memory-bytes`. An empty `exit-code`, `queued-ms` or +`peak-memory-bytes` means no reading, never a zero. `pace-cc -s` prints the wait and the +peak-memory lines only over records that carry them. A line with no version was written by an older launcher and is read by how many columns it has, as it always was; its exit code is *absent*, never a zero that would call every old compile a success. A line naming a version this launcher does not know — a newer launcher appending to the same @@ -844,6 +847,47 @@ none of them is a caching failure. | `a worker compile failed and was retried locally` | The remote compiler exited non-zero, so the result was discarded and the translation unit compiled here to confirm. Broken code produces this on every machine and is not a fleet problem; a *rising* count against a build that keeps succeeding is a worker producing failures that are not real, and the verbose line names the machine. | | `the dispatched object could not be written`, `the depfile for a dispatched compile could not be written` | The object came back and this machine could not store it — a full disk or a read-only output path. Local, not a fleet problem, and the compile is redone here. | +## The compile queue + +Since #1728 every compile the launcher runs on a miss, and every compile it declines to cache, +asks the fleet's queue first (`ADMIT`, always to this machine's own node at `PACE_ADDR`, which +answers when it leads and forwards to the leader when it does not). `PACE_SCHEDULER` does **not** +redirect it: the leader charges a compile to the machine that asked, which it can name only by the +key this machine's node proves on the forwarded connection, so a launcher asking a remote +scheduler directly would be charged to no machine at all. `PACE_SCHEDULER` keeps its one meaning, +where a `LEASE` goes when no queue answers. The queue exists so that a build with more compiler processes than the machine has +memory for waits instead of being killed by the kernel: a miss is **admitted, never refused for +capacity**. The reply says where the compile runs: on a worker, on this machine, or nowhere -- +*stored*, because another compile of the same key finished and stored its object while this one +waited, in which case the launcher fetches it. + +- **The connection is the lease.** The launcher keeps the admitting connection open for the + whole compile and ends it with `SETTLE` (stored, not stored, failed, or *the worker refused + the job*). `kill -9` of the launcher or of ninja closes the connection and the leader gives the + memory back at once; nothing has to expire. +- **A worker that declines keeps your place.** A refused dispatch goes back to the front of the + queue on the same connection and is counted; after three refusals the compile runs locally. +- **A wait is not a hang.** The leader pulses while a request is queued, so the launcher tells a + long queue from a dead leader; `PACE_VERBOSE` says when it is waiting. +- **It fails open.** A node that is stopped, unreachable, refuses the verb, or answers something + unreadable costs the build nothing: the compile runs as it did before the queue, and a launcher + that already found the node unreachable on this invocation does not dial it a second time. + A launcher that learns this from a FETCH also writes it to the reachability memo, so for the + next fifteen seconds no launcher of the build dials the queue either. The one compile that pays + is one with no such evidence: a command line the launcher cannot parse makes no FETCH, so with + the node stopped and nothing in the memo it dials the queue once -- on Windows, where a refused + loopback connect can take the whole one-second connect bound, up to one second -- and that + failure is then remembered. So a stopped node costs a build at most one such connect per + fifteen-second memo window, not one per compile. + A *local* grant that says the machine has no room yet is an ordinary queue decision and is not + counted as a refusal. +- **`pace-cc -s`** reports what the queue cost: `waited s for a compile slot ( compiles)` + and `peak compile memory GiB`, each only over records that carry the reading (the + `queued-ms` and `peak-memory-bytes` columns above). + +The budget the queue charges against is the node's, see +[the machine's compile budget](pace-node.md#the-machines-compile-budget). + ## Exit codes | Situation | Code | diff --git a/docs/tools/pace-node.md b/docs/tools/pace-node.md index 8880086b0..8a8fc9d28 100644 --- a/docs/tools/pace-node.md +++ b/docs/tools/pace-node.md @@ -3339,6 +3339,57 @@ tally that also covers a full connection cap, so read the node's own counter: a `pace_worker_jobs_refused_endpoint_busy_total` is the signal to raise `--compile-inflight-max`. +#### The machine's compile budget + +Separate from the worker's in-flight bytes above, every node also states what its +**machine** may be charged for compiles, to the fleet's queue. It is sent every +presence round (`MACHINE-BUDGET`), **whether or not the node runs a worker**: a node +started with `--slots=0` runs no worker and still has a machine. + +| Option | Default | What it states | +| --- | --- | --- | +| `--machine-slots` (`machine_slots`) | the machine's logical cores | Compiles the machine runs at once, every one on it counted, local ones included. Not `--slots`, which sizes the worker and is `0` on a node with none. `0` is refused. | +| `--machine-memory-budget` (`machine_memory_budget`) | 75% of RAM, less this node's cache tier, never below `--compile-estimate-default` | The memory those compiles may hold at once. `k`/`m`/`g` or `N%` of host RAM. A typed figure is used as given. `0` is refused. | +| `--compile-estimate-default` (`compile_estimate_default`) | `2g` | What a translation unit nobody has measured is assumed to need. Too low is the out-of-memory kill the queue exists to prevent. `0` is refused. | + +All three are **restart-only**, and a service registration replays only the ones typed. +Beside the figures, the report carries what the machine's compiles hold now and a +bounded sample of the peaks its compiler processes measured, so the leader's estimates +learn from the machines that ran the compiles. A leader that cannot take the report +refuses it like any other verb; the node logs that once, when it starts, and again when +it clears. + +#### The queue, and who answers it + +The queue lives at the fleet's **leader**, which is the one node that knows every machine's +budget. A launcher asks its own node (`127.0.0.1:6674`); a node that does not lead **forwards** +`ADMIT` and `SETTLE` inline to the leader, over its own proven identity, and relays the leader's +pulses and answer back. This holds for every non-leader, a voting follower and a learner alike, +and a node never forwards to itself. The relay keeps one upstream connection per launcher +connection for as long as it lives, so a launcher that goes away releases its charge at the +leader at once (`pace_admission_forward_abandoned_total`). A leader that cannot be reached is +counted (`pace_admission_forward_failed_total`) and the launcher compiles without the queue. + +What the queue does, in the order it matters to an operator: + +- **A miss is queued, never refused for capacity.** Waiting is the product; the depth is + `pace_admission_queue_depth`. +- **A compile larger than a whole machine's budget runs alone** on an idle machine + (`pace_admission_ran_alone_total`) rather than never. +- **Duplicates wait for the object, not for each other.** Compiles of one cache key that share a + cache are answered *stored* when the first one STOREs, never when it merely finishes, and only + for compiles that can be stored. +- **A worker that refuses an admitted compile keeps the request's place and count**; after three + refusals it runs on the machine that asked. +- **A failover admits nothing on a machine until that machine's budget report arrives**, and the + new leader's ledger charges the larger of what it granted and what the machine says it holds, + so a leader change cannot over-commit a machine. Expect a short pause after one. +- **A worker that fails a compile settles it as *not stored*,** which keeps a broken toolchain + from teaching the queue that every translation unit is small. + +`pacectl live-stats fleet` shows the queue's depth and, per client, the machine and build tree, +how many compiles are running and how many are queued. + The node's one `0xFC` listener holds its own pool of bytes for frames still being read, sized to the largest budget any of its components declares — so it follows the compile budget and never undercuts it. It is ONE pool: the cache and shared-cache @@ -3915,7 +3966,7 @@ lease, and answers every other scheduling verb `not-leader` naming the leader's endpoint, once the caller is admitted, so a launcher pointed at its own machine's learner follows it to the leader; it names nobody once the leader has been silent past the consensus tier's election timeout. A node running no consensus at all refuses the capacity verbs -(`REGISTER`, `HEARTBEAT`, `WITHDRAW`, `LEASE`, `RELEASE`, `NODE-ANNOUNCE`) +(`REGISTER`, `HEARTBEAT`, `WITHDRAW`, `LEASE`, `RELEASE`, `NODE-ANNOUNCE`, `ADMIT`, `SETTLE`, `MACHINE-BUDGET`) `dispatch-not-permitted` and the cluster verbs `no-cluster`, the codes `pace-kv` gives for the same verbs. None of these is counted. @@ -4107,6 +4158,7 @@ What it shows, and why each part is split the way it is: | Leases outstanding | One row per lease the scheduler has handed out and not yet seen resolved: the object key, the worker it went to, that worker's endpoint, and how long it has been held. **Oldest first, and bounded** — the header says `the 50 oldest of 900` when there are more, so a truncated table is not read as the whole fleet's work. A client hands its lease back when the job ends however it ended, so a row that has been there for minutes is a client that died mid-build, and the endpoint is where its work was going. A row with no endpoint is a lease against a worker that is no longer registered. | | Why requests were refused | Granted, and refused split four ways, each with what it tells you to do. | | Cache tiers | Items, bytes, budget, evictions and index RAM **per tier**. A tier no member runs has no column at all, and a fleet where nobody runs one says so rather than showing an empty table. `index-ram` is what the tier's key index costs in memory: always RAM, even for a disk tier whose budget is bytes on a filesystem, so the two are not comparable and must not be added. | +| Compile queue | The leader's admission queue (#1728): how many compiles are **waiting**, which machine and build tree each client is compiling for with how many **running** and **queued**, what the queue has **charged** each machine, what is **in use** on it (the larger of that charge and what the machine itself reports holding: after a failover the charge starts at zero while the machine is still full, and in-use is the figure admission refuses on), and the memory **budget** it stated. In `/fleet.json` this is the `admission` object (`depth`, `clients`, `machines`), `null` on a node that leads no queue, and `/fleet.txt` serves the clients as the `clients` section beside the `queue-depth` headline figure (the per-machine ledger is in the JSON and on the page). A node that has not heard a budget or a request shows `–`, never `0 waiting`: zero would say every compile was admitted at once. A machine that stated a budget but no worker name is shown by the first bytes of its key in hex. A budget cell is `–` until the machine has reported one. | | Over time | Four charts over 24 hours or 7 days: compiles dispatched, refusals stacked four ways, offerable capacity against jobs in flight, and cache hit rate per bucket. | | Members | Who the cluster has agreed on, and where each answers. A member that has announced none shows no scheduler endpoint, because it has not said. | @@ -4556,7 +4608,7 @@ it on every heartbeat round, and a node running consensus verifies the proofs pr it against the cluster's roster. **The verbs a machine JOINS the fleet with require it.** `REGISTER`, `NODE-ANNOUNCE`, -`HEARTBEAT` and `WITHDRAW` are refused `node-identity-required` on a connection that proved +`HEARTBEAT`, `WITHDRAW` and `MACHINE-BUDGET` are refused `node-identity-required` on a connection that proved no identity the cluster holds -- from any address, loopback included. A **client** is admitted without a proof: on this machine by loopback, and from another machine by the machine ticket its own node mints ([a machine that only asks](#a-machine-that-only-asks-tickets)), diff --git a/docs/tools/pacectl.md b/docs/tools/pacectl.md index 1c93a19c8..f349ecfba 100644 --- a/docs/tools/pacectl.md +++ b/docs/tools/pacectl.md @@ -147,7 +147,7 @@ the help renders `Verbs()` grouped by `WireSpec::heading`, and ### The fleet verb Every fleet table — machines, workers, outstanding leases, members, revoked keys, -conditions, cache tiers — used +conditions, cache tiers, the compile queue's clients — used to be reachable from a browser and from nowhere else. `/fleet.json` is the only other door and it needs a JSON parser the operator supplies; `jq` is not on a Windows build box, and this tool has no JSON *parser* of its own — it only emits one. @@ -837,7 +837,7 @@ $ pacectl live-stats node --format=tsv --samples=30 > node.tsv |---|---|---| | `cache` | a `pace-kv`, or a `pace-node` for its own cache tier | the hit rate, and operations, connections, evictions and expiries per second, each with its trend; connections, items and bytes in use against their limits, per storage tier | | `node` | a `pace-node` | compiles and refusals per minute, the mean compile and its trend; the slots in use against the slots available and the limit that bounds them; the cache tier's fill; the host's CPU and free memory; the node's identity, the conditions it has raised (each marked `latched` or `live`, `none raised` when there are none, the absent marker from a node too old to say), toolchains, registrars and the leader; on a consensus node, `dialled at` — the address its peers dial, the string `--cluster-admit`'s receipt asks you to compare | -| `fleet` | the node that leads the fleet | the headline figures as tiles; one table per section (machines, workers, leases, members, revoked, conditions, tiers); each machine's CPU over time | +| `fleet` | the node that leads the fleet | the headline figures as tiles; one table per section (machines, workers, leases, members, revoked, conditions, tiers, clients); the compile queue's depth tile (`Queued compiles`) and, per client, the machine and build tree compiling with how many compiles are running and queued; each machine's CPU over time | The subject may be left out, and then it is what `--addr` is: a `pace-kv` is watched as `cache`, and a compile node as `node`. **`fleet` is never inferred.** The @@ -866,7 +866,7 @@ A terminal smaller than a panel's minimum gets one line, `needs x | `q`, `Q`, `Esc`, `Ctrl-C` | quit, restoring the terminal; `Esc` closes a machine's detail instead while one is open | | `Tab`, `→` / `←` | the fleet's next / previous section | | `1` … `9` | the fleet's sections by position | -| `m` `w` `l` `c` `r` `!` `t` | the fleet's sections by name: machines, workers, leases, members (`c`, for cluster: `m` is taken), revoked keys, conditions (`!`: `c` is taken) and tiers | +| `m` `w` `l` `c` `r` `!` `t` `u` | the fleet's sections by name: machines, workers, leases, members (`c`, for cluster: `m` is taken), revoked keys, conditions (`!`: `c` is taken), tiers, the compile queue's clients (`u`, for use: `q` quits) | | `PgDn` / `PgUp`, `Home` | scroll the fleet's table a page, or back to its top | | `/` | filter the fleet's table: type, `Enter` keeps the filter, `Esc` clears it, `Backspace` edits it | | `↑` / `↓` | highlight the fleet chart's lanes, then the table's rows, and the machine they name everywhere it appears | diff --git a/packaging/config/pace-node.yaml b/packaging/config/pace-node.yaml index 1f4b97f30..b618ae533 100644 --- a/packaging/config/pace-node.yaml +++ b/packaging/config/pace-node.yaml @@ -182,6 +182,25 @@ # Read at startup only. #compile_inflight_max: +# compile slots this MACHINE shares between every compile on it, local ones +# included, as the fleet's queue counts them. Default: its logical cores. Not +# slots, which sizes the worker and is 0 on a node with none: this is reported +# whether or not a worker runs. Read at startup only. +#machine_slots: + +# the memory the fleet's queue may have this machine's compiles hold at once, +# local ones included; k/m/g = KiB/MiB/GiB or N% of host RAM. Default: 75% of +# RAM less this node's cache tier, never below compile_estimate_default. A +# compile larger than the whole budget runs alone rather than never. Read at +# startup only. +#machine_memory_budget: + +# what a translation unit nobody has measured is assumed to need (default 2g; +# k/m/g = KiB/MiB/GiB). Too low is the out-of-memory kill the queue exists to +# prevent; too high is a compile waiting longer than it must until its first +# measurement. Read at startup only. +#compile_estimate_default: + # cores never offered to the fleet, overriding what the node class reserves. # 0 is a real answer and is not the same as omitting the flag. Ignored when # --slots names a number, which is the operator's answer already. diff --git a/scripts/agent-md-budget.txt b/scripts/agent-md-budget.txt index bc348d580..af0721e9b 100644 --- a/scripts/agent-md-budget.txt +++ b/scripts/agent-md-budget.txt @@ -67,4 +67,4 @@ # paragraphs below the sentence forbidding exactly that. max-bullet-lines|8 -total-lines|1635 +total-lines|1641 diff --git a/scripts/check-ranges-seam-selftest.cmake b/scripts/check-ranges-seam-selftest.cmake index a26ad3fd3..f982f3c96 100644 --- a/scripts/check-ranges-seam-selftest.cmake +++ b/scripts/check-ranges-seam-selftest.cmake @@ -93,6 +93,13 @@ inline constexpr auto FoldLeft = std::ranges::fold_left; #else inline constexpr auto FoldLeft = Detail::FoldLeftFallback; #endif + +// A views:: row, as core-cpp's Stride block is. +#if defined(__cpp_lib_ranges_stride) && !defined(PACE_RANGES_FORCE_FALLBACK) +inline constexpr auto Stride = std::views::stride; +#else +inline constexpr auto Stride = Detail::StrideFallback; +#endif } // namespace Pace::Ranges ") @@ -285,7 +292,7 @@ endfunction() # case expects the check to PASS. The field count is asserted per row below. set(PaceRangesSeamCases # The baselines, in both enumeration modes. The vendored direct call is DECLINED here. - "baseline via walk|none|-|-|none called directly && directory walk && 1 declined under third-party roots && std::ranges::iota -> Ranges::Iota && std::ranges::fold_left -> Ranges::FoldLeft|CMake Error" + "baseline via walk|none|-|-|none called directly && directory walk && 1 declined under third-party roots && std::ranges::iota -> Ranges::Iota && std::ranges::fold_left -> Ranges::FoldLeft && std::views::stride -> Ranges::Stride|CMake Error" "baseline via git|none-git|-|-|none called directly && git ls-files && 1 declined under third-party roots|CMake Error" # --- THE RED ARM --- @@ -297,11 +304,18 @@ set(PaceRangesSeamCases "two calls on two lines are both named|newfile|src/Pace/Core/Two.cpp|void Two(std::vector& v)~n~{~n~ std::ranges::iota(v, 0)~sc~~n~ (void) std::ranges::fold_left(v, 0, std::plus {})~sc~~n~}~n~|Two.cpp:3: calls `std::ranges::iota` && Two.cpp:4: calls `std::ranges::fold_left` && 2 site(s)|-" "line number survives brackets semicolons and a trailing backslash|newfile|src/Pace/Core/Drift.cpp|int a~lb~4~rb~ = {1, 2, 3, 4}~sc~~n~auto b = a~lb~0~rb~~sc~ auto c = a~lb~1~rb~~sc~~n~// a trailing backslash \\~n~void Drift(std::vector& v) { std::ranges::iota(v, 0)~sc~ }~n~|src/Pace/Core/Drift.cpp:4: calls|-" + # A `views::` row: `std::views::stride` is no libc++'s, so it is refused by every spelling. + "planted views stride|newfile|src/apps/pace-cc/Walk.cpp|void Walk(std::vector& v)~n~{~n~ for (int x : std::views::stride(v, 4)) ~lb~~rb~~n~}~n~|src/apps/pace-cc/Walk.cpp:3: calls `std::views::stride` directly -- write `Ranges::Stride` (gated by __cpp_lib_ranges_stride) && 1 site(s) call a wrapped range algorithm directly|-" + "planted ranges views stride|newfile|src/apps/pace-cc/Walk.cpp|void Walk(std::vector& v)~n~{~n~ auto s = std::ranges::views::stride(v, 4)~sc~~n~}~n~|src/apps/pace-cc/Walk.cpp:3: calls `std::views::stride` directly|-" + "bare views stride under using namespace std|newfile|src/apps/pace-cc/Walk.cpp|using namespace std~sc~~n~auto Walk(vector& v) { return views::stride(v, 4)~sc~ }~n~|src/apps/pace-cc/Walk.cpp:2: calls `std::views::stride` directly|-" + # --- THE QUIET ARM --- "C++20 names sharing the prefix are not wrapped facilities|newfile|src/Pace/Core/Views.cpp|auto Indices(int n) { return std::views::iota(0, n)~sc~ }~n~using Span = std::ranges::iota_view~sc~~n~auto Tail = std::ranges::fold_left_first~sc~~n~|none called directly|CMake Error" "the seam spelling and the macro are not direct calls|newfile|src/Pace/Core/Seam.cpp|#if defined(__cpp_lib_ranges_iota)~n~#endif~n~void Seam(std::vector& v) { Pace::Ranges::Iota(v, 0)~sc~ (void) Ranges::FoldLeft(v, 0, std::plus {})~sc~ }~n~|none called directly|CMake Error" "comments are not calls|newfile|src/Pace/Core/Note.cpp|// Never std::ranges::iota(v, 0) here.~n~/// Nor `std::ranges::fold_left(v, 0, op)`.~n~/*~n~ std::ranges::iota(v, 0) is missing from Apple's libc++.~n~*/~n~int Note() { return 0~sc~ }~n~|none called directly|CMake Error" + "the seam stride, stride_view and a views iota are not direct calls|newfile|src/apps/pace-cc/Step.cpp|auto Step(std::vector& v) { return core::ranges::Stride(v, 4)~sc~ }~n~using V = std::ranges::stride_view>~sc~~n~auto I = std::views::iota(0, 3)~sc~~n~// std::views::stride is not shipped by libc++.~n~|none called directly|CMake Error" + # --- DERIVATION: a block added to the header is enforced with no edit to the check --- "a selection block added to the header is enforced|seamrow:src/Pace/Core/Right.cpp|#if defined(__cpp_lib_ranges_fold) && !defined(PACE_RANGES_FORCE_FALLBACK)~n~inline constexpr auto FoldRight = std::ranges::fold_right~sc~~n~#else~n~inline constexpr auto FoldRight = Detail::FoldRightFallback~sc~~n~#endif|int Right(std::vector const& v) { return std::ranges::fold_right(v, 0, std::minus {})~sc~ }~n~|Right.cpp:1: calls `std::ranges::fold_right` directly -- write `Ranges::FoldRight` && std::ranges::fold_right -> Ranges::FoldRight|-" # ...and its CONTROL: the same call with no block in the header is not this check's business. diff --git a/scripts/check-ranges-seam.cmake b/scripts/check-ranges-seam.cmake index 82743a6d5..5c87a081c 100644 --- a/scripts/check-ranges-seam.cmake +++ b/scripts/check-ranges-seam.cmake @@ -38,17 +38,20 @@ # block of the shape # # #if defined(__cpp_lib_) ... -# inline constexpr auto = std::ranges::; +# inline constexpr auto = std::ranges::; (or std::views::) # # contributes one row -- the standard name refused, the seam name offered, the macro that gates -# it. A restated list would catch a facility going away and be blind to one ARRIVING, which is +# it. The scope is part of the row: `std::views::stride` (core::ranges::Stride, which no libc++ +# ships) is refused as `views::stride`, and a row for `std::ranges::iota` does not make +# `std::views::iota` -- C++20, shipped everywhere -- a violation. A restated list would catch a facility going away and be blind to one ARRIVING, which is # the direction that matters: a row added to the header is enforced here with no edit. A header # this check derives no row from is REFUSED, never read as "nothing to enforce". # -# A USE is `ranges::` followed by something that cannot continue an identifier, in code +# A USE is `::` (`ranges::` or `views::`) followed by something that cannot continue an identifier, in code # with comments stripped (`pace_scan_code_lines`, `scripts/lib/CheckCommon.cmake`). So # `std::ranges::iota` and a bare `ranges::iota` under a `using namespace std;` are refused, and # `std::views::iota` and `std::ranges::iota_view` -- C++20, shipped everywhere -- are not. +# `std::views::stride`, `std::ranges::views::stride` and a bare `views::stride` are refused. # The seam spelling differs by CASE (`Ranges::Iota`), which is what lets one case-sensitive # pattern tell the two apart. # @@ -142,7 +145,7 @@ endforeach() # the `;` that ends the declaration, so no element of the result list is split by one. file(READ "${seamHeaderPath}" seamContent) string(REPLACE "\r\n" "\n" seamContent "${seamContent}") -set(selectionShape "#if defined\\((__cpp_lib_[a-z0-9_]+)\\)[^\n]*\ninline constexpr auto ([A-Z][A-Za-z0-9]*) = std::ranges::([a-z_][a-z0-9_]*)") +set(selectionShape "#if defined\\((__cpp_lib_[a-z0-9_]+)\\)[^\n]*\ninline constexpr auto ([A-Z][A-Za-z0-9]*) = std::(ranges::views|ranges|views)::([a-z_][a-z0-9_]*)") string(REGEX MATCHALL "${selectionShape}" selections "${seamContent}") set(rowNames "") @@ -153,11 +156,17 @@ foreach(selection IN LISTS selections) endif() set(macro "${CMAKE_MATCH_1}") set(seamName "${CMAKE_MATCH_2}") - set(standardName "${CMAKE_MATCH_3}") + set(scope "${CMAKE_MATCH_3}") + # `std::ranges::views::x` is `std::views::x`: one row, scanned as `views::x`. + if(scope STREQUAL "ranges::views") + set(scope "views") + endif() + # The row's key is the token a use contains, so the scope separates `ranges::x` from `views::x`. + set(standardName "${scope}::${CMAKE_MATCH_4}") list(APPEND rowNames "${standardName}") set("seamNameOf_${standardName}" "${seamName}") set("macroOf_${standardName}" "${macro}") - list(APPEND rowSummaries "std::ranges::${standardName} -> Ranges::${seamName}") + list(APPEND rowSummaries "std::${standardName} -> Ranges::${seamName}") endforeach() if(NOT rowNames) @@ -167,7 +176,7 @@ if(NOT rowNames) message("Each wrapped facility is expected as two consecutive lines:") message("") message(" #if defined(__cpp_lib_) && !defined(PACE_RANGES_FORCE_FALLBACK)") - message(" inline constexpr auto = std::ranges::;") + message(" inline constexpr auto = std::ranges::; (or std::views::)") message("") message("Reading none is not \"nothing to enforce\": it is a header whose shape changed") message("under this check, and every direct call would pass unseen. Restore the shape, or") @@ -224,11 +233,11 @@ foreach(relative IN LISTS sourceFiles) file(READ "${PACE_SOURCE_DIR}/${relative}" wholeFile) foreach(standardName IN LISTS rowNames) # A whole-file test first; the line walk is quadratic and almost no file needs it. - string(FIND "${wholeFile}" "ranges::${standardName}" tokenPosition) + string(FIND "${wholeFile}" "${standardName}" tokenPosition) if(tokenPosition EQUAL -1) continue() endif() - pace_scan_code_lines("${wholeFile}" "ranges::${standardName}([^A-Za-z0-9_]|$)" hits) + pace_scan_code_lines("${wholeFile}" "${standardName}([^A-Za-z0-9_]|$)" hits) foreach(hit IN LISTS hits) if(hit MATCHES "^use:([0-9]+)$") list(APPEND violations "${relative}:${CMAKE_MATCH_1}|${standardName}") @@ -257,7 +266,7 @@ if(violations) string(REPLACE "|" ";" fields "${violation}") list(GET fields 0 where) list(GET fields 1 standardName) - message(" ${where}: calls `std::ranges::${standardName}` directly -- " + message(" ${where}: calls `std::${standardName}` directly -- " "write `Ranges::${seamNameOf_${standardName}}` (gated by ${macroOf_${standardName}})") endforeach() message("") diff --git a/src/Pace/CMakeLists.txt b/src/Pace/CMakeLists.txt index 6e7b3f28d..109e6008e 100644 --- a/src/Pace/CMakeLists.txt +++ b/src/Pace/CMakeLists.txt @@ -70,12 +70,15 @@ target_sources(Pace Cluster/SeedSources.cpp Core/Logger.cpp Distributed/LeaseTable.cpp + Distributed/PeakEstimates.cpp + Distributed/AdmissionQueue.cpp Distributed/WorkerRegistry.cpp Distributed/UnservedToolchains.cpp Distributed/FleetVersions.cpp Distributed/NodeProof.cpp Distributed/StateLeaseRoster.cpp Distributed/TicketVerifier.cpp + Distributed/SchedulerAdmission.cpp Distributed/SchedulerService.cpp Distributed/SchedulerProtocol.cpp Distributed/FleetHistory.cpp diff --git a/src/Pace/Distributed/AdmissionQueue.cpp b/src/Pace/Distributed/AdmissionQueue.cpp new file mode 100644 index 000000000..7a0bc585f --- /dev/null +++ b/src/Pace/Distributed/AdmissionQueue.cpp @@ -0,0 +1,350 @@ +// SPDX-License-Identifier: Apache-2.0 +#include + +#include +#include +#include + +namespace Pace::Distributed +{ + +namespace +{ + + /// @return a - b, or zero when b exceeds a. + template + constexpr T Saturating(T a, T b) noexcept + { + return a > b ? static_cast(a - b) : T {}; + } + +} // namespace + +AdmissionQueue::AdmissionQueue(PeakEstimates& estimates, IPlacementSource const& placement, SharesCache sharesCache): + _estimates { estimates }, + _sharesCache { sharesCache ? std::move(sharesCache) : SharesCache { std::equal_to<> {} } }, + _placement { placement } +{ +} + +Ticket AdmissionQueue::Enqueue(QueueRequest request) +{ + auto const ticket = _nextTicket++; + auto const charged = _estimates.Estimate(request.estimate); + _queue.push_back(Entry { .ticket = ticket, + .request = std::move(request), + .chargedBytes = charged, + .refusals = 0, + .order = static_cast(ticket) }); + return ticket; +} + +void AdmissionQueue::ReportMachine(MachineId const& machine, MachineReport const& report) +{ + auto& ledger = _ledgers[machine]; + ledger.report = report; + ledger.refused = false; +} + +std::size_t AdmissionQueue::Depth() const noexcept +{ + return _queue.size(); +} + +std::vector> AdmissionQueue::ChargedByMachine() const +{ + std::vector> out; + for (auto const& [machine, ledger]: _ledgers) + if (ledger.chargedBytes != 0) + out.emplace_back(machine, ledger.chargedBytes); + return out; +} + +QueueView AdmissionQueue::Snapshot() const +{ + auto view = QueueView { .depth = _queue.size(), .clients = {}, .machines = {} }; + + auto const clientOf = [&view](MachineId const& machine, std::string const& tree) -> ClientLoad& { + auto const found = std::ranges::find_if( + view.clients, [&](ClientLoad const& load) { return load.machine == machine && load.tree == tree; }); + if (found != view.clients.end()) + return *found; + return view.clients.emplace_back(ClientLoad { .machine = machine, .tree = tree }); + }; + for (auto const& entry: _queue) + ++clientOf(entry.request.caller, entry.request.treeLabel).queued; + for (auto const& [ticket, charge]: _charges) + ++clientOf(charge.entry.request.caller, charge.entry.request.treeLabel).running; + + for (auto const& [machine, ledger]: _ledgers) + { + if (!ledger.report.has_value() && ledger.chargedBytes == 0) + continue; + view.machines.push_back(MachineLedgerView { + .machine = machine, + .chargedBytes = ledger.chargedBytes, + .inUseBytes = InUseBytes(ledger), + .budgetBytes = ledger.report.has_value() ? std::optional { ledger.report->memoryBudgetBytes } : std::nullopt }); + } + return view; +} + +std::optional AdmissionQueue::Free(Ticket ticket) +{ + auto const found = _charges.find(ticket); + if (found == _charges.end()) + return std::nullopt; + auto charge = std::move(found->second); + _charges.erase(found); + auto& ledger = _ledgers[charge.machine]; + ledger.chargedBytes = Saturating(ledger.chargedBytes, charge.bytes); + ledger.chargedSlots = Saturating(ledger.chargedSlots, 1); + return charge; +} + +std::vector AdmissionQueue::TakeInflight(Ticket owner) +{ + auto const found = _inflight.find(owner); + if (found == _inflight.end()) + return {}; + auto waiters = std::move(found->second.waiters); + auto const range = _inflightByKey.equal_range(found->second.cacheKey); + _inflightByKey.erase(std::ranges::find(range.first, range.second, owner, &decltype(_inflightByKey)::value_type::second)); + _inflight.erase(found); + return waiters; +} + +void AdmissionQueue::Requeue(std::vector entries) +{ + for (auto& entry: entries) + { + auto const place = std::ranges::find_if(_queue, [&](Entry const& queued) { return queued.order > entry.order; }); + _queue.insert(place, std::move(entry)); + } +} + +void AdmissionQueue::Settle(Ticket ticket, Settlement const& settlement) +{ + auto charge = Free(ticket); + if (!charge) + return; + auto waiters = TakeInflight(ticket); + + if (settlement.outcome == SettleOutcome::WorkerRefused) + { + auto& entry = charge->entry; + ++entry.refusals; + _ledgers[charge->machine].refused = true; + entry.order = --_frontOrder; // below every key in the queue, the older refusals' included + _queue.insert(_queue.begin(), std::move(entry)); + Requeue(std::move(waiters)); + return; + } + + if (settlement.peakBytes) + _estimates.Record(charge->entry.request.estimate, *settlement.peakBytes); + + if (settlement.outcome == SettleOutcome::Stored) + for (auto& waiter: waiters) + _stored.push_back(Decision { .ticket = waiter.ticket, + .kind = GrantKind::Stored, + .placement = std::nullopt, + .chargedBytes = 0, + .ranAlone = false }); + else + Requeue(std::move(waiters)); +} + +void AdmissionQueue::ReturnToFront(Ticket ticket) +{ + auto charge = Free(ticket); + if (!charge) + return; + auto waiters = TakeInflight(ticket); + auto& entry = charge->entry; + entry.order = --_frontOrder; // the place a refused ticket takes, without the refusal + _queue.insert(_queue.begin(), std::move(entry)); + Requeue(std::move(waiters)); +} + +void AdmissionQueue::Abandon(Ticket ticket) +{ + if (auto const queued = std::ranges::find(_queue, ticket, &Entry::ticket); queued != _queue.end()) + { + _queue.erase(queued); + _headTickets.erase(ticket); + return; + } + if (auto const woken = std::ranges::find(_stored, ticket, &Decision::ticket); woken != _stored.end()) + { + _stored.erase(woken); + return; + } + for (auto& [owner, inflight]: _inflight) + if (std::erase_if(inflight.waiters, [&](Entry const& waiter) { return waiter.ticket == ticket; }) > 0) + return; + + auto charge = Free(ticket); + if (!charge) + return; + _estimates.RecordKilled(charge->entry.request.estimate); + Requeue(TakeInflight(ticket)); +} + +std::vector AdmissionQueue::LoseLeadership() +{ + std::vector parked; + parked.reserve(_queue.size() + _stored.size()); + for (auto const& entry: _queue) + parked.push_back(entry.ticket); + for (auto const& decision: _stored) + parked.push_back(decision.ticket); + for (auto const& [owner, inflight]: _inflight) + for (auto const& waiter: inflight.waiters) + parked.push_back(waiter.ticket); + std::ranges::sort(parked); + + _queue.clear(); + _stored.clear(); + _inflight.clear(); + _inflightByKey.clear(); + _headTickets.clear(); + for (auto& [machine, ledger]: _ledgers) + ledger.report.reset(); + return parked; +} + +std::uint64_t AdmissionQueue::InUseBytes(Ledger const& ledger) noexcept +{ + return std::max(ledger.chargedBytes, ledger.report.has_value() ? ledger.report->heldBytes : std::uint64_t {}); +} + +std::uint64_t AdmissionQueue::BudgetOf(Ledger const& ledger) noexcept +{ + return ledger.report.has_value() ? ledger.report->memoryBudgetBytes : std::uint64_t {}; +} + +AdmissionQueue::Room AdmissionQueue::RoomOn(MachineId const& machine, Reserved const& hold) const +{ + auto const& ledger = _ledgers.at(machine); + if (!ledger.report.has_value()) + return Room {}; // unreported: nothing may be placed here + auto const& report = *ledger.report; + Room room; + room.usedBytes = InUseBytes(ledger); + room.usedSlots = std::max(ledger.chargedSlots, report.heldSlots); + room.freeBytes = Saturating(Saturating(report.memoryBudgetBytes, room.usedBytes), hold.bytes); + // A machine running an over-budget compile has nothing left for anybody. + room.freeSlots = + room.usedBytes > report.memoryBudgetBytes ? 0U : Saturating(Saturating(report.slots, room.usedSlots), hold.slots); + return room; +} + +std::vector AdmissionQueue::Pump() +{ + /// A machine an entry may run on; `placement` is engaged for a worker. + struct Target + { + MachineId machine; + std::optional placement; + }; + + std::vector decisions = std::exchange(_stored, {}); + std::vector remaining; + std::unordered_map reserved; + std::unordered_set claimed; // machines whose head was already found this pump + std::unordered_set heads; + + auto const reported = [this](MachineId const& machine) { + auto const found = _ledgers.find(machine); + return found != _ledgers.end() && found->second.report.has_value() && !found->second.refused; + }; + + for (auto& entry: _queue) + { + auto const& request = entry.request; + + // Another compile of this key is running where this one's cache is: wait for its store. + if (request.storable && !request.estimate.cacheKey.empty()) + { + auto const [first, last] = _inflightByKey.equal_range(request.estimate.cacheKey); + auto const owner = std::ranges::find_if(first, last, [&](auto const& item) { + return _sharesCache(_inflight.at(item.second).machine, request.caller); + }); + if (owner != last) + { + _inflight.at(owner->second).waiters.push_back(std::move(entry)); + continue; + } + } + + // Reported candidates in placement order, then the caller: the CPU never idles while the + // caller has room. + std::vector targets; + if (request.kind == AdmitKind::Dispatchable && entry.refusals < MaxWorkerRefusals) + for (auto& placement: _placement.Candidates(request.fingerprint, request.excluded)) + if (reported(placement.machine)) + targets.push_back(Target { .machine = placement.machine, .placement = std::move(placement) }); + if (reported(request.caller)) + targets.push_back(Target { .machine = request.caller, .placement = std::nullopt }); + + auto const granted = std::ranges::find_if(targets, [&](Target const& target) { + auto const& report = *_ledgers.at(target.machine).report; + auto const& hold = reserved[target.machine]; + auto const room = RoomOn(target.machine, hold); + if (entry.chargedBytes > report.memoryBudgetBytes) + return report.slots > 0 && room.usedBytes == 0 && room.usedSlots == 0 && hold.bytes == 0 && hold.slots == 0; + return room.freeBytes >= entry.chargedBytes && room.freeSlots >= 1; + }); + + if (granted != targets.end()) + { + auto& ledger = _ledgers.at(granted->machine); + ledger.chargedBytes += entry.chargedBytes; + ledger.chargedSlots += 1; + auto const ticket = entry.ticket; + auto const chargedBytes = entry.chargedBytes; + if (request.storable && !request.estimate.cacheKey.empty()) + { + _inflight.emplace( + ticket, Inflight { .cacheKey = request.estimate.cacheKey, .machine = request.caller, .waiters = {} }); + _inflightByKey.emplace(request.estimate.cacheKey, ticket); + } + _charges.emplace(ticket, + Charge { .machine = granted->machine, .bytes = chargedBytes, .entry = std::move(entry) }); + decisions.push_back(Decision { + .ticket = ticket, + .kind = granted->placement.has_value() ? GrantKind::Worker : GrantKind::Local, + .placement = granted->placement, + .chargedBytes = chargedBytes, + .ranAlone = chargedBytes > BudgetOf(ledger), + }); + continue; + } + + // Blocked everywhere. On each machine of its target set where nothing older is blocked it + // is the head; having been head across a pump already, it holds back everything free there. + auto const wasHead = _headTickets.contains(entry.ticket); + for (auto const& target: targets) + { + if (!claimed.insert(target.machine).second) + continue; + heads.insert(entry.ticket); + if (!wasHead) + continue; + auto& hold = reserved[target.machine]; + auto const room = RoomOn(target.machine, hold); + auto const budget = BudgetOf(_ledgers.at(target.machine)); + auto const alone = entry.chargedBytes > budget; + hold.bytes += std::min({ entry.chargedBytes, budget, room.freeBytes }); + hold.slots += alone ? room.freeSlots : std::min(1, room.freeSlots); + } + remaining.push_back(std::move(entry)); + } + + _queue = std::move(remaining); + _headTickets = std::move(heads); + return decisions; +} + +} // namespace Pace::Distributed diff --git a/src/Pace/Distributed/AdmissionQueue.hpp b/src/Pace/Distributed/AdmissionQueue.hpp new file mode 100644 index 000000000..05f9b8a58 --- /dev/null +++ b/src/Pace/Distributed/AdmissionQueue.hpp @@ -0,0 +1,299 @@ +// SPDX-License-Identifier: Apache-2.0 +#pragma once + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace Pace::Distributed +{ + +/// The machine a charge lands on: the node's identity public key, as the registry stores it. +using MachineId = std::string; +/// Names one queued request until it is settled. +using Ticket = std::uint64_t; + +using CompileCacheWire::AdmitKind; +using CompileCacheWire::GrantKind; +using CompileCacheWire::SettleOutcome; + +/// One compile asking to be admitted. +struct QueueRequest +{ + AdmitKind kind {}; + MachineId caller; ///< Machine the request came from (local leases charge it). + std::string fingerprint; ///< Toolchain; empty for LocalOnly with an unknown toolchain. + EstimateKey estimate; + bool storable { true }; ///< Joins the dedup in-flight set only when true. + std::string treeLabel; ///< Display only. + std::vector excluded; +}; + +/// One candidate worker for a dispatchable request. +struct Placement +{ + std::string workerId; + MachineId machine; + std::string endpoint; +}; + +/// Injected: who can run a toolchain. +class IPlacementSource +{ + public: + IPlacementSource() = default; + IPlacementSource(IPlacementSource const&) = delete; + IPlacementSource& operator=(IPlacementSource const&) = delete; + IPlacementSource(IPlacementSource&&) = delete; + IPlacementSource& operator=(IPlacementSource&&) = delete; + virtual ~IPlacementSource() = default; + + /// @param fingerprint The toolchain to run. + /// @param excluded Worker ids that must not be returned. + /// @return The workers that can run `fingerprint` and have a free worker slot, best first. + [[nodiscard]] virtual std::vector Candidates(std::string_view fingerprint, + std::span excluded) const = 0; +}; + +/// A machine's own statement of its budget and what it currently holds. +struct MachineReport +{ + std::uint64_t memoryBudgetBytes {}; + std::uint32_t slots {}; + std::uint32_t heldSlots {}; ///< Every compile running on the machine (local leases + worker jobs). + std::uint64_t heldBytes {}; ///< Their charged bytes. The ledger uses max(charged, held), never the sum. +}; + +/// An admission: what to run, and where. +struct Decision +{ + Ticket ticket {}; + GrantKind kind {}; + std::optional placement; ///< Engaged for `Worker`. + std::uint64_t chargedBytes {}; + bool ranAlone {}; ///< Its estimate exceeded the whole budget. +}; + +/// How a granted compile ended. +struct Settlement +{ + SettleOutcome outcome {}; + std::optional peakBytes; ///< The compile's measured peak, when known. +}; + +/// One client's load on the queue: a machine and the build tree it compiles for. +struct ClientLoad +{ + MachineId machine; ///< The machine the compiles came from. + std::string tree; ///< The build tree's label, as the client stated it; possibly empty. + std::uint32_t running {}; ///< Granted compiles not yet settled. + std::uint32_t queued {}; ///< Requests still waiting. +}; + +/// One machine's ledger, as a report reads it. +struct MachineLedgerView +{ + MachineId machine; ///< The machine. + std::uint64_t chargedBytes {}; ///< What this ledger has charged it. + std::uint64_t inUseBytes {}; ///< What the queue admits against: `max(charged, held by its report)`. + std::optional budgetBytes; ///< Its own stated budget; absent until it reports one. +}; + +/// Everything a fleet report shows of the queue, read at one moment. +struct QueueView +{ + std::size_t depth {}; ///< Requests waiting (`Depth()`). + std::vector clients; ///< One per (machine, tree) with anything running or queued. + std::vector machines; ///< One per machine that reported a budget or holds a charge. +}; + +/// The leader's FIFO of compiles waiting for memory and slots, with a per-machine ledger. +/// +/// A request is admitted only onto a machine that has reported its budget. In-use memory and +/// slots on a machine are `max(charged by this ledger, held by the machine's own report)` per +/// resource, so a report that lags the ledger does not count one compile twice and, after a +/// failover (ledger empty), the report alone counts. +/// +/// Admission is FIFO with backfill: a later request may be admitted past a blocked earlier one +/// only from memory and slots that are neither in use nor reserved. A dispatchable request +/// targets its reported candidates in placement order, then its caller; a local-only one its +/// caller. The oldest blocked request on a machine is that machine's head; once a request has +/// been head across a `Pump()` it reserves whatever is free (up to what it needs) on EVERY +/// reported machine of its target set, so a stream of small requests cannot starve it however +/// the candidate order changes. A request estimated above the whole budget runs alone: it +/// waits for an idle machine and reserves all free memory and slots meanwhile. +/// +/// A storable request whose `estimate.cacheKey` is already in flight from a machine that +/// shares its cache waits on that compile instead of running: it is never charged and holds no +/// reservation. When the owner stores, the waiter is answered `GrantKind::Stored` on the next +/// `Pump()`; when it does not, the waiter rejoins the queue at its original FIFO position. +/// +/// Not thread-safe: the owner locks. +class AdmissionQueue +{ + public: + /// A worker refusing the same request this many times sends it to its caller. + static constexpr std::uint32_t MaxWorkerRefusals = 3; + + /// Whether two machines read and write one compile cache. + using SharesCache = std::function; + + /// @param estimates Read at enqueue time; the figure is stored on the entry. + /// @param placement Who can run a toolchain. + /// @param sharesCache Whether a compile on one machine stores what the other would look up; + /// empty means only the same machine does. + AdmissionQueue(PeakEstimates& estimates, IPlacementSource const& placement, SharesCache sharesCache = {}); + + /// @param request The compile to queue. + /// @return Its ticket. + [[nodiscard]] Ticket Enqueue(QueueRequest request); + + /// Admits whatever fits, in FIFO order with backfill. + /// @return The admissions made, in grant order. + [[nodiscard]] std::vector Pump(); + + /// Records a machine's own statement; replaces the previous one. + /// @param machine The reporting machine. + /// @param report Its budget and what it holds. + void ReportMachine(MachineId const& machine, MachineReport const& report); + + /// A granted ticket finished: frees its charge, records its peak, and answers or releases + /// whatever waited on it. An unknown ticket is ignored. + /// @param ticket A ticket `Pump()` granted. + /// @param settlement How it ended. + void Settle(Ticket ticket, Settlement const& settlement); + + /// A grant `Pump()` made could not be handed over -- the worker it placed on left between the + /// placement and the lease. Frees its charge and puts it back at the FRONT, ahead of every queued + /// request, WITHOUT counting a refusal: nobody refused it, and a vanished worker must not spend the + /// request's refusal bound or mark its machine refused. Whatever waited on it rejoins the queue at + /// its own place. A ticket holding no charge is ignored. + /// @param ticket A ticket the last `Pump()` granted. + void ReturnToFront(Ticket ticket); + + /// The ticket's connection closed, at any stage. A granted ticket frees its charge and teaches + /// the estimator it was killed; a queued or waiting one simply leaves. An unknown ticket is ignored. + /// @param ticket Any ticket. + void Abandon(Ticket ticket); + + /// This node is no longer the leader. + /// @return Every queued and dedup-waiting ticket, now forgotten. Granted tickets stay tracked + /// so their `Settle`/`Abandon` still releases; every machine becomes unreported. + [[nodiscard]] std::vector LoseLeadership(); + + /// @return How many requests are waiting. + [[nodiscard]] std::size_t Depth() const noexcept; + + /// @return What this ledger has charged each machine, in bytes, machines in no particular order. + /// A machine with nothing charged is omitted. + [[nodiscard]] std::vector> ChargedByMachine() const; + + /// @return The queue as a report reads it: depth, who is compiling or waiting, and each machine's + /// charge against its budget. Order is unspecified; the caller names and sorts. + [[nodiscard]] QueueView Snapshot() const; + + private: + struct Entry + { + Ticket ticket {}; + QueueRequest request; + std::uint64_t chargedBytes {}; ///< The estimate taken at enqueue. + std::uint32_t refusals {}; ///< Worker refusals so far. + std::int64_t order {}; ///< Queue position key: the ticket, or below every key for a refused one. + }; + + /// What this ledger has charged a machine, beside its own report. + struct Ledger + { + std::optional report; + std::uint64_t chargedBytes {}; + std::uint32_t chargedSlots {}; + bool refused {}; ///< A worker here declined a job; admits nothing until the next report. + }; + + /// A granted entry and what it charged, to give back on `Settle`/`Abandon`. + struct Charge + { + MachineId machine; + std::uint64_t bytes {}; + Entry entry; + }; + + /// A granted storable compile and the requests waiting for its store. + struct Inflight + { + std::string cacheKey; + MachineId machine; ///< The owner's caller. + std::vector waiters; + }; + + /// Memory and slots held back from later requests on one machine during one pump. + struct Reserved + { + std::uint64_t bytes {}; + std::uint32_t slots {}; + }; + + /// What is free on a machine right now. + struct Room + { + std::uint64_t usedBytes {}; + std::uint32_t usedSlots {}; + std::uint64_t freeBytes {}; + std::uint32_t freeSlots {}; + }; + + /// The memory in use on a machine, as admission counts it: `max(charged by this ledger, held by its + /// own report)`, so after a failover (empty ledger) the report alone counts. The ONE place this is + /// computed: `RoomOn` and `Snapshot` both ask it. + /// @param ledger The machine's ledger. + /// @return Bytes in use. + [[nodiscard]] static std::uint64_t InUseBytes(Ledger const& ledger) noexcept; + + /// The memory a machine states it may be charged. + /// @param ledger The machine's ledger. + /// @return Its reported budget, or zero while it has reported none. + [[nodiscard]] static std::uint64_t BudgetOf(Ledger const& ledger) noexcept; + + /// @param machine A machine that has reported. + /// @param hold What earlier entries of this pump reserved there. + /// @return Its in-use and free figures. + [[nodiscard]] Room RoomOn(MachineId const& machine, Reserved const& hold) const; + + /// Frees a granted ticket's charge. + /// @return What it held, or nothing when the ticket holds no charge. + [[nodiscard]] std::optional Free(Ticket ticket); + + /// Puts entries back in the queue at their original FIFO place. + void Requeue(std::vector entries); + + /// Forgets an in-flight owner. + /// @return The requests that were waiting on it. + [[nodiscard]] std::vector TakeInflight(Ticket owner); + + PeakEstimates& _estimates; + SharesCache _sharesCache; + IPlacementSource const& _placement; + Ticket _nextTicket { 1 }; + std::vector _queue; + std::unordered_map _ledgers; + std::unordered_map _charges; + std::unordered_map _inflight; ///< Keyed by the owner's ticket. + std::unordered_multimap _inflightByKey; ///< cacheKey -> the owners running it. + std::int64_t _frontOrder {}; ///< Last key handed to a refused ticket; counts down from zero. + std::vector _stored; ///< Waiters woken by a store, answered by the next pump. + std::unordered_set _headTickets; ///< Entries that were a blocked head in the last pump. +}; + +} // namespace Pace::Distributed diff --git a/src/Pace/Distributed/AdmissionQueue_test.cpp b/src/Pace/Distributed/AdmissionQueue_test.cpp new file mode 100644 index 000000000..12a3dabfe --- /dev/null +++ b/src/Pace/Distributed/AdmissionQueue_test.cpp @@ -0,0 +1,834 @@ +// SPDX-License-Identifier: Apache-2.0 +#include + +#include + +#include +#include +#include + +#include + +using Pace::Testing::Unwrap; + +using namespace Pace::Distributed; + +namespace +{ +constexpr std::uint64_t GiB = 1ULL << 30; + +/// Returns what the test scripted per fingerprint, ignoring `excluded`. +class FakePlacement final: public IPlacementSource +{ + public: + std::map, std::less<>> byFingerprint; + + [[nodiscard]] std::vector Candidates(std::string_view fingerprint, + std::span /*excluded*/) const override + { + auto const found = byFingerprint.find(fingerprint); + return found == byFingerprint.end() ? std::vector {} : found->second; + } +}; + +struct Fixture +{ + PeakEstimates estimates { 3 * GiB, 64 }; + FakePlacement placement; + AdmissionQueue queue { estimates, placement }; + + /// Enqueues a request of `bytes`, the figure being taught to the estimator under its own key. + Ticket Add(AdmitKind kind, std::string caller, std::uint64_t bytes, std::string fingerprint = {}) + { + auto key = EstimateKey { .cacheKey = "key-" + std::to_string(++serial), + .sourcePath = "a.cpp", + .compilerIdentity = fingerprint }; + estimates.Record(key, bytes); + QueueRequest request; + request.kind = kind; + request.caller = std::move(caller); + request.fingerprint = std::move(fingerprint); + request.estimate = std::move(key); + return queue.Enqueue(std::move(request)); + } + + Ticket Local(std::uint64_t bytes, std::string caller = "m") + { + return Add(AdmitKind::LocalOnly, std::move(caller), bytes); + } + + /// Enqueues `count` local-only requests of `bytes` each. + std::vector LocalMany(int count, std::uint64_t bytes, std::string const& caller = "m") + { + std::vector tickets; + tickets.reserve(static_cast(count)); + for ([[maybe_unused]] auto const _: std::views::iota(0, count)) + tickets.push_back(Local(bytes, caller)); + return tickets; + } + + /// Enqueues a request whose estimate key is `cacheKey`, so two of them collide in the dedup set. + Ticket Keyed(AdmitKind kind, std::string caller, std::string cacheKey, bool storable, std::uint64_t bytes) + { + auto key = EstimateKey { .cacheKey = std::move(cacheKey), .sourcePath = "a.cpp", .compilerIdentity = {} }; + estimates.Record(key, bytes); + QueueRequest request; + request.kind = kind; + request.caller = std::move(caller); + request.estimate = std::move(key); + request.storable = storable; + return queue.Enqueue(std::move(request)); + } + + int serial {}; ///< Numbers the keys `Add` mints. +}; + +constexpr Settlement Done { .outcome = SettleOutcome::NotStored, .peakBytes = std::nullopt }; + +MachineReport Report(std::uint64_t budget, std::uint32_t slots, std::uint32_t heldSlots = 0, std::uint64_t heldBytes = 0) +{ + return MachineReport { .memoryBudgetBytes = budget, .slots = slots, .heldSlots = heldSlots, .heldBytes = heldBytes }; +} + +std::vector Tickets(std::vector const& decisions) +{ + std::vector result; + result.reserve(decisions.size()); + for (auto const& decision: decisions) + result.push_back(decision.ticket); + return result; +} +} // namespace + +TEST_CASE("AdmissionQueue.unreported-machine-admits-nothing", "[distributed][admission]") +{ + Fixture f; + f.Local(1 * GiB); + CHECK(f.queue.Pump().empty()); + CHECK(f.queue.Depth() == 1); + + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].kind == Pace::CompileCacheWire::GrantKind::Local); + CHECK(f.queue.Depth() == 0); +} + +TEST_CASE("AdmissionQueue.memory-bounds-admission", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 32)); + f.LocalMany(4, 3 * GiB); + CHECK(f.queue.Pump().size() == 2); + CHECK(f.queue.Depth() == 2); +} + +TEST_CASE("AdmissionQueue.slots-bound-admission", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(64 * GiB, 2)); + f.LocalMany(3, 3 * GiB); + CHECK(f.queue.Pump().size() == 2); + CHECK(f.queue.Depth() == 1); +} + +TEST_CASE("AdmissionQueue.held-charges-count", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4, 1, 6 * GiB)); + f.Local(3 * GiB); + CHECK(f.queue.Pump().empty()); + CHECK(f.queue.Depth() == 1); +} + +TEST_CASE("AdmissionQueue.held-and-charged-not-double-counted", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + f.Local(3 * GiB); + REQUIRE(f.queue.Pump().size() == 1); + + // The machine now reports the very compile the ledger charged. + f.queue.ReportMachine("m", Report(8 * GiB, 4, 1, 3 * GiB)); + f.Local(4 * GiB); + CHECK(f.queue.Pump().size() == 1); // 8 - max(3, 3) = 5 free; summing would leave 2 + + // And a report alone counts when the ledger knows nothing (failover): 5 GiB held, 3 free. + Fixture fresh; + fresh.queue.ReportMachine("m", Report(8 * GiB, 4, 1, 5 * GiB)); + fresh.Local(4 * GiB); + CHECK(fresh.queue.Pump().empty()); +} + +TEST_CASE("AdmissionQueue.backfill-behind-large-head", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 32)); + f.Local(4 * GiB); + REQUIRE(f.queue.Pump().size() == 1); + + auto const head = f.Local(6 * GiB); + auto const a = f.Local(1 * GiB); + auto const b = f.Local(1 * GiB); + auto const c = f.Local(1 * GiB); + CHECK(Tickets(f.queue.Pump()) == std::vector { a, b, c }); + CHECK(f.queue.Depth() == 1); + for (auto const ticket: Tickets(f.queue.Pump())) + CHECK(ticket != head); +} + +TEST_CASE("AdmissionQueue.head-reservation-prevents-starvation", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 32)); + auto const holder = f.Local(4 * GiB); + REQUIRE(f.queue.Pump().size() == 1); + auto const head = f.Local(6 * GiB); + auto const smalls = f.LocalMany(3, 1 * GiB); + REQUIRE(f.queue.Pump().size() == 3); + + f.queue.Settle(holder, Done); // 3 GiB in use, 5 free, head needs 6 + auto const late = f.Local(1 * GiB); + CHECK(f.queue.Pump().empty()); // the free 5 GiB is the head's, not the latecomer's + + for (auto const small: smalls) + f.queue.Settle(small, Done); + CHECK(Tickets(f.queue.Pump()) == std::vector { head, late }); // head first, then the 1 GiB fits too +} + +TEST_CASE("AdmissionQueue.over-budget-runs-alone", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4, 1, 1 * GiB)); + auto const big = f.Local(12 * GiB); + CHECK(f.queue.Pump().empty()); // the machine is not idle + + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].ticket == big); + CHECK(granted[0].ranAlone); + + f.Local(1 * GiB); + CHECK(f.queue.Pump().empty()); // nothing else while it runs + CHECK(f.queue.Depth() == 1); +} + +TEST_CASE("AdmissionQueue.over-budget-head-is-not-starved-by-backfill", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4, 1, 2 * GiB)); + auto const big = f.Local(12 * GiB); + REQUIRE(f.queue.Pump().empty()); // becomes head, the machine is busy + + f.LocalMany(3, 1 * GiB); + f.queue.ReportMachine("m", Report(8 * GiB, 4, 1, 1 * GiB)); // memory is freed while it waits + CHECK(f.queue.Pump().empty()); // ...and is the head's, not the small ones' + CHECK(f.queue.Depth() == 4); + + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].ticket == big); + CHECK(granted[0].ranAlone); + CHECK(f.queue.Depth() == 3); +} + +TEST_CASE("AdmissionQueue.fifo-among-equals", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(3 * GiB, 4)); + auto const first = f.Local(3 * GiB); + auto const second = f.Local(3 * GiB); + auto const third = f.Local(3 * GiB); + + CHECK(Tickets(f.queue.Pump()) == std::vector { first }); + f.queue.Settle(first, Done); + CHECK(Tickets(f.queue.Pump()) == std::vector { second }); + f.queue.Settle(second, Done); + CHECK(Tickets(f.queue.Pump()) == std::vector { third }); +} + +TEST_CASE("AdmissionQueue.dispatchable-uses-placement", "[distributed][admission]") +{ + Fixture f; + Placement const worker { .workerId = "worker-1", .machine = "w", .endpoint = "10.0.0.2:6680" }; + f.placement.byFingerprint["fp"] = { worker }; + f.queue.ReportMachine("w", Report(8 * GiB, 4)); + f.queue.ReportMachine("c", Report(8 * GiB, 4)); + + f.Add(AdmitKind::Dispatchable, "c", 5 * GiB, "fp"); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].kind == Pace::CompileCacheWire::GrantKind::Worker); + REQUIRE(granted[0].placement.has_value()); + CHECK(Unwrap(granted[0].placement).workerId == worker.workerId); + CHECK(Unwrap(granted[0].placement).machine == "w"); + CHECK(Unwrap(granted[0].placement).endpoint == worker.endpoint); + + // Machine w is charged: a second 5 GiB request does not fit there, and runs on the caller + // (which has room) rather than waiting; a third fits nowhere. + f.Add(AdmitKind::Dispatchable, "c", 5 * GiB, "fp"); + auto const second = f.queue.Pump(); + REQUIRE(second.size() == 1); + CHECK(second[0].kind == Pace::CompileCacheWire::GrantKind::Local); + CHECK_FALSE(second[0].placement.has_value()); + f.Add(AdmitKind::Dispatchable, "c", 5 * GiB, "fp"); + CHECK(f.queue.Pump().empty()); +} + +TEST_CASE("AdmissionQueue.dispatchable-with-full-candidates-runs-on-caller-with-room", "[distributed][admission]") +{ + Fixture f; + f.placement.byFingerprint["fp"] = { Placement { .workerId = "worker-1", .machine = "w", .endpoint = "10.0.0.2:6680" } }; + f.queue.ReportMachine("w", Report(8 * GiB, 4, 4, 8 * GiB)); + f.queue.ReportMachine("c", Report(8 * GiB, 4)); + f.Add(AdmitKind::Dispatchable, "c", 5 * GiB, "fp"); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].kind == Pace::CompileCacheWire::GrantKind::Local); +} + +TEST_CASE("AdmissionQueue.dispatchable-head-reserves-across-changing-candidate-order", "[distributed][admission]") +{ + Fixture f; + Placement const w1 { .workerId = "worker-1", .machine = "w1", .endpoint = "10.0.0.2:6680" }; + Placement const w2 { .workerId = "worker-2", .machine = "w2", .endpoint = "10.0.0.3:6680" }; + f.placement.byFingerprint["fp"] = { w1, w2 }; + f.queue.ReportMachine("w1", Report(8 * GiB, 32)); + f.queue.ReportMachine("w2", Report(8 * GiB, 32)); + f.queue.ReportMachine("c", Report(8 * GiB, 4, 4, 8 * GiB)); // the caller is full + // Each worker machine holds 4 GiB and 2 GiB: 6 GiB in use, so a 6 GiB head fits nowhere. + auto const big1 = f.Local(4 * GiB, "w1"); + auto const big2 = f.Local(4 * GiB, "w2"); + auto const small1 = f.Local(2 * GiB, "w1"); + auto const small2 = f.Local(2 * GiB, "w2"); + REQUIRE(f.queue.Pump().size() == 4); + + auto const head = f.Add(AdmitKind::Dispatchable, "c", 6 * GiB, "fp"); + REQUIRE(f.queue.Pump().empty()); // it becomes the head + + // The candidate order flips and the 2 GiB holders finish: 4 GiB is free on both machines, and + // the head (needing 6) must keep it from the latecomers on BOTH, not only the first candidate. + f.placement.byFingerprint["fp"] = { w2, w1 }; + f.queue.Settle(small1, Done); + f.queue.Settle(small2, Done); + f.Local(1 * GiB, "w1"); + f.Local(1 * GiB, "w2"); + CHECK(f.queue.Pump().empty()); + + f.queue.Settle(big1, Done); + f.queue.Settle(big2, Done); + auto const granted = f.queue.Pump(); + REQUIRE_FALSE(granted.empty()); + CHECK(granted.front().ticket == head); + CHECK(granted.front().kind == Pace::CompileCacheWire::GrantKind::Worker); +} + +TEST_CASE("AdmissionQueue.dispatchable-without-candidate-runs-local", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("c", Report(8 * GiB, 4)); + f.Add(AdmitKind::Dispatchable, "c", 5 * GiB, "unknown"); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].kind == Pace::CompileCacheWire::GrantKind::Local); + CHECK_FALSE(granted[0].placement.has_value()); + + // Charged to the caller: another 5 GiB does not fit there. + f.Add(AdmitKind::Dispatchable, "c", 5 * GiB, "unknown"); + CHECK(f.queue.Pump().empty()); +} + +using Pace::CompileCacheWire::GrantKind; + +TEST_CASE("AdmissionQueue.dedup-waiter-woken-by-store", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const first = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const second = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + CHECK(Tickets(f.queue.Pump()) == std::vector { first }); + + f.queue.Settle(first, Settlement { .outcome = SettleOutcome::Stored, .peakBytes = std::nullopt }); + auto const woken = f.queue.Pump(); + REQUIRE(woken.size() == 1); + CHECK(woken[0].ticket == second); + CHECK(woken[0].kind == GrantKind::Stored); + CHECK(f.queue.Depth() == 0); + CHECK(f.queue.Pump().empty()); +} + +TEST_CASE("AdmissionQueue.dedup-waiter-holds-no-charge", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(4 * GiB, 4)); + f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 3 * GiB); + f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 3 * GiB); + f.Local(1 * GiB); + CHECK(f.queue.Pump().size() == 2); // the owner and the 1 GiB one; the waiter took nothing + CHECK(f.queue.Depth() == 0); +} + +TEST_CASE("AdmissionQueue.dedup-not-woken-by-not-stored", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const first = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const second = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + REQUIRE(f.queue.Pump().size() == 1); + + f.queue.Settle(first, Settlement { .outcome = SettleOutcome::NotStored, .peakBytes = std::nullopt }); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].ticket == second); + CHECK(granted[0].kind == GrantKind::Local); +} + +TEST_CASE("AdmissionQueue.dedup-failed-owner-requeues-waiters", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const first = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const second = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + REQUIRE(f.queue.Pump().size() == 1); + + f.queue.Settle(first, Settlement { .outcome = SettleOutcome::Failed, .peakBytes = std::nullopt }); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].ticket == second); + CHECK(granted[0].kind == GrantKind::Local); +} + +TEST_CASE("AdmissionQueue.dedup-owner-abandon-requeues-waiters", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const first = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const second = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + REQUIRE(f.queue.Pump().size() == 1); + + f.queue.Abandon(first); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].ticket == second); + CHECK(granted[0].kind == GrantKind::Local); +} + +TEST_CASE("AdmissionQueue.requeued-waiter-keeps-fifo-position", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(4 * GiB, 4)); + auto const first = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 4 * GiB); + auto const waiter = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 4 * GiB); + auto const later = f.Local(4 * GiB); + REQUIRE(Tickets(f.queue.Pump()) == std::vector { first }); + + f.queue.Settle(first, Settlement { .outcome = SettleOutcome::NotStored, .peakBytes = std::nullopt }); + CHECK(Tickets(f.queue.Pump()) == std::vector { waiter }); // not `later`, which is younger + f.queue.Settle(waiter, Done); + CHECK(Tickets(f.queue.Pump()) == std::vector { later }); +} + +TEST_CASE("AdmissionQueue.abandoned-dedup-waiter-is-not-woken", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const first = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const second = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + REQUIRE(f.queue.Pump().size() == 1); + + f.queue.Abandon(second); + f.queue.Settle(first, Settlement { .outcome = SettleOutcome::Stored, .peakBytes = std::nullopt }); + CHECK(f.queue.Pump().empty()); +} + +TEST_CASE("AdmissionQueue.non-storable-not-deduplicated", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + f.Keyed(AdmitKind::LocalOnly, "m", "k", false, 1 * GiB); + f.Keyed(AdmitKind::LocalOnly, "m", "k", false, 1 * GiB); + CHECK(f.queue.Pump().size() == 2); +} + +TEST_CASE("AdmissionQueue.dedup-only-same-cache", "[distributed][admission]") +{ + { + Fixture f; + f.queue.ReportMachine("a", Report(8 * GiB, 4)); + f.queue.ReportMachine("b", Report(8 * GiB, 4)); + f.Keyed(AdmitKind::LocalOnly, "a", "k", true, 1 * GiB); + f.Keyed(AdmitKind::LocalOnly, "b", "k", true, 1 * GiB); + CHECK(f.queue.Pump().size() == 2); // default: only the same machine shares a cache + } + + PeakEstimates estimates { 3 * GiB, 64 }; + FakePlacement placement; + AdmissionQueue queue { estimates, placement, [](MachineId const&, MachineId const&) { return true; } }; + queue.ReportMachine("a", Report(8 * GiB, 4)); + queue.ReportMachine("b", Report(8 * GiB, 4)); + for (auto const* machine: { "a", "b" }) + { + QueueRequest request; + request.kind = AdmitKind::LocalOnly; + request.caller = machine; + request.estimate = EstimateKey { .cacheKey = "k", .sourcePath = "a.cpp", .compilerIdentity = {} }; + static_cast(queue.Enqueue(std::move(request))); + } + CHECK(queue.Pump().size() == 1); +} + +TEST_CASE("AdmissionQueue.abandon-while-queued", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(4 * GiB, 4)); + auto const holder = f.Local(4 * GiB); + REQUIRE(f.queue.Pump().size() == 1); + auto const head = f.Local(4 * GiB); + REQUIRE(f.queue.Pump().empty()); // becomes head + REQUIRE(f.queue.Pump().empty()); // reserving + f.queue.Abandon(head); + CHECK(f.queue.Depth() == 0); + + f.queue.Settle(holder, Done); + auto const next = f.Local(4 * GiB); // the whole budget, no reservation left behind + CHECK(Tickets(f.queue.Pump()) == std::vector { next }); +} + +TEST_CASE("AdmissionQueue.abandon-granted-releases-charge", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(4 * GiB, 4)); + auto key = EstimateKey { .cacheKey = "key-x", .sourcePath = "a.cpp", .compilerIdentity = {} }; + f.estimates.Record(key, 1 * GiB); + QueueRequest request; + request.kind = AdmitKind::LocalOnly; + request.caller = "m"; + request.estimate = key; + auto const ticket = f.queue.Enqueue(request); + REQUIRE(f.queue.Pump().size() == 1); + REQUIRE(f.estimates.Estimate(key) == 1 * GiB); + + f.queue.Abandon(ticket); + CHECK(f.estimates.Estimate(key) == 3 * GiB); // RecordKilled raised it to the default + + f.Local(4 * GiB); + CHECK(f.queue.Pump().size() == 1); // the charge went back +} + +TEST_CASE("AdmissionQueue.abandon-queued-records-nothing", "[distributed][admission]") +{ + Fixture f; + auto key = EstimateKey { .cacheKey = "key-y", .sourcePath = "a.cpp", .compilerIdentity = {} }; + f.estimates.Record(key, 1 * GiB); + QueueRequest request; + request.kind = AdmitKind::LocalOnly; + request.caller = "m"; + request.estimate = key; + auto const ticket = f.queue.Enqueue(request); + f.queue.Abandon(ticket); + CHECK(f.estimates.Estimate(key) == 1 * GiB); +} + +TEST_CASE("AdmissionQueue.settle-records-peak", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto key = EstimateKey { .cacheKey = "key-z", .sourcePath = "a.cpp", .compilerIdentity = {} }; + f.estimates.Record(key, 1 * GiB); + QueueRequest request; + request.kind = AdmitKind::LocalOnly; + request.caller = "m"; + request.estimate = key; + auto const ticket = f.queue.Enqueue(request); + REQUIRE(f.queue.Pump().size() == 1); + + f.queue.Settle(ticket, Settlement { .outcome = SettleOutcome::Stored, .peakBytes = 3 * GiB }); + CHECK(f.estimates.Estimate(key) == 3 * GiB); +} + +TEST_CASE("AdmissionQueue.worker-refusal-reparks-at-head", "[distributed][admission]") +{ + Fixture f; + f.placement.byFingerprint["fp"] = { Placement { .workerId = "worker-1", .machine = "w", .endpoint = "10.0.0.2:6680" } }; + f.queue.ReportMachine("w", Report(8 * GiB, 4)); + f.queue.ReportMachine("c", Report(8 * GiB, 4, 4, 8 * GiB)); // the caller is full + auto const ticket = f.Add(AdmitKind::Dispatchable, "c", 1 * GiB, "fp"); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + REQUIRE(granted[0].kind == GrantKind::Worker); + auto const younger = f.Add(AdmitKind::Dispatchable, "c", 1 * GiB, "fp"); + + f.queue.Settle(ticket, Settlement { .outcome = SettleOutcome::WorkerRefused, .peakBytes = std::nullopt }); + CHECK(f.queue.Depth() == 2); + CHECK(f.queue.Pump().empty()); // `w` admits nothing until it reports again + + f.queue.ReportMachine("w", Report(8 * GiB, 4)); + auto const again = f.queue.Pump(); + REQUIRE(again.size() == 2); + CHECK(again[0].ticket == ticket); // first in the queue + CHECK(again[1].ticket == younger); +} + +TEST_CASE("AdmissionQueue.worker-refusals-bounded", "[distributed][admission]") +{ + Fixture f; + f.placement.byFingerprint["fp"] = { Placement { .workerId = "worker-1", .machine = "w", .endpoint = "10.0.0.2:6680" } }; + f.queue.ReportMachine("c", Report(8 * GiB, 4)); + auto const ticket = f.Add(AdmitKind::Dispatchable, "c", 1 * GiB, "fp"); + for ([[maybe_unused]] auto const round: std::views::iota(0U, AdmissionQueue::MaxWorkerRefusals)) + { + f.queue.ReportMachine("w", Report(8 * GiB, 4)); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + REQUIRE(granted[0].kind == GrantKind::Worker); + f.queue.Settle(ticket, Settlement { .outcome = SettleOutcome::WorkerRefused, .peakBytes = std::nullopt }); + } + f.queue.ReportMachine("w", Report(8 * GiB, 4)); + auto const last = f.queue.Pump(); + REQUIRE(last.size() == 1); + CHECK(last[0].kind == GrantKind::Local); + CHECK_FALSE(last[0].placement.has_value()); +} + +TEST_CASE("AdmissionQueue.returned-grant-goes-to-the-front-uncounted", "[distributed][admission]") +{ + // A grant whose worker left before it could be handed over: back at the front, its charge freed, + // and NOT a refusal -- the machine is not marked refused and the request keeps its whole bound. + Fixture f; + f.placement.byFingerprint["fp"] = { Placement { .workerId = "worker-1", .machine = "w", .endpoint = "10.0.0.2:6680" } }; + f.queue.ReportMachine("w", Report(8 * GiB, 1)); + f.queue.ReportMachine("c", Report(8 * GiB, 4, 4, 8 * GiB)); // the caller is full + auto const ticket = f.Add(AdmitKind::Dispatchable, "c", 1 * GiB, "fp"); + REQUIRE(f.queue.Pump().size() == 1); + auto const younger = f.Add(AdmitKind::Dispatchable, "c", 1 * GiB, "fp"); + REQUIRE(f.queue.Pump().empty()); // the one slot is taken + + f.queue.ReturnToFront(ticket); + CHECK(f.queue.Depth() == 2); + CHECK(f.queue.ChargedByMachine().empty()); + + // No report in between: a refusal would have kept `w` shut until one. And ahead of the younger one. + auto const again = f.queue.Pump(); + REQUIRE(again.size() == 1); + CHECK(again[0].ticket == ticket); + CHECK(again[0].kind == GrantKind::Worker); + + // The bound is whole: two refusals later it still goes to a worker, which a counted return would not. + for ([[maybe_unused]] auto const round: std::views::iota(1U, AdmissionQueue::MaxWorkerRefusals)) + { + f.queue.Settle(ticket, Settlement { .outcome = SettleOutcome::WorkerRefused, .peakBytes = std::nullopt }); + f.queue.ReportMachine("w", Report(8 * GiB, 1)); + auto const granted = f.queue.Pump(); + REQUIRE(granted.size() == 1); + CHECK(granted[0].ticket == ticket); + CHECK(granted[0].kind == GrantKind::Worker); + } + CHECK(f.queue.Depth() == 1); + CHECK(Tickets(f.queue.Pump()).empty()); + (void) younger; +} + +TEST_CASE("AdmissionQueue.leadership-loss", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(4 * GiB, 4)); + auto const owner = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const waiter = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const queued = f.Local(4 * GiB); + REQUIRE(Tickets(f.queue.Pump()) == std::vector { owner }); + REQUIRE(f.queue.Depth() == 1); + + auto const parked = f.queue.LoseLeadership(); + CHECK(parked == std::vector { waiter, queued }); + CHECK(f.queue.Depth() == 0); + + f.Local(1 * GiB); + CHECK(f.queue.Pump().empty()); // 3 GiB would be free, but the machine is unreported + + f.queue.ReportMachine("m", Report(4 * GiB, 4)); + CHECK(f.queue.Pump().size() == 1); + + // The granted ticket is still tracked, so its settle releases cleanly: 1 + 1 GiB charged, one given back. + f.queue.Settle(owner, Done); + f.Local(3 * GiB); + CHECK(f.queue.Pump().size() == 1); +} + +TEST_CASE("AdmissionQueue.requeued-waiter-stays-behind-refused-front", "[distributed][admission]") +{ + Fixture f; + f.placement.byFingerprint["fp"] = { Placement { .workerId = "worker-1", .machine = "w", .endpoint = "10.0.0.2:6680" } }; + f.queue.ReportMachine("m", Report(4 * GiB, 4)); + f.queue.ReportMachine("w", Report(4 * GiB, 4)); + f.queue.ReportMachine("c", Report(4 * GiB, 4, 4, 4 * GiB)); // the caller is full + auto const owner = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 4 * GiB); + auto const older = f.Local(4 * GiB, "m"); + auto const waiter = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 4 * GiB); + auto const younger = f.Local(4 * GiB, "m"); + REQUIRE(Tickets(f.queue.Pump()) == std::vector { owner }); + auto const refused = f.Add(AdmitKind::Dispatchable, "c", 1 * GiB, "fp"); // a higher ticket than all of them + REQUIRE(Tickets(f.queue.Pump()) == std::vector { refused }); + + f.queue.Settle( + refused, + Settlement { .outcome = SettleOutcome::WorkerRefused, .peakBytes = std::nullopt }); // queue: refused, older, younger + f.queue.Settle(owner, + Settlement { .outcome = SettleOutcome::NotStored, .peakBytes = std::nullopt }); // the waiter rejoins + f.queue.ReportMachine("w", Report(4 * GiB, 4)); + CHECK(Tickets(f.queue.Pump()) == std::vector { refused, older }); // the refused ticket is still first + + f.queue.Settle(older, Done); + CHECK(Tickets(f.queue.Pump()) == std::vector { waiter }); // between `older` and `younger` + f.queue.Settle(waiter, Done); + CHECK(Tickets(f.queue.Pump()) == std::vector { younger }); +} + +TEST_CASE("AdmissionQueue.refused-machine-admits-no-local-request-until-reported", "[distributed][admission]") +{ + Fixture f; + f.placement.byFingerprint["fp"] = { Placement { .workerId = "worker-1", .machine = "w", .endpoint = "10.0.0.2:6680" } }; + f.queue.ReportMachine("w", Report(8 * GiB, 4)); + f.queue.ReportMachine("c", Report(8 * GiB, 4, 4, 8 * GiB)); + auto const job = f.Add(AdmitKind::Dispatchable, "c", 1 * GiB, "fp"); + REQUIRE(Tickets(f.queue.Pump()) == std::vector { job }); + auto const local = f.Local(1 * GiB, "w"); // its caller is the machine about to refuse + + f.queue.Settle(job, Settlement { .outcome = SettleOutcome::WorkerRefused, .peakBytes = std::nullopt }); + CHECK(f.queue.Pump().empty()); + + f.queue.ReportMachine("w", Report(8 * GiB, 4)); + auto const granted = Tickets(f.queue.Pump()); + CHECK(std::ranges::find(granted, local) != granted.end()); +} + +TEST_CASE("AdmissionQueue.woken-waiter-is-parked-by-leadership-loss", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const owner = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const waiter = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + REQUIRE(Tickets(f.queue.Pump()) == std::vector { owner }); + f.queue.Settle(owner, Settlement { .outcome = SettleOutcome::Stored, .peakBytes = std::nullopt }); + + CHECK(f.queue.LoseLeadership() == std::vector { waiter }); + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + CHECK(f.queue.Pump().empty()); // nothing is delivered after the loss +} + +TEST_CASE("AdmissionQueue.woken-waiter-abandoned-before-pump-is-not-delivered", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const owner = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const waiter = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + auto const other = f.Keyed(AdmitKind::LocalOnly, "m", "k", true, 1 * GiB); + REQUIRE(Tickets(f.queue.Pump()) == std::vector { owner }); + f.queue.Settle(owner, Settlement { .outcome = SettleOutcome::Stored, .peakBytes = std::nullopt }); + + f.queue.Abandon(waiter); + auto const delivered = f.queue.Pump(); + REQUIRE(delivered.size() == 1); + CHECK(delivered[0].ticket == other); + CHECK(delivered[0].kind == GrantKind::Stored); +} + +TEST_CASE("AdmissionQueue.not-stored-without-a-peak-teaches-the-estimator-nothing", "[distributed][admission]") +{ + // What a launcher settles when the worker its grant named failed the compile: the charge is freed, + // and neither the key's estimate nor its shape's moves -- unlike the abandonment it replaces, which + // reads as a KILLED compile and raises both to the default. + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 4)); + auto const key = EstimateKey { .cacheKey = "key-w", .sourcePath = "w.cpp", .compilerIdentity = "gcc-14" }; + auto const sameShape = EstimateKey { .cacheKey = "key-never-seen", .sourcePath = "w.cpp", .compilerIdentity = "gcc-14" }; + f.estimates.Record(key, 1 * GiB); + QueueRequest request; + request.kind = AdmitKind::LocalOnly; + request.caller = "m"; + request.fingerprint = "gcc-14"; + request.estimate = key; + auto const ticket = f.queue.Enqueue(request); + REQUIRE(f.queue.Pump().size() == 1); + auto expected = std::uint64_t { 0 }; + + SECTION("settled NotStored with no peak") + { + f.queue.Settle(ticket, Settlement { .outcome = SettleOutcome::NotStored, .peakBytes = std::nullopt }); + expected = 1 * GiB; + } + SECTION("control: abandoned") + { + f.queue.Abandon(ticket); + expected = 3 * GiB; + } + + CHECK(f.estimates.Estimate(key) == expected); + CHECK(f.estimates.Estimate(sameShape) == expected); + // Either way the charge went back. + f.Local(8 * GiB); + CHECK(f.queue.Pump().size() == 1); +} + +TEST_CASE("AdmissionQueue.snapshot-names-who-runs-and-who-waits", "[distributed][admission]") +{ + Fixture f; + f.queue.ReportMachine("m", Report(8 * GiB, 32)); + auto const enqueue = [&f](std::string caller, std::string tree, std::uint64_t bytes) { + auto key = EstimateKey { .cacheKey = "key-" + caller + tree + std::to_string(bytes), + .sourcePath = "a.cpp", + .compilerIdentity = {} }; + f.estimates.Record(key, bytes); + QueueRequest request; + request.kind = AdmitKind::LocalOnly; + request.caller = std::move(caller); + request.estimate = std::move(key); + request.treeLabel = std::move(tree); + request.storable = false; // Identical keys would otherwise wait on one another. + return f.queue.Enqueue(std::move(request)); + }; + enqueue("m", "wt7", 1 * GiB); + enqueue("m", "wt3", 3 * GiB); + enqueue("m", "wt3", 3 * GiB); + enqueue("m", "wt3", 3 * GiB); // 1 + 3 + 3 of 8 GiB are granted; this one does not fit. + REQUIRE(f.queue.Pump().size() == 3); + + auto const view = f.queue.Snapshot(); + CHECK(view.depth == 1); + auto const find = [&view](std::string const& tree) { + return std::ranges::find(view.clients, tree, &ClientLoad::tree); + }; + REQUIRE(find("wt3") != view.clients.end()); + CHECK(find("wt3")->running == 2); + CHECK(find("wt3")->queued == 1); + REQUIRE(find("wt7") != view.clients.end()); + CHECK(find("wt7")->running == 1); + CHECK(find("wt7")->queued == 0); + REQUIRE(view.machines.size() == 1); + CHECK(view.machines[0].chargedBytes == 7 * GiB); + CHECK(view.machines[0].budgetBytes == std::optional { 8 * GiB }); +} + +TEST_CASE("AdmissionQueue.snapshot-is-empty-before-anything-happens", "[distributed][admission]") +{ + Fixture f; + auto const view = f.queue.Snapshot(); + CHECK(view.depth == 0); + CHECK(view.clients.empty()); + CHECK(view.machines.empty()); +} + +TEST_CASE("AdmissionQueue.snapshot-in-use-is-what-admission-counts-after-a-failover", "[distributed][admission]") +{ + Fixture f; + // A new leader: the ledger is empty, the machine says it holds 7 GiB of its 8. + f.queue.ReportMachine("m", Report(8 * GiB, 32, 2, 7 * GiB)); + auto const view = f.queue.Snapshot(); + REQUIRE(view.machines.size() == 1); + CHECK(view.machines[0].chargedBytes == 0); + CHECK(view.machines[0].inUseBytes == 7 * GiB); + // And it is the figure admission refuses on: 2 GiB does not fit in the 1 GiB left. + f.Local(2 * GiB); + CHECK(f.queue.Pump().empty()); +} diff --git a/src/Pace/Distributed/FleetView.cpp b/src/Pace/Distributed/FleetView.cpp index cfa129c54..405c3df8c 100644 --- a/src/Pace/Distributed/FleetView.cpp +++ b/src/Pace/Distributed/FleetView.cpp @@ -30,6 +30,7 @@ #include #include #include +#include #include #include @@ -366,6 +367,81 @@ namespace } }, }; + /// What a compile-queue client row shows (#1728): who is compiling, for which build tree, and how + /// much of it is running against waiting. + constexpr std::array, 4> ClientColumns { + FleetColumn { + .name = "machine", + .help = "The machine the compiles came from, by its display name.", + .format = CellFormat::Text, + .keep = ColumnKeep::Identity, + .project = [](AdmissionClientRow const& row) { return FleetCell::Of(row.machine); } }, + FleetColumn { .name = "tree", + .help = "The build tree the client compiles for, as it labelled itself.", + .format = CellFormat::Text, + .keep = ColumnKeep::Vital, + .project = [](AdmissionClientRow const& row) { return FleetCell::Of(row.tree); } }, + FleetColumn { + .name = "running", + .help = "Compiles this client holds a grant for right now.", + .format = CellFormat::Count, + .keep = ColumnKeep::Vital, + .project = [](AdmissionClientRow const& row) { return FleetCell::Of(std::uint64_t { row.running }); } }, + FleetColumn { + .name = "queued", + .help = "Compiles of this client still waiting for memory or a slot.", + .format = CellFormat::Count, + .keep = ColumnKeep::Vital, + .project = [](AdmissionClientRow const& row) { return FleetCell::Of(std::uint64_t { row.queued }); } }, + }; + + /// What a compile-queue ledger row shows (#1728). + constexpr std::array, 4> BudgetColumns { + FleetColumn { + .name = "machine", + .help = "The machine, by its display name.", + .format = CellFormat::Text, + .keep = ColumnKeep::Identity, + .project = [](AdmissionMachineRow const& row) { return FleetCell::Of(row.machine); } }, + FleetColumn { + .name = "charged", + .help = "Memory this queue has charged the machine for compiles it granted. After a failover the " + "ledger starts empty, so this can read 0 on a full machine: read in-use for what admission sees.", + .format = CellFormat::Bytes, + .keep = ColumnKeep::Vital, + .project = [](AdmissionMachineRow const& row) { return FleetCell::Of(row.chargedBytes); } }, + FleetColumn { + .name = "in-use", + .help = "Memory admission counts as used on the machine: the larger of what this queue charged and " + "what the machine itself reports holding. A new request is admitted only if it fits what " + "is left of the budget after this.", + .format = CellFormat::Bytes, + .keep = ColumnKeep::Vital, + .project = [](AdmissionMachineRow const& row) { return FleetCell::Of(row.inUseBytes); } }, + FleetColumn { + .name = "budget", + .help = "The memory budget the machine stated. Absent until it has reported one.", + .format = CellFormat::Bytes, + .keep = ColumnKeep::Vital, + .project = [](AdmissionMachineRow const& row) { return FleetCell::Maybe(row.budgetBytes); } }, + }; + + /// The rows of the queue's two tables, or none where this node leads no queue: the marker line of the + /// full document and the header carry that, as they do for `Members`. + /// @param snapshot What to read. + /// @return The client rows. + [[nodiscard]] std::vector ClientRowsOf(FleetSnapshot const& snapshot) + { + return snapshot.admission.has_value() ? snapshot.admission->clients : std::vector {}; + } + + /// @param snapshot What to read. + /// @return The ledger rows. + [[nodiscard]] std::vector BudgetRowsOf(FleetSnapshot const& snapshot) + { + return snapshot.admission.has_value() ? snapshot.admission->machines : std::vector {}; + } + /// One row of the conditions section: a machine, and one condition it reported (#1364). /// /// **A machine that reported NO list is one row with no condition**, every cell but the @@ -1036,6 +1112,61 @@ std::optional PickCoverageFor(FleetSnapshot const& snapsh return ToolchainPickCoverage { .toolchains = reached.size(), .neverPicked = static_cast(never) }; } +namespace +{ + /// What to call a machine the queue knows only by its key: the first bytes in hex, so a row is + /// never nameless and two machines are still told apart. + /// @param key The machine's identity key, as raw bytes. + /// @return A short, printable form. + [[nodiscard]] std::string ShortMachineName(std::string_view key) + { + constexpr std::size_t ShownBytes = 4; + std::string out; + for (auto const byte: key.substr(0, ShownBytes)) + out += std::format("{:02x}", static_cast(static_cast(byte))); + return out.empty() ? std::string { "unknown" } : out; + } + + /// A machine's display name: the one its worker registered with, else the short form of its key. + /// @param key The machine's identity key. + /// @param workers The registry entries, which carry the key and the name. + /// @return The name; never empty. + [[nodiscard]] std::string MachineNameOf(std::string_view key, std::vector const& workers) + { + auto const named = std::ranges::find_if(workers, [key](WorkerReport const& worker) { + return worker.info.identityKey == key && !worker.info.displayName.empty(); + }); + return named != workers.end() ? named->info.displayName : ShortMachineName(key); + } + + /// The queue's own view, with machines named for a person and the rows put in a stable order. + /// @param view The queue, or none when this node leads none. + /// @param workers Where the names come from. + /// @return The section, or none. + [[nodiscard]] std::optional AdmissionSectionOf(std::optional const& view, + std::vector const& workers) + { + if (!view.has_value()) + return std::nullopt; + AdmissionSection section { .depth = view->depth, .clients = {}, .machines = {} }; + for (auto const& client: view->clients) + section.clients.push_back(AdmissionClientRow { .machine = MachineNameOf(client.machine, workers), + .tree = client.tree, + .running = client.running, + .queued = client.queued }); + for (auto const& machine: view->machines) + section.machines.push_back(AdmissionMachineRow { .machine = MachineNameOf(machine.machine, workers), + .chargedBytes = machine.chargedBytes, + .inUseBytes = machine.inUseBytes, + .budgetBytes = machine.budgetBytes }); + std::ranges::sort(section.clients, [](AdmissionClientRow const& a, AdmissionClientRow const& b) { + return std::tie(a.machine, a.tree) < std::tie(b.machine, b.tree); + }); + std::ranges::sort(section.machines, {}, &AdmissionMachineRow::machine); + return section; + } +} // namespace + FleetSnapshot CollectFleet(FleetSources const& sources) { FleetSnapshot snapshot; @@ -1089,6 +1220,8 @@ FleetSnapshot CollectFleet(FleetSources const& sources) for (auto const& tier: StorageTierTable) snapshot.tiersPresent[static_cast(tier.tier)] = AnyNodeHasTier(snapshot.nodes, tier.tier); + snapshot.admission = AdmissionSectionOf(sources.scheduler->AdmissionDetailNow(), snapshot.workers); + return snapshot; } @@ -1432,6 +1565,21 @@ std::string RenderFleetJson(FleetSnapshot const& snapshot, FleetHistoryView cons else out += "null"; // Runs no cluster -- not "a cluster that has revoked nothing". + // The compile queue (#1728): `null` where this node leads none, never a queue of zero. + out += ','; + AppendJsonString(out, "admission"); + out += ':'; + if (snapshot.admission.has_value()) + { + out += std::format(R"({{"depth":{},"clients":)", snapshot.admission->depth); + AppendJsonRows(out, ClientColumns, snapshot.admission->clients); + out += ",\"machines\":"; + AppendJsonRows(out, BudgetColumns, snapshot.admission->machines); + out += '}'; + } + else + out += "null"; + out += ','; AppendJsonString(out, "nodes"); out += ':'; @@ -1618,6 +1766,9 @@ namespace case FleetSection::Tiers: AppendTierText(out, snapshot); return; + case FleetSection::Clients: + AppendTextRows(out, ClientColumns, ClientRowsOf(snapshot)); + return; case FleetSection::Series: AppendSeriesText(out, history.buckets, history.range); return; @@ -1676,6 +1827,8 @@ std::vector FleetColumnNames(FleetSection section, FleetSnapshot co } return names; } + case FleetSection::Clients: + return namesOf(ClientColumns); case FleetSection::Series: // The bucket arrays and the series, from the tables `AppendSeriesText` walks; not a // `FleetColumn` set, and `tabular` says so. @@ -1751,6 +1904,8 @@ namespace .decor = CellDecor::Plain, .keep = ColumnKeep::Useful }; return std::nullopt; + case FleetSection::Clients: + return factsOf(ClientColumns); case FleetSection::Series: // Raw per-bucket values: a rate is per minute, a share a fraction, a level a count, so no // column of this section has one scale a human surface could apply. @@ -2340,6 +2495,29 @@ footer { margin-top:2.4rem; padding-top:1rem; border-top:1px solid var(--line); .sub = std::format("granted, {}", LeasesNote) }; } + /// Compiles waiting in the leader's queue (#1728). + /// + /// Absent, never zero, on a node that leads no queue: `0` would read as every compile admitted at + /// once, said for a fleet that has not reported a budget. + KpiReadout KpiQueueDepth(FleetSnapshot const& snapshot, FleetHistoryView const& /*history*/) + { + if (!snapshot.admission.has_value()) + return KpiReadout { .value = FleetCell::Nothing(), + .of = FleetCell::Nothing(), + .format = CellFormat::Count, + .unit = {}, + .sub = snapshot.role == SchedulerRole::Leader ? "no queue activity yet" + : "no compile queue on this node" }; + auto running = std::uint64_t { 0 }; + for (auto const& row: snapshot.admission->clients) + running += row.running; + return KpiReadout { .value = FleetCell::Of(std::uint64_t { snapshot.admission->depth }), + .of = FleetCell::Nothing(), + .format = CellFormat::Count, + .unit = {}, + .sub = std::format("waiting; {} running", running) }; + } + KpiReadout KpiOldestHeartbeat(FleetSnapshot const& snapshot, FleetHistoryView const& /*history*/) { if (snapshot.nodes.empty()) @@ -2430,7 +2608,7 @@ footer { margin-top:2.4rem; padding-top:1rem; border-top:1px solid var(--line); /// and one added after them rather than among them: the mockup's order is what a /// reader of this page already knows, and an insertion would move six tiles to /// place one. - constexpr std::array KpiTable { + constexpr std::array KpiTable { KpiRow { .label = "Dispatched", .key = "dispatched", .project = KpiDispatched, .sparkline = true }, KpiRow { .label = "Compiling now", .key = "compiling-now", @@ -2450,6 +2628,7 @@ footer { margin-top:2.4rem; padding-top:1rem; border-top:1px solid var(--line); .project = KpiNeverPicked, .ofNoun = ToolchainsNoun, .sparkline = false }, + KpiRow { .label = "Queued compiles", .key = "queue-depth", .project = KpiQueueDepth, .sparkline = false }, }; /// Every tile states a key, no two share one, and each is kebab-case. @@ -3245,6 +3424,24 @@ std::string RenderFleetHtml(FleetSnapshot const& snapshot, FleetHistoryView cons out += R"(

This node runs no cluster: it leads itself, and has no replicated state.

)"; out += ""; + // ---- compile queue -------------------------------------------------------- + out += R"(

Compile queue

)" + R"(the leader's admission queue
)"; + // The tables are drawn whether or not there is a queue, exactly as Members is: a section that + // renders its columns only in some states is a coverage hole no test over the tables can see. + out += R"(
)"; + AppendHtmlRows(out, ClientColumns, ClientRowsOf(snapshot)); + out += "
"; + out += R"(
)"; + AppendHtmlRows(out, BudgetColumns, BudgetRowsOf(snapshot)); + out += "
"; + if (snapshot.admission.has_value()) + out += std::format(R"(

{} waiting.

)", snapshot.admission->depth); + else + out += R"(

This node leads no compile queue, so there is nothing to show: )" + R"(not a queue of zero.

)"; + out += "
"; + out += R"(