diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 98a99ffb056..6c065a19bb4 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -6554,6 +6554,15 @@ void server_routes::init_routes() { } } task.learn_grants["op"] = op; + if (task.learn_grants.value("ticket", false)) { + // a ticketed mutation answers at once; its result arrives on GET /learn/events (card 7fc7e2ce) + task.id = queue_tasks.get_new_id(); + const int ticket = task.id; + queue_results.add_ticket(ticket); + queue_tasks.post(std::move(task)); + res->ok({{"ticket", ticket}}); + return res; + } auto & rd = res->rd; task.id = rd.get_new_id(); rd.post_task(std::move(task)); @@ -6590,6 +6599,47 @@ 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"); }; + // The learning event stream (card 7fc7e2ce): every ticketed mutation's terminal result, pushed to + // the one live subscriber, then gone. Opened once per engine incarnation by the core; its end is + // the incarnation gate (the stream drops with the engine). Idle, it sends SSE keepalives. + this->get_learn_events = [this](const server_http_req & req) { + struct ticket_stream : server_res_generator { + server_response & results; + ticket_stream(server_queue & tasks, server_response & results, int sleep_idle_seconds) + : server_res_generator(tasks, results, sleep_idle_seconds, true), results(results) { + results.subscribe_tickets(); + } + void on_complete() override { + results.unsubscribe_tickets(); + server_res_generator::on_complete(); + } + }; + auto res = std::make_unique(queue_tasks, queue_results, params.sleep_idle_seconds); + res->set_req(&req); // its connection ends the stream (should_stop) + res->status = 200; + res->content_type = "text/event-stream"; + res->data = "data: " + safe_json_to_str({{"engine", ctx_server.learning_engine_identity()}, {"subscribed", true}}) + "\n\n"; + const int keepalive = params.sse_ping_interval > 0 ? params.sse_ping_interval : 1; + res->set_next([res_this = res.get(), keepalive](std::string & output) -> bool { + if (res_this->should_stop()) { + return false; + } + if (!res_this->data.empty()) { + output = std::move(res_this->data); + res_this->data.clear(); + return true; + } + auto result = res_this->results.next_ticket(keepalive); + if (!result) { + output = ":\n\n"; // keepalive + return true; + } + output = "data: " + safe_json_to_str({{"ticket", result->id}, {"error", result->is_error()}, {"result", result->to_json()}}) + "\n\n"; + return true; + }); + return res; + }; + // 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(); @@ -6708,6 +6758,18 @@ void server_routes::init_routes() { res->error(format_error_response("\"id\" (an integer adapter id) is required", ERROR_TYPE_INVALID_REQUEST)); return res; } + if (body.value("ticket", false)) { + // a ticketed unload answers at once; its result arrives on GET /learn/events (card 7fc7e2ce) + server_task task(SERVER_TASK_TYPE_UNLOAD_LORA); + task.id = queue_tasks.get_new_id(); + task.unload_lora = body.at("id").get(); + task.unload_engine = body.at("engine").get(); + const int ticket = task.id; + queue_results.add_ticket(ticket); + queue_tasks.post(std::move(task)); + res->ok({{"ticket", ticket}}); + return res; + } auto & rd = res->rd; { server_task task(SERVER_TASK_TYPE_UNLOAD_LORA); diff --git a/tools/server/server-context.h b/tools/server/server-context.h index a810c3e3a1b..8a854c93963 100644 --- a/tools/server/server-context.h +++ b/tools/server/server-context.h @@ -177,6 +177,7 @@ struct server_routes { server_http_context::handler_t post_learn; server_http_context::handler_t post_learn_residency, release_learn_residency, revoke_learn_residency; server_http_context::handler_t get_learn_grants, post_learn_grants, revoke_learn_grant, post_learn_preflight; + server_http_context::handler_t get_learn_events; server_http_context::handler_t get_train; server_http_context::handler_t post_train; server_http_context::handler_t post_train_cancel; diff --git a/tools/server/server-queue.cpp b/tools/server/server-queue.cpp index 78169e9a5d8..8d3ae9350be 100644 --- a/tools/server/server-queue.cpp +++ b/tools/server/server-queue.cpp @@ -481,6 +481,14 @@ void server_response::send(server_task_result_ptr && result) { RES_DBG("sending result for task id = %d\n", result->id); std::unique_lock lock(mutex_results); + if (ticketed_ids.erase(result->id) > 0) { + // stored before the notify, under the same lock the stream waits with (no lost wakeup) + if (ticket_subscribers > 0) { + ticket_results.emplace_back(std::move(result)); + condition_results.notify_all(); + } + return; + } for (const auto & id_task : waiting_task_ids) { if (result->id == id_task) { RES_DBG("task id = %d pushed to result queue\n", result->id); @@ -619,3 +627,32 @@ void server_response_reader::stop() { SRV_DBG("%s", "all tasks already finished, no need to cancel\n"); } } + +void server_response::add_ticket(int id_task) { + std::unique_lock lock(mutex_results); + ticketed_ids.insert(id_task); +} + +void server_response::subscribe_tickets() { + std::unique_lock lock(mutex_results); + ++ticket_subscribers; +} + +void server_response::unsubscribe_tickets() { + std::unique_lock lock(mutex_results); + if (--ticket_subscribers <= 0) { + ticket_subscribers = 0; + ticket_results.clear(); // nobody is listening: undelivered results are gone + } +} + +server_task_result_ptr server_response::next_ticket(int timeout) { + std::unique_lock lock(mutex_results); + condition_results.wait_for(lock, std::chrono::seconds(timeout), [&] { return !running || !ticket_results.empty(); }); + if (!running || ticket_results.empty()) { + return nullptr; + } + server_task_result_ptr res = std::move(ticket_results.front()); + ticket_results.erase(ticket_results.begin()); + return res; +} diff --git a/tools/server/server-queue.h b/tools/server/server-queue.h index e17733a743f..16426ebc591 100644 --- a/tools/server/server-queue.h +++ b/tools/server/server-queue.h @@ -164,7 +164,24 @@ struct server_response { std::mutex mutex_results; std::condition_variable condition_results; + // Continuum fork (card 7fc7e2ce): ticketed loop mutations. A ticketed task's result goes to the + // learning event stream's subscriber, then it is gone (delivery is the drop); with no subscriber + // it is dropped, and a core that (re)connects reads the physical state, never a replay. + std::unordered_set ticketed_ids; + std::vector ticket_results; + int ticket_subscribers = 0; + public: + // register `id_task` as ticketed: its result is delivered on the event stream, not to a waiter + void add_ticket(int id_task); + + // the learning event stream opens / closes; the last close drops anything undelivered + void subscribe_tickets(); + void unsubscribe_tickets(); + + // blocks until a ticketed result arrives (returned) or `timeout` seconds pass (nullptr: keepalive) + server_task_result_ptr next_ticket(int timeout); + // add the id_task to the list of tasks waiting for response void add_waiting_task_id(int id_task); diff --git a/tools/server/server.cpp b/tools/server/server.cpp index 4a4ec63ab39..5c93424cc7d 100644 --- a/tools/server/server.cpp +++ b/tools/server/server.cpp @@ -224,6 +224,7 @@ int llama_server(common_params & params, int argc, char ** argv) { routes.post_lora_adapters_unload = models_routes->proxy_post; routes.post_learn = models_routes->proxy_post; routes.get_learn_grants = models_routes->proxy_get; + routes.get_learn_events = models_routes->proxy_get; routes.post_learn_grants = models_routes->proxy_post; routes.post_learn_preflight = models_routes->proxy_post; routes.post_learn_residency = models_routes->proxy_post; @@ -295,6 +296,7 @@ int llama_server(common_params & params, int argc, char ** argv) { ctx_http.post("/learn/residency/revoke", ex_wrapper(routes.revoke_learn_residency)); ctx_http.post("/learn/preflight", ex_wrapper(routes.post_learn_preflight)); ctx_http.post("/learn/grants", ex_wrapper(routes.post_learn_grants)); + ctx_http.get ("/learn/events", ex_wrapper(routes.get_learn_events)); ctx_http.post("/learn/grants/revoke", ex_wrapper(routes.revoke_learn_grant)); ctx_http.get ("/train", ex_wrapper(routes.get_train)); ctx_http.post("/train", ex_wrapper(routes.post_train)); diff --git a/tools/server/tests/learn-tickets.py b/tools/server/tests/learn-tickets.py new file mode 100644 index 00000000000..bbe181ce0ef --- /dev/null +++ b/tools/server/tests/learn-tickets.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Real server test; requires an isolated server. Usage: learn-tickets.py URL + +What this catches (card 7fc7e2ce): a learning mutation the core must wait out behind the work it +controls. On a CPU node one prefill chunk ran ~40 s while the core waited 4 s, so every learner +stop read as unreachable. A ticketed mutation answers at once with its ticket; its terminal result +is pushed on GET /learn/events (one stream per subscriber, no polling) and then gone. A result +made while nobody subscribes is dropped, never replayed to a later subscriber: a reconnecting core +reads the physical state instead. +Run with --train-dir FIXTURE --no-repack. +""" +import json +import os +import queue +import sys +import threading +import urllib.request + +sys.path.insert(0, os.path.dirname(__file__)) +from learn_fixture import Engine + +engine = Engine(sys.argv[1], "tickets", timeout=120) + + +def subscribe(): + """Open the event stream; its data events land on the returned queue.""" + events = queue.Queue() + stream = urllib.request.urlopen(engine.url + "/learn/events", timeout=engine.timeout) + + def read(): + try: + for raw in stream: + line = raw.decode().strip() + if line.startswith("data: "): + events.put(json.loads(line[len("data: "):])) + except Exception as error: # the stream closed under us + events.put({"closed": str(error)}) + + threading.Thread(target=read, daemon=True).start() + first = events.get(timeout=engine.timeout) + assert first.get("subscribed") is True and first.get("engine") == engine.engine, first + return stream, events + + +def ticketed(route, body): + reply = engine.ok(route, dict(body, ticket=True)) + assert set(reply) == {"ticket"}, ("a ticketed mutation answers only its ticket, at once", reply) + return reply["ticket"] + + +# with nobody subscribed, a ticketed result is dropped: never replayed later +gid_unseen = engine.register_grant("-unseen") +engine.grants.remove(gid_unseen) +unseen = ticketed("/learn/grants/revoke", {"id": gid_unseen}) + +stream, events = subscribe() + +# a ticketed grant revoke: its terminal result arrives on the stream under its ticket +gid = engine.register_grant("-revoked") +engine.grants.remove(gid) +ticket = ticketed("/learn/grants/revoke", {"id": gid}) +event = events.get(timeout=engine.timeout) +assert event["ticket"] != unseen, ("a result made while nobody listened is never replayed", event) +assert event["ticket"] == ticket and not event["error"], event +assert event["result"]["revoked"] is True, event + +# a ticketed unload: the adapter's retirement arrives the same way +shadow = engine.fresh() +ticket = ticketed("/lora-adapters/unload", {"id": shadow["id"], "engine": engine.engine}) +event = events.get(timeout=engine.timeout) +assert event["ticket"] == ticket and not event["error"], event +engine.shadows.remove(shadow["id"]) + +stream.close() +engine.close() +print("ok: ticketed mutations answer at once and their results arrive on the event stream")