mirror of
https://github.com/ggml-org/llama.cpp.git
synced 2026-10-27 06:57:35 -05:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fe8eb721d8 | ||
|
|
c9c3154524 |
@@ -286,7 +286,7 @@ Instead of building everything from the ground up (like what most AI agents will
|
||||
|
||||
The flow for downloading a new model:
|
||||
- POST request comes in --> `post_router_models` --> validation
|
||||
- A new `llama-server` subprocess will be spawned with special `SERVER_CHILD_MODE_DOWNLOAD`
|
||||
- A new `llama-server` subprocess will be spawned with the `SERVER_ROLE_DOWNLOAD` role (see `server_role_type` in `server-common.h`)
|
||||
- Child process runs the download and report status back to router via stdin/out
|
||||
- If a stop request comes in, the router asks the child process to stop (same mechanism as running a model in child process)
|
||||
- Otherwise, upon completion, we call `load_models()` to refresh the list of models
|
||||
|
||||
@@ -16,7 +16,9 @@
|
||||
#include <limits>
|
||||
#include <cstring>
|
||||
#include <type_traits>
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
|
||||
#ifdef _WIN32
|
||||
@@ -2210,3 +2212,64 @@ void server_subproc::waiter::wait(const std::vector<server_subproc *> & procs, s
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
//
|
||||
// server_role
|
||||
//
|
||||
|
||||
server_role::server_role() {
|
||||
if (is_managed_by_router()) {
|
||||
cmd_out = server_reserve_stdout();
|
||||
}
|
||||
}
|
||||
|
||||
server_role::~server_role() {
|
||||
if (cmd_out) {
|
||||
fclose(cmd_out);
|
||||
}
|
||||
}
|
||||
|
||||
bool server_role::is_managed_by_router() {
|
||||
return std::getenv(SERVER_ENV_ROUTER_PORT) != nullptr;
|
||||
}
|
||||
|
||||
server_role_type server_role::get_role() {
|
||||
const char * role = std::getenv(SERVER_ENV_CHILD_ROLE);
|
||||
return server_role_type_from_str(role ? role : "");
|
||||
}
|
||||
|
||||
std::thread server_role::setup(const std::function<void(int)> & shutdown_handler) {
|
||||
// setup thread for monitoring stdin
|
||||
return std::thread([shutdown_handler]() {
|
||||
// wait for EOF on stdin
|
||||
SRV_INF("%s", "child server monitoring thread started, waiting for EOF on stdin...\n");
|
||||
bool eof = false;
|
||||
while (true) {
|
||||
std::string line;
|
||||
if (!std::getline(std::cin, line)) {
|
||||
// EOF detected, that means the router server is unexpectedly exit or killed
|
||||
eof = true;
|
||||
break;
|
||||
}
|
||||
if (line.find(CMD_ROUTER_TO_CHILD_EXIT) != std::string::npos) {
|
||||
SRV_INF("%s", "exit command received, exiting...\n");
|
||||
shutdown_handler(0);
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (eof) {
|
||||
SRV_INF("%s", "EOF on stdin detected, forcing shutdown...\n");
|
||||
exit(1);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void server_role::notify_to_router(const std::string & state, const json & payload) {
|
||||
json data = {
|
||||
{"state", state},
|
||||
{"payload", payload},
|
||||
};
|
||||
std::lock_guard<std::mutex> lk(mtx_stdout);
|
||||
fprintf(cmd_out, "%s%s\n", CMD_CHILD_TO_ROUTER_STATE, safe_json_to_str(data).c_str());
|
||||
fflush(cmd_out);
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@
|
||||
#include <filesystem>
|
||||
#include <functional>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
#include <queue>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
@@ -725,3 +726,97 @@ private:
|
||||
intptr_t out_handles[SERVER_SUBPROC_STREAMS] = { -1, -1 }; // fd on POSIX, HANDLE on Windows; taken lazily from sproc
|
||||
bool out_closed [SERVER_SUBPROC_STREAMS] = { false, false };
|
||||
};
|
||||
|
||||
//
|
||||
// router <-> child protocol
|
||||
//
|
||||
|
||||
// what this llama-server process does. a router spawns the other roles as child processes
|
||||
// and talks to them over their stdin/stdout
|
||||
enum server_role_type {
|
||||
SERVER_ROLE_ROUTER, // no model: lists the models, spawns instances on demand and proxies requests to them
|
||||
SERVER_ROLE_INSTANCE, // serves one model, standalone or as a child of a router
|
||||
SERVER_ROLE_DOWNLOAD, // child of a router: download a model, report the progress, exit
|
||||
};
|
||||
|
||||
static inline const char * server_role_type_to_str(server_role_type role) {
|
||||
switch (role) {
|
||||
case SERVER_ROLE_ROUTER: return "router";
|
||||
case SERVER_ROLE_INSTANCE: return "instance";
|
||||
case SERVER_ROLE_DOWNLOAD: return "download";
|
||||
default: GGML_ASSERT(false && "invalid server_role_type");
|
||||
}
|
||||
}
|
||||
|
||||
// the role a router asks of a child; anything it does not name is an instance
|
||||
static inline server_role_type server_role_type_from_str(const std::string & str) {
|
||||
if (str == "download") {
|
||||
return SERVER_ROLE_DOWNLOAD;
|
||||
}
|
||||
return SERVER_ROLE_INSTANCE;
|
||||
}
|
||||
|
||||
// set by the router in the environment of every child
|
||||
#define SERVER_ENV_ROUTER_PORT "LLAMA_SERVER_ROUTER_PORT" // its presence makes the process a child
|
||||
#define SERVER_ENV_CHILD_ROLE "LLAMA_SERVER_CHILD_ROLE" // server_role_type_to_str(), an instance when unset
|
||||
|
||||
// lines on the child's stdin (router -> child) and stdout (child -> router),
|
||||
// see server_role::setup() and server_role::notify_to_router()
|
||||
#define CMD_ROUTER_TO_CHILD_EXIT "cmd_router_to_child:exit"
|
||||
#define CMD_CHILD_TO_ROUTER_STATE "cmd_child_to_router:state:" // followed by json string
|
||||
|
||||
// address a child binds to: the router may listen on 0.0.0.0, but reaches its children locally
|
||||
// ref: https://github.com/ggml-org/llama.cpp/issues/17862
|
||||
#define CHILD_ADDR "127.0.0.1"
|
||||
|
||||
// the states a child reports to the router
|
||||
enum server_state {
|
||||
SERVER_STATE_DOWNLOADING,
|
||||
SERVER_STATE_LOADING,
|
||||
SERVER_STATE_READY,
|
||||
SERVER_STATE_SLEEPING,
|
||||
};
|
||||
|
||||
static inline std::string server_state_to_str(server_state state) {
|
||||
switch (state) {
|
||||
case SERVER_STATE_DOWNLOADING: return "downloading";
|
||||
case SERVER_STATE_LOADING: return "loading";
|
||||
case SERVER_STATE_READY: return "ready";
|
||||
case SERVER_STATE_SLEEPING: return "sleeping";
|
||||
default: GGML_ASSERT(false && "invalid server_state");
|
||||
}
|
||||
}
|
||||
|
||||
static inline server_state server_state_from_str(const std::string & str) {
|
||||
if (str == "downloading") return SERVER_STATE_DOWNLOADING;
|
||||
if (str == "loading") return SERVER_STATE_LOADING;
|
||||
if (str == "ready") return SERVER_STATE_READY;
|
||||
if (str == "sleeping") return SERVER_STATE_SLEEPING;
|
||||
GGML_ASSERT(false && "invalid server_state string");
|
||||
}
|
||||
|
||||
// the role of this process and, when it is a child of a router, its side of the protocol
|
||||
struct server_role {
|
||||
// serializes the notify_to_router writes
|
||||
std::mutex mtx_stdout;
|
||||
|
||||
// in a child, keeps stdout for the commands to the router, so it is created before anything is written;
|
||||
// everything else written to stdout goes to stderr with the logs
|
||||
server_role();
|
||||
~server_role();
|
||||
|
||||
// true if the current process was spawned by a router
|
||||
bool is_managed_by_router();
|
||||
server_role_type get_role();
|
||||
|
||||
// register the shutdown_handler to be called by the router
|
||||
// return the monitoring thread (to be joined by the caller)
|
||||
std::thread setup(const std::function<void(int)> & shutdown_handler);
|
||||
|
||||
// notify router server for status changes (e.g. loading, downloading, sleeping, etc.)
|
||||
// message will be handled by server_models::handle_child_state() on the router side
|
||||
void notify_to_router(const std::string & state_name, const json & payload);
|
||||
|
||||
private:
|
||||
FILE * cmd_out = nullptr; // the stdout the router reads the commands from
|
||||
};
|
||||
|
||||
@@ -58,31 +58,6 @@ struct server_context_meta {
|
||||
std::string model_ftype;
|
||||
};
|
||||
|
||||
enum server_state {
|
||||
SERVER_STATE_DOWNLOADING,
|
||||
SERVER_STATE_LOADING,
|
||||
SERVER_STATE_READY,
|
||||
SERVER_STATE_SLEEPING,
|
||||
};
|
||||
|
||||
static std::string server_state_to_str(server_state state) {
|
||||
switch (state) {
|
||||
case SERVER_STATE_DOWNLOADING: return "downloading";
|
||||
case SERVER_STATE_LOADING: return "loading";
|
||||
case SERVER_STATE_READY: return "ready";
|
||||
case SERVER_STATE_SLEEPING: return "sleeping";
|
||||
default: GGML_ASSERT(false && "invalid server_state");
|
||||
}
|
||||
}
|
||||
|
||||
static server_state server_state_from_str(const std::string & str) {
|
||||
if (str == "downloading") return SERVER_STATE_DOWNLOADING;
|
||||
if (str == "loading") return SERVER_STATE_LOADING;
|
||||
if (str == "ready") return SERVER_STATE_READY;
|
||||
if (str == "sleeping") return SERVER_STATE_SLEEPING;
|
||||
GGML_ASSERT(false && "invalid server_state string");
|
||||
}
|
||||
|
||||
using server_state_callback_t = std::function<void(server_state, json /* payload */)>;
|
||||
|
||||
struct server_context {
|
||||
|
||||
+15
-165
@@ -37,9 +37,6 @@ extern char **environ;
|
||||
|
||||
#define DEFAULT_STOP_TIMEOUT 10 // seconds
|
||||
|
||||
#define CMD_ROUTER_TO_CHILD_EXIT "cmd_router_to_child:exit"
|
||||
#define CMD_CHILD_TO_ROUTER_STATE "cmd_child_to_router:state:" // followed by json string
|
||||
|
||||
// note: SIGPIPE is ignored by the server
|
||||
static void request_child_exit(server_subproc & proc) {
|
||||
FILE * stdin_file = proc.sproc.stdin_file();
|
||||
@@ -49,10 +46,6 @@ static void request_child_exit(server_subproc & proc) {
|
||||
}
|
||||
}
|
||||
|
||||
// address for child process, this is needed because router may run on 0.0.0.0
|
||||
// ref: https://github.com/ggml-org/llama.cpp/issues/17862
|
||||
#define CHILD_ADDR "127.0.0.1"
|
||||
|
||||
// single-threaded, watching all child processes at once
|
||||
struct server_monitor {
|
||||
server_monitor(server_models & models) : models(models) {
|
||||
@@ -65,11 +58,11 @@ struct server_monitor {
|
||||
}
|
||||
|
||||
// thread-safe
|
||||
void watch(const std::string & name, std::shared_ptr<server_subproc> proc, server_child_mode mode, int port) {
|
||||
void watch(const std::string & name, std::shared_ptr<server_subproc> proc, server_role_type role, int port) {
|
||||
child_t c;
|
||||
c.name = name;
|
||||
c.proc = std::move(proc);
|
||||
c.mode = mode;
|
||||
c.role = role;
|
||||
c.port = port;
|
||||
if (!c.proc->has_output()) {
|
||||
SRV_ERR("failed to get stdout/stderr of child process for name=%s\n", name.c_str());
|
||||
@@ -87,7 +80,7 @@ private:
|
||||
struct child_t {
|
||||
std::string name;
|
||||
std::shared_ptr<server_subproc> proc;
|
||||
server_child_mode mode = SERVER_CHILD_MODE_NORMAL;
|
||||
server_role_type role = SERVER_ROLE_INSTANCE;
|
||||
int port = 0;
|
||||
std::string buf[SERVER_SUBPROC_STREAMS]; // partial line of each pipe
|
||||
bool eof = false; // output closed, waiting for the process to be reaped
|
||||
@@ -241,7 +234,7 @@ private:
|
||||
if (it->eof && !it->proc->is_alive()) {
|
||||
int exit_code = it->proc->join();
|
||||
it->proc->stopped.store(true, std::memory_order_release);
|
||||
models.on_child_exit(it->name, it->proc, it->mode, exit_code);
|
||||
models.on_child_exit(it->name, it->proc, it->role, exit_code);
|
||||
SRV_INF("instance name=%s exited with status %d\n", it->name.c_str(), exit_code);
|
||||
it = children.erase(it);
|
||||
} else {
|
||||
@@ -1324,7 +1317,7 @@ void server_models::load(const std::string & name) {
|
||||
|
||||
void server_models::load(const std::string & name, const load_options & opts) {
|
||||
// the fake load time of a model; a download worker loads none
|
||||
if (debug_fake_timing && opts.mode == SERVER_CHILD_MODE_NORMAL) {
|
||||
if (debug_fake_timing && opts.role == SERVER_ROLE_INSTANCE) {
|
||||
// do not hold the mutex here, other requests must keep making progress
|
||||
std::this_thread::sleep_for(std::chrono::seconds(2));
|
||||
}
|
||||
@@ -1345,7 +1338,7 @@ void server_models::load(const std::string & name, const load_options & opts) {
|
||||
// same model cannot start in between: a paused download or an idle cached model
|
||||
// is downloaded again with its files kept, a download on its way and a running
|
||||
// model are refused, and a preset or local model is not a download
|
||||
if (opts.mode == SERVER_CHILD_MODE_DOWNLOAD) {
|
||||
if (opts.role == SERVER_ROLE_DOWNLOAD) {
|
||||
auto it = mapping.find(name);
|
||||
if (it != mapping.end()) {
|
||||
const auto & cur = it->second.meta;
|
||||
@@ -1389,7 +1382,7 @@ void server_models::load(const std::string & name, const load_options & opts) {
|
||||
// releasing its lock and this lock_guard acquiring allows multiple
|
||||
// threads to each observe capacity and all proceed to load.
|
||||
// Download workers do not use models_max slots.
|
||||
if (opts.mode == SERVER_CHILD_MODE_NORMAL && base_params.models_max > 0) {
|
||||
if (opts.role == SERVER_ROLE_INSTANCE && base_params.models_max > 0) {
|
||||
size_t count_active = 0;
|
||||
for (const auto & m : mapping) {
|
||||
if (m.second.meta.is_running()) {
|
||||
@@ -1421,11 +1414,11 @@ void server_models::load(const std::string & name, const load_options & opts) {
|
||||
|
||||
std::vector<std::string> child_args = inst.meta.args; // copy
|
||||
std::vector<std::string> child_env = base_env; // copy
|
||||
child_env.push_back("LLAMA_SERVER_ROUTER_PORT=" + std::to_string(base_params.port));
|
||||
child_env.push_back(std::string(SERVER_ENV_ROUTER_PORT) + "=" + std::to_string(base_params.port));
|
||||
|
||||
if (opts.mode == SERVER_CHILD_MODE_DOWNLOAD) {
|
||||
if (opts.role == SERVER_ROLE_DOWNLOAD) {
|
||||
inst.meta.status = SERVER_MODEL_STATUS_DOWNLOADING;
|
||||
child_env.push_back("LLAMA_SERVER_CHILD_MODE=download");
|
||||
child_env.push_back(std::string(SERVER_ENV_CHILD_ROLE) + "=" + server_role_type_to_str(SERVER_ROLE_DOWNLOAD));
|
||||
child_env.push_back("LLAMA_ARG_HF_REPO=" + name);
|
||||
|
||||
// the sidecars as -hf would pull them: the draft types, and the mmproj switch
|
||||
@@ -1460,7 +1453,7 @@ void server_models::load(const std::string & name, const load_options & opts) {
|
||||
}
|
||||
|
||||
// remember the download: the partial files carry a content hash, not the name
|
||||
if (opts.mode == SERVER_CHILD_MODE_DOWNLOAD) {
|
||||
if (opts.role == SERVER_ROLE_DOWNLOAD) {
|
||||
unfinished_downloads.add({ name, opts.sidecars, opts.partials });
|
||||
}
|
||||
|
||||
@@ -1480,7 +1473,7 @@ void server_models::load(const std::string & name, const load_options & opts) {
|
||||
auto proc = inst.subproc;
|
||||
int port = inst.meta.port;
|
||||
mapping[name] = std::move(inst);
|
||||
monitor->watch(name, proc, opts.mode, port);
|
||||
monitor->watch(name, proc, opts.role, port);
|
||||
cv.notify_all();
|
||||
}
|
||||
|
||||
@@ -1493,7 +1486,7 @@ void server_models::request_stop(const std::string & name, bool send_exit) {
|
||||
monitor->stop(name, it->second.meta.stop_timeout, send_exit);
|
||||
}
|
||||
|
||||
void server_models::on_child_exit(const std::string & name, const std::shared_ptr<server_subproc> & proc, server_child_mode mode, int exit_code) {
|
||||
void server_models::on_child_exit(const std::string & name, const std::shared_ptr<server_subproc> & proc, server_role_type role, int exit_code) {
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(mutex);
|
||||
auto it = mapping.find(name);
|
||||
@@ -1502,7 +1495,7 @@ void server_models::on_child_exit(const std::string & name, const std::shared_pt
|
||||
return; // entry erased, or a newer instance took the name
|
||||
}
|
||||
}
|
||||
if (mode == SERVER_CHILD_MODE_DOWNLOAD) {
|
||||
if (role == SERVER_ROLE_DOWNLOAD) {
|
||||
// instance will be cleaned up on next load_models() call
|
||||
std::lock_guard<std::mutex> lk(mutex);
|
||||
stopping_models.erase(name);
|
||||
@@ -2046,149 +2039,6 @@ void server_models::handle_child_state(const std::string & name, const std::stri
|
||||
}
|
||||
}
|
||||
|
||||
//
|
||||
// server_child
|
||||
//
|
||||
|
||||
server_child::server_child() {
|
||||
if (is_child()) {
|
||||
cmd_out = server_reserve_stdout();
|
||||
}
|
||||
}
|
||||
|
||||
server_child::~server_child() {
|
||||
if (cmd_out) {
|
||||
fclose(cmd_out);
|
||||
}
|
||||
}
|
||||
|
||||
bool server_child::is_child() {
|
||||
const char * router_port = std::getenv("LLAMA_SERVER_ROUTER_PORT");
|
||||
return router_port != nullptr;
|
||||
}
|
||||
|
||||
server_child_mode server_child::get_mode() {
|
||||
const char * mode = std::getenv("LLAMA_SERVER_CHILD_MODE");
|
||||
std::string mode_str(mode ? mode : "");
|
||||
if (mode_str == "download") {
|
||||
return SERVER_CHILD_MODE_DOWNLOAD;
|
||||
} else {
|
||||
return SERVER_CHILD_MODE_NORMAL;
|
||||
}
|
||||
}
|
||||
|
||||
struct server_download_state : public common_download_callback {
|
||||
server_child * self;
|
||||
std::function<bool()> should_stop;
|
||||
std::atomic<int64_t> last_progress_time{0}; // multiple files downloading in different threads
|
||||
bool is_ok = false;
|
||||
|
||||
server_download_state(server_child * s) : self(s) {}
|
||||
|
||||
bool run(common_params & params) {
|
||||
try {
|
||||
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();
|
||||
SRV_ERR("download failed for model name=%s: %s\n", model_name.c_str(), e.what());
|
||||
is_ok = false;
|
||||
}
|
||||
return is_ok;
|
||||
}
|
||||
void on_progress(const common_download_progress & p) {
|
||||
json data = {
|
||||
{"url", p.url},
|
||||
{"downloaded", p.downloaded},
|
||||
{"total", p.total},
|
||||
};
|
||||
self->notify_to_router(server_state_to_str(SERVER_STATE_DOWNLOADING), data);
|
||||
}
|
||||
void on_start(const common_download_progress & p) override {
|
||||
on_progress(p);
|
||||
}
|
||||
void on_update(const common_download_progress & p) override {
|
||||
int64_t now = ggml_time_ms();
|
||||
// throttle progress updates to avoid flooding logs
|
||||
if (now - last_progress_time.load(std::memory_order_relaxed) >= 100) {
|
||||
on_progress(p);
|
||||
last_progress_time.store(now, std::memory_order_relaxed);
|
||||
}
|
||||
}
|
||||
void on_done(const common_download_progress & p, bool) override {
|
||||
on_progress(p);
|
||||
}
|
||||
bool is_cancelled() const override {
|
||||
return should_stop ? should_stop() : false;
|
||||
}
|
||||
};
|
||||
|
||||
int server_child::run_download(common_params & params) {
|
||||
auto cancelled = std::make_shared<std::atomic<bool>>(false);
|
||||
|
||||
// monitor stdin for cancellation command from the router
|
||||
std::thread signal_thread = setup([cancelled](int) {
|
||||
cancelled->store(true, std::memory_order_relaxed);
|
||||
});
|
||||
|
||||
server_download_state dl(this);
|
||||
dl.should_stop = [cancelled]() {
|
||||
return cancelled->load(std::memory_order_relaxed);
|
||||
};
|
||||
|
||||
bool ok = dl.run(params);
|
||||
|
||||
notify_to_router(server_state_to_str(SERVER_STATE_DOWNLOADING), {
|
||||
{"result", ok ? "download_finished" : "download_failed"},
|
||||
});
|
||||
|
||||
// router should send CMD_ROUTER_TO_CHILD_EXIT after receiving the result
|
||||
if (signal_thread.joinable()) {
|
||||
signal_thread.join();
|
||||
}
|
||||
|
||||
SRV_INF("download completed %s\n", ok ? "successfully" : "with errors");
|
||||
return 0;
|
||||
}
|
||||
|
||||
std::thread server_child::setup(const std::function<void(int)> & shutdown_handler) {
|
||||
// setup thread for monitoring stdin
|
||||
return std::thread([shutdown_handler]() {
|
||||
// wait for EOF on stdin
|
||||
SRV_INF("%s", "child server monitoring thread started, waiting for EOF on stdin...\n");
|
||||
bool eof = false;
|
||||
while (true) {
|
||||
std::string line;
|
||||
if (!std::getline(std::cin, line)) {
|
||||
// EOF detected, that means the router server is unexpectedly exit or killed
|
||||
eof = true;
|
||||
break;
|
||||
}
|
||||
if (line.find(CMD_ROUTER_TO_CHILD_EXIT) != std::string::npos) {
|
||||
SRV_INF("%s", "exit command received, exiting...\n");
|
||||
shutdown_handler(0);
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (eof) {
|
||||
SRV_INF("%s", "EOF on stdin detected, forcing shutdown...\n");
|
||||
exit(1);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void server_child::notify_to_router(const std::string & state, const json & payload) {
|
||||
json data = {
|
||||
{"state", state},
|
||||
{"payload", payload},
|
||||
};
|
||||
std::lock_guard<std::mutex> lk(mtx_stdout);
|
||||
fprintf(cmd_out, "%s%s\n", CMD_CHILD_TO_ROUTER_STATE, safe_json_to_str(data).c_str());
|
||||
fflush(cmd_out);
|
||||
}
|
||||
|
||||
|
||||
//
|
||||
// server_models_routes
|
||||
//
|
||||
@@ -2617,7 +2467,7 @@ void server_models_routes::init_routes() {
|
||||
SRV_INF("starting download for model '%s'\n", name.c_str());
|
||||
{
|
||||
server_models::load_options load_opts;
|
||||
load_opts.mode = SERVER_CHILD_MODE_DOWNLOAD;
|
||||
load_opts.role = SERVER_ROLE_DOWNLOAD;
|
||||
load_opts.custom_meta = server_model_meta{};
|
||||
load_opts.custom_meta->source = SERVER_MODEL_SOURCE_CACHE;
|
||||
load_opts.custom_meta->name = name;
|
||||
|
||||
@@ -46,11 +46,6 @@ enum server_model_source {
|
||||
SERVER_MODEL_SOURCE_CACHE,
|
||||
};
|
||||
|
||||
enum server_child_mode {
|
||||
SERVER_CHILD_MODE_NORMAL, // load the model and run normally
|
||||
SERVER_CHILD_MODE_DOWNLOAD, // download the model and exit
|
||||
};
|
||||
|
||||
static std::string server_model_status_to_string(server_model_status status) {
|
||||
switch (status) {
|
||||
case SERVER_MODEL_STATUS_DOWNLOADING: return "downloading";
|
||||
@@ -171,7 +166,7 @@ private:
|
||||
server_model_meta meta;
|
||||
int req_count = 0; // number of active proxy requests
|
||||
|
||||
// ask the child to exit (it handles the command on its stdin, see server_child::setup)
|
||||
// ask the child to exit (it handles the command on its stdin, see server_role::setup)
|
||||
void request_exit() const;
|
||||
};
|
||||
|
||||
@@ -276,7 +271,7 @@ private:
|
||||
void request_stop(const std::string & name, bool send_exit = true);
|
||||
|
||||
// called by the monitor once a child exited and was reaped
|
||||
void on_child_exit(const std::string & name, const std::shared_ptr<server_subproc> & proc, server_child_mode mode, int exit_code);
|
||||
void on_child_exit(const std::string & name, const std::shared_ptr<server_subproc> & proc, server_role_type role, int exit_code);
|
||||
|
||||
// notify SSE clients
|
||||
void notify_sse(const std::string & event, const std::string & model_id, const json & data = nullptr);
|
||||
@@ -320,7 +315,7 @@ public:
|
||||
std::vector<server_model_meta> get_all_meta();
|
||||
|
||||
struct load_options {
|
||||
server_child_mode mode = SERVER_CHILD_MODE_NORMAL;
|
||||
server_role_type role = SERVER_ROLE_INSTANCE;
|
||||
// used for spawning a downloading child process
|
||||
std::optional<server_model_meta> custom_meta = std::nullopt;
|
||||
// download mode: the sidecar kinds pulled with the model, as -hf would, and
|
||||
@@ -369,7 +364,7 @@ public:
|
||||
// forward a websocket connection to the child, returns when either side closes
|
||||
void proxy_ws(const server_http_req & req, server_http_context::ws_conn & conn, const std::string & name);
|
||||
|
||||
// handle message sent from server_child::notify_to_router()
|
||||
// handle message sent from server_role::notify_to_router()
|
||||
// raw input must starts with CMD_CHILD_TO_ROUTER_STATE, followed by a JSON string
|
||||
// called from the monitor thread
|
||||
// payload per state:
|
||||
@@ -383,33 +378,6 @@ private:
|
||||
std::unique_ptr<server_monitor> monitor;
|
||||
};
|
||||
|
||||
struct server_child {
|
||||
// serializes the notify_to_router writes
|
||||
std::mutex mtx_stdout;
|
||||
std::atomic<bool> is_finished_downloading = false; // set by run_download
|
||||
|
||||
// in a child, keeps stdout for the commands to the router, so it is created before anything is written;
|
||||
// everything else written to stdout goes to stderr with the logs
|
||||
server_child();
|
||||
~server_child();
|
||||
|
||||
// return true if the current process is a child server instance
|
||||
bool is_child();
|
||||
server_child_mode get_mode();
|
||||
int run_download(common_params & params);
|
||||
|
||||
// register the shutdown_handler to be called by the router
|
||||
// return the monitoring thread (to be joined by the caller)
|
||||
std::thread setup(const std::function<void(int)> & shutdown_handler);
|
||||
|
||||
// notify router server for status changes (e.g. loading, downloading, sleeping, etc.)
|
||||
// message will be handled by server_models::handle_child_state() on the router side
|
||||
void notify_to_router(const std::string & state_name, const json & payload);
|
||||
|
||||
private:
|
||||
FILE * cmd_out = nullptr; // the stdout the router reads the commands from
|
||||
};
|
||||
|
||||
struct server_models_routes {
|
||||
common_params params;
|
||||
json ui_settings = json::object(); // Primary: new name
|
||||
|
||||
+473
-300
@@ -37,12 +37,17 @@ static inline void signal_handler(int signal) {
|
||||
shutdown_handler(signal);
|
||||
}
|
||||
|
||||
// how the server is run; everything is on when it is the llama-server binary
|
||||
struct server_run_opts {
|
||||
bool install_signal_handlers = true; // off when embedded in llama-cli, which handles them itself
|
||||
bool resolve_model = true; // download / resolve -hf and docker models; off when the caller did it already
|
||||
};
|
||||
|
||||
// satisfies -Wmissing-declarations (used by llama command)
|
||||
int llama_server(int argc, char ** argv);
|
||||
|
||||
// to be used via CLI (argc / argv are used by router mode only)
|
||||
int llama_server(common_params & params, int argc, char ** argv);
|
||||
int llama_server(common_params & params, int argc, char ** argv, server_child & child);
|
||||
void llama_server_terminate();
|
||||
void llama_server_terminate() {
|
||||
if (shutdown_handler) {
|
||||
@@ -91,72 +96,24 @@ static server_http_context::handler_t ex_wrapper(server_http_context::handler_t
|
||||
};
|
||||
}
|
||||
|
||||
int llama_server(int argc, char ** argv) {
|
||||
server_child child;
|
||||
//
|
||||
// shared by every role
|
||||
//
|
||||
|
||||
std::setlocale(LC_NUMERIC, "C");
|
||||
|
||||
#ifndef _WIN32
|
||||
// Ignore SIGPIPE so the server does not crash if a child (MCP server, tools runtime) exits while we are writing to its stdin
|
||||
signal(SIGPIPE, SIG_IGN);
|
||||
#endif
|
||||
|
||||
// own arguments required by this example
|
||||
common_params params;
|
||||
|
||||
common_init();
|
||||
|
||||
// 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
|
||||
server_stream_session_manager_start();
|
||||
|
||||
if (!common_params_parse(argc, argv, params, LLAMA_EXAMPLE_SERVER)) {
|
||||
return 1;
|
||||
static server_role_type server_role_detect(const common_params & params, server_role & role) {
|
||||
if (role.is_managed_by_router()) {
|
||||
return role.get_role();
|
||||
}
|
||||
|
||||
SRV_INF("%s", "initializing ...\n");
|
||||
|
||||
llama_backend_init();
|
||||
llama_numa_init(params.numa);
|
||||
|
||||
const int result = llama_server(params, argc, argv, child);
|
||||
common_log_flush(common_log_main());
|
||||
return result;
|
||||
// a server given no model is a router
|
||||
const bool has_model = !params.model.path.empty()
|
||||
|| !params.model.hf_repo.empty()
|
||||
|| !params.model.docker_repo.empty();
|
||||
return has_model ? SERVER_ROLE_INSTANCE : SERVER_ROLE_ROUTER;
|
||||
}
|
||||
|
||||
int llama_server(common_params & params, int argc, char ** argv) {
|
||||
server_child child;
|
||||
return llama_server(params, argc, argv, child);
|
||||
}
|
||||
|
||||
int llama_server(common_params & params, int argc, char ** argv, server_child & child) {
|
||||
bool is_run_by_cli = (argv == nullptr);
|
||||
|
||||
common_models_handler models_handler;
|
||||
|
||||
// note: router mode also accepts -hf remote-preset, so we need to check that first
|
||||
if (!is_run_by_cli && !params.model.hf_repo.empty()) {
|
||||
try {
|
||||
models_handler = common_models_handler_init(params, LLAMA_EXAMPLE_SERVER);
|
||||
if (common_models_handler_is_preset_repo(models_handler)) {
|
||||
// apply the preset and start the server in router mode
|
||||
common_models_handler_apply(models_handler, params);
|
||||
}
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("failed to fetch model metadata: %s\n", e.what());
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
// router server never loads a model and must not touch the GPU
|
||||
const bool is_router_server = params.model.path.empty()
|
||||
&& params.model.hf_repo.empty()
|
||||
&& params.model.docker_repo.empty();
|
||||
|
||||
// skip device enumeration so the CUDA primary context stays uncreated
|
||||
common_params_print_info(params, !is_router_server);
|
||||
|
||||
if (!is_router_server) {
|
||||
// adjust the params for the role; returns false on a configuration error
|
||||
static bool server_prepare_params(common_params & params, server_role_type role_type) {
|
||||
if (role_type != SERVER_ROLE_ROUTER) {
|
||||
// validate batch size for embeddings
|
||||
// embeddings require all tokens to be processed in a single ubatch
|
||||
// see https://github.com/ggml-org/llama.cpp/issues/12836
|
||||
@@ -197,7 +154,7 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
const std::string reason = server_connect::unavailable_reason(params);
|
||||
if (!reason.empty()) {
|
||||
SRV_ERR("--connect is not available: %s\n", reason.c_str());
|
||||
return 1;
|
||||
return false;
|
||||
}
|
||||
params.server_connect_code = server_connect::resolve_code(params.server_connect_code);
|
||||
} else if (!params.server_connect_code.empty()) {
|
||||
@@ -206,77 +163,11 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
params.server_connect_code.clear();
|
||||
}
|
||||
|
||||
// note: this is guaranteed to out-live ctx_http and tools
|
||||
server_mcp mcp_mgr;
|
||||
|
||||
// struct that contains llama context and inference
|
||||
server_context ctx_server;
|
||||
|
||||
//
|
||||
// Router
|
||||
//
|
||||
|
||||
// register API routes
|
||||
server_routes routes(params, ctx_server);
|
||||
server_tools tools;
|
||||
|
||||
std::optional<server_models_routes> models_routes{};
|
||||
|
||||
server_http_context ctx_http;
|
||||
if (!ctx_http.init(params)) {
|
||||
SRV_ERR("%s", "failed to initialize HTTP server\n");
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (is_router_server) {
|
||||
// setup server instances manager
|
||||
try {
|
||||
models_routes.emplace(params, argc, argv);
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("failed to initialize router models: %s\n", e.what());
|
||||
return 1;
|
||||
}
|
||||
|
||||
// proxy handlers
|
||||
// note: routes.get_health stays the same
|
||||
routes.get_metrics = models_routes->proxy_get;
|
||||
routes.post_props = models_routes->proxy_post;
|
||||
routes.post_completions = models_routes->proxy_post;
|
||||
routes.post_completions_oai = models_routes->proxy_post;
|
||||
routes.post_chat_completions = models_routes->proxy_post;
|
||||
routes.post_control = models_routes->proxy_post;
|
||||
routes.post_responses_oai = models_routes->proxy_post;
|
||||
routes.post_transcriptions_oai = models_routes->proxy_post;
|
||||
routes.post_tts = models_routes->proxy_post;
|
||||
routes.ws_realtime = models_routes->proxy_ws;
|
||||
routes.post_anthropic_messages = models_routes->proxy_post;
|
||||
routes.post_anthropic_count_tokens = models_routes->proxy_post;
|
||||
routes.post_infill = models_routes->proxy_post;
|
||||
routes.post_embeddings = models_routes->proxy_post;
|
||||
routes.post_embeddings_oai = models_routes->proxy_post;
|
||||
routes.post_rerank = models_routes->proxy_post;
|
||||
routes.post_systemone = models_routes->proxy_post;
|
||||
routes.post_tokenize = models_routes->proxy_post;
|
||||
routes.post_detokenize = models_routes->proxy_post;
|
||||
routes.post_apply_template = models_routes->proxy_post;
|
||||
routes.post_chat_completions_tok = models_routes->proxy_post;
|
||||
routes.post_responses_tok_oai = models_routes->proxy_post;
|
||||
routes.get_lora_adapters = models_routes->proxy_get;
|
||||
routes.post_lora_adapters = models_routes->proxy_post;
|
||||
routes.get_slots = models_routes->proxy_get;
|
||||
routes.post_slots = models_routes->proxy_post;
|
||||
|
||||
// custom routes for router
|
||||
routes.get_props = models_routes->get_router_props;
|
||||
routes.get_models = models_routes->get_router_models;
|
||||
|
||||
ctx_http.post("/models", ex_wrapper(models_routes->post_router_models));
|
||||
ctx_http.post("/models/load", ex_wrapper(models_routes->post_router_models_load));
|
||||
ctx_http.post("/models/unload", ex_wrapper(models_routes->post_router_models_unload));
|
||||
ctx_http.get ("/models/sse", ex_wrapper(models_routes->get_router_models_sse));
|
||||
ctx_http.del ("/models", ex_wrapper(models_routes->del_router_models));
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
// the API of every server; a router points the handlers in routes at its proxies before calling this
|
||||
static void server_register_routes(server_http_context & ctx_http, server_routes & routes) {
|
||||
ctx_http.get ("/health", ex_wrapper(routes.get_health)); // public endpoint (no API key check)
|
||||
ctx_http.get ("/v1/health", ex_wrapper(routes.get_health)); // public endpoint (no API key check)
|
||||
ctx_http.get ("/metrics", ex_wrapper(routes.get_metrics));
|
||||
@@ -322,27 +213,30 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
ctx_http.get ("/slots", ex_wrapper(routes.get_slots));
|
||||
ctx_http.post("/slots/:id_slot", ex_wrapper(routes.post_slots));
|
||||
|
||||
// resumable streaming: a child binds the local session factories, the router binds
|
||||
// proxies that resolve the owning child, see server-stream.h
|
||||
server_http_context::handler_t stream_get_h;
|
||||
server_http_context::handler_t streams_lookup_h;
|
||||
server_http_context::handler_t stream_delete_h;
|
||||
if (is_router_server) {
|
||||
stream_get_h = models_routes->router_stream_get;
|
||||
streams_lookup_h = models_routes->router_streams_lookup;
|
||||
stream_delete_h = models_routes->router_stream_delete;
|
||||
} else {
|
||||
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", ex_wrapper(stream_get_h));
|
||||
ctx_http.post("/v1/streams/lookup", ex_wrapper(streams_lookup_h));
|
||||
ctx_http.del ("/v1/stream", ex_wrapper(stream_delete_h));
|
||||
|
||||
// Google Cloud Platform (Vertex AI) compat
|
||||
ctx_http.register_gcp_compat();
|
||||
}
|
||||
|
||||
// resumable streaming, see server-stream.h: an instance binds the local session factories,
|
||||
// a router binds proxies that resolve the owning child
|
||||
static void server_register_stream_routes(
|
||||
server_http_context & ctx_http,
|
||||
const server_http_context::handler_t & stream_get,
|
||||
const server_http_context::handler_t & streams_lookup,
|
||||
const server_http_context::handler_t & stream_delete) {
|
||||
ctx_http.get ("/v1/stream", ex_wrapper(stream_get));
|
||||
ctx_http.post("/v1/streams/lookup", ex_wrapper(streams_lookup));
|
||||
ctx_http.del ("/v1/stream", ex_wrapper(stream_delete));
|
||||
}
|
||||
|
||||
// the features behind a flag, and the warning about the ones that must not face untrusted clients
|
||||
// returns false when one of them fails to start
|
||||
static bool server_setup_features(
|
||||
server_http_context & ctx_http,
|
||||
const common_params & params,
|
||||
server_mcp & mcp_mgr,
|
||||
server_tools & tools,
|
||||
server_role_type role_type) {
|
||||
// return 403 for disabled features
|
||||
server_http_context::handler_t res_403 = [](const server_http_req &) {
|
||||
auto res = std::make_unique<server_http_res>();
|
||||
@@ -360,9 +254,8 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
SRV_WRN("%s", "security: no API key is set and CORS allows all origins (see https://github.com/ggml-org/llama.cpp/pull/25655)\n");
|
||||
}
|
||||
|
||||
// CORS proxy (EXPERIMENTAL, only used by the Web UI for MCP)
|
||||
std::vector<std::string> warn_names;
|
||||
if (is_router_server) {
|
||||
if (role_type == SERVER_ROLE_ROUTER) {
|
||||
warn_names.push_back("router mode");
|
||||
}
|
||||
|
||||
@@ -370,6 +263,7 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
warn_names.push_back("peer-to-peer tunnel (llama-connect, experimental)");
|
||||
}
|
||||
|
||||
// CORS proxy (EXPERIMENTAL, only used by the Web UI for MCP)
|
||||
if (params.ui_mcp_proxy) {
|
||||
ctx_http.get ("/cors-proxy", ex_wrapper(proxy_handler_get));
|
||||
ctx_http.post("/cors-proxy", ex_wrapper(proxy_handler_post));
|
||||
@@ -383,7 +277,7 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
mcp_mgr.start(params);
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("MCP starting failed: %s\n", e.what());
|
||||
return 1;
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!params.server_tools.empty() || !mcp_mgr.empty()) {
|
||||
@@ -391,7 +285,7 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
tools.setup(params.server_tools, mcp_mgr, params.server_tools_runtime);
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("tools setup failed: %s\n", e.what());
|
||||
return 1;
|
||||
return false;
|
||||
}
|
||||
ctx_http.get ("/tools", ex_wrapper(tools.handle_get));
|
||||
ctx_http.post("/tools", ex_wrapper(tools.handle_post));
|
||||
@@ -418,15 +312,267 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
SRV_WRN("security: %s enabled - do not expose to untrusted environments\n", features.c_str());
|
||||
}
|
||||
|
||||
//
|
||||
// Handle downloading model
|
||||
//
|
||||
return true;
|
||||
}
|
||||
|
||||
if (child.is_child() && child.get_mode() == SERVER_CHILD_MODE_DOWNLOAD) {
|
||||
return child.run_download(params);
|
||||
} else if (!is_router_server && !is_run_by_cli) {
|
||||
// single-model mode (NOT spawned by router)
|
||||
// if this is invoked by CLI, model downloading should be already handled
|
||||
static void server_install_signal_handlers() {
|
||||
#if defined (__unix__) || (defined (__APPLE__) && defined (__MACH__))
|
||||
struct sigaction sigint_action;
|
||||
sigint_action.sa_handler = signal_handler;
|
||||
sigemptyset (&sigint_action.sa_mask);
|
||||
sigint_action.sa_flags = 0;
|
||||
sigaction(SIGINT, &sigint_action, NULL);
|
||||
sigaction(SIGTERM, &sigint_action, NULL);
|
||||
#elif defined (_WIN32)
|
||||
auto console_ctrl_handler = +[](DWORD ctrl_type) -> BOOL {
|
||||
return (ctrl_type == CTRL_C_EVENT) ? (signal_handler(SIGINT), true) : false;
|
||||
};
|
||||
SetConsoleCtrlHandler(reinterpret_cast<PHANDLER_ROUTINE>(console_ctrl_handler), true);
|
||||
#endif
|
||||
}
|
||||
|
||||
// once the HTTP server listens: log the addresses and spawn llama-connect if asked, so the port is
|
||||
// bound and the child startup health check reaches us. the destructor of connect_proc stops it
|
||||
static bool server_on_listening(server_http_context & ctx_http, const common_params & params, server_connect & connect_proc) {
|
||||
for (const auto & address : ctx_http.listening_addresses) {
|
||||
SRV_INF("listening on %s\n", address.c_str());
|
||||
}
|
||||
|
||||
if (params.server_connect && !connect_proc.start(params, ctx_http.port)) {
|
||||
SRV_ERR("%s", "exiting due to llama-connect error\n");
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
//
|
||||
// download worker: a child of the router that pulls a model into the cache and reports the progress
|
||||
//
|
||||
|
||||
struct server_download_state : public common_download_callback {
|
||||
server_role * role;
|
||||
std::function<bool()> should_stop;
|
||||
std::atomic<int64_t> last_progress_time{0}; // multiple files downloading in different threads
|
||||
bool is_ok = false;
|
||||
|
||||
server_download_state(server_role * r) : role(r) {}
|
||||
|
||||
bool run(common_params & params) {
|
||||
try {
|
||||
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();
|
||||
SRV_ERR("download failed for model name=%s: %s\n", model_name.c_str(), e.what());
|
||||
is_ok = false;
|
||||
}
|
||||
return is_ok;
|
||||
}
|
||||
void on_progress(const common_download_progress & p) {
|
||||
json data = {
|
||||
{"url", p.url},
|
||||
{"downloaded", p.downloaded},
|
||||
{"total", p.total},
|
||||
};
|
||||
role->notify_to_router(server_state_to_str(SERVER_STATE_DOWNLOADING), data);
|
||||
}
|
||||
void on_start(const common_download_progress & p) override {
|
||||
on_progress(p);
|
||||
}
|
||||
void on_update(const common_download_progress & p) override {
|
||||
int64_t now = ggml_time_ms();
|
||||
// throttle progress updates to avoid flooding logs
|
||||
if (now - last_progress_time.load(std::memory_order_relaxed) >= 100) {
|
||||
on_progress(p);
|
||||
last_progress_time.store(now, std::memory_order_relaxed);
|
||||
}
|
||||
}
|
||||
void on_done(const common_download_progress & p, bool) override {
|
||||
on_progress(p);
|
||||
}
|
||||
bool is_cancelled() const override {
|
||||
return should_stop ? should_stop() : false;
|
||||
}
|
||||
};
|
||||
|
||||
static int server_run_download(common_params & params, server_role & role) {
|
||||
auto cancelled = std::make_shared<std::atomic<bool>>(false);
|
||||
|
||||
// monitor stdin for cancellation command from the router
|
||||
std::thread signal_thread = role.setup([cancelled](int) {
|
||||
cancelled->store(true, std::memory_order_relaxed);
|
||||
});
|
||||
|
||||
server_download_state dl(&role);
|
||||
dl.should_stop = [cancelled]() {
|
||||
return cancelled->load(std::memory_order_relaxed);
|
||||
};
|
||||
|
||||
bool ok = dl.run(params);
|
||||
|
||||
role.notify_to_router(server_state_to_str(SERVER_STATE_DOWNLOADING), {
|
||||
{"result", ok ? "download_finished" : "download_failed"},
|
||||
});
|
||||
|
||||
// router should send CMD_ROUTER_TO_CHILD_EXIT after receiving the result
|
||||
if (signal_thread.joinable()) {
|
||||
signal_thread.join();
|
||||
}
|
||||
|
||||
SRV_INF("download completed %s\n", ok ? "successfully" : "with errors");
|
||||
return 0;
|
||||
}
|
||||
|
||||
//
|
||||
// router
|
||||
//
|
||||
|
||||
static int server_run_router(common_params & params, int argc, char ** argv, const server_run_opts & opts) {
|
||||
SRV_INF("%s", "starting server in router mode. models will be automatically loaded on-demand\n");
|
||||
|
||||
// note: this is guaranteed to out-live ctx_http and tools
|
||||
server_mcp mcp_mgr;
|
||||
|
||||
// the router loads no model, its context stays empty; the routes it does not proxy still need one
|
||||
server_context ctx_server;
|
||||
server_routes routes(params, ctx_server);
|
||||
server_tools tools;
|
||||
|
||||
server_http_context ctx_http;
|
||||
if (!ctx_http.init(params)) {
|
||||
SRV_ERR("%s", "failed to initialize HTTP server\n");
|
||||
return 1;
|
||||
}
|
||||
|
||||
// setup server instances manager
|
||||
std::optional<server_models_routes> models_routes;
|
||||
try {
|
||||
models_routes.emplace(params, argc, argv);
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("failed to initialize router models: %s\n", e.what());
|
||||
return 1;
|
||||
}
|
||||
|
||||
// proxy handlers
|
||||
// note: routes.get_health stays the same
|
||||
routes.get_metrics = models_routes->proxy_get;
|
||||
routes.post_props = models_routes->proxy_post;
|
||||
routes.post_completions = models_routes->proxy_post;
|
||||
routes.post_completions_oai = models_routes->proxy_post;
|
||||
routes.post_chat_completions = models_routes->proxy_post;
|
||||
routes.post_control = models_routes->proxy_post;
|
||||
routes.post_responses_oai = models_routes->proxy_post;
|
||||
routes.post_transcriptions_oai = models_routes->proxy_post;
|
||||
routes.post_tts = models_routes->proxy_post;
|
||||
routes.ws_realtime = models_routes->proxy_ws;
|
||||
routes.post_anthropic_messages = models_routes->proxy_post;
|
||||
routes.post_anthropic_count_tokens = models_routes->proxy_post;
|
||||
routes.post_infill = models_routes->proxy_post;
|
||||
routes.post_embeddings = models_routes->proxy_post;
|
||||
routes.post_embeddings_oai = models_routes->proxy_post;
|
||||
routes.post_rerank = models_routes->proxy_post;
|
||||
routes.post_systemone = models_routes->proxy_post;
|
||||
routes.post_tokenize = models_routes->proxy_post;
|
||||
routes.post_detokenize = models_routes->proxy_post;
|
||||
routes.post_apply_template = models_routes->proxy_post;
|
||||
routes.post_chat_completions_tok = models_routes->proxy_post;
|
||||
routes.post_responses_tok_oai = models_routes->proxy_post;
|
||||
routes.get_lora_adapters = models_routes->proxy_get;
|
||||
routes.post_lora_adapters = models_routes->proxy_post;
|
||||
routes.get_slots = models_routes->proxy_get;
|
||||
routes.post_slots = models_routes->proxy_post;
|
||||
|
||||
// custom routes for router
|
||||
routes.get_props = models_routes->get_router_props;
|
||||
routes.get_models = models_routes->get_router_models;
|
||||
|
||||
ctx_http.post("/models", ex_wrapper(models_routes->post_router_models));
|
||||
ctx_http.post("/models/load", ex_wrapper(models_routes->post_router_models_load));
|
||||
ctx_http.post("/models/unload", ex_wrapper(models_routes->post_router_models_unload));
|
||||
ctx_http.get ("/models/sse", ex_wrapper(models_routes->get_router_models_sse));
|
||||
ctx_http.del ("/models", ex_wrapper(models_routes->del_router_models));
|
||||
|
||||
server_register_routes(ctx_http, routes);
|
||||
server_register_stream_routes(ctx_http,
|
||||
models_routes->router_stream_get,
|
||||
models_routes->router_streams_lookup,
|
||||
models_routes->router_stream_delete);
|
||||
|
||||
if (!server_setup_features(ctx_http, params, mcp_mgr, tools, SERVER_ROLE_ROUTER)) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
auto clean_up = [&models_routes, &mcp_mgr]() {
|
||||
SRV_INF("%s: cleaning up before exit...\n", __func__);
|
||||
// stop the session GC first, it finalizes live sessions and wakes pending readers
|
||||
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();
|
||||
}
|
||||
mcp_mgr.shutdown();
|
||||
llama_backend_free();
|
||||
};
|
||||
|
||||
if (!ctx_http.start()) {
|
||||
clean_up();
|
||||
SRV_ERR("%s", "exiting due to HTTP server error\n");
|
||||
return 1;
|
||||
}
|
||||
ctx_http.is_ready.store(true);
|
||||
|
||||
shutdown_handler = [&](int) {
|
||||
if (models_routes.has_value()) {
|
||||
// important to disconnect any SSE clients
|
||||
models_routes->stopping.store(true);
|
||||
}
|
||||
mcp_mgr.shutdown();
|
||||
ctx_http.stop();
|
||||
};
|
||||
|
||||
try {
|
||||
models_routes->models.load_startup_models();
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("failed to load models on startup: %s\n", e.what());
|
||||
ctx_http.stop();
|
||||
ctx_http.join();
|
||||
clean_up();
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (opts.install_signal_handlers) {
|
||||
server_install_signal_handlers();
|
||||
}
|
||||
|
||||
server_connect connect_proc;
|
||||
if (!server_on_listening(ctx_http, params, connect_proc)) {
|
||||
ctx_http.stop();
|
||||
ctx_http.join();
|
||||
clean_up();
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (!params.models_preset_hf.empty()) {
|
||||
SRV_WRN( "NOTE: using preset.ini from HF repo '%s'\n", params.models_preset_hf.c_str());
|
||||
SRV_WRN("%s", " please only use presets that you can trust! Unknown presets may be unsafe\n");
|
||||
}
|
||||
|
||||
ctx_http.join(); // keep the main thread alive
|
||||
|
||||
// when the HTTP server stops, clean up and exit
|
||||
clean_up();
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
//
|
||||
// instance
|
||||
//
|
||||
|
||||
static int server_run_instance(common_params & params, server_role & role, common_models_handler & models_handler, const server_run_opts & opts) {
|
||||
// -hf / docker: download if needed and resolve the local path. a child resolves what the router downloaded
|
||||
if (opts.resolve_model) {
|
||||
try {
|
||||
common_models_handler_apply(models_handler, params);
|
||||
} catch (const std::exception & e) {
|
||||
@@ -435,161 +581,188 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
|
||||
}
|
||||
}
|
||||
|
||||
//
|
||||
// Start the server
|
||||
//
|
||||
// note: this is guaranteed to out-live ctx_http and tools
|
||||
server_mcp mcp_mgr;
|
||||
|
||||
std::function<void()> clean_up;
|
||||
// struct that contains llama context and inference
|
||||
server_context ctx_server;
|
||||
server_routes routes(params, ctx_server);
|
||||
server_tools tools;
|
||||
|
||||
if (is_router_server) {
|
||||
SRV_INF("%s", "starting server in router mode. models will be automatically loaded on-demand\n");
|
||||
|
||||
clean_up = [&models_routes, &mcp_mgr]() {
|
||||
SRV_INF("%s: cleaning up before exit...\n", __func__);
|
||||
// stop the session GC first, it finalizes live sessions and wakes pending readers
|
||||
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();
|
||||
}
|
||||
mcp_mgr.shutdown();
|
||||
llama_backend_free();
|
||||
};
|
||||
|
||||
if (!ctx_http.start()) {
|
||||
clean_up();
|
||||
SRV_ERR("%s", "exiting due to HTTP server error\n");
|
||||
return 1;
|
||||
}
|
||||
ctx_http.is_ready.store(true);
|
||||
|
||||
shutdown_handler = [&](int) {
|
||||
if (models_routes.has_value()) {
|
||||
// important to disconnect any SSE clients
|
||||
models_routes->stopping.store(true);
|
||||
}
|
||||
mcp_mgr.shutdown();
|
||||
ctx_http.stop();
|
||||
};
|
||||
|
||||
try {
|
||||
models_routes->models.load_startup_models();
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("failed to load models on startup: %s\n", e.what());
|
||||
ctx_http.stop();
|
||||
ctx_http.join();
|
||||
clean_up();
|
||||
return 1;
|
||||
}
|
||||
|
||||
} else {
|
||||
// setup clean up function, to be called before exit
|
||||
clean_up = [&ctx_http, &ctx_server, &mcp_mgr]() {
|
||||
SRV_INF("%s: cleaning up before exit...\n", __func__);
|
||||
// stop the session GC first, it finalizes live sessions and wakes pending readers
|
||||
server_stream_session_manager_stop();
|
||||
ctx_http.stop();
|
||||
ctx_server.terminate();
|
||||
mcp_mgr.shutdown();
|
||||
llama_backend_free();
|
||||
};
|
||||
|
||||
// start the HTTP server before loading the model to be able to serve /health requests
|
||||
if (!ctx_http.start()) {
|
||||
clean_up();
|
||||
SRV_ERR("%s", "exiting due to HTTP server error\n");
|
||||
return 1;
|
||||
}
|
||||
|
||||
// setup communication child --> router if necessary
|
||||
if (child.is_child()) {
|
||||
ctx_server.set_state_callback([&](server_state state, json payload) {
|
||||
child.notify_to_router(server_state_to_str(state), payload);
|
||||
});
|
||||
}
|
||||
|
||||
if (!ctx_server.load_model(params)) {
|
||||
clean_up();
|
||||
ctx_http.join();
|
||||
SRV_ERR("%s", "exiting due to model loading error\n");
|
||||
return 1;
|
||||
}
|
||||
|
||||
routes.update_meta(ctx_server);
|
||||
ctx_http.is_ready.store(true);
|
||||
|
||||
SRV_INF("%s", "model loaded\n");
|
||||
|
||||
shutdown_handler = [&](int) {
|
||||
mcp_mgr.shutdown();
|
||||
// this will unblock start_loop()
|
||||
ctx_server.terminate();
|
||||
};
|
||||
server_http_context ctx_http;
|
||||
if (!ctx_http.init(params)) {
|
||||
SRV_ERR("%s", "failed to initialize HTTP server\n");
|
||||
return 1;
|
||||
}
|
||||
|
||||
// register signal handler if not running by CLI
|
||||
if (!is_run_by_cli) {
|
||||
#if defined (__unix__) || (defined (__APPLE__) && defined (__MACH__))
|
||||
struct sigaction sigint_action;
|
||||
sigint_action.sa_handler = signal_handler;
|
||||
sigemptyset (&sigint_action.sa_mask);
|
||||
sigint_action.sa_flags = 0;
|
||||
sigaction(SIGINT, &sigint_action, NULL);
|
||||
sigaction(SIGTERM, &sigint_action, NULL);
|
||||
#elif defined (_WIN32)
|
||||
auto console_ctrl_handler = +[](DWORD ctrl_type) -> BOOL {
|
||||
return (ctrl_type == CTRL_C_EVENT) ? (signal_handler(SIGINT), true) : false;
|
||||
};
|
||||
SetConsoleCtrlHandler(reinterpret_cast<PHANDLER_ROUTINE>(console_ctrl_handler), true);
|
||||
#endif
|
||||
server_register_routes(ctx_http, routes);
|
||||
server_register_stream_routes(ctx_http,
|
||||
server_stream_make_get_handler(),
|
||||
server_stream_make_lookup_handler(),
|
||||
server_stream_make_delete_handler());
|
||||
|
||||
if (!server_setup_features(ctx_http, params, mcp_mgr, tools, SERVER_ROLE_INSTANCE)) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
for (const auto & address : ctx_http.listening_addresses) {
|
||||
SRV_INF("listening on %s\n", address.c_str());
|
||||
// setup clean up function, to be called before exit
|
||||
auto clean_up = [&ctx_http, &ctx_server, &mcp_mgr]() {
|
||||
SRV_INF("%s: cleaning up before exit...\n", __func__);
|
||||
// stop the session GC first, it finalizes live sessions and wakes pending readers
|
||||
server_stream_session_manager_stop();
|
||||
ctx_http.stop();
|
||||
ctx_server.terminate();
|
||||
mcp_mgr.shutdown();
|
||||
llama_backend_free();
|
||||
};
|
||||
|
||||
// start the HTTP server before loading the model to be able to serve /health requests
|
||||
if (!ctx_http.start()) {
|
||||
clean_up();
|
||||
SRV_ERR("%s", "exiting due to HTTP server error\n");
|
||||
return 1;
|
||||
}
|
||||
|
||||
// setup communication child --> router if necessary
|
||||
if (role.is_managed_by_router()) {
|
||||
ctx_server.set_state_callback([&](server_state state, json payload) {
|
||||
role.notify_to_router(server_state_to_str(state), payload);
|
||||
});
|
||||
}
|
||||
|
||||
if (!ctx_server.load_model(params)) {
|
||||
clean_up();
|
||||
ctx_http.join();
|
||||
SRV_ERR("%s", "exiting due to model loading error\n");
|
||||
return 1;
|
||||
}
|
||||
|
||||
routes.update_meta(ctx_server);
|
||||
ctx_http.is_ready.store(true);
|
||||
|
||||
SRV_INF("%s", "model loaded\n");
|
||||
|
||||
shutdown_handler = [&](int) {
|
||||
mcp_mgr.shutdown();
|
||||
// this will unblock start_loop()
|
||||
ctx_server.terminate();
|
||||
};
|
||||
|
||||
if (opts.install_signal_handlers) {
|
||||
server_install_signal_handlers();
|
||||
}
|
||||
|
||||
// spawn only once listening, so the port is bound and the child startup health check reaches us. the destructor stops it
|
||||
server_connect connect_proc;
|
||||
if (params.server_connect && !connect_proc.start(params, ctx_http.port)) {
|
||||
SRV_ERR("%s", "exiting due to llama-connect error\n");
|
||||
if (!server_on_listening(ctx_http, params, connect_proc)) {
|
||||
ctx_http.stop();
|
||||
ctx_http.join();
|
||||
clean_up();
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (is_router_server) {
|
||||
if (!params.models_preset_hf.empty()) {
|
||||
SRV_WRN( "NOTE: using preset.ini from HF repo '%s'\n", params.models_preset_hf.c_str());
|
||||
SRV_WRN("%s", " please only use presets that you can trust! Unknown presets may be unsafe\n");
|
||||
}
|
||||
// optionally, notify router server that this instance is ready
|
||||
std::thread monitor_thread;
|
||||
if (role.is_managed_by_router()) {
|
||||
monitor_thread = role.setup(shutdown_handler);
|
||||
role.notify_to_router(server_state_to_str(SERVER_STATE_READY), routes.get_model_info());
|
||||
}
|
||||
|
||||
ctx_http.join(); // keep the main thread alive
|
||||
// this call blocks the main thread until queue_tasks.terminate() is called
|
||||
ctx_server.start_loop();
|
||||
|
||||
// when the HTTP server stops, clean up and exit
|
||||
clean_up();
|
||||
} else {
|
||||
// optionally, notify router server that this instance is ready
|
||||
std::thread monitor_thread;
|
||||
if (child.is_child()) {
|
||||
monitor_thread = child.setup(shutdown_handler);
|
||||
child.notify_to_router(server_state_to_str(SERVER_STATE_READY), routes.get_model_info());
|
||||
}
|
||||
clean_up();
|
||||
ctx_http.join();
|
||||
if (monitor_thread.joinable()) {
|
||||
monitor_thread.join();
|
||||
}
|
||||
|
||||
// this call blocks the main thread until queue_tasks.terminate() is called
|
||||
ctx_server.start_loop();
|
||||
|
||||
clean_up();
|
||||
ctx_http.join();
|
||||
if (monitor_thread.joinable()) {
|
||||
monitor_thread.join();
|
||||
}
|
||||
|
||||
auto * ll_ctx = ctx_server.get_llama_context();
|
||||
if (ll_ctx != nullptr) {
|
||||
common_memory_breakdown_print(ll_ctx);
|
||||
}
|
||||
auto * ll_ctx = ctx_server.get_llama_context();
|
||||
if (ll_ctx != nullptr) {
|
||||
common_memory_breakdown_print(ll_ctx);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
//
|
||||
// entry points
|
||||
//
|
||||
|
||||
static int server_run(common_params & params, int argc, char ** argv, server_role & role, const server_run_opts & opts) {
|
||||
common_models_handler models_handler;
|
||||
|
||||
// -hf may name a preset repo, which makes this a router: settle that before deciding the role
|
||||
if (opts.resolve_model && !params.model.hf_repo.empty()) {
|
||||
try {
|
||||
models_handler = common_models_handler_init(params, LLAMA_EXAMPLE_SERVER);
|
||||
if (common_models_handler_is_preset_repo(models_handler)) {
|
||||
// apply the preset and start the server in router mode
|
||||
common_models_handler_apply(models_handler, params);
|
||||
}
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("failed to fetch model metadata: %s\n", e.what());
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
const server_role_type role_type = server_role_detect(params, role);
|
||||
|
||||
// only an instance loads a model; the others must not touch the GPU, so skip the
|
||||
// device enumeration for them and the CUDA primary context stays uncreated
|
||||
common_params_print_info(params, role_type == SERVER_ROLE_INSTANCE);
|
||||
|
||||
if (!server_prepare_params(params, role_type)) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
switch (role_type) {
|
||||
case SERVER_ROLE_DOWNLOAD: return server_run_download(params, role);
|
||||
case SERVER_ROLE_ROUTER: return server_run_router(params, argc, argv, opts);
|
||||
case SERVER_ROLE_INSTANCE: return server_run_instance(params, role, models_handler, opts);
|
||||
}
|
||||
|
||||
GGML_ABORT("invalid server role");
|
||||
}
|
||||
|
||||
int llama_server(int argc, char ** argv) {
|
||||
server_role role;
|
||||
|
||||
std::setlocale(LC_NUMERIC, "C");
|
||||
|
||||
#ifndef _WIN32
|
||||
// Ignore SIGPIPE so the server does not crash if a child (MCP server, tools runtime) exits while we are writing to its stdin
|
||||
signal(SIGPIPE, SIG_IGN);
|
||||
#endif
|
||||
|
||||
// own arguments required by this example
|
||||
common_params params;
|
||||
|
||||
common_init();
|
||||
|
||||
// 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
|
||||
server_stream_session_manager_start();
|
||||
|
||||
if (!common_params_parse(argc, argv, params, LLAMA_EXAMPLE_SERVER)) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
SRV_INF("%s", "initializing ...\n");
|
||||
|
||||
llama_backend_init();
|
||||
llama_numa_init(params.numa);
|
||||
|
||||
const int result = server_run(params, argc, argv, role, server_run_opts{});
|
||||
common_log_flush(common_log_main());
|
||||
return result;
|
||||
}
|
||||
|
||||
int llama_server(common_params & params, int argc, char ** argv) {
|
||||
// embedded in llama-cli: it handles the signals, and the model is resolved already
|
||||
server_run_opts opts;
|
||||
opts.install_signal_handlers = false;
|
||||
opts.resolve_model = false;
|
||||
|
||||
server_role role;
|
||||
return server_run(params, argc, argv, role, opts);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user