diff --git a/tools/server/server-context-learn.inc b/tools/server/server-context-learn.inc index 925adcfcf01..b39a0789c92 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,55 @@ } } 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; + 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; + { + 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..98a99ffb056 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,27 @@ 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); + // 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); + if (continuation) { + publish_ledger_if_changed(); + } else { + publish_ledger(); + } + return handled; }); queue_tasks.on_update_slots([this]() { update_slots(); + 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); }); 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 +6590,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/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 new file mode 100644 index 00000000000..8b3abfc7a7f --- /dev/null +++ b/tools/server/tests/learn-ledger-off-loop.py @@ -0,0 +1,50 @@ +#!/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: 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": 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") +assert not done.is_set(), "the ledger read waited for the completion it ran beside" +worker.join() +assert during["sequence"] >= seen["sequence"], during + +engine.close() +print("ok: ledger reads answer from the loop's publication")