From 87ed0cb1c21bfcbb4cc474901c130139ae6c689c Mon Sep 17 00:00:00 2001 From: joelteply Date: Wed, 11 Nov 2026 07:25:54 -0600 Subject: [PATCH 1/3] server: a learning ledger read answers from the loop's last publication, never queued behind the chunk it reports on The loop publishes its half of the ledger (grants, use, committed steps, the optimizer's memory) at every task and slot-update boundary, stamped with sequence and clock_us, under one mutex with a condvar signalled per publication (the wake a ticket wait will block on). GET /learn/grants answers from that publication plus the residency mirror (its own mutex) from the HTTP thread. Mutations stay loop tasks. Card 7fc7e2ce: on the IntelMac a 2,048-token prefill chunk ran ~40 s while the core waited 4 s, so every learner stop read as unreachable. --- tools/server/server-context-learn.inc | 55 +++++++++++++++++++-- tools/server/server-context.cpp | 15 +++++- tools/server/tests/learn-ledger-off-loop.py | 45 +++++++++++++++++ 3 files changed, 108 insertions(+), 7 deletions(-) create mode 100644 tools/server/tests/learn-ledger-off-loop.py diff --git a/tools/server/server-context-learn.inc b/tools/server/server-context-learn.inc index 925adcfcf01..422076c13bb 100644 --- a/tools/server/server-context-learn.inc +++ b/tools/server/server-context-learn.inc @@ -226,9 +226,18 @@ learn_ledger.revoke(body.value("id", std::string()), now); out["revoked"] = true; // idempotent: this id can admit no new work } - out["max_period_ms"] = server_learn_ledger::kMaxPeriodMs; - out["max_expiry_ms"] = server_learn_ledger::kMaxPeriodMs; - out["grants"] = json::array(); + const json receipt = ledger_receipt(now); + for (const auto & entry : receipt.items()) { out[entry.key()] = entry.value(); } + out["residencies"] = residency_snapshot(); + result->receipt = std::move(out); + queue_results.send(std::move(result)); + } + + // The loop-owned half of the ledger (grants, their use, the memory the optimizer holds), read + // ON THE LOOP: the reply to a mutation, and the snapshot every boundary publishes. + json ledger_receipt(int64_t now) const { + json out = {{"max_period_ms", server_learn_ledger::kMaxPeriodMs}, {"max_expiry_ms", server_learn_ledger::kMaxPeriodMs}, + {"grants", json::array()}}; for (const auto & a : learn_ledger.snapshot()) { out["grants"].push_back({{"id", a.grant.id}, {"period_ms", a.grant.period_ms}, {"max_chunk_us", a.grant.max_chunk_us}, {"share_ppm", a.grant.share_ppm}, {"expires_at_us", a.grant.expires_at_us}, @@ -240,9 +249,45 @@ } } out["memory"] = learn_memory_receipt(); + return out; + } + + // A control read never queues behind the work it reports on (card 7fc7e2ce, Fable's design): + // the loop publishes its half of the ledger at every task and slot-update boundary, stamped, + // and a reader on any thread takes the last publication plus the residency mirror (its own + // mutex). At most one boundary stale, and the stamp says when. + struct ledger_publication { + std::mutex mutex; + std::condition_variable published; // signalled at every publication: the wait-on-ticket wake + json receipt; + uint64_t sequence = 0; + }; + mutable ledger_publication ledger_published; + + // ON THE LOOP, at a boundary. + void publish_ledger() { + const auto now = ggml_time_us(); + json receipt = ledger_receipt(now); + receipt["clock_us"] = now; + { + std::lock_guard lock(ledger_published.mutex); + receipt["sequence"] = ++ledger_published.sequence; + ledger_published.receipt = std::move(receipt); + } + ledger_published.published.notify_all(); + } + + // ANY THREAD: the last publication, the live residency mirror and the engine identity (const). + json published_ledger() const { + json out; + { + std::lock_guard lock(ledger_published.mutex); + out = ledger_published.receipt; + } + out["engine"] = learn_engine; + out["refusal"] = "none"; out["residencies"] = residency_snapshot(); - result->receipt = std::move(out); - queue_results.send(std::move(result)); + return out; } diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index b03adacd3d3..7324b253dd0 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -910,6 +910,8 @@ struct server_context_impl { catch (const std::exception &) { return "invalid"; } } const std::string & learning_engine_identity() const { return learn_engine; } + // the ledger as the loop last published it; readable from any thread (card 7fc7e2ce) + json ledger_as_published() const { return published_ledger(); } json lora_report_snapshot() const { std::lock_guard lock(lora_report_mutex); @@ -1571,16 +1573,20 @@ struct server_context_impl { // wiring up server queues queue_tasks.on_new_task([this](server_task && task, bool is_yielding) { - return process_single_task(std::move(task), is_yielding); + auto handled = process_single_task(std::move(task), is_yielding); + publish_ledger(); // a task boundary: what it changed is readable off the loop + return handled; }); queue_tasks.on_update_slots([this]() { update_slots(); + publish_ledger(); // a chunk boundary }); queue_tasks.on_sleeping_state([this](bool sleeping) { handle_sleeping_state(sleeping); }); metrics.init(); + publish_ledger(); // readable before the loop's first boundary if (params_base.cache_idle_slots) { if (params_base.cache_ram_mib == 0) { @@ -6577,7 +6583,12 @@ void server_routes::init_routes() { this->post_learn_residency = [grant_route](const server_http_req & req) { return grant_route(req, "residency_register"); }; this->release_learn_residency = [grant_route](const server_http_req & req) { return grant_route(req, "residency_release"); }; this->revoke_learn_residency = [grant_route](const server_http_req & req) { return grant_route(req, "residency_revoke"); }; - this->get_learn_grants = [grant_route](const server_http_req & req) { return grant_route(req, "list"); }; + // a read answers from the loop's last publication, never queued behind the chunk it reports on + this->get_learn_grants = [this](const server_http_req &) { + auto res = create_response(); + res->ok(ctx_server.ledger_as_published()); + return res; + }; this->post_learn_grants = [grant_route](const server_http_req & req) { return grant_route(req, "register"); }; this->revoke_learn_grant = [grant_route](const server_http_req & req) { return grant_route(req, "revoke"); }; diff --git a/tools/server/tests/learn-ledger-off-loop.py b/tools/server/tests/learn-ledger-off-loop.py new file mode 100644 index 00000000000..41394996a2b --- /dev/null +++ b/tools/server/tests/learn-ledger-off-loop.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python3 +"""Real server test; requires an isolated server. Usage: learn-ledger-off-loop.py URL + +What this catches (card 7fc7e2ce): a ledger read queued behind the work it controls. On a CPU +node one prefill chunk ran ~40 s while the core waited 4 s for GET /learn/grants, so every +learner stop read as unreachable. The read answers from the loop's last publication: it carries +the publication's sequence and clock, the sequence advances at boundaries, a grant registered +through the loop is in the next read, and a read answers while a completion is in flight. +Run with --train-dir FIXTURE. +""" +import os +import sys +import threading + +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine + +engine = Engine(sys.argv[1], "ledger", timeout=120) + +first = engine.get("/learn/grants") +for key in ("sequence", "clock_us", "grants", "memory", "residencies", "engine", "max_expiry_ms"): + assert key in first, ("a read carries the publication's stamp and both halves of the ledger", key, first) +assert first["engine"] == engine.engine, first + +# a mutation goes through the loop; the publication after it holds its result +gid = engine.register_grant("-seen") +seen = engine.get("/learn/grants") +assert seen["sequence"] > first["sequence"], ("a boundary publishes again", first["sequence"], seen["sequence"]) +assert any(g["id"] == gid for g in seen["grants"]), ("the registered grant is in the next read", gid, seen["grants"]) + +# a read answers while a completion holds the loop +done = threading.Event() +def complete(): + engine.ok("/completion", {"prompt": "Count the windows in the room:", "n_predict": 256, "temperature": 0}) + done.set() +worker = threading.Thread(target=complete) +worker.start() +during = engine.get("/learn/grants") +answered_while_busy = not done.is_set() +worker.join() +assert during["sequence"] >= seen["sequence"], during +print("read during a completion, before it ended:", answered_while_busy) + +engine.close() +print("ok: ledger reads answer from the loop's publication") From 26bcec870b1d9aba33a5836471dc108ca8bfcfbd Mon Sep 17 00:00:00 2001 From: joelteply Date: Wed, 11 Nov 2026 08:03:52 -0600 Subject: [PATCH 2/3] server: the ledger publishes at a slot-update boundary only when it changed; the fixture asserts the read beside a running completion server_learn_ledger::revision() is bumped by every mutation; on_update_slots publishes only when it moved (task boundaries and init still publish), so no ledger JSON or memory visit runs per decode batch. The fixture waits for /slots to report the completion processing, reads, and asserts the completion is still running (Fable on fork #109). --- tools/server/server-context-learn.inc | 10 ++++++++++ tools/server/server-context.cpp | 2 +- tools/server/server-learn-ledger.h | 13 +++++++++++++ tools/server/tests/learn-ledger-off-loop.py | 13 +++++++++---- 4 files changed, 33 insertions(+), 5 deletions(-) diff --git a/tools/server/server-context-learn.inc b/tools/server/server-context-learn.inc index 422076c13bb..b39a0789c92 100644 --- a/tools/server/server-context-learn.inc +++ b/tools/server/server-context-learn.inc @@ -263,9 +263,19 @@ uint64_t sequence = 0; }; mutable ledger_publication ledger_published; + uint64_t ledger_published_revision = 0; // loop-owned: the ledger revision the last publication holds + + // ON THE LOOP, at a slot-update boundary: publish only when the ledger changed since the last + // publication (a task boundary, below, always publishes). + void publish_ledger_if_changed() { + if (learn_ledger.revision() != ledger_published_revision) { + publish_ledger(); + } + } // ON THE LOOP, at a boundary. void publish_ledger() { + ledger_published_revision = learn_ledger.revision(); const auto now = ggml_time_us(); json receipt = ledger_receipt(now); receipt["clock_us"] = now; diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 7324b253dd0..d7b9cae6ad4 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -1579,7 +1579,7 @@ struct server_context_impl { }); queue_tasks.on_update_slots([this]() { update_slots(); - publish_ledger(); // a chunk boundary + publish_ledger_if_changed(); // a chunk boundary: a learn chunk moved the ledger }); queue_tasks.on_sleeping_state([this](bool sleeping) { handle_sleeping_state(sleeping); diff --git a/tools/server/server-learn-ledger.h b/tools/server/server-learn-ledger.h index 2e1dc5cd0cf..06642dcc487 100644 --- a/tools/server/server-learn-ledger.h +++ b/tools/server/server-learn-ledger.h @@ -76,8 +76,13 @@ class server_learn_ledger { explicit server_learn_ledger(std::string engine) : engine(std::move(engine)) {} + // Bumped by every mutation: the loop publishes the ledger at a slot-update boundary only when + // this moved since its last publication (card 7fc7e2ce; Fable on fork #109: no JSON per batch). + uint64_t revision() const { return revision_; } + // The authority registers (or re-confirms) a grant. server_learn_refusal register_grant(const server_learn_grant & g, int64_t now_us) { + ++revision_; // every mutation is a new ledger the loop must publish if (!in_clock(now_us) || g.id.empty() || (g.period_ms < 0 || (g.period_ms == 0 && g.share_ppm != 0)) || g.period_ms > kMaxPeriodMs || g.share_ppm < 0 || g.share_ppm > 1000000 || g.max_chunk_us < 0 || g.max_chunk_us > kMaxCostUs || g.expires_at_us <= now_us || !in_clock(g.expires_at_us)) { return server_learn_refusal::invalid; @@ -102,6 +107,7 @@ class server_learn_ledger { // The authority ends a grant. Its account goes once its last reservation settles. void revoke(const std::string & id, int64_t now_us) { + ++revision_; // every mutation is a new ledger the loop must publish if (!in_clock(now_us)) { return; } @@ -116,6 +122,7 @@ class server_learn_ledger { // serving at admission, so the step's time is debited; an uncontended step is admitted // without a debit but is still a live reservation. server_learn_admission admit(const std::string & grant, int64_t est_us, int64_t now_us, bool contended) { + ++revision_; // every mutation is a new ledger the loop must publish server_learn_admission out; if (est_us < 0 || est_us > kMaxCostUs || !in_clock(now_us)) { out.refusal = server_learn_refusal::invalid; @@ -155,6 +162,7 @@ class server_learn_ledger { // Latency admission is separate from duty and restoration custody. A new // serving arrival may wait for this segment even when admission is idle. server_learn_admission admit_chunk(const std::string & grant, int64_t measured_us, int64_t now_us, bool contended) { + ++revision_; // every mutation is a new ledger the loop must publish if (measured_us < 0 || measured_us > kMaxCostUs || !in_clock(now_us)) { return {server_learn_refusal::invalid}; } @@ -179,6 +187,7 @@ class server_learn_ledger { // Restoration is an obligation, not permission for a gradient. Keep its // measured debit even on an idle-only grant and after revocation. server_learn_admission hold_restoration(const std::string & grant, int64_t now_us) { + ++revision_; // every mutation is a new ledger the loop must publish auto out = admit(grant, 0, now_us, false); if (out.refusal == server_learn_refusal::none) { reservations.at(out.reservation).contended = true; @@ -190,6 +199,7 @@ class server_learn_ledger { // grant was revoked or expired after admission (unknown_grant): the reservation stays held, so // the caller cancels it (it never ran) and nothing is erased silently. server_learn_refusal start(uint64_t id, int64_t now_us) { + ++revision_; // every mutation is a new ledger the loop must publish if (!in_clock(now_us)) { return server_learn_refusal::invalid; } @@ -211,6 +221,7 @@ class server_learn_ledger { // The step quiesced after measured_us with committed_steps optimizer steps: debit the measured // time (when it was admitted contended) and release the rest of its reservation. server_learn_refusal finish(uint64_t id, int64_t measured_us, int64_t committed_steps, int64_t now_us) { + ++revision_; // every mutation is a new ledger the loop must publish if (measured_us < 0 || measured_us > kMaxCostUs || committed_steps < 0 || committed_steps > kMaxStepsPerSettle || !in_clock(now_us)) { return server_learn_refusal::invalid; @@ -220,6 +231,7 @@ class server_learn_ledger { // The step never ran: its whole reservation returns. Refused once it started. server_learn_refusal cancel(uint64_t id, int64_t now_us) { + ++revision_; // every mutation is a new ledger the loop must publish if (!in_clock(now_us)) { return server_learn_refusal::invalid; } @@ -263,6 +275,7 @@ class server_learn_ledger { } private: + uint64_t revision_ = 0; struct account { server_learn_grant grant; int64_t period_start_us; diff --git a/tools/server/tests/learn-ledger-off-loop.py b/tools/server/tests/learn-ledger-off-loop.py index 41394996a2b..8b3abfc7a7f 100644 --- a/tools/server/tests/learn-ledger-off-loop.py +++ b/tools/server/tests/learn-ledger-off-loop.py @@ -28,18 +28,23 @@ assert seen["sequence"] > first["sequence"], ("a boundary publishes again", first["sequence"], seen["sequence"]) assert any(g["id"] == gid for g in seen["grants"]), ("the registered grant is in the next read", gid, seen["grants"]) -# a read answers while a completion holds the loop +# a read answers while a completion holds the loop: the completion fills its slot's window +# (ignore_eos), the read is taken only once /slots reports it processing, and the completion must +# still be running when the read returns done = threading.Event() +budget = engine.get("/slots")[0]["n_ctx"] // 2 def complete(): - engine.ok("/completion", {"prompt": "Count the windows in the room:", "n_predict": 256, "temperature": 0}) + engine.ok("/completion", {"prompt": "Count the windows in the room:", "n_predict": budget, + "temperature": 0, "ignore_eos": True, "id_slot": 0}) done.set() worker = threading.Thread(target=complete) worker.start() +while not any(slot.get("is_processing") for slot in engine.get("/slots")): + assert not done.is_set(), "the completion ended before it was ever seen processing" during = engine.get("/learn/grants") -answered_while_busy = not done.is_set() +assert not done.is_set(), "the ledger read waited for the completion it ran beside" worker.join() assert during["sequence"] >= seen["sequence"], during -print("read during a completion, before it ended:", answered_while_busy) engine.close() print("ok: ledger reads answer from the loop's publication") From d991ff344b78a84aa8865a3e8f954a7889fe0759 Mon Sep 17 00:00:00 2001 From: joelteply Date: Wed, 11 Nov 2026 08:11:30 -0600 Subject: [PATCH 3/3] server: the loop's per-token continuation publishes the ledger only when it changed Each decode step posts a SERVER_TASK_TYPE_NEXT_RESPONSE task, so publishing at every task boundary still published per token (1,025 publications for a 512-token completion). A continuation now publishes only when the ledger revision moved; real tasks (requests, ledger and adapter operations) still publish at their boundary. Measured: 1 publication for a 512-token completion. --- tools/server/server-context.cpp | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index d7b9cae6ad4..98a99ffb056 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -1573,8 +1573,15 @@ struct server_context_impl { // wiring up server queues queue_tasks.on_new_task([this](server_task && task, bool is_yielding) { + // the loop's own per-step continuation changes the ledger only through it (revision); + // every other task (a request, a ledger or adapter operation) publishes at its boundary + const bool continuation = task.type == SERVER_TASK_TYPE_NEXT_RESPONSE; auto handled = process_single_task(std::move(task), is_yielding); - publish_ledger(); // a task boundary: what it changed is readable off the loop + if (continuation) { + publish_ledger_if_changed(); + } else { + publish_ledger(); + } return handled; }); queue_tasks.on_update_slots([this]() {