diff --git a/tools/server/README-dev.md b/tools/server/README-dev.md index dfc9004de5..882adca09b 100644 --- a/tools/server/README-dev.md +++ b/tools/server/README-dev.md @@ -57,7 +57,7 @@ The core architecture consists of the following components: - `server_tokens`: Unified representation of token sequences (supports both text and multimodal tokens); used by `server_task` and `server_slot`. - `server_prompt_checkpoint`: For recurrent (e.g., RWKV) and SWA models, stores snapshots of KV cache state. Enables reuse when subsequent requests share the same prompt prefix, saving redundant computation. - `server_models`: Standalone component for managing multiple backend instances (used in router mode). It is completely independent of `server_context`. -- `stream_session_manager`: Process wide owner of resumable SSE stream sessions (`g_stream_sessions`), keyed by conversation id. Backs the replay buffer that lets a client reattach to a generation after an HTTP disconnect. See the "Resumable streaming" section below. +- `stream_session_manager`: process wide owner of resumable SSE stream sessions, keyed by conversation id. A file-static singleton inside `server-stream.cpp`, driven through `server_stream_session_manager_start/stop`. Backs the replay buffer that lets a client reattach to a generation after an HTTP disconnect. See the "Resumable streaming" section below. ```mermaid graph TD @@ -127,10 +127,12 @@ It is opt in via the `X-Conversation-Id` header on `POST /v1/chat/completions`. The feature lives entirely in `server-stream.{h,cpp}` and rests on three types: - `stream_session`: a bounded ring buffer (4 MiB cap, oldest bytes drop first) plus a condvar. `append` pushes raw SSE bytes, `read_from` drains from any offset and blocks for live bytes or finalize, `finalize` wakes readers, `cancel` stops the producer. One conv maps to at most one live session. -- `stream_session_manager` (`g_stream_sessions`): owns all sessions keyed by conv id, enforces the one conv one session invariant via `create_or_replace`, and runs a GC thread that drops completed sessions past their TTL. +- `stream_session_manager`: a file-static singleton (`g_stream_sessions`) inside `server-stream.cpp`, owns all sessions keyed by conv id, enforces the one conv one session invariant via `create_or_replace`, and runs a GC thread that drops completed sessions past their TTL. Exposed to main only through `server_stream_session_manager_start/stop`. - `stream_pipe_producer` / `stream_pipe_consumer`: the write and read ends. The producer owns the session lifetime and finalizes it on destruction; the consumer is read only and never finalizes, so a reader detaching cannot kill a running generation. -Producer side: `server_res_generator` attaches a producer pipe when the header is present. The HTTP content provider mirrors every chunk into the ring before writing it to the socket. While a pipe is attached, `stream_aware_should_stop` ignores peer disconnect, so a dropped socket does not stop generation: only an explicit `DELETE` does. When the peer leaves early, `on_complete` calls `close()`, which drains the rest of the generation into the ring on the http worker. +The implementation is hidden in `server-stream.cpp` (pimpl). The header exposes only the route handler factories, `server_stream_session_attach_pipe`, `server_stream_aware_should_stop`, `server_stream_conv_id_from_headers` and the GC lifecycle; the session, manager and consumer types stay in the `.cpp`. + +Producer side: `server_res_generator` attaches a producer pipe when the header is present. The HTTP content provider mirrors every chunk into the ring before writing it to the socket. While a pipe is attached, `server_stream_aware_should_stop` ignores peer disconnect, so a dropped socket does not stop generation: only an explicit `DELETE` does. When the peer leaves early, `on_complete` calls `close()`, which drains the rest of the generation into the ring on the http worker. Lifetime safety: the producer pipe holds a shared `alive` flag also captured by the session cancel hook. `~server_res_generator` calls `cleanup()` to clear that hook while the reader is still alive, so a `cancel` arriving during teardown can never call `stop()` on a freed response. This ordering is the most fragile part of the feature: finalizing or destroying the producer before `cleanup()` runs reintroduces a use after free. @@ -144,7 +146,7 @@ Routes: Router mode binds the same paths to proxy handlers. A `conv_id -> child` map (`conv_models`), populated when a POST is routed, resolves the owning child in one lookup with no polling. The lookup groups ids per child; GET and DELETE proxy straight to the owner. This loopback REST hop is expected to move to a websocket IPC later, swapping only the transport. -Lifecycle: `g_stream_sessions.start_gc()` runs in main after common init, `stop_gc()` runs first in `clean_up()` and finalizes every live session so no reader hangs. Reader blocking and the post drop drain both run on httplib worker threads, which block on a condvar rather than spin. +Lifecycle: `server_stream_session_manager_start()` runs in main after common init, `server_stream_session_manager_stop()` runs first in `clean_up()` and finalizes every live session so no reader hangs. Reader blocking and the post drop drain both run on httplib worker threads, which block on a condvar rather than spin. | Constant | Value | Role | | --- | --- | --- | diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index df56665815..aa5d0a2abb 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -4188,7 +4188,7 @@ std::unique_ptr server_routes::handle_completions_impl( } }; - auto effective_should_stop = stream_aware_should_stop(res_this, req.should_stop); + auto effective_should_stop = server_stream_aware_should_stop(res_this, req.should_stop); try { if (effective_should_stop()) { @@ -4284,7 +4284,7 @@ std::unique_ptr server_routes::handle_completions_impl( // attach a producer pipe to the response when X-Conversation-Id is present. // the pipe mirrors SSE chunks into the ring buffer and wires up the cancel hook. - stream_session_attach_pipe(*res, req.headers); + server_stream_session_attach_pipe(*res, req.headers); return res; } diff --git a/tools/server/server-models.cpp b/tools/server/server-models.cpp index 6a8eb2a2be..0cbf520af1 100644 --- a/tools/server/server-models.cpp +++ b/tools/server/server-models.cpp @@ -1681,7 +1681,7 @@ void server_models_routes::init_routes() { } // remember which child serves this conversation so the stream routes can route straight // to it without polling, keyed on the exact conv id from the header - std::string conv_id = stream_conv_id_from_headers(req.headers); + std::string conv_id = server_stream_conv_id_from_headers(req.headers); if (!conv_id.empty()) { models.conv_models.remember(conv_id, name); } @@ -1896,7 +1896,7 @@ void server_models_routes::init_routes() { if (!from.empty()) { child_path += "?from=" + from; } - SRV_INF("proxying stream resume to model %s on port %d, path=%s\n", + SRV_TRC("proxying stream resume to model %s on port %d, path=%s\n", owner->name.c_str(), owner->port, child_path.c_str()); auto proxy = std::make_unique( "GET", diff --git a/tools/server/server-stream.cpp b/tools/server/server-stream.cpp index c2bba8ec46..553ac26b1e 100644 --- a/tools/server/server-stream.cpp +++ b/tools/server/server-stream.cpp @@ -6,6 +6,12 @@ #include #include #include +#include + +enum class stream_read_status { + OK, + OFFSET_LOST, +}; namespace { constexpr int64_t STREAM_SESSION_TTL_SECONDS = 300; @@ -13,7 +19,6 @@ constexpr size_t STREAM_SESSION_MAX_BYTES = 4 * 1024 * 1024; constexpr int64_t STREAM_SESSION_GC_INTERVAL_SECONDS = 60; constexpr int64_t STREAM_READ_WAKE_INTERVAL_MS = 200; -// returns unix time in seconds int64_t now_seconds() { return std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch() @@ -21,6 +26,91 @@ int64_t now_seconds() { } } +// owns all live sessions keyed by conversation_id, one conv = at most one live session. +// a periodic GC evicts expired ones +class stream_session_manager { +public: + stream_session_manager(); + ~stream_session_manager(); + + stream_session_manager(const stream_session_manager &) = delete; + stream_session_manager & operator=(const stream_session_manager &) = delete; + + // install a new session, evicting and cancelling any previous one. conversation_id must be non empty + stream_session_ptr create_or_replace(const std::string & conversation_id); + + stream_session_ptr get(const std::string & conversation_id); + + std::vector list_all() const; + + void evict(const std::string & conversation_id); + + void evict_and_cancel(const std::string & conversation_id); + + void start_gc(); + void stop_gc(); + +private: + void gc_loop(); + + mutable std::shared_mutex map_mu; + std::unordered_map sessions; // key: conversation_id + std::thread gc_thread; + bool running; + std::mutex gc_wake_mu; + std::condition_variable gc_wake_cv; +}; + +// process wide manager, lifecycle controlled by llama-server main() via start_gc/stop_gc +static stream_session_manager g_stream_sessions; + +void server_stream_session_manager_start() { + g_stream_sessions.start_gc(); +} + +void server_stream_session_manager_stop() { + g_stream_sessions.stop_gc(); +} + +struct stream_session { + std::string conversation_id; + int64_t started_ts; // unix seconds at construction + + stream_session(std::string conversation_id_, size_t max_bytes_); + stream_session(const stream_session &) = delete; + stream_session & operator=(const stream_session &) = delete; + + bool append(const char * data, size_t len); + + void finalize(); + + // drain from offset into sink, blocking for more bytes or finalize. OFFSET_LOST if offset + // fell below the dropped prefix + stream_read_status read_from(size_t offset, + const std::function & sink, + const std::function & should_stop); + + bool is_done() const; + bool is_cancelled() const; + size_t total_size() const; // bytes that ever entered the session + size_t dropped_prefix() const; // bytes evicted from the front due to cap + int64_t completed_at() const; // 0 while alive, unix seconds after finalize + + void set_stop_producer(std::function fn); + + void cancel(); + +private: + mutable std::mutex mu; + std::condition_variable cv; + std::vector buffer; + size_t prefix_dropped; + size_t cap_bytes; + bool done; + std::atomic cancelled; // polled lock-free by the should_stop closure, no mu + int64_t completed_ts; + std::function stop_producer; +}; stream_session::stream_session(std::string conversation_id_, size_t max_bytes_) : conversation_id(std::move(conversation_id_)) , started_ts(now_seconds()) @@ -38,7 +128,7 @@ bool stream_session::append(const char * data, size_t len) { } { std::lock_guard lock(mu); - if (done.load(std::memory_order_relaxed)) { + if (done) { return false; } if (len >= cap_bytes) { @@ -62,11 +152,14 @@ bool stream_session::append(const char * data, size_t len) { } void stream_session::finalize() { - bool was_done = done.exchange(true, std::memory_order_acq_rel); - if (was_done) { - return; + { + std::lock_guard lock(mu); + if (done) { + return; + } + done = true; + completed_ts = now_seconds(); } - completed_ts.store(now_seconds(), std::memory_order_release); cv.notify_all(); } @@ -96,7 +189,7 @@ stream_read_status stream_session::read_from(size_t offset, lock.lock(); continue; } - if (done.load(std::memory_order_acquire)) { + if (done) { return stream_read_status::OK; } // wait for new bytes, finalize, or a periodic wake to re check should_stop @@ -105,7 +198,8 @@ stream_read_status stream_session::read_from(size_t offset, } bool stream_session::is_done() const { - return done.load(std::memory_order_acquire); + std::lock_guard lock(mu); + return done; } size_t stream_session::total_size() const { @@ -119,7 +213,8 @@ size_t stream_session::dropped_prefix() const { } int64_t stream_session::completed_at() const { - return completed_ts.load(std::memory_order_acquire); + std::lock_guard lock(mu); + return completed_ts; } void stream_session::set_stop_producer(std::function fn) { @@ -128,7 +223,7 @@ void stream_session::set_stop_producer(std::function fn) { } void stream_session::cancel() { - // flip cancelled first so the producer-side stream_aware_should_stop can break out of the + // flip cancelled first so the producer-side server_stream_aware_should_stop can break out of the // recv() wait even if remove_waiting_task_ids does not notify the condvar (the cancel task // posted by rd.stop() will eventually notify, but we do not want to depend on that timing) cancelled.store(true, std::memory_order_release); @@ -237,18 +332,24 @@ void stream_session_manager::evict_and_cancel(const std::string & conversation_i } void stream_session_manager::start_gc() { - if (running.exchange(true)) { - return; + { + std::lock_guard lock(gc_wake_mu); + if (running) { + return; + } + running = true; } gc_thread = std::thread([this] { gc_loop(); }); } void stream_session_manager::stop_gc() { - bool was_running = running.exchange(false); + bool was_running; + { + std::lock_guard lock(gc_wake_mu); + was_running = running; + running = false; + } if (was_running) { - { - std::lock_guard lock(gc_wake_mu); - } gc_wake_cv.notify_all(); if (gc_thread.joinable()) { gc_thread.join(); @@ -270,15 +371,15 @@ void stream_session_manager::stop_gc() { } void stream_session_manager::gc_loop() { - while (running.load(std::memory_order_acquire)) { + while (true) { { std::unique_lock lock(gc_wake_mu); gc_wake_cv.wait_for(lock, std::chrono::seconds(STREAM_SESSION_GC_INTERVAL_SECONDS), - [this] { return !running.load(std::memory_order_acquire); }); - } - if (!running.load(std::memory_order_acquire)) { - return; + [this] { return !running; }); + if (!running) { + return; + } } int64_t cutoff = now_seconds() - STREAM_SESSION_TTL_SECONDS; std::vector to_drop; @@ -301,10 +402,19 @@ void stream_session_manager::gc_loop() { } } -// process wide manager, lifecycle controlled by llama-server main() via start_gc/stop_gc -stream_session_manager g_stream_sessions; +// stream_pipe -// stream_pipe --------------------------------------------------------------------------------- +// consumer end: read-only replay of the ring buffer, the destructor does not finalize the session +struct stream_pipe_consumer : stream_pipe { + stream_read_status read(size_t & offset, + const std::function & sink, + const std::function & should_stop); + + static std::shared_ptr create(stream_session_ptr session); + +private: + explicit stream_pipe_consumer(stream_session_ptr session); +}; stream_pipe::stream_pipe(stream_session_ptr session) : session_(std::move(session)) { @@ -408,12 +518,10 @@ static server_http_res_ptr make_error_response(int status, const std::string & m return res; } -server_http_context::handler_t make_stream_get_handler() { +server_http_context::handler_t server_stream_make_get_handler() { return [](const server_http_req & req) -> server_http_res_ptr { - // GET /v1/stream/?from=N replays the SSE bytes already buffered for the - // session, blocks for more bytes when the session is still running, returns when - // the session is finalized. the body is streamed back as text/event-stream so the - // browser EventSource can attach to it like a fresh request + // GET /v1/stream/?from=N replays buffered SSE bytes then blocks for live + // bytes until the session finalizes, streamed as text/event-stream for EventSource std::string conv_id = req.get_param("conv_id"); if (conv_id.empty()) { return make_error_response(400, "Missing conversation id in path", ERROR_TYPE_INVALID_REQUEST); @@ -459,11 +567,10 @@ server_http_context::handler_t make_stream_get_handler() { }; } -server_http_context::handler_t make_streams_lookup_handler() { +server_http_context::handler_t server_stream_make_lookup_handler() { return [](const server_http_req & req) -> server_http_res_ptr { - // POST /v1/streams/lookup with body {"conversation_ids": ["X", "Y", ...]} returns the - // matching sessions, only for ids the caller already knows. each id matches the exact key - // and any "::" variant, so one lookup covers every per model session for a conv + // POST /v1/streams/lookup returns the matching sessions, only for ids the caller already + // knows. each id matches the exact key and any "::" per model variant std::vector requested; try { json body = json::parse(req.body); @@ -518,11 +625,10 @@ server_http_context::handler_t make_streams_lookup_handler() { }; } -server_http_context::handler_t make_stream_delete_handler() { +server_http_context::handler_t server_stream_make_delete_handler() { return [](const server_http_req & req) -> server_http_res_ptr { - // DELETE /v1/stream/ is the explicit user Stop, cancels the producer hook - // wired by handle_completions_impl and evicts the buffer. idempotent, a session that - // already finalized or was never created returns 204 either way + // DELETE /v1/stream/ is the explicit user Stop, cancels the producer and evicts + // the buffer. idempotent, returns 204 even if the session was already gone std::string conv_id = req.get_param("conv_id"); if (conv_id.empty()) { return make_error_response(400, "Missing conversation id in path", ERROR_TYPE_INVALID_REQUEST); @@ -536,7 +642,7 @@ server_http_context::handler_t make_stream_delete_handler() { }; } -std::string stream_conv_id_from_headers(const std::map & headers) { +std::string server_stream_conv_id_from_headers(const std::map & headers) { // case-insensitive scan for x-conversation-id static constexpr char target[] = "x-conversation-id"; static constexpr size_t target_len = sizeof(target) - 1; @@ -555,8 +661,8 @@ std::string stream_conv_id_from_headers(const std::map return std::string(); } -void stream_session_attach_pipe(server_http_res & res, const std::map & headers) { - std::string conversation_id = stream_conv_id_from_headers(headers); +void server_stream_session_attach_pipe(server_http_res & res, const std::map & headers) { + std::string conversation_id = server_stream_conv_id_from_headers(headers); SRV_TRC("conv_id=%s (empty=%d)\n", conversation_id.c_str(), conversation_id.empty() ? 1 : 0); if (conversation_id.empty()) { return; @@ -565,7 +671,7 @@ void stream_session_attach_pipe(server_http_res & res, const std::map stream_aware_should_stop(server_http_res * res, std::function fallback) { +std::function server_stream_aware_should_stop(server_http_res * res, std::function fallback) { return [res, fallback = std::move(fallback)]() -> bool { if (res->spipe) { return res->spipe->is_cancelled(); diff --git a/tools/server/server-stream.h b/tools/server/server-stream.h index ff363bb4cd..c0c3e924fa 100644 --- a/tools/server/server-stream.h +++ b/tools/server/server-stream.h @@ -3,81 +3,23 @@ #include "server-http.h" #include -#include #include -#include #include #include -#include -#include #include -#include -#include -#include -enum class stream_read_status { - OK, - OFFSET_LOST, -}; +// streaming buffer for one generation, survives HTTP disconnect. the producer appends SSE bytes, +// readers drain from any offset via read_from. keyed by conversation_id, one conv = one live session -// streaming buffer for one generation, survives HTTP disconnect. the producer appends raw SSE -// bytes, readers drain from any offset via read_from and block until more bytes or finalize. -// keyed by conversation_id: one conv = at most one live session -struct stream_session { - std::string conversation_id; - int64_t started_ts; // unix seconds at construction, used by /v1/streams listing - - stream_session(std::string conversation_id_, size_t max_bytes_); - stream_session(const stream_session &) = delete; - stream_session & operator=(const stream_session &) = delete; - - // append raw bytes, drops from the front if the cap is reached. - // returns false if the session is already finalized - bool append(const char * data, size_t len); - - // mark the session as complete, wakes all pending readers - void finalize(); - - // drain bytes from offset, calling sink for each chunk. blocks until more - // bytes arrive or finalize is called. returns OK on clean exit, OFFSET_LOST - // if offset falls below the dropped prefix - stream_read_status read_from(size_t offset, - const std::function & sink, - const std::function & should_stop); - - bool is_done() const; - bool is_cancelled() const; - size_t total_size() const; // bytes that ever entered the session - size_t dropped_prefix() const; // bytes evicted from the front due to cap - int64_t completed_at() const; // 0 while alive, unix seconds after finalize - - // attach the producer stop hook used to cancel its reader, pass an empty function to detach - void set_stop_producer(std::function fn); - - // signal the producer to abort its inference asap via the stop hook, idempotent - void cancel(); - -private: - mutable std::mutex mu; - std::condition_variable cv; - std::vector buffer; - size_t prefix_dropped; - size_t cap_bytes; - std::atomic done; - std::atomic cancelled; - std::atomic completed_ts; - std::function stop_producer; // protected by mu -}; +struct stream_session; using stream_session_ptr = std::shared_ptr; -// one end of a stream_session pipe. the base holds the session and the shared query, the -// producer and consumer ends derive from it. virtual dtor so each end runs its own teardown: +// base of the producer/consumer pipe ends. virtual dtor so each runs its own teardown: // the producer finalizes the session, the consumer leaves it untouched struct stream_pipe { virtual ~stream_pipe() = default; - // true if the session was cancelled (e.g. via DELETE /v1/stream/) bool is_cancelled() const; protected: @@ -95,7 +37,6 @@ protected: struct stream_pipe_producer : stream_pipe { ~stream_pipe_producer() override; - // append raw bytes to the session's ring buffer, returns false if already finalized bool write(const char * data, size_t len); // mark the natural end on the wire so a later close() is a no-op @@ -121,83 +62,21 @@ private: server_http_res * res_ = nullptr; }; -// consumer end: read-only replay of the ring buffer, the destructor does not finalize the session -struct stream_pipe_consumer : stream_pipe { - // drain bytes from offset, calling sink for each available chunk. blocks until more data - // arrives or the session finalizes. should_stop is polled, returns OFFSET_LOST if offset - // fell below the dropped prefix - stream_read_status read(size_t & offset, - const std::function & sink, - const std::function & should_stop); +void server_stream_session_manager_start(); +void server_stream_session_manager_stop(); - static std::shared_ptr create(stream_session_ptr session); +// route handler factories wired under /v1/stream/* by server.cpp +server_http_context::handler_t server_stream_make_get_handler(); +server_http_context::handler_t server_stream_make_lookup_handler(); +server_http_context::handler_t server_stream_make_delete_handler(); -private: - explicit stream_pipe_consumer(stream_session_ptr session); -}; +// extract the X-Conversation-Id header value (case-insensitive), empty when absent +std::string server_stream_conv_id_from_headers(const std::map & headers); -// owns all live sessions, runs a periodic GC to evict expired ones. -// the map is keyed by conversation_id, so the invariant "one conv = at most one -// live session" is enforced at the type level -class stream_session_manager { -public: - stream_session_manager(); - ~stream_session_manager(); - - stream_session_manager(const stream_session_manager &) = delete; - stream_session_manager & operator=(const stream_session_manager &) = delete; - - // install a new session for this conversation, evicting and cancelling any previous one. - // the conversation_id must be non empty, the caller is responsible for that check. - // returns the new session - stream_session_ptr create_or_replace(const std::string & conversation_id); - - // lookup, returns null if unknown or already evicted - stream_session_ptr get(const std::string & conversation_id); - - // list every live or recently completed session, used by GET /v1/streams without filter - std::vector list_all() const; - - // remove from the map and finalize, wakes any pending readers - void evict(const std::string & conversation_id); - - // signal the producer to cancel asap then evict, used by the explicit user Stop path - void evict_and_cancel(const std::string & conversation_id); - - void start_gc(); - void stop_gc(); - -private: - void gc_loop(); - - mutable std::shared_mutex map_mu; - std::unordered_map sessions; // key: conversation_id - std::thread gc_thread; - std::atomic running; - std::mutex gc_wake_mu; - std::condition_variable gc_wake_cv; -}; - -// process wide manager, linked by both llama-server and llama-cli. llama-server main() drives -// start_gc/stop_gc, llama-cli leaves it idle. the dtor calls stop_gc() unconditionally so exit -// is safe whether or not the GC thread ran -extern stream_session_manager g_stream_sessions; - -// route handler factories operating on g_stream_sessions, wired under /v1/stream/* by server.cpp. -// keeps the resumable stream surface confined to server-stream -server_http_context::handler_t make_stream_get_handler(); -server_http_context::handler_t make_streams_lookup_handler(); -server_http_context::handler_t make_stream_delete_handler(); - -// extract the X-Conversation-Id header value (case-insensitive), empty when absent. exposed so -// the router can track which child serves a forwarded POST -std::string stream_conv_id_from_headers(const std::map & headers); - -// on an X-Conversation-Id header, create or replace the session and attach a producer pipe to -// res. no-op when absent, called from the server_res_generator constructor -void stream_session_attach_pipe(server_http_res & res, const std::map & headers); +// on an X-Conversation-Id header, create or replace the session and attach a producer pipe to res +void server_stream_session_attach_pipe(server_http_res & res, const std::map & headers); // should_stop closure that ignores peer disconnect when a pipe is attached, so only an explicit // DELETE stops the producer and generation keeps flowing into the ring buffer. without a pipe it // delegates to fallback, the legacy non-resumable flow -std::function stream_aware_should_stop(server_http_res * res, std::function fallback); +std::function server_stream_aware_should_stop(server_http_res * res, std::function fallback); diff --git a/tools/server/server.cpp b/tools/server/server.cpp index eafef86bac..9e8603be66 100644 --- a/tools/server/server.cpp +++ b/tools/server/server.cpp @@ -85,7 +85,7 @@ int llama_server(int argc, char ** argv) { // start the stream session manager GC right after common init, before any HTTP route can // touch it. lifecycle is symmetric, stop_gc() runs in clean_up() before backend free - g_stream_sessions.start_gc(); + server_stream_session_manager_start(); if (!common_params_parse(argc, argv, params, LLAMA_EXAMPLE_SERVER)) { return 1; @@ -245,8 +245,8 @@ int llama_server(int argc, char ** argv) { ctx_http.post("/slots/:id_slot", ex_wrapper(routes.post_slots)); // resumable streaming, the conversation_id is the session identity end to end. router and - // child wire different handlers under the same paths: a child binds the local g_stream_sessions - // backed factories, the router binds proxies that resolve the owning child through the + // child wire different handlers under the same paths: a child binds the local session + // factories, the router binds proxies that resolve the owning child through the // conv_id -> model map server_http_context::handler_t stream_get_h; server_http_context::handler_t streams_lookup_h; @@ -256,9 +256,9 @@ int llama_server(int argc, char ** argv) { streams_lookup_h = models_routes->router_streams_lookup; stream_delete_h = models_routes->router_stream_delete; } else { - stream_get_h = make_stream_get_handler(); - streams_lookup_h = make_streams_lookup_handler(); - stream_delete_h = make_stream_delete_handler(); + stream_get_h = server_stream_make_get_handler(); + streams_lookup_h = server_stream_make_lookup_handler(); + stream_delete_h = server_stream_make_delete_handler(); } ctx_http.get ("/v1/stream/:conv_id", ex_wrapper(stream_get_h)); // POST /v1/streams/lookup with body {"conversation_ids": [...]}. you can only ask for ids @@ -343,7 +343,7 @@ int llama_server(int argc, char ** argv) { clean_up = [&models_routes]() { SRV_INF("%s: cleaning up before exit...\n", __func__); // stop the session GC first, it finalizes live sessions and wakes pending readers - g_stream_sessions.stop_gc(); + server_stream_session_manager_stop(); if (models_routes.has_value()) { models_routes->stopping.store(true); // maybe redundant, but just to be safe models_routes->models.unload_all(); @@ -371,7 +371,7 @@ int llama_server(int argc, char ** argv) { clean_up = [&ctx_http, &ctx_server]() { SRV_INF("%s: cleaning up before exit...\n", __func__); // stop the session GC first, it finalizes live sessions and wakes pending readers - g_stream_sessions.stop_gc(); + server_stream_session_manager_stop(); ctx_http.stop(); ctx_server.terminate(); llama_backend_free();