wylab/llama.cpp
0
0
Fork 1
mirror of https://github.com/ggml-org/llama.cpp.git synced 2026-09-14 18:02:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

server: refactor subproc handling (#28555)

Browse Source 
* server: refactor subproc handling

* fix Windows build

* download: keep concurrent downloads of one blob apart

Every process writes the same path + .downloadInProgress, so a second
download of the same blob finds that file, takes it for its own partial
transfer and asks for the bytes after it, which produces a corrupt
result. The in-progress file now carries the pid of the process writing
it.

std::rename also replaces an existing destination on POSIX but fails on
Windows, so a download whose blob appeared in the meantime is dropped
after every retry and an etag rewrite silently keeps the old value.
std::filesystem::rename has the POSIX behaviour everywhere, and the
error now carries the reason reported by the system.

* Revert "download: keep concurrent downloads of one blob apart"

This reverts commit 917b83f149.

* tests: serialize the router tests that download the same model

Parallel workers share one cache, so the two tests fetch the same blob
into the same in-progress file and race to rename it. They now take a
file lock around the download, like the session fixture does for the
preset models.

* Revert "tests: serialize the router tests that download the same model"

This reverts commit c368a4a98c.

---------

Co-authored-by: Pascal <admin@serveurperso.com>
This commit is contained in:
Xuan-Son Nguyen
2026-09-12 00:53:07 +02:00
committed by GitHub
co-authored by Pascal
parent 8ea290247c
commit 82d6bb284d
4 changed files with 473 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/server/server-models.cpp
+272 -199
View File
@@ -44,30 +44,215 @@ extern char **environ;
#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();
if (stdin_file) {
fprintf(stdin_file, "%s\n", CMD_ROUTER_TO_CHILD_EXIT);
fflush(stdin_file);
}
}
// 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"
struct server_subproc {
common_subproc sproc; // not yet spawned while in DOWNLOADING state
std::atomic<bool> stopped{false}; // set to cancel a download or signal child process exit
bool is_alive() {
return sproc.alive();
// single-threaded, watching all child processes at once
struct server_monitor {
server_monitor(server_models & models) : models(models) {
th = std::thread([this]() { run(); });
}
void request_exit() {
FILE * stdin_file = sproc.stdin_file();
if (stdin_file) {
fprintf(stdin_file, "%s\n", CMD_ROUTER_TO_CHILD_EXIT);
fflush(stdin_file);
~server_monitor() {
push({ cmd_t::QUIT, {}, "", 0, false });
th.join();
}
// thread-safe
void watch(const std::string & name, std::shared_ptr<server_subproc> proc, server_child_mode mode, int port) {
child_t c;
c.name = name;
c.proc = std::move(proc);
c.mode = mode;
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());
c.eof = true;
}
stopped.store(true, std::memory_order_relaxed);
push({ cmd_t::WATCH, std::move(c), "", 0, false });
}
void terminate() {
sproc.terminate();
// thread-safe
void stop(const std::string & name, int stop_timeout, bool send_exit) {
push({ cmd_t::STOP, {}, name, stop_timeout, send_exit });
}
private:
struct child_t {
std::string name;
std::shared_ptr<server_subproc> proc;
server_child_mode mode = SERVER_CHILD_MODE_NORMAL;
int port = 0;
std::string buf; // partial line
bool eof = false; // output closed, waiting for the process to be reaped
int64_t deadline = 0; // force-kill time in ms, 0 when no stop is pending
};
struct cmd_t {
enum { WATCH, STOP, QUIT } type;
child_t child;
std::string name;
int stop_timeout;
bool send_exit;
};
void push(cmd_t && cmd) {
{
std::lock_guard<std::mutex> lk(mu);
cmds.push_back(std::move(cmd));
}
waiter.wake();
}
// returns true if the loop should exit
bool handle_commands() {
std::deque<cmd_t> batch;
{
std::lock_guard<std::mutex> lk(mu);
batch.swap(cmds);
}
for (auto & cmd : batch) {
switch (cmd.type) {
case cmd_t::WATCH:
children.push_back(std::move(cmd.child));
break;
case cmd_t::STOP:
// the newest child with this name is the one the registry knows
for (auto it = children.rbegin(); it != children.rend(); ++it) {
if (it->name != cmd.name) {
continue;
}
if (cmd.send_exit && !it->eof) {
request_child_exit(*it->proc);
}
it->deadline = ggml_time_ms() + (int64_t) cmd.stop_timeout * 1000;
break;
}
break;
case cmd_t::QUIT:
return true;
}
}
return false;
}
// read what the child wrote, forward complete lines
void read_output(child_t & c) {
char chunk[4096];
while (!c.eof) {
int n = c.proc->read_output(chunk, sizeof(chunk));
if (n < 0) {
c.eof = true;
break;
}
if (n == 0) {
break;
}
c.buf.append(chunk, (size_t) n);
size_t start = 0;
while (true) {
size_t nl = c.buf.find('\n', start);
if (nl == std::string::npos) {
break;
}
std::string line = c.buf.substr(start, nl + 1 - start);
start = nl + 1;
on_line(c, line);
}
c.buf.erase(0, start);
if (c.buf.size() > max_line) {
c.buf.clear(); // a child that never writes a newline must not grow this without bound
}
}
if (c.eof && !c.buf.empty()) {
on_line(c, c.buf);
c.buf.clear();
}
}
void on_line(child_t & c, const std::string & line) {
if (string_starts_with(line, CMD_CHILD_TO_ROUTER_STATE)) {
LOG_DBG("[%5d] %s", c.port, line.c_str()); // prevent spamming the log
models.handle_child_state(c.name, line);
} else {
LOG("[%5d] %s", c.port, line.c_str()); // forward log
}
}
void run() {
while (true) {
if (handle_commands()) {
return;
}
// wait for output, a wakeup, or the next deadline;
// a child whose output closed is polled for its exit every 50 ms
int64_t now = ggml_time_ms();
int64_t timeout = -1;
for (const auto & c : children) {
if (c.eof) {
timeout = timeout < 0 ? 50 : std::min<int64_t>(timeout, 50);
}
if (c.deadline) {
int64_t d = std::max<int64_t>(0, c.deadline - now);
timeout = timeout < 0 ? d : std::min(timeout, d);
}
}
std::vector<server_subproc *> procs;
std::vector<child_t *> owners;
for (auto & c : children) {
if (!c.eof) {
procs.push_back(c.proc.get());
owners.push_back(&c);
}
}
std::vector<bool> ready;
waiter.wait(procs, ready, timeout);
for (size_t i = 0; i < owners.size(); i++) {
if (ready[i]) {
read_output(*owners[i]);
}
}
// deadlines and exits
now = ggml_time_ms();
for (auto it = children.begin(); it != children.end();) {
if (it->deadline && now >= it->deadline && !it->proc->stopped.load(std::memory_order_acquire)) {
SRV_WRN("force-killing model instance name=%s after timeout\n", it->name.c_str());
it->proc->terminate();
it->deadline = 0;
}
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);
SRV_INF("instance name=%s exited with status %d\n", it->name.c_str(), exit_code);
it = children.erase(it);
} else {
++it;
}
}
}
}
static constexpr size_t max_line = 1024 * 1024;
server_models & models;
std::mutex mu;
std::deque<cmd_t> cmds;
std::vector<child_t> children; // monitor thread only
server_subproc::waiter waiter;
std::thread th;
};
struct server_lru_sched {
@@ -395,7 +580,8 @@ server_models::server_models(
base_params(params),
base_env(get_environment()),
base_preset(ctx_preset.load_from_args(argc, argv)),
sched(std::make_unique<server_lru_sched>(*this)) {
sched(std::make_unique<server_lru_sched>(*this)),
monitor(std::make_unique<server_monitor>(*this)) {
// clean up base preset
unset_reserved_args(base_preset, true);
// set binary path
@@ -412,6 +598,10 @@ server_models::server_models(
server_models::~server_models() = default;
void server_models::instance_t::request_exit() const {
request_child_exit(*subproc);
}
void server_models::add_model(server_model_meta && meta) {
if (mapping.find(meta.name) != mapping.end()) {
throw std::runtime_error(string_format("model '%s' appears multiple times", meta.name.c_str()));
@@ -466,7 +656,6 @@ void server_models::add_model(server_model_meta && meta) {
std::string name = meta.name;
mapping[name] = instance_t{
/* subproc */ std::make_shared<server_subproc>(),
/* th */ std::thread(),
/* meta */ std::move(meta)
};
}
@@ -621,9 +810,7 @@ void server_models::load_models() {
};
// Phase 2: acquire the lock once for all mapping mutations.
// We temporarily release it only when calling functions that acquire it internally
// (unload, load) or when joining threads (the monitoring thread calls update_status
// which locks the mutex, so joining while holding it would deadlock).
// We temporarily release it only when calling functions that acquire it internally (unload)
std::unique_lock<std::mutex> lk(mutex);
need_reload = false;
@@ -708,49 +895,15 @@ void server_models::load_models() {
return true;
});
// 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) {
auto it = mapping.find(name);
if (it != mapping.end() && it->second.th.joinable()) {
threads_to_join.push_back(std::move(it->second.th));
}
}
for (auto & [name, inst] : mapping) {
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));
}
}
// join outside the lock - monitoring thread calls update_status (needs lock)
lk.unlock();
for (auto & th : threads_to_join) th.join();
lk.lock();
// erase models no longer in any source
for (auto it = mapping.begin(); it != mapping.end(); ) {
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, thread is joined above, safe to erase
GGML_ASSERT(!it->second.th.joinable());
// download finished, safe to erase
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());
GGML_ASSERT(!it->second.th.joinable()); // must have been joined above
it = mapping.erase(it);
} else {
++it;
@@ -1030,117 +1183,12 @@ void server_models::load(const std::string & name, const load_options & opts) {
}
}
// start a thread to manage the child process
// captured variables are guaranteed to be destroyed only after the thread is joined
inst.th = std::thread([
this, name,
child_proc = inst.subproc,
port = inst.meta.port,
stop_timeout = inst.meta.stop_timeout,
child_mode = opts.mode
]() {
FILE * stdin_file = child_proc->sproc.stdin_file();
FILE * stdout_file = child_proc->sproc.stdout_file(); // combined stdout/stderr
std::thread log_thread([&]() {
// read stdout/stderr and forward to main server log
// also handle status report from child process
std::vector<char> vec_buf(128 * 1024); // large buffer for storing info
char * buffer = vec_buf.data();
if (stdout_file) {
while (fgets(buffer, vec_buf.size(), stdout_file) != nullptr) {
std::string str(buffer);
if (string_starts_with(buffer, CMD_CHILD_TO_ROUTER_STATE)) {
LOG_DBG("[%5d] %s", port, buffer); // prevent spamming the log
this->handle_child_state(name, str);
} else {
// forward log
LOG("[%5d] %s", port, buffer);
}
}
} else {
SRV_ERR("failed to get stdout/stderr of child process for name=%s\n", name.c_str());
}
});
std::thread stopping_thread([&]() {
// thread to monitor explicit stop requests; child crash is signalled via child_proc->stopped
auto is_stopping = [this, &name]() {
return this->stopping_models.find(name) != this->stopping_models.end();
};
{
std::unique_lock<std::mutex> lk(this->mutex);
this->cv_stop.wait(lk, [&]() {
return is_stopping() || child_proc->stopped.load(std::memory_order_acquire);
});
}
// child crashed or finished on its own, skip graceful shutdown sequence
if (child_proc->stopped.load(std::memory_order_acquire)) {
return;
}
SRV_INF("stopping model instance name=%s\n", name.c_str());
fprintf(stdin_file, "%s\n", CMD_ROUTER_TO_CHILD_EXIT);
fflush(stdin_file);
int64_t start_time = ggml_time_ms();
while (true) {
std::unique_lock<std::mutex> lk(this->mutex);
if (!is_stopping() || child_proc->stopped.load(std::memory_order_acquire)) {
return;
}
int64_t elapsed = ggml_time_ms() - start_time;
if (elapsed >= stop_timeout * 1000) {
lk.unlock();
SRV_WRN("force-killing model instance name=%s after %d seconds timeout\n", name.c_str(), stop_timeout);
child_proc->terminate();
return;
}
this->cv_stop.wait_for(lk, std::chrono::seconds(1), [&]() {
return !is_stopping() || child_proc->stopped.load(std::memory_order_acquire);
});
}
});
// we reach here when the child process exits (stdout EOF)
// note: we cannot join() prior to this point because it will close stdin_file
if (log_thread.joinable()) {
log_thread.join();
}
child_proc->stopped.store(true, std::memory_order_release);
{
std::lock_guard<std::mutex> lk(this->mutex);
stopping_models.erase(name);
cv_stop.notify_all();
}
if (stopping_thread.joinable()) {
stopping_thread.join();
}
// get the exit code
int exit_code = child_proc->sproc.join();
// update status and exit code
if (child_mode == SERVER_CHILD_MODE_DOWNLOAD) {
// instance will be cleaned up on next load_models() call
} else {
this->update_status(name, {
SERVER_MODEL_STATUS_UNLOADED,
exit_code
});
}
SRV_INF("instance name=%s exited with status %d\n", name.c_str(), exit_code);
});
// clean up old process/thread if exists
// old process should have exited already, but just in case, we clean it up here
{
auto & old_instance = mapping[name];
// old process should have exited already, but just in case, we clean it up here
if (old_instance.subproc && old_instance.subproc->is_alive()) {
auto it = mapping.find(name);
if (it != mapping.end() && it->second.subproc && it->second.subproc->is_alive()) {
SRV_WRN("old process for model name=%s is still alive, this is unexpected\n", name.c_str());
old_instance.subproc->terminate(); // force kill
}
if (old_instance.th.joinable()) {
old_instance.th.join();
it->second.subproc->terminate(); // force kill
}
}
@@ -1148,13 +1196,41 @@ void server_models::load(const std::string & name, const load_options & opts) {
{"status", server_model_status_to_string(inst.meta.status)},
});
auto proc = inst.subproc;
int port = inst.meta.port;
mapping[name] = std::move(inst);
monitor->watch(name, proc, opts.mode, port);
cv.notify_all();
}
void server_models::request_stop(const std::string & name) {
void server_models::request_stop(const std::string & name, bool send_exit) {
auto it = mapping.find(name);
if (it == mapping.end() || stopping_models.count(name)) {
return;
}
stopping_models.insert(name);
cv_stop.notify_all();
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) {
{
std::lock_guard<std::mutex> lk(mutex);
stopping_models.erase(name);
auto it = mapping.find(name);
if (it == mapping.end() || it->second.subproc != proc) {
return; // entry erased, or a newer instance took the name
}
}
if (mode == SERVER_CHILD_MODE_DOWNLOAD) {
// instance will be cleaned up on next load_models() call
std::lock_guard<std::mutex> lk(mutex);
cv.notify_all();
} else {
update_status(name, {
SERVER_MODEL_STATUS_UNLOADED,
exit_code
});
}
}
void server_models::unload(const std::string & name) {
@@ -1163,20 +1239,21 @@ void server_models::unload(const std::string & name) {
if (it != mapping.end()) {
if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
SRV_INF("cancelling download for model name=%s\n", name.c_str());
it->second.subproc->request_exit();
it->second.request_exit();
// for convenience, we wait the status change here
wait(lk, name, [](const server_model_meta & new_meta) {
return new_meta.status != SERVER_MODEL_STATUS_DOWNLOADING;
});
} else if (it->second.meta.is_running()) {
SRV_INF("stopping model instance name=%s\n", name.c_str());
if (it->second.meta.status == SERVER_MODEL_STATUS_LOADING) {
bool loading = it->second.meta.status == SERVER_MODEL_STATUS_LOADING;
if (loading) {
// special case: if model is in loading state, unloading means force-killing it
SRV_WRN("model name=%s is still loading, force-killing\n", name.c_str());
it->second.subproc->terminate();
}
request_stop(name);
// status change will be handled by the managing thread
request_stop(name, !loading);
// status change will be handled by the monitor
} else {
SRV_WRN("model instance name=%s is not running\n", name.c_str());
}
@@ -1184,27 +1261,29 @@ void server_models::unload(const std::string & name) {
}
void server_models::unload_all() {
std::vector<std::thread> to_join;
{
std::lock_guard<std::mutex> lk(mutex);
for (auto & [name, inst] : mapping) {
if (inst.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
SRV_INF("cancelling download for model name=%s\n", name.c_str());
inst.subproc->stopped.store(true, std::memory_order_relaxed);
} else if (inst.meta.is_running()) {
SRV_INF("stopping model instance name=%s\n", name.c_str());
request_stop(name);
// status change will be handled by the managing thread
std::unique_lock<std::mutex> lk(mutex);
for (auto & [name, inst] : mapping) {
if (inst.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
SRV_INF("cancelling download for model name=%s\n", name.c_str());
inst.request_exit();
} else if (inst.meta.is_running()) {
SRV_INF("stopping model instance name=%s\n", name.c_str());
bool loading = inst.meta.status == SERVER_MODEL_STATUS_LOADING;
if (loading) {
inst.subproc->terminate();
}
// moving the thread to join list to avoid deadlock
to_join.push_back(std::move(inst.th));
request_stop(name, !loading);
}
}
for (auto & th : to_join) {
if (th.joinable()) {
th.join();
// wait for every child to exit, the monitor force-kills the ones that ignore the exit command
cv.wait(lk, [this]() {
for (const auto & [name, inst] : mapping) {
if (inst.meta.is_running() || inst.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
return false;
}
}
}
return true;
});
}
void server_models::update_status(const std::string & name, const update_status_args & args) {
@@ -1291,18 +1370,18 @@ bool server_models::remove(const std::string & name) {
if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
// cancel in-flight download
SRV_INF("cancelling download for model name=%s\n", name.c_str());
it->second.subproc->request_exit();
it->second.request_exit();
} else if (it->second.meta.is_running()) {
// stop running instance
SRV_INF("stopping model instance name=%s\n", name.c_str());
stopping_models.insert(name);
if (it->second.meta.status == SERVER_MODEL_STATUS_LOADING) {
bool loading = it->second.meta.status == SERVER_MODEL_STATUS_LOADING;
if (loading) {
it->second.subproc->terminate();
}
cv_stop.notify_all();
request_stop(name, !loading);
}
// wait until the monitoring thread finishes
// wait until the child is gone
wait(lk, name, [](const server_model_meta & meta) {
return meta.status == SERVER_MODEL_STATUS_UNLOADED
|| meta.status == SERVER_MODEL_STATUS_DOWNLOADED;
@@ -1311,8 +1390,7 @@ bool server_models::remove(const std::string & name) {
// re-find after wait - load_models() may have erased the entry during the wait
it = mapping.find(name);
if (it == mapping.end()) {
// load_models() already joined the thread and erased the entry;
// we just need to clean up the cached files on disk
// load_models() already erased the entry; we just need to clean up the cached files on disk
lk.unlock();
bool ok = common_download_remove(name);
SRV_INF("removing model name=%s from cache (%s)\n", name.c_str(), ok ? "succeeded" : "partial");
@@ -1320,11 +1398,6 @@ bool server_models::remove(const std::string & name) {
return true;
}
// join before erasing - thread no longer acquires this mutex
if (it->second.th.joinable()) {
it->second.th.join();
}
// remove from disk (best-effort: cancelled downloads may have no cached files)
bool ok = common_download_remove(name);
mapping.erase(name);
@@ -1539,7 +1612,7 @@ void server_models::handle_child_state(const std::string & name, const std::stri
std::lock_guard<std::mutex> lk(mutex);
auto it = mapping.find(name);
if (it != mapping.end()) {
return it->second.subproc->request_exit();
return it->second.request_exit();
}
};
if (result == "download_finished") {