From 82d6bb284d1ff1c6ef37f29a4c3b63d1a8b11806 Mon Sep 17 00:00:00 2001 From: Xuan-Son Nguyen Date: Sat, 12 Sep 2026 00:53:07 +0200 Subject: [PATCH] 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 --- tools/server/server-common.cpp | 145 ++++++++++ tools/server/server-common.h | 38 +++ tools/server/server-models.cpp | 471 +++++++++++++++++++-------------- tools/server/server-models.h | 26 +- 4 files changed, 473 insertions(+), 207 deletions(-) diff --git a/tools/server/server-common.cpp b/tools/server/server-common.cpp index 2ac98b6fdd..eade7db212 100644 --- a/tools/server/server-common.cpp +++ b/tools/server/server-common.cpp @@ -16,6 +16,21 @@ #include #include +#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 +#include +#else +#include +#include +#include +#include +#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 & procs, std::vector & 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(timeout_ms, 50); + std::this_thread::sleep_for(std::chrono::milliseconds(step)); + } +#else + std::vector 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(timeout_ms, std::numeric_limits::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 6c681a2cf5..9894f5f06f 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 #include #include +#include #include #include #include @@ -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 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 & procs, std::vector & 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 4d2592b259..f1783c0830 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 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 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 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 lk(mu); + cmds.push_back(std::move(cmd)); + } + waiter.wake(); + } + + // returns true if the loop should exit + bool handle_commands() { + std::deque batch; + { + std::lock_guard 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(timeout, 50); + } + if (c.deadline) { + int64_t d = std::max(0, c.deadline - now); + timeout = timeout < 0 ? d : std::min(timeout, d); + } + } + std::vector procs; + std::vector owners; + for (auto & c : children) { + if (!c.eof) { + procs.push_back(c.proc.get()); + owners.push_back(&c); + } + } + std::vector 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 cmds; + std::vector 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(*this)) { + sched(std::make_unique(*this)), + monitor(std::make_unique(*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(), - /* 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 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 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 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 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 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 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 & proc, server_child_mode mode, int exit_code) { + { + std::lock_guard 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 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 to_join; - { - std::lock_guard 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 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 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 7f6c26b358..90161bf34a 100644 --- a/tools/server/server-models.h +++ b/tools/server/server-models.h @@ -9,6 +9,7 @@ #include #include +#include #include #include #include @@ -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 subproc; // shared between main thread and monitoring thread - std::thread th; + std::shared_ptr 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 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 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 & 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 monitor; }; struct server_child {