Commit 82d6bb284 for llama.cpp
commit 82d6bb284d1ff1c6ef37f29a4c3b63d1a8b11806
Author: Xuan-Son Nguyen <son@huggingface.co>
Date: Sat Sep 12 00:53:07 2026 +0200
server: refactor subproc handling (#28555)
* 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 917b83f149c625527f872fb2cf41289358fa5371.
* 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 c368a4a98c677938ca87002edb6186ff2c02fd83.
---------
Co-authored-by: Pascal <admin@serveurperso.com>
diff --git a/tools/server/server-common.cpp b/tools/server/server-common.cpp
index 2ac98b6fd..eade7db21 100644
--- a/tools/server/server-common.cpp
+++ b/tools/server/server-common.cpp
@@ -16,6 +16,21 @@
#include <cstring>
#include <type_traits>
+#ifdef _WIN32
+// windows.h defines min and max as macros, which breaks std::min and std::max
+#define WIN32_LEAN_AND_MEAN
+#ifndef NOMINMAX
+# define NOMINMAX
+#endif
+#include <windows.h>
+#include <io.h>
+#else
+#include <errno.h>
+#include <fcntl.h>
+#include <poll.h>
+#include <unistd.h>
+#endif
+
json format_error_response(const std::string & message, const enum error_type type) {
std::string type_str;
int code = 500;
@@ -1832,3 +1847,133 @@ server_tokens format_prompt_rerank(
return result;
}
+
+//
+// server_subproc
+//
+
+bool server_subproc::has_output() {
+ if (out_handle >= 0) {
+ return true;
+ }
+ FILE * f = sproc.stdout_file(); // combined stdout/stderr
+ if (!f) {
+ return false;
+ }
+#ifdef _WIN32
+ HANDLE h = (HANDLE) _get_osfhandle(_fileno(f));
+ if (h != INVALID_HANDLE_VALUE) {
+ out_handle = (intptr_t) h;
+ }
+#else
+ int fd = fileno(f);
+ if (fd >= 0) {
+ fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
+ out_handle = fd;
+ }
+#endif
+ return out_handle >= 0;
+}
+
+int server_subproc::read_output(char * buf, size_t len) {
+ if (!has_output()) {
+ return -1;
+ }
+#ifdef _WIN32
+ HANDLE h = (HANDLE) out_handle;
+ DWORD avail = 0;
+ if (!PeekNamedPipe(h, NULL, 0, NULL, &avail, NULL)) {
+ return -1; // pipe broken, child gone
+ }
+ if (avail == 0) {
+ return 0;
+ }
+ DWORD to_read = avail < (DWORD) len ? avail : (DWORD) len;
+ DWORD got = 0;
+ if (!ReadFile(h, buf, to_read, &got, NULL) || got == 0) {
+ return -1;
+ }
+ return (int) got;
+#else
+ while (true) {
+ ssize_t r = read((int) out_handle, buf, len);
+ if (r > 0) {
+ return (int) r;
+ }
+ if (r == 0) {
+ return -1; // EOF
+ }
+ if (errno == EINTR) {
+ continue;
+ }
+ if (errno == EAGAIN || errno == EWOULDBLOCK) {
+ return 0;
+ }
+ return -1;
+ }
+#endif
+}
+
+server_subproc::waiter::waiter() {
+#ifndef _WIN32
+ int fds[2];
+ GGML_ASSERT(pipe(fds) == 0);
+ for (int fd : fds) {
+ fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
+ }
+ wake_fd[0] = fds[0];
+ wake_fd[1] = fds[1];
+#endif
+}
+
+server_subproc::waiter::~waiter() {
+#ifndef _WIN32
+ close((int) wake_fd[0]);
+ close((int) wake_fd[1]);
+#endif
+}
+
+void server_subproc::waiter::wake() {
+#ifndef _WIN32
+ char c = 1;
+ (void) !write((int) wake_fd[1], &c, 1);
+#endif
+}
+
+void server_subproc::waiter::wait(const std::vector<server_subproc *> & procs, std::vector<bool> & ready, int64_t timeout_ms) {
+ ready.assign(procs.size(), false);
+#ifdef _WIN32
+ // no waitable wait exists for anonymous pipes, so poll them in 50 ms steps
+ bool any = false;
+ for (size_t i = 0; i < procs.size(); i++) {
+ DWORD avail = 0;
+ if (!procs[i]->has_output() || !PeekNamedPipe((HANDLE) procs[i]->out_handle, NULL, 0, NULL, &avail, NULL) || avail > 0) {
+ ready[i] = true; // data or broken pipe, read_output() tells which
+ any = true;
+ }
+ }
+ if (!any) {
+ int64_t step = timeout_ms < 0 ? 50 : std::min<int64_t>(timeout_ms, 50);
+ std::this_thread::sleep_for(std::chrono::milliseconds(step));
+ }
+#else
+ std::vector<pollfd> pfds;
+ pfds.reserve(procs.size() + 1);
+ pfds.push_back({ (int) wake_fd[0], POLLIN, 0 });
+ for (auto * p : procs) {
+ pfds.push_back({ p->has_output() ? (int) p->out_handle : -1, POLLIN, 0 }); // poll() skips negative fds
+ }
+ int timeout = timeout_ms < 0 ? -1 : (int) std::min<int64_t>(timeout_ms, std::numeric_limits<int>::max());
+ int r = poll(pfds.data(), pfds.size(), timeout);
+ if (r < 0 && errno != EINTR) {
+ LOG_ERR("%s: poll() failed: %s\n", __func__, strerror(errno));
+ }
+ if (pfds[0].revents) {
+ char buf[64];
+ while (read((int) wake_fd[0], buf, sizeof(buf)) > 0) {}
+ }
+ for (size_t i = 0; i < procs.size(); i++) {
+ ready[i] = pfds[i + 1].fd < 0 || pfds[i + 1].revents != 0;
+ }
+#endif
+}
diff --git a/tools/server/server-common.h b/tools/server/server-common.h
index 6c681a2cf..9894f5f06 100644
--- a/tools/server/server-common.h
+++ b/tools/server/server-common.h
@@ -6,6 +6,7 @@
#include "chat.h"
#include "mtmd.h"
#include "mtmd-helper.h"
+#include "subproc.h"
#include "json.h"
@@ -13,6 +14,7 @@
#include <chrono>
#include <condition_variable>
#include <cinttypes>
+#include <cstdio>
#include <functional>
#include <mutex>
#include <queue>
@@ -611,3 +613,39 @@ struct server_pipe {
return true;
}
};
+
+// wrapper around common_subproc to manage a child server process
+// mainly used by router mode
+struct server_subproc {
+ common_subproc sproc;
+ std::atomic<bool> stopped{false}; // set by the monitor once the process exited and was reaped
+
+ bool is_alive() { return sproc.alive(); }
+ void terminate() { sproc.terminate(); }
+ int join() { return sproc.join(); }
+
+ // true if the child's combined stdout/stderr pipe is available (call after create())
+ bool has_output();
+
+ // non-blocking read
+ // returns the number of bytes read, 0 when nothing is available, -1 when the pipe is closed or broken
+ int read_output(char * buf, size_t len);
+
+ // wait until one of a set of children has output, wake() is called, or a timeout passes
+ struct waiter {
+ waiter();
+ ~waiter();
+
+ // thread-safe; on Windows this is a no-op, wait() returns within 50 ms anyway
+ void wake();
+
+ // timeout_ms < 0 waits until data or wake(); ready[i] is set for each child with data (or a broken pipe)
+ void wait(const std::vector<server_subproc *> & procs, std::vector<bool> & ready, int64_t timeout_ms);
+
+ private:
+ intptr_t wake_fd[2] = { -1, -1 }; // POSIX self-pipe
+ };
+
+private:
+ intptr_t out_handle = -1; // fd on POSIX, HANDLE on Windows; taken lazily from sproc
+};
diff --git a/tools/server/server-models.cpp b/tools/server/server-models.cpp
index 4d2592b25..f1783c083 100644
--- a/tools/server/server-models.cpp
+++ b/tools/server/server-models.cpp
@@ -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
+// single-threaded, watching all child processes at once
+struct server_monitor {
+ server_monitor(server_models & models) : models(models) {
+ th = std::thread([this]() { run(); });
+ }
+
+ ~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;
+ }
+ push({ cmd_t::WATCH, std::move(c), "", 0, false });
+ }
+
+ // 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;
+ }
- bool is_alive() {
- return sproc.alive();
+ // 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 request_exit() {
- FILE * stdin_file = sproc.stdin_file();
- if (stdin_file) {
- fprintf(stdin_file, "%s\n", CMD_ROUTER_TO_CHILD_EXIT);
- fflush(stdin_file);
+ 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
}
- stopped.store(true, std::memory_order_relaxed);
}
- void terminate() {
- sproc.terminate();
+ 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") {
diff --git a/tools/server/server-models.h b/tools/server/server-models.h
index 7f6c26b35..90161bf34 100644
--- a/tools/server/server-models.h
+++ b/tools/server/server-models.h
@@ -9,6 +9,7 @@
#include <mutex>
#include <condition_variable>
+#include <thread>
#include <functional>
#include <memory>
#include <optional>
@@ -107,27 +108,29 @@ struct server_model_meta {
};
struct server_models_routes;
-struct server_subproc; // defined in server-models.cpp
struct server_lru_sched; // defined in server-models.cpp
+struct server_monitor; // defined in server-models.cpp
struct server_models {
friend struct server_models_routes;
friend struct server_lru_sched;
+ friend struct server_monitor;
private:
struct instance_t {
- std::shared_ptr<server_subproc> subproc; // shared between main thread and monitoring thread
- std::thread th;
+ std::shared_ptr<server_subproc> subproc; // shared with the monitor thread
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)
+ void request_exit() const;
};
std::mutex mutex;
std::condition_variable cv;
std::map<std::string, instance_t> mapping;
- // for stopping models
- std::condition_variable cv_stop;
+ // models asked to stop, still counted as running until the monitor records their exit
std::set<std::string> stopping_models;
// set to true while load_models() is executing a reload; load() will wait until clear
@@ -216,9 +219,12 @@ private:
// not thread-safe, caller must hold mutex
void add_model(server_model_meta && meta);
- // ask the monitoring thread to stop a running instance
+ // ask the monitor to stop a running instance; send_exit is false for a child that was already force-killed
// not thread-safe, caller must hold mutex
- void request_stop(const std::string & name);
+ 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);
// notify SSE clients
void notify_sse(const std::string & event, const std::string & model_id, const json & data = nullptr);
@@ -297,12 +303,16 @@ public:
// handle message sent from server_child::notify_to_router()
// raw input must starts with CMD_CHILD_TO_ROUTER_STATE, followed by a JSON string
- // this function is not thread-safe, must be called from instance's monitoring thread
+ // called from the monitor thread
// payload per state:
// state = loading -> payload = {} (TODO: add progress info)
// state = ready -> payload = model_info (json), or {} if wakeup from sleeping
// state = sleeping -> payload = {}
void handle_child_state(const std::string & name, const std::string & raw_input);
+
+private:
+ // one thread watching every child; keep last, the destructor joins the thread
+ std::unique_ptr<server_monitor> monitor;
};
struct server_child {