mirror of
https://github.com/ggml-org/llama.cpp.git
synced 2026-07-30 22:50:48 -05:00
d73c1d6b22
* server + ui: refactor resumable stream routes to query string conv_id The conversation id can embed a model name containing slashes (ggml-org/...) in router mode, which the decoded path splits before the :conv_id param is captured, so stop and resume never matched the session. Move the id to the conv_id query string on the public routes and on the internal router -> child hop, where slashes survive encoding. Handlers are unchanged since query and path params land in the same map. Add a regression test with a slashed model name. * server: move stream route docs to server-stream.h Address review: ngxson wants the main server.cpp registration code kept clean and simple, with route-level explanations living in the header. Move the query string rationale and the lookup ownership note next to the handler declarations in server-stream.h, and shorten the wiring comment to a pointer. * server: cancel a pending request when its stream is stopped during model load The conversation was registered in the conv map only after the blocking autoload wait, so a stop issued while the model loaded found nothing to cancel and the request went on to generate an orphan once the load ended. Register the conversation before the wait and give the entry a ticket: a stop erases the entry, and the parked request checks its ticket after the wait and aborts with 400 instead of starting. A newer request on the same conversation replaces the entry, so only the stopped request is cancelled. Add a regression test that stops during the load window. * server + ui: resume a stream after a page reload during model load A pending request died with the client socket when the page was reloaded while its model was loading, so no session ever existed and the conversation had nothing to recover. A session request that waited for a load now detaches from the client socket and reaches the child regardless, the session buffer receives the generation, and the resume route answers 503 while the owner is loading so the client retries instead of dropping its state. The WebUI persists the pending stream at send time, quietly polls on 503, and attaches once the session exists. Add a regression test that drops the client during the load window. * ui: show the model load progress again after a page refresh The resume wait was invisible, so a conversation refreshed while its model was loading showed nothing until the first byte. On a 503 from the resume probe, mark the conversation as loading again so the assistant row persisted at send time renders the processing info, and target the model frozen in the persisted stream state for the progress, since the row has no model yet and the dropdown may not be restored. * fix CI * fix CI bis
80 lines
3.0 KiB
C++
80 lines
3.0 KiB
C++
#pragma once
|
|
|
|
#include "server-http.h"
|
|
|
|
#include <atomic>
|
|
#include <cstddef>
|
|
#include <functional>
|
|
#include <memory>
|
|
#include <string>
|
|
|
|
// 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
|
|
|
|
struct stream_session;
|
|
|
|
using stream_session_ptr = std::shared_ptr<stream_session>;
|
|
|
|
// 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;
|
|
|
|
bool is_cancelled() const;
|
|
|
|
protected:
|
|
explicit stream_pipe(stream_session_ptr session);
|
|
|
|
stream_session_ptr session_;
|
|
};
|
|
|
|
// producer end: writes chunks into the ring buffer and owns the session lifetime, finalizing it
|
|
// on destruction.
|
|
struct stream_pipe_producer : stream_pipe {
|
|
~stream_pipe_producer() override;
|
|
|
|
bool write(const char * data, size_t len);
|
|
|
|
static stream_pipe_producer * create(stream_session_ptr session);
|
|
|
|
private:
|
|
explicit stream_pipe_producer(stream_session_ptr session);
|
|
};
|
|
|
|
void server_stream_session_manager_start();
|
|
void server_stream_session_manager_stop();
|
|
|
|
// route handler factories wired under /v1/stream/* by server.cpp
|
|
// child-side handlers for the resumable stream routes. the conv id travels in the conv_id
|
|
// query string because it can embed a model name containing slashes (org/repo), which the
|
|
// decoded path would split before the param is captured
|
|
server_http_context::handler_t server_stream_make_get_handler();
|
|
// POST /v1/streams/lookup with body {"conversation_ids": [...]}: only answers for ids the
|
|
// caller already owns (the WebUI passes the convs visible in its sidebar), the server never
|
|
// lists ids it has not been asked about, so a random caller cannot enumerate live sessions
|
|
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
|
|
std::string server_stream_conv_id_from_headers(const std::map<std::string, std::string> & headers);
|
|
|
|
// implement tee-style pipe (spipe) for "stream replay" functionality
|
|
struct server_res_spipe : server_http_res {
|
|
private:
|
|
// if set, the stream survives a client disconnect:
|
|
// connection kept alive, output is forwarded to spipe and reuse later
|
|
std::unique_ptr<stream_pipe_producer> spipe;
|
|
// if spipe is set, use this next_orig to implement tee-style pipe
|
|
std::function<bool(std::string &)> next_orig;
|
|
const server_http_req * req = nullptr;
|
|
// set once next_orig reports no more data, so on_complete() doesn't re-drain a finished stream
|
|
bool next_finished = false;
|
|
|
|
public:
|
|
void set_req(const server_http_req * req);
|
|
bool conn_alive();
|
|
bool should_stop();
|
|
void on_complete() override;
|
|
void set_next(std::function<bool(std::string &)> next_fn);
|
|
};
|