kanshan/llama.cpp
1
0
Fork 0
mirror of https://github.com/ggml-org/llama.cpp.git synced 2026-07-10 13:10:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into xsn/cli_http_based

Browse Source
This commit is contained in:
Xuan Son Nguyen
2026-07-07 21:11:05 +02:00
parent a432e6f863 bec4772f6a
commit 7cd7832297
444 changed files with 44486 additions and 18746 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/server/server-models.cpp
+197 -24
View File
@@ -1,6 +1,8 @@
#include "server-common.h"
#include "http.h"
#include "server-models.h"
#include "server-context.h"
#include "server-stream.h"
#include "build-info.h"
#include "preset.h"
@@ -8,6 +10,7 @@
#include "http.h"
#include <cpp-httplib/httplib.h> // TODO: remove this once we use HTTP client from download.h
#include <optional>
#include <sheredom/subprocess.h>
#include <functional>
@@ -86,6 +89,9 @@ struct server_subproc {
}
};
// short loopback budget for the resumable stream router to child JSON calls (probe, lookup,
// delete). distinct from params.timeout_read/write which only applies to the generation proxy
static constexpr int STREAM_LOOKUP_TIMEOUT_MS = 250;
static std::filesystem::path get_server_exec_path() {
#if defined(_WIN32)
@@ -217,8 +223,8 @@ void server_model_meta::update_caps() {
"LLAMA_ARG_HF_REPO_FILE",
});
params.offline = true;
// params.skip_download = true; // TODO: ideally, we should validate the model here, but it takes too much time
common_params_handle_models(params, LLAMA_EXAMPLE_SERVER, {});
common_models_handler handler = common_models_handler_init(params, LLAMA_EXAMPLE_SERVER);
common_models_handler_apply(handler, params); // note: this won't download the model because offline=true
if (params.mmproj.path.empty()) {
multimodal = { false, false };
} else {
@@ -511,6 +517,7 @@ void server_models::load_models() {
// collect all threads to join in one pass while the lock is held:
// - monitoring threads from just-unloaded models (to_unload)
// - threads of finished downloads (DOWNLOADED), they acquire the mutex on exit
// - threads of already-UNLOADED models that are being removed from source
std::vector<std::thread> threads_to_join;
for (const auto & name : to_unload) {
@@ -523,6 +530,13 @@ void server_models::load_models() {
if (inst.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
continue; // downloading models are not from config sources, leave them alone
}
if (inst.meta.status == SERVER_MODEL_STATUS_DOWNLOADED) {
// joining this thread under the lock deadlocks: it locks the mutex on its way out
if (inst.th.joinable()) {
threads_to_join.push_back(std::move(inst.th));
}
continue;
}
if (final_presets.find(name) == final_presets.end() && !inst.meta.is_running() && inst.th.joinable()) {
threads_to_join.push_back(std::move(inst.th));
}
@@ -538,10 +552,8 @@ void server_models::load_models() {
if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
++it; // download thread is still busy, skip
} else if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADED) {
// download finished, safe to erase
if (it->second.th.joinable()) {
it->second.th.join();
}
// download finished, thread is joined above, safe to erase
GGML_ASSERT(!it->second.th.joinable());
it = mapping.erase(it);
} else if (final_presets.find(it->first) == final_presets.end()) {
SRV_INF("(reload) removing model name=%s (no longer in source)\n", it->first.c_str());
@@ -1327,9 +1339,8 @@ struct server_download_state : public common_download_callback {
bool run(common_params & params) {
try {
common_params_handle_models_params p;
p.callback = this;
common_params_handle_models(params, LLAMA_EXAMPLE_SERVER, p);
common_models_handler handler = common_models_handler_init(params, LLAMA_EXAMPLE_SERVER);
common_models_handler_apply(handler, params, this);
is_ok = true;
} catch (const std::exception & e) {
auto model_name = params.model.get_name();
@@ -1515,6 +1526,45 @@ static bool is_autoload(const common_params & params, const server_http_req & re
}
}
// percent encode one query or path component, covers reserved chars without pulling in
// httplib::detail. used by the stream routes to forward conversation_id to children safely
static std::string encode_qs(const std::string & in) {
std::string out;
out.reserve(in.size() * 3);
for (unsigned char c : in) {
bool safe = (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9')
|| c == '-' || c == '_' || c == '.' || c == '~';
if (safe) {
out.push_back(char(c));
} else {
char buf[4];
std::snprintf(buf, sizeof(buf), "%%%02X", c);
out.append(buf, 3);
}
}
return out;
}
// resolve the child that owns a conversation's stream session via the conv_id -> model map
// populated when the POST was routed. single map lookup then a meta lookup, no polling, no
// parsing of the conv id. returns nullopt when nothing maps, the caller answers not found and
// the client recovers
static std::optional<server_model_meta> resolve_child_for_conv(
server_models & models, const std::string & conversation_id) {
if (conversation_id.empty()) {
return std::nullopt;
}
auto tracked = models.conv_models.lookup(conversation_id);
if (!tracked.has_value()) {
return std::nullopt;
}
auto meta = models.get_meta(*tracked);
if (meta.has_value() && meta->is_ready()) {
return meta;
}
return std::nullopt;
}
void server_models_routes::init_routes() {
this->get_router_props = [this](const server_http_req & req) {
std::string name = req.get_param("model");
@@ -1563,6 +1613,12 @@ void server_models_routes::init_routes() {
if (!router_validate_model(name, models, autoload, error_res)) {
return error_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);
if (!conv_id.empty()) {
models.conv_models.remember(conv_id, name);
}
return models.proxy_request(req, method, name, true); // update last usage for POST request only
};
@@ -1702,23 +1758,14 @@ void server_models_routes::init_routes() {
throw std::invalid_argument("model must be a non-empty string");
}
common_params_model model;
common_download_opts opts;
common_params p;
p.model.hf_repo = name;
p.hf_token = params.hf_token;
model.hf_repo = name;
opts.bearer_token = params.hf_token;
// note: we only check main model, no need sidecar here
opts.download_mmproj = false;
opts.download_mtp = false;
// first, only check if the model is valid and can be downloaded
opts.skip_download = true;
// validate by fetching metadata
bool ok = false;
try {
auto validation = common_download_model(model, opts);
ok = !validation.model_path.empty();
} catch (const common_skip_download_exception &) {
// model is valid and will be downloaded
common_models_handler_init(p, LLAMA_EXAMPLE_SERVER);
ok = true;
} catch (...) {
SRV_ERR("unknown error while validating model '%s'\n", name.c_str());
@@ -1763,6 +1810,131 @@ void server_models_routes::init_routes() {
res_ok(res, {{"success", true}});
return res;
};
this->router_stream_get = [this](const server_http_req & req) {
// GET /v1/stream/<conv_id>?from=N. resolve the owning child from the conv_id -> model
// map, 404 when nothing maps
auto res = std::make_unique<server_http_res>();
std::string conv_id = req.get_param("conv_id");
if (conv_id.empty()) {
res_err(res, format_error_response("Missing conversation id in path", ERROR_TYPE_INVALID_REQUEST));
return res;
}
std::optional<server_model_meta> owner = resolve_child_for_conv(models, conv_id);
if (!owner.has_value()) {
res_err(res, format_error_response("Stream not found or expired", ERROR_TYPE_NOT_FOUND));
return res;
}
std::string from = req.get_param("from");
std::string child_path = "/v1/stream/" + encode_qs(conv_id);
if (!from.empty()) {
child_path += "?from=" + from;
}
SRV_INF("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",
"http",
CHILD_ADDR,
owner->port,
child_path,
req.headers,
req.body,
req.files,
req.should_stop,
params.timeout_read,
params.timeout_write);
return std::unique_ptr<server_http_res>(std::move(proxy));
};
this->router_streams_lookup = [this](const server_http_req & req) {
// POST /v1/streams/lookup. resolve each requested conv id to its owning child via the
// map, group the ids per child, and query only the children that actually own some of
// them instead of fanning out to every ready child. a child only answers for the ids
// it owns, never lists anything else
auto res = std::make_unique<server_http_res>();
std::vector<std::string> requested;
try {
json body = json::parse(req.body);
if (body.contains("conversation_ids") && body["conversation_ids"].is_array()) {
for (const auto & v : body["conversation_ids"]) {
if (v.is_string() && !v.get<std::string>().empty()) {
requested.push_back(v.get<std::string>());
}
}
}
} catch (const std::exception &) {
res_ok(res, json::array());
return res;
}
// group requested ids by the child port that owns them, drop ids that map to nothing
std::unordered_map<int, json> per_child;
for (const auto & cid : requested) {
auto owner = resolve_child_for_conv(models, cid);
if (!owner.has_value()) {
continue;
}
per_child[owner->port].push_back(cid);
}
json aggregated = json::array();
for (auto & [port, ids] : per_child) {
json child_body = {{"conversation_ids", ids}};
httplib::Client cli(CHILD_ADDR, port);
cli.set_connection_timeout(0, STREAM_LOOKUP_TIMEOUT_MS * 1000);
cli.set_read_timeout(0, STREAM_LOOKUP_TIMEOUT_MS * 1000);
cli.set_write_timeout(0, STREAM_LOOKUP_TIMEOUT_MS * 1000);
auto resp = cli.Post("/v1/streams/lookup", child_body.dump(), "application/json");
if (!resp || resp->status != 200) {
continue;
}
try {
json child_arr = json::parse(resp->body);
if (!child_arr.is_array()) {
continue;
}
for (auto & entry : child_arr) {
if (entry.is_object()) {
aggregated.push_back(entry);
}
}
} catch (const std::exception &) {
continue;
}
}
res_ok(res, aggregated);
return res;
};
this->router_stream_delete = [this](const server_http_req & req) {
// DELETE /v1/stream/<conv_id>. resolve the owning child via the map and forward only to
// it, evict_and_cancel is idempotent on the child
auto res = std::make_unique<server_http_res>();
std::string conv_id = req.get_param("conv_id");
if (conv_id.empty()) {
res_err(res, format_error_response("Missing conversation id in path", ERROR_TYPE_INVALID_REQUEST));
return res;
}
std::string child_path = "/v1/stream/" + encode_qs(conv_id);
auto owner = resolve_child_for_conv(models, conv_id);
if (owner.has_value()) {
httplib::Client cli(CHILD_ADDR, owner->port);
cli.set_connection_timeout(0, STREAM_LOOKUP_TIMEOUT_MS * 1000);
cli.set_read_timeout(0, STREAM_LOOKUP_TIMEOUT_MS * 1000);
cli.set_write_timeout(0, STREAM_LOOKUP_TIMEOUT_MS * 1000);
auto resp = cli.Delete(child_path.c_str());
(void) resp; // the child logs its own miss when the session is unknown there
} else {
SRV_WRN("router stop for unknown conv_id=%s, no owning child in the conv map\n",
conv_id.c_str());
}
// drop the tracking entry, the session is being torn down
models.conv_models.forget(conv_id);
res->status = 204;
res->content_type = "application/json";
return res;
};
}
@@ -2032,7 +2204,8 @@ server_http_proxy::server_http_proxy(
}
if (lowered == "host") {
bool is_default_port = (scheme == "https" && port == 443) || (scheme == "http" && port == 80);
req.set_header(key, is_default_port ? host : host + ":" + std::to_string(port));
const std::string url_host = common_http_format_host(host);
req.set_header(key, is_default_port ? url_host : url_host + ":" + std::to_string(port));
} else {
req.set_header(key, value);
}