- `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
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.
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 |
| --- | --- | --- |
}
};
- 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()) {
// 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;
}
}
// 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);
}
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<server_http_proxy>(
"GET",
#include <chrono>
#include <memory>
#include <utility>
+#include <shared_mutex>
+
+enum class stream_read_status {
+ OK,
+ OFFSET_LOST,
+};
namespace {
constexpr int64_t STREAM_SESSION_TTL_SECONDS = 300;
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::seconds>(
std::chrono::system_clock::now().time_since_epoch()
}
}
+// 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<stream_session_ptr> 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<std::string, stream_session_ptr> 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<bool(const char *, size_t)> & sink,
+ const std::function<bool()> & 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<void()> fn);
+
+ void cancel();
+
+private:
+ mutable std::mutex mu;
+ std::condition_variable cv;
+ std::vector<char> buffer;
+ size_t prefix_dropped;
+ size_t cap_bytes;
+ bool done;
+ std::atomic<bool> cancelled; // polled lock-free by the should_stop closure, no mu
+ int64_t completed_ts;
+ std::function<void()> stop_producer;
+};
stream_session::stream_session(std::string conversation_id_, size_t max_bytes_)
: conversation_id(std::move(conversation_id_))
, started_ts(now_seconds())
}
{
std::lock_guard<std::mutex> lock(mu);
- if (done.load(std::memory_order_relaxed)) {
+ if (done) {
return false;
}
if (len >= cap_bytes) {
}
void stream_session::finalize() {
- bool was_done = done.exchange(true, std::memory_order_acq_rel);
- if (was_done) {
- return;
+ {
+ std::lock_guard<std::mutex> lock(mu);
+ if (done) {
+ return;
+ }
+ done = true;
+ completed_ts = now_seconds();
}
- completed_ts.store(now_seconds(), std::memory_order_release);
cv.notify_all();
}
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
}
bool stream_session::is_done() const {
- return done.load(std::memory_order_acquire);
+ std::lock_guard<std::mutex> lock(mu);
+ return done;
}
size_t stream_session::total_size() const {
}
int64_t stream_session::completed_at() const {
- return completed_ts.load(std::memory_order_acquire);
+ std::lock_guard<std::mutex> lock(mu);
+ return completed_ts;
}
void stream_session::set_stop_producer(std::function<void()> 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);
}
void stream_session_manager::start_gc() {
- if (running.exchange(true)) {
- return;
+ {
+ std::lock_guard<std::mutex> 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<std::mutex> lock(gc_wake_mu);
+ was_running = running;
+ running = false;
+ }
if (was_running) {
- {
- std::lock_guard<std::mutex> lock(gc_wake_mu);
- }
gc_wake_cv.notify_all();
if (gc_thread.joinable()) {
gc_thread.join();
}
void stream_session_manager::gc_loop() {
- while (running.load(std::memory_order_acquire)) {
+ while (true) {
{
std::unique_lock<std::mutex> 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<stream_session_ptr> to_drop;
}
}
-// process wide manager, lifecycle controlled by llama-server main() via start_gc/stop_gc
-stream_session_manager g_stream_sessions;
+// 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<bool(const char *, size_t)> & sink,
+ const std::function<bool()> & should_stop);
+
+ static std::shared_ptr<stream_pipe_consumer> create(stream_session_ptr session);
-// stream_pipe ---------------------------------------------------------------------------------
+private:
+ explicit stream_pipe_consumer(stream_session_ptr session);
+};
stream_pipe::stream_pipe(stream_session_ptr session)
: session_(std::move(session)) {
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/<conv_id>?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/<conv_id>?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);
};
}
-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 "<id>::<model>" 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 "<id>::<model>" per model variant
std::vector<std::string> requested;
try {
json body = json::parse(req.body);
};
}
-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/<conv_id> 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/<conv_id> 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);
};
}
-std::string stream_conv_id_from_headers(const std::map<std::string, std::string> & headers) {
+std::string server_stream_conv_id_from_headers(const std::map<std::string, std::string> & headers) {
// case-insensitive scan for x-conversation-id
static constexpr char target[] = "x-conversation-id";
static constexpr size_t target_len = sizeof(target) - 1;
return std::string();
}
-void stream_session_attach_pipe(server_http_res & res, const std::map<std::string, std::string> & headers) {
- std::string conversation_id = stream_conv_id_from_headers(headers);
+void server_stream_session_attach_pipe(server_http_res & res, const std::map<std::string, std::string> & 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;
res.spipe = stream_pipe_producer::create(session, res);
}
-std::function<bool()> stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback) {
+std::function<bool()> server_stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback) {
return [res, fallback = std::move(fallback)]() -> bool {
if (res->spipe) {
return res->spipe->is_cancelled();
#include "server-http.h"
#include <atomic>
-#include <condition_variable>
#include <cstddef>
-#include <cstdint>
#include <functional>
#include <memory>
-#include <mutex>
-#include <shared_mutex>
#include <string>
-#include <thread>
-#include <unordered_map>
-#include <vector>
-enum class stream_read_status {
- OK,
- OFFSET_LOST,
-};
-
-// 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<bool(const char *, size_t)> & sink,
- const std::function<bool()> & 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
+// 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
- // attach the producer stop hook used to cancel its reader, pass an empty function to detach
- void set_stop_producer(std::function<void()> 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<char> buffer;
- size_t prefix_dropped;
- size_t cap_bytes;
- std::atomic<bool> done;
- std::atomic<bool> cancelled;
- std::atomic<int64_t> completed_ts;
- std::function<void()> stop_producer; // protected by mu
-};
+struct stream_session;
using stream_session_ptr = std::shared_ptr<stream_session>;
-// 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/<conv_id>)
bool is_cancelled() const;
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
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<bool(const char *, size_t)> & sink,
- const std::function<bool()> & should_stop);
-
- static std::shared_ptr<stream_pipe_consumer> create(stream_session_ptr session);
-
-private:
- explicit stream_pipe_consumer(stream_session_ptr session);
-};
-
-// 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<stream_session_ptr> 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<std::string, stream_session_ptr> sessions; // key: conversation_id
- std::thread gc_thread;
- std::atomic<bool> 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;
+void server_stream_session_manager_start();
+void server_stream_session_manager_stop();
-// 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();
+// 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();
-// 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<std::string, std::string> & headers);
+// extract the X-Conversation-Id header value (case-insensitive), empty when absent
+std::string server_stream_conv_id_from_headers(const std::map<std::string, std::string> & 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<std::string, std::string> & 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<std::string, std::string> & 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<bool()> stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback);
+std::function<bool()> server_stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback);
// 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;
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;
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
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();
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();
overview / pkg / ggml / sources / llama.cpp / commitdiff