return sproc.has_value() && subprocess_alive(&sproc.value());
}
+ void request_exit() {
+ if (sproc.has_value()) {
+ FILE * stdin_file = subprocess_stdin(&sproc.value());
+ if (stdin_file) {
+ fprintf(stdin_file, "%s\n", CMD_ROUTER_TO_CHILD_EXIT);
+ fflush(stdin_file);
+ }
+ }
+ stopped.store(true, std::memory_order_relaxed);
+ }
+
void terminate() {
if (!sproc.has_value()) {
return;
}
void server_models::load_models() {
- // Phase 1: load presets from all sources — pure I/O, no lock needed
+ // Phase 1: load presets from all sources - pure I/O, no lock needed
// 1. cached models
common_presets cached_models = ctx_preset.load_from_cache();
SRV_INF("Loaded %zu cached model presets\n", cached_models.size());
return source_map.count(name) ? source_map.at(name) : SERVER_MODEL_SOURCE_PRESET;
};
- // Helpers that read `mapping` — must be called while holding the lock.
+ // Helpers that read `mapping` - must be called while holding the lock.
std::unordered_set<std::string> custom_names;
for (const auto & [name, preset] : custom_presets) custom_names.insert(name);
auto join_set = [](const std::set<std::string> & s) {
}
}
- // join outside the lock — monitoring thread calls update_status (needs lock)
+ // join outside the lock - monitoring thread calls update_status (needs lock)
lk.unlock();
for (auto & th : threads_to_join) th.join();
lk.lock();
apply_stop_timeout();
- // clear reload flag before unlocking for autoload — load() blocks on !is_reloading,
+ // clear reload flag before unlocking for autoload - load() blocks on !is_reloading,
// so clearing it here (while still locked) prevents a deadlock in the autoload calls below
is_reloading = false;
cv.notify_all();
}
void server_models::load(const std::string & name) {
- if (!has_model(name)) {
- throw std::runtime_error("model name=" + name + " is not found");
+ load(name, load_options{});
+}
+
+void server_models::load(const std::string & name, const load_options & opts) {
+ if (!opts.custom_meta.has_value()) {
+ if (!has_model(name)) {
+ throw std::runtime_error("model name=" + name + " is not found");
+ }
+ unload_lru();
}
- unload_lru();
std::unique_lock<std::mutex> lk(mutex);
// edge case: block until any in-progress reload has finished so we always load
// against the freshest preset and a consistent mapping state
cv.wait(lk, [this]() { return !is_reloading; });
- auto meta = mapping[name].meta;
+ auto meta = opts.custom_meta.has_value() ? *opts.custom_meta : mapping[name].meta;
if (meta.status != SERVER_MODEL_STATUS_UNLOADED) {
SRV_INF("model %s is not ready\n", name.c_str());
return;
std::vector<std::string> child_env = base_env; // copy
child_env.push_back("LLAMA_SERVER_ROUTER_PORT=" + std::to_string(base_params.port));
+ if (opts.mode == SERVER_CHILD_MODE_DOWNLOAD) {
+ inst.meta.status = SERVER_MODEL_STATUS_DOWNLOADING;
+ child_env.push_back("LLAMA_SERVER_CHILD_MODE=download");
+ child_env.push_back("LLAMA_ARG_HF_REPO=" + name);
+ }
+
SRV_INF("%s", "spawning server instance with args:\n");
for (const auto & arg : child_args) {
SRV_INF(" %s\n", arg.c_str());
if (result != 0) {
throw std::runtime_error("failed to spawn server instance");
}
-
- inst.stdin_file = subprocess_stdin(&inst.subproc->get());
}
// 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]() {
+ 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 = subprocess_stdin(&child_proc->get());
FILE * stdout_file = subprocess_stdout(&child_proc->get()); // combined stdout/stderr
return is_stopping() || child_proc->stopped.load(std::memory_order_acquire);
});
}
- // child crashed or finished on its own — skip graceful shutdown sequence
+ // child crashed or finished on its own, skip graceful shutdown sequence
if (child_proc->stopped.load(std::memory_order_acquire)) {
return;
}
subprocess_destroy(&child_proc->get());
// update status and exit code
- this->update_status(name, {
- SERVER_MODEL_STATUS_UNLOADED,
- 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);
});
{
auto & old_instance = mapping[name];
// old process should have exited already, but just in case, we clean it up here
- if (old_instance.subproc->is_alive()) {
+ if (old_instance.subproc && old_instance.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
}
cv.notify_all();
}
-// callback for model downloading functionality
-struct server_models_download_res : public common_download_callback {
- common_params_model model;
- common_download_opts opts;
-
- std::function<bool()> should_stop;
- std::function<void(const common_download_progress & p)> on_progress;
-
- bool is_ok = false;
-
- bool run() {
- try {
- common_download_model(model, opts);
- is_ok = true;
- } catch (const std::exception & e) {
- auto model_name = 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_start(const common_download_progress & p) override {
- on_progress(p);
- }
- void on_update(const common_download_progress & p) override {
- on_progress(p);
- }
- void on_done(const common_download_progress &, bool ok) override {
- is_ok = ok;
- }
- bool is_cancelled() const override {
- return should_stop();
- }
-};
-
-void server_models::download(common_params_model && model, common_download_opts && opts) {
- std::string name = model.get_name();
- GGML_ASSERT(name == model.hf_repo);
-
- std::unique_lock<std::mutex> lk(mutex);
- if (mapping.find(name) != mapping.end()) {
- throw std::runtime_error("model name=" + name + " already exists");
- }
-
- instance_t inst;
- inst.meta.name = name;
- inst.meta.status = SERVER_MODEL_STATUS_DOWNLOADING;
- inst.subproc = std::make_shared<server_subproc>();
-
- auto dl = std::make_unique<server_models_download_res>();
- dl->model = model; // copy
- dl->opts = opts; // copy
-
- dl->should_stop = [sp = inst.subproc]() {
- return sp->stopped.load(std::memory_order_relaxed);
- };
-
- dl->on_progress = [this, name](const common_download_progress & p) {
- update_download_progress(name, p, false);
- };
-
- inst.th = std::thread([this, dl = std::move(dl)]() {
- dl->opts.callback = dl.get();
- bool ok = dl->run();
- auto model_name = dl->model.get_name();
- SRV_INF("download finished for model name=%s with status=%s\n",
- model_name.c_str(), ok ? "success" : "failure");
- update_download_progress(model_name, {}, true, ok);
- // need_reload is set inside update_download_progress under the mutex;
- // the next load_models() call will clean up this instance
- });
-
- mapping[name] = std::move(inst);
- notify_sse("status_update", name, {
- {"status", server_model_status_to_string(SERVER_MODEL_STATUS_DOWNLOADING)},
- });
- cv.notify_all();
-}
-
void server_models::unload(const std::string & name) {
std::unique_lock<std::mutex> lk(mutex);
auto it = mapping.find(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->stopped.store(true, std::memory_order_relaxed);
+ it->second.subproc->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;
}
bool server_models::remove(const std::string & name) {
- auto meta = get_meta(name);
+ // do everything under one lock acquisition; avoid get_meta() /
+ // unload() because they can trigger load_models() which erases
+ // transient DOWNLOADING / DOWNLOADED entries as a side-effect
+ std::unique_lock<std::mutex> lk(mutex);
- if (!meta.has_value()) {
+ auto it = mapping.find(name);
+ if (it == mapping.end()) {
throw std::runtime_error("model name=" + name + " is not found");
}
- if (meta->source != SERVER_MODEL_SOURCE_CACHE) {
+ if (it->second.meta.source != SERVER_MODEL_SOURCE_CACHE) {
throw std::runtime_error("model name=" + name + " is not removable (not from cache)");
}
- unload(name); // cancel download or stop running instance
- {
- std::unique_lock<std::mutex> lk(mutex);
- // a cancelled download lands on DOWNLOADED; a stopped instance lands on UNLOADED
- wait(lk, name, [](const server_model_meta & new_meta) {
- return new_meta.status == SERVER_MODEL_STATUS_UNLOADED
- || new_meta.status == SERVER_MODEL_STATUS_DOWNLOADED;
- });
- // join before erasing - after status reaches UNLOADED/DOWNLOADED the thread no
- // longer acquires this mutex, so joining while holding it is safe
- if (mapping[name].th.joinable()) {
- mapping[name].th.join();
+ 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();
+ } 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) {
+ it->second.subproc->terminate();
}
- // remove the model from disk (hold lock to prevent concurrent load)
+ cv_stop.notify_all();
+ }
+
+ // wait until the monitoring thread finishes
+ wait(lk, name, [](const server_model_meta & meta) {
+ return meta.status == SERVER_MODEL_STATUS_UNLOADED
+ || meta.status == SERVER_MODEL_STATUS_DOWNLOADED;
+ });
+
+ // 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
+ lk.unlock();
bool ok = common_download_remove(name);
- if (ok) {
- mapping.erase(name);
- }
- SRV_INF("removing model name=%s from cache (%s)\n", name.c_str(), ok ? "succeeded" : "failed");
+ SRV_INF("removing model name=%s from cache (%s)\n", name.c_str(), ok ? "succeeded" : "partial");
notify_sse("model_remove", name, {});
- return ok;
+ 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);
+ if (!ok) {
+ SRV_WRN("removing model name=%s from disk returned false (no cached files?)\n", name.c_str());
}
+ SRV_INF("removing model name=%s from cache (%s)\n", name.c_str(), ok ? "succeeded" : "partial");
+ notify_sse("model_remove", name, {});
+ return true;
}
void server_models::wait(const std::string & name, std::function<bool(const server_model_meta &)> predicate) {
return predicate(it->second.meta);
}
- return false;
+ // model was removed from mapping by another code path (e.g. load_models()).
+ // nothing left to wait for - tell the caller to proceed.
+ return true;
});
}
}
switch (state) {
+ case SERVER_STATE_DOWNLOADING:
+ {
+ std::string result = json_value(payload, "result", std::string());
+ std::string url = json_value(payload, "url", std::string());
+ auto request_exit = [&]() {
+ std::lock_guard<std::mutex> lk(mutex);
+ auto it = mapping.find(name);
+ if (it != mapping.end()) {
+ return it->second.subproc->request_exit();
+ }
+ };
+ if (result == "download_finished") {
+ update_download_progress(name, {}, true, true);
+ request_exit();
+ } else if (result == "download_failed") {
+ update_download_progress(name, {}, true, false);
+ request_exit();
+ } else if (!url.empty()) {
+ common_download_progress p;
+ p.url = url;
+ p.downloaded = json_value(payload, "downloaded", (size_t)0);
+ p.total = json_value(payload, "total", (size_t)0);
+ update_download_progress(name, p, false);
+ }
+ } break;
case SERVER_STATE_LOADING:
{
update_status(name, {
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_params_handle_models(params, LLAMA_EXAMPLE_SERVER, 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]() {
res_err(res, format_error_response("model is not found", ERROR_TYPE_INVALID_REQUEST));
return res;
}
- if (!model->is_running()) {
+ if (!model->is_running() && model->status != SERVER_MODEL_STATUS_DOWNLOADING) {
res_err(res, format_error_response("model is not running", ERROR_TYPE_INVALID_REQUEST));
return res;
}
model.hf_repo = name;
opts.bearer_token = params.hf_token;
- opts.download_mmproj = true;
- opts.download_mtp = true;
+ // note: we only check main model, no need sidecar here
+ opts.download_mmproj = false;
+ opts.download_mtp = false;
// first, only check if the model is valid and can be downloaded
opts.skip_download = true;
throw std::invalid_argument("model validation failed, unable to download");
}
+ // reject if model already exists
+ if (models.has_model(name)) {
+ throw std::invalid_argument("model '" + name + "' already exists");
+ }
+
// then, proceed with the actual download
- opts.skip_download = false;
SRV_INF("starting download for model '%s'\n", name.c_str());
- models.download(std::move(model), std::move(opts));
+ {
+ server_models::load_options load_opts;
+ load_opts.mode = SERVER_CHILD_MODE_DOWNLOAD;
+ load_opts.custom_meta = server_model_meta{};
+ load_opts.custom_meta->source = SERVER_MODEL_SOURCE_CACHE;
+ load_opts.custom_meta->name = name;
+ models.load(name, load_opts);
+ }
res_ok(res, {{"success", true}});
return res;
throw std::invalid_argument("model must be a non-empty string");
}
- bool ok = models.remove(name);
- if (!ok) {
- throw std::runtime_error("failed to remove model '" + name + "'");
- }
+ models.remove(name); // throws on error
res_ok(res, {{"success", true}});
return res;
overview / pkg / ggml / sources / llama.cpp / commitdiff