#include "model_manager.h" #include #include #include #include "http_helpers.h" #include #include #include #include #include namespace asio = boost::asio; namespace beast = boost::beast; namespace http = beast::http; bool is_alive(pid_t pid) { if (pid <= 0) return false; // A worker that exited on its own is a zombie until reaped, and a zombie // still passes kill(pid, 0). Reap it here so crashes are detected instead // of reporting a dead worker as ready. int status = 0; const pid_t r = waitpid(pid, &status, WNOHANG); if (r == pid) return false; // exited; now reaped if (r == 0) return true; // still running // r < 0: not our child or already reaped — fall back to a liveness probe. return kill(pid, 0) == 0; } void shutdown_worker(pid_t pid, int wait_seconds) { if (pid <= 0) return; if (kill(pid, SIGTERM) != 0 && errno == ESRCH) { waitpid(pid, nullptr, WNOHANG); // reap in case it is a zombie return; } const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(wait_seconds); while (std::chrono::steady_clock::now() < deadline) { int status = 0; pid_t r = waitpid(pid, &status, WNOHANG); if (r == pid) return; if (r < 0 && errno == ECHILD) return; // already reaped elsewhere std::this_thread::sleep_for(std::chrono::milliseconds(200)); } kill(pid, SIGKILL); int status = 0; waitpid(pid, &status, 0); } ModelManager::ModelManager(const ManagerConfig &config) : default_model_(config.worker.default_model), llama_server_bin_(config.worker.llama_server_bin), worker_host_(config.worker.host), worker_bind_host_(config.worker.bind_host), base_port_(config.worker.base_port), switch_timeout_sec_(config.worker.switch_timeout_sec), n_ctx_(config.llama.n_ctx), n_threads_(config.llama.threads), n_gpu_layers_(config.llama.ngl), n_batch_(config.llama.batch), n_ubatch_(config.llama.ubatch), next_port_(base_port_) {} bool ModelManager::initialize_default(std::string &error) { return switch_model(default_model_, error); } bool ModelManager::switch_model(const std::string &model, std::string &error) { { std::lock_guard lock(mu_); if (switch_in_progress_) { error = "Switch already in progress"; return false; } if (active_ && active_->model == model && is_alive(active_->pid)) { return true; } switch_in_progress_ = true; } std::optional old_worker; { std::lock_guard lock(mu_); if (active_) old_worker = active_; } int port = allocate_port(); pid_t pid = spawn_worker(model, port); if (pid <= 0) { finish_switch(false); error = "Failed to start worker process"; return false; } if (!wait_until_ready(pid, port, switch_timeout_sec_)) { shutdown_worker(pid); finish_switch(false); error = "New model did not become ready in time"; return false; } WorkerInfo new_worker{model, port, pid, now_utc_iso()}; { std::lock_guard lock(mu_); active_ = new_worker; switch_in_progress_ = false; } if (old_worker && old_worker->pid != pid) { shutdown_worker(old_worker->pid); } return true; } bool ModelManager::restart_active(std::string &error) { std::optional old_worker; std::string model; { std::lock_guard lock(mu_); if (switch_in_progress_) { error = "Switch already in progress"; return false; } if (!active_ || !is_alive(active_->pid)) { error = "No active model"; return false; } switch_in_progress_ = true; old_worker = active_; model = active_->model; } shutdown_worker(old_worker->pid); int port = allocate_port(); pid_t pid = spawn_worker(model, port); if (pid <= 0) { std::lock_guard lock(mu_); active_ = std::nullopt; switch_in_progress_ = false; error = "Failed to start worker process"; return false; } if (!wait_until_ready(pid, port, switch_timeout_sec_)) { shutdown_worker(pid); std::lock_guard lock(mu_); active_ = std::nullopt; switch_in_progress_ = false; error = "New model did not become ready in time"; return false; } WorkerInfo new_worker{model, port, pid, now_utc_iso()}; { std::lock_guard lock(mu_); active_ = new_worker; switch_in_progress_ = false; } return true; } std::optional ModelManager::active_worker() { std::lock_guard lock(mu_); if (active_ && is_alive(active_->pid)) return active_; return std::nullopt; } json ModelManager::models_view() { std::lock_guard lock(mu_); json out; out["status"] = (active_ && is_alive(active_->pid)) ? "ready" : "no_active_model"; out["switch_in_progress"] = switch_in_progress_; if (active_ && is_alive(active_->pid)) { out["current_model"] = active_->model; out["last_loaded"] = active_->last_loaded; out["active_pid"] = active_->pid; out["active_port"] = active_->port; } else { out["current_model"] = nullptr; out["last_loaded"] = nullptr; out["active_pid"] = nullptr; out["active_port"] = nullptr; } return out; } int ModelManager::allocate_port() { std::lock_guard lock(mu_); // Cycle within a small window above base_port so restarts/switches don't // walk off into ports used by other services (or past 65535). Skip the // active worker's port: during a switch both processes run briefly. for (int i = 0; i < kPortWindow; ++i) { if (next_port_ >= base_port_ + kPortWindow) next_port_ = base_port_; const int port = next_port_++; if (active_ && active_->port == port) continue; return port; } return base_port_; } void ModelManager::finish_switch(bool ok) { std::lock_guard lock(mu_); if (!ok) switch_in_progress_ = false; } pid_t ModelManager::spawn_worker(const std::string &model, int port) { // Build argv before forking: other threads exist, so the child may only // call async-signal-safe functions between fork() and exec() — no heap // allocation (std::string/std::to_string malloc and can deadlock). std::vector args = { llama_server_bin_, "-hf", model, "--host", worker_bind_host_, "--port", std::to_string(port), "-c", std::to_string(n_ctx_), "-t", std::to_string(n_threads_), "-ngl", std::to_string(n_gpu_layers_), "--cont-batching", "-b", std::to_string(n_batch_), "--ubatch-size", std::to_string(n_ubatch_) }; std::vector argv; argv.reserve(args.size() + 1); for (auto &s : args) argv.push_back(const_cast(s.c_str())); argv.push_back(nullptr); pid_t pid = fork(); if (pid < 0) return -1; if (pid == 0) { setsid(); execvp(argv[0], argv.data()); _exit(127); } return pid; } bool ModelManager::wait_until_ready(pid_t pid, int port, int timeout_sec) { const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(timeout_sec); // Exponential backoff: fast detection when the worker comes up (or dies) // quickly, without hammering it during a minutes-long model load. auto poll_interval = std::chrono::milliseconds(100); const auto max_interval = std::chrono::milliseconds(800); while (std::chrono::steady_clock::now() < deadline) { if (!is_alive(pid)) return false; try { // /health is load-state aware (503 while the model loads); the // static index at / can answer 200 before the model is usable. auto [status, _] = http_get(port, "/health"); if (status == 200) return true; } catch (...) { } std::this_thread::sleep_for(poll_interval); poll_interval = std::min(poll_interval * 2, max_interval); } return false; } std::pair ModelManager::http_get(int port, const std::string &target) { asio::io_context ioc; asio::ip::tcp::resolver resolver(ioc); beast::tcp_stream stream(ioc); auto const results = resolver.resolve(worker_host_, std::to_string(port)); stream.connect(results); http::request req{http::verb::get, target, 11}; req.set(http::field::host, worker_host_); req.set(http::field::user_agent, "llm-manager"); http::write(stream, req); beast::flat_buffer buffer; http::response res; http::read(stream, buffer, res); beast::error_code ec; stream.socket().shutdown(asio::ip::tcp::socket::shutdown_both, ec); return {res.result_int(), res.body()}; }