#include <sheredom/subprocess.h>
#include <functional>
+#include <optional>
#include <algorithm>
#include <thread>
#include <mutex>
// ref: https://github.com/ggml-org/llama.cpp/issues/17862
#define CHILD_ADDR "127.0.0.1"
+struct server_subproc {
+ std::optional<subprocess_s> sproc; // empty while in DOWNLOADING state
+ std::atomic<bool> stop_download{false}; // flag to signal download cancellation
+
+ subprocess_s & get() {
+ GGML_ASSERT(sproc.has_value() && "subprocess not initialized");
+ return sproc.value();
+ }
+
+ bool is_alive() {
+ return sproc.has_value() && subprocess_alive(&sproc.value());
+ }
+};
+
+
static std::filesystem::path get_server_exec_path() {
#if defined(_WIN32)
wchar_t buf[32768] = { 0 }; // Large buffer to handle long paths
meta.update_caps();
std::string name = meta.name;
mapping[name] = instance_t{
- /* subproc */ std::make_shared<subprocess_s>(),
+ /* subproc */ std::make_shared<server_subproc>(),
/* th */ std::thread(),
/* meta */ std::move(meta)
};
}
+void server_models::notify_sse(const std::string & event, const std::string & model_id, const json & data) {
+ std::unique_ptr<server_task_result_router> result = std::make_unique<server_task_result_router>();
+ result->data = {
+ {"model", model_id},
+ {"event", event},
+ };
+ if (!data.is_null()) {
+ result->data["data"] = data;
+ }
+ SRV_DBG("notifying SSE clients about event '%s' for model '%s': %s\n", event.c_str(), model_id.c_str(), safe_json_to_str(result->data).c_str());
+ sse.broadcast(std::move(result));
+}
+
void server_models::load_models() {
// Phase 1: load presets from all sources — pure I/O, no lock needed
// 1. cached models
// note: if a model exists in both cached and local, local takes precedence
common_presets final_presets;
- for (const auto & [name, preset] : cached_models) final_presets[name] = preset;
- for (const auto & [name, preset] : local_models) final_presets[name] = preset;
+ std::unordered_map<std::string, server_model_source> source_map;
+ for (const auto & [name, preset] : cached_models) {
+ final_presets[name] = preset;
+ source_map[name] = SERVER_MODEL_SOURCE_CACHE;
+ }
+ for (const auto & [name, preset] : local_models) {
+ final_presets[name] = preset;
+ source_map[name] = SERVER_MODEL_SOURCE_MODELS_DIR;
+ }
for (const auto & [name, custom] : custom_presets) {
if (final_presets.find(name) != final_presets.end()) {
final_presets[name].merge(custom);
} else {
final_presets[name] = custom;
}
+ source_map[name] = SERVER_MODEL_SOURCE_PRESET;
}
- // server base preset from CLI args takes highest precedence
- for (auto & [name, preset] : final_presets) {
- preset.merge(base_preset);
- }
+
+ auto get_source = [&](const std::string & name) {
+ return source_map.count(name) ? source_map.at(name) : SERVER_MODEL_SOURCE_PRESET;
+ };
// Helpers that read `mapping` — must be called while holding the lock.
std::unordered_set<std::string> custom_names;
// (unload, load) or when joining threads (the monitoring thread calls update_status
// which locks the mutex, so joining while holding it would deadlock).
std::unique_lock<std::mutex> lk(mutex);
+
+ need_reload = false;
bool is_first_load = mapping.empty();
if (is_first_load) {
// FIRST LOAD: add all models, then unlock for autoloading
for (const auto & [name, preset] : final_presets) {
server_model_meta meta{
+ /* source */ get_source(name),
/* preset */ preset,
/* name */ name,
/* aliases */ {},
/* exit_code */ 0,
/* stop_timeout */ DEFAULT_STOP_TIMEOUT,
/* multimodal */ mtmd_caps{false, false},
- /* need_download */ false,
+ // /* need_download */ false,
};
add_model(std::move(meta));
}
}
}
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 (final_presets.find(name) == final_presets.end() && !inst.meta.is_running() && inst.th.joinable()) {
threads_to_join.push_back(std::move(inst.th));
}
// erase models no longer in any source
for (auto it = mapping.begin(); it != mapping.end(); ) {
- if (final_presets.find(it->first) == final_presets.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, safe to erase
+ if (it->second.th.joinable()) {
+ it->second.th.join();
+ }
+ 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);
for (const auto & [name, preset] : final_presets) {
if (mapping.find(name) == mapping.end()) {
server_model_meta meta{
+ /* source */ get_source(name),
/* preset */ preset,
/* name */ name,
/* aliases */ {},
/* exit_code */ 0,
/* stop_timeout */ DEFAULT_STOP_TIMEOUT,
/* multimodal */ mtmd_caps{false, false},
- /* need_download */ false,
+ // /* need_download */ false,
};
add_model(std::move(meta));
newly_added.push_back(name);
SRV_INF("(reload) loading new model %s\n", name.c_str());
load(name);
}
+
+ notify_sse("models_reload", "*");
}
}
}
std::optional<server_model_meta> server_models::get_meta(const std::string & name) {
- std::lock_guard<std::mutex> lk(mutex);
+ std::unique_lock<std::mutex> lk(mutex);
+ if (need_reload) {
+ lk.unlock();
+ load_models();
+ lk.lock();
+ }
+
auto it = mapping.find(name);
if (it != mapping.end()) {
return it->second.meta;
}
std::vector<server_model_meta> server_models::get_all_meta() {
- std::lock_guard<std::mutex> lk(mutex);
+ std::unique_lock<std::mutex> lk(mutex);
+ if (need_reload) {
+ lk.unlock();
+ load_models();
+ lk.lock();
+ }
+
std::vector<server_model_meta> result;
result.reserve(mapping.size());
for (const auto & [name, inst] : mapping) {
throw std::runtime_error("failed to get a port number");
}
- inst.subproc = std::make_shared<subprocess_s>();
+ inst.subproc = std::make_shared<server_subproc>();
{
SRV_INF("spawning server instance with name=%s on port %d\n", inst.meta.name.c_str(), inst.meta.port);
// TODO @ngxson : maybe separate stdout and stderr in the future
// so that we can use stdout for commands and stderr for logging
int options = subprocess_option_no_window | subprocess_option_combined_stdout_stderr;
- int result = subprocess_create_ex(argv.data(), options, envp.data(), inst.subproc.get());
+ inst.subproc->sproc.emplace();
+ int result = subprocess_create_ex(argv.data(), options, envp.data(), &inst.subproc->get());
if (result != 0) {
throw std::runtime_error("failed to spawn server instance");
}
- inst.stdin_file = subprocess_stdin(inst.subproc.get());
+ 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]() {
- FILE * stdin_file = subprocess_stdin(child_proc.get());
- FILE * stdout_file = subprocess_stdout(child_proc.get()); // combined stdout/stderr
+ FILE * stdin_file = subprocess_stdin(&child_proc->get());
+ FILE * stdout_file = subprocess_stdout(&child_proc->get()); // combined stdout/stderr
std::thread log_thread([&]() {
// read stdout/stderr and forward to main server log
return this->stopping_models.find(name) != this->stopping_models.end();
};
auto should_wake = [&]() {
- return is_stopping() || !subprocess_alive(child_proc.get());
+ return is_stopping() || !child_proc->is_alive();
};
{
std::unique_lock<std::mutex> lk(this->mutex);
this->cv_stop.wait(lk, should_wake);
}
// child may have already exited (e.g. crashed) — skip shutdown sequence
- if (!subprocess_alive(child_proc.get())) {
+ if (!child_proc->is_alive()) {
return;
}
SRV_INF("stopping model instance name=%s\n", name.c_str());
if (elapsed >= stop_timeout * 1000) {
// timeout, force kill
SRV_WRN("force-killing model instance name=%s after %d seconds timeout\n", name.c_str(), stop_timeout);
- subprocess_terminate(child_proc.get());
+ subprocess_terminate(&child_proc->get());
return;
}
this->cv_stop.wait_for(lk, std::chrono::seconds(1));
// get the exit code
int exit_code = 0;
- subprocess_join(child_proc.get(), &exit_code);
- subprocess_destroy(child_proc.get());
+ subprocess_join(&child_proc->get(), &exit_code);
+ subprocess_destroy(&child_proc->get());
// update status and exit code
this->update_status(name, SERVER_MODEL_STATUS_UNLOADED, exit_code);
{
auto & old_instance = mapping[name];
// old process should have exited already, but just in case, we clean it up here
- if (subprocess_alive(old_instance.subproc.get())) {
+ if (old_instance.subproc->is_alive()) {
SRV_WRN("old process for model name=%s is still alive, this is unexpected\n", name.c_str());
- subprocess_terminate(old_instance.subproc.get()); // force kill
+ subprocess_terminate(&old_instance.subproc->get()); // force kill
}
if (old_instance.th.joinable()) {
old_instance.th.join();
}
}
+ notify_sse("model_status", name, {
+ {"status", server_model_status_to_string(inst.meta.status)},
+ });
+
mapping[name] = std::move(inst);
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) {
+ 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.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->stop_download.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();
+ SRV_INF("download finished for model name=%s with status=%s\n",
+ dl->model.name.c_str(), ok ? "success" : "failure");
+ update_download_progress(dl->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::lock_guard<std::mutex> lk(mutex);
+ std::unique_lock<std::mutex> lk(mutex);
auto it = mapping.find(name);
if (it != mapping.end()) {
- if (it->second.meta.is_running()) {
+ if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
+ SRV_INF("cancelling download for model name=%s\n", name.c_str());
+ it->second.subproc->stop_download.store(true, std::memory_order_relaxed);
+ // 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());
stopping_models.insert(name);
if (it->second.meta.status == SERVER_MODEL_STATUS_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());
- subprocess_terminate(it->second.subproc.get());
+ subprocess_terminate(&it->second.subproc->get());
}
cv_stop.notify_all();
// status change will be handled by the managing thread
{
std::lock_guard<std::mutex> lk(mutex);
for (auto & [name, inst] : mapping) {
- if (inst.meta.is_running()) {
+ if (inst.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {
+ SRV_INF("cancelling download for model name=%s\n", name.c_str());
+ inst.subproc->stop_download.store(true, std::memory_order_relaxed);
+ } else if (inst.meta.is_running()) {
SRV_INF("stopping model instance name=%s\n", name.c_str());
stopping_models.insert(name);
cv_stop.notify_all();
meta.status = status;
meta.exit_code = exit_code;
}
+ // broadcast status change to SSE
+ {
+ json data = {
+ {"status", server_model_status_to_string(status)},
+ };
+ if (status == SERVER_MODEL_STATUS_UNLOADED) {
+ data["exit_code"] = exit_code;
+ }
+ // note: notify_sse doesn't acquire the lock, so no deadlock here
+ notify_sse("status_change", name, data);
+ }
cv.notify_all();
}
cv.notify_all();
}
-void server_models::wait_until_loading_finished(const std::string & name) {
+void server_models::update_download_progress(const std::string & name, const common_download_progress & progress, bool done, bool ok) {
+ json curr;
+ {
+ std::lock_guard<std::mutex> lk(mutex);
+ auto it = mapping.find(name);
+ if (it != mapping.end()) {
+ if (done) {
+ // mark the instance to be erased on next load_models() call
+ it->second.meta.status = SERVER_MODEL_STATUS_DOWNLOADED;
+ need_reload = true;
+ } else {
+ json & info = it->second.meta.loaded_info;
+ if (!info.contains("progress")) {
+ info["progress"] = json{};
+ }
+ info["progress"][progress.url] = {
+ {"done", progress.downloaded},
+ {"total", progress.total},
+ };
+ curr = it->second.meta.loaded_info; // copy
+ }
+ }
+ }
+ if (done) {
+ cv.notify_all(); // notify in case unload() is waiting for download to be cancelled
+ notify_sse(ok ? "download_finished" : "download_failed", name, {});
+ } else {
+ notify_sse("download_progress", name, curr);
+ }
+}
+
+bool server_models::remove(const std::string & name) {
+ auto meta = get_meta(name);
+
+ if (!meta.has_value()) {
+ throw std::runtime_error("model name=" + name + " is not found");
+ }
+ if (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();
+ }
+ // remove the model from disk (hold lock to prevent concurrent load)
+ 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");
+ notify_sse("model_remove", name, {});
+ return ok;
+ }
+}
+
+void server_models::wait(const std::string & name, std::function<bool(const server_model_meta &)> predicate) {
std::unique_lock<std::mutex> lk(mutex);
- cv.wait(lk, [this, &name]() {
+ wait(lk, name, predicate);
+}
+
+void server_models::wait(std::unique_lock<std::mutex> & lk, const std::string & name, std::function<bool(const server_model_meta &)> predicate) {
+ cv.wait(lk, [this, &name, &predicate]() {
auto it = mapping.find(name);
if (it != mapping.end()) {
- return it->second.meta.status != SERVER_MODEL_STATUS_LOADING;
+ return predicate(it->second.meta);
+
}
return false;
});
// wait for loading to complete
SRV_INF("waiting until model name=%s is fully loaded...\n", name.c_str());
- wait_until_loading_finished(name);
+ wait(name, [&meta](const server_model_meta & new_meta) {
+ if (new_meta.status != SERVER_MODEL_STATUS_LOADING) {
+ meta = new_meta; // update meta for final check after wait
+ return true;
+ }
+ return false;
+ });
// check final status
- meta = get_meta(name);
if (!meta.has_value() || meta->is_failed()) {
throw std::runtime_error("model name=" + name + " failed to load");
}
// server_models_routes
//
+// RAII wrapper similar to server_response_reader, but doesn't use server_queue
+static std::atomic<int> sse_client_id_counter = 0;
+struct server_models_sse_client {
+ server_response & queue_results;
+ int client_id;
+ server_models_sse_client(server_response & q)
+ : queue_results(q), client_id(sse_client_id_counter.fetch_add(1, std::memory_order_relaxed)) {
+ SRV_DBG("new SSE client connected, assigned client_id=%d\n", client_id);
+ queue_results.add_waiting_task_id(client_id);
+ }
+ ~server_models_sse_client() {
+ SRV_DBG("SSE client disconnected, removing client_id=%d\n", client_id);
+ queue_results.remove_waiting_task_id(client_id);
+ }
+
+ // return nullptr if should_stop() is true before receiving a result
+ // note: if one error is received, it will stop further processing and return error result
+ server_task_result_ptr next(const std::function<bool()> & should_stop) {
+ while (true) {
+ static const int http_polling_seconds = 1; // check should_stop every 1 second
+ server_task_result_ptr result = queue_results.recv_with_timeout({client_id}, http_polling_seconds);
+ if (result == nullptr) {
+ // timeout, check stop condition
+ if (should_stop()) {
+ return nullptr;
+ }
+ // continue waiting otherwise
+ } else {
+ SRV_DBG("recv result for client_id=%d: %s\n", client_id, safe_json_to_str(result->to_json()).c_str());
+ return result;
+ }
+ }
+ // should not reach here
+ }
+};
+
static void res_ok(std::unique_ptr<server_http_res> & res, const json & response_data) {
res->status = 200;
res->data = safe_json_to_str(response_data);
{"created", t}, // for OAI-compat
{"status", status},
{"architecture", architecture},
- {"need_download", meta.need_download},
+ {"source", server_model_source_to_string(meta.source)},
+ {"can_remove", meta.source == SERVER_MODEL_SOURCE_CACHE},
+ // {"need_download", meta.need_download},
// TODO: add other fields, may require reading GGUF metadata
};
res_ok(res, {{"success", true}});
return res;
};
+
+ this->get_router_models_sse = [this](const server_http_req & req) {
+ auto res = std::make_unique<server_http_res>();
+ res->status = 200;
+ res->content_type = "text/event-stream";
+ auto sse_client = std::make_shared<server_models_sse_client>(models.sse);
+ res->next = [this, sse_client, &req](std::string & output) -> bool {
+ auto result = sse_client->next([&]() {
+ return stopping.load(std::memory_order_relaxed) || req.should_stop();
+ });
+ if (result == nullptr) {
+ return false; // client disconnected or should_stop
+ }
+ output = "data: " + safe_json_to_str(result->to_json()) + "\n\n";
+ return true; // listen for the next event
+ };
+ return res;
+ };
+
+ this->post_router_models = [this](const server_http_req & req) {
+ auto res = std::make_unique<server_http_res>();
+
+ json body = json::parse(req.body);
+ std::string name = json_value(body, "model", std::string());
+ if (name.empty()) {
+ throw std::invalid_argument("model must be a non-empty string");
+ }
+
+ common_params_model model;
+ common_download_opts opts;
+
+ model.name = name;
+ model.hf_repo = name;
+ opts.bearer_token = params.hf_token;
+ opts.download_mmproj = true;
+ opts.download_mtp = true;
+
+ // first, only check if the model is valid and can be downloaded
+ opts.skip_download = true;
+ bool ok = false;
+ try {
+ auto validation = common_download_model(model, opts);
+ ok = !validation.model_path.empty();
+ } catch (const common_skip_download_exception &) {
+ // model is valid and will be downloaded
+ ok = true;
+ } catch (...) {
+ SRV_ERR("unknown error while validating model '%s'\n", name.c_str());
+ // other exceptions will be handled by the outer ex_wrapper()
+ throw;
+ }
+
+ if (!ok) {
+ throw std::invalid_argument("model validation failed, unable to download");
+ }
+
+ // 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));
+
+ res_ok(res, {{"success", true}});
+ return res;
+ };
+
+ this->del_router_models = [this](const server_http_req & req) {
+ auto res = std::make_unique<server_http_res>();
+
+ std::string name = req.get_param("model");
+ if (name.empty()) {
+ 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 + "'");
+ }
+
+ res_ok(res, {{"success", true}});
+ return res;
+ };
}
#pragma once
#include "common.h"
+#include "download.h"
#include "preset.h"
#include "server-common.h"
#include "server-http.h"
+#include "server-queue.h"
#include <mutex>
#include <condition_variable>
/**
* state diagram:
*
+ * DOWNLOADING ──► DOWNLOADED ──► (replaced by new instance)
+ *
* UNLOADED ──► LOADING ──► LOADED ◄──── SLEEPING
* ▲ │ │ ▲
* └───failed───┘ │ │
*/
enum server_model_status {
// TODO: also add downloading state when the logic is added
+ SERVER_MODEL_STATUS_DOWNLOADING,
+ SERVER_MODEL_STATUS_DOWNLOADED,
SERVER_MODEL_STATUS_UNLOADED,
SERVER_MODEL_STATUS_LOADING,
SERVER_MODEL_STATUS_LOADED,
SERVER_MODEL_STATUS_SLEEPING
};
-static server_model_status server_model_status_from_string(const std::string & status_str) {
- if (status_str == "unloaded") {
- return SERVER_MODEL_STATUS_UNLOADED;
- }
- if (status_str == "loading") {
- return SERVER_MODEL_STATUS_LOADING;
- }
- if (status_str == "loaded") {
- return SERVER_MODEL_STATUS_LOADED;
- }
- if (status_str == "sleeping") {
- return SERVER_MODEL_STATUS_SLEEPING;
- }
- throw std::runtime_error("invalid server model status");
-}
+enum server_model_source {
+ SERVER_MODEL_SOURCE_PRESET,
+ SERVER_MODEL_SOURCE_MODELS_DIR,
+ SERVER_MODEL_SOURCE_CACHE,
+};
static std::string server_model_status_to_string(server_model_status status) {
switch (status) {
- case SERVER_MODEL_STATUS_UNLOADED: return "unloaded";
- case SERVER_MODEL_STATUS_LOADING: return "loading";
- case SERVER_MODEL_STATUS_LOADED: return "loaded";
- case SERVER_MODEL_STATUS_SLEEPING: return "sleeping";
- default: return "unknown";
+ case SERVER_MODEL_STATUS_DOWNLOADING: return "downloading";
+ case SERVER_MODEL_STATUS_DOWNLOADED: return "downloaded";
+ case SERVER_MODEL_STATUS_UNLOADED: return "unloaded";
+ case SERVER_MODEL_STATUS_LOADING: return "loading";
+ case SERVER_MODEL_STATUS_LOADED: return "loaded";
+ case SERVER_MODEL_STATUS_SLEEPING: return "sleeping";
+ default: return "unknown";
+ }
+}
+
+static std::string server_model_source_to_string(server_model_source source) {
+ switch (source) {
+ case SERVER_MODEL_SOURCE_PRESET: return "preset";
+ case SERVER_MODEL_SOURCE_MODELS_DIR: return "models_dir";
+ case SERVER_MODEL_SOURCE_CACHE: return "cache";
+ default: return "unknown";
}
}
struct server_model_meta {
+ server_model_source source = SERVER_MODEL_SOURCE_CACHE;
common_preset preset;
std::string name;
std::set<std::string> aliases; // additional names that resolve to this model
server_model_status status = SERVER_MODEL_STATUS_UNLOADED;
int64_t last_used = 0; // for LRU unloading
std::vector<std::string> args; // args passed to the model instance, will be populated by render_args()
- json loaded_info; // info to be reflected via /v1/models endpoint
+ json loaded_info; // info to be reflected via /v1/models endpoint ; if in DOWNLOADING state, it should contain download progress info
int exit_code = 0; // exit code of the model instance process (only valid if status == FAILED)
int stop_timeout = 0; // seconds to wait before force-killing the model instance during shutdown
mtmd_caps multimodal; // multimodal capabilities
- bool need_download = false; // whether the model needs to be downloaded before loading
+ // bool need_download = false; // whether the model needs to be downloaded before loading // TODO @ngxson: implement this
bool is_ready() const {
return status == SERVER_MODEL_STATUS_LOADED;
void update_caps();
};
-struct subprocess_s;
+struct server_models_routes;
+struct server_subproc; // defined in server-models.cpp
struct server_models {
+ friend struct server_models_routes;
+
private:
struct instance_t {
- std::shared_ptr<subprocess_s> subproc; // shared between main thread and monitoring thread
+ std::shared_ptr<server_subproc> subproc; // shared between main thread and monitoring thread
std::thread th;
server_model_meta meta;
FILE * stdin_file = nullptr;
// set to true while load_models() is executing a reload; load() will wait until clear
bool is_reloading = false;
+ // if true, the next get_meta() will trigger a reload of model list
+ bool need_reload = false;
+
common_preset_context ctx_preset;
common_params base_params;
// not thread-safe, caller must hold mutex
void add_model(server_model_meta && meta);
+ // notify SSE clients
+ void notify_sse(const std::string & event, const std::string & model_id, const json & data = nullptr);
+
public:
server_models(const common_params & params, int argc, char ** argv);
+ server_response sse; // for real-time updates via SSE endpoint
+
// (re-)load the list of models from various sources and prepare the metadata mapping
// - if this is called the first time, simply populate the metadata
// - if this is called subsequently (e.g. when refreshing from disk):
void unload(const std::string & name);
void unload_all();
+ // download a new model, progress is reported via SSE
+ // to stop the download, call unload()
+ void download(common_params_model && model, common_download_opts && opts);
+
// update the status of a model instance (thread-safe)
void update_status(const std::string & name, server_model_status status, int exit_code);
void update_loaded_info(const std::string & name, std::string & raw_info);
+ void update_download_progress(const std::string & name, const common_download_progress & progress, bool done, bool ok = true);
+
+ // remove a cache model from disk and update the list (thread-safe)
+ // note: only cache models can be removed; returns false if the model doesn't exist or is not a cache model
+ bool remove(const std::string & name);
// wait until the model instance is fully loaded (thread-safe)
+ // note: predicate is called while holding the lock
// return when the model no longer in "loading" state
- void wait_until_loading_finished(const std::string & name);
+ void wait(const std::string & name, std::function<bool(const server_model_meta &)> predicate);
+ void wait(std::unique_lock<std::mutex> & lk, const std::string & name, std::function<bool(const server_model_meta &)> predicate);
// ensure the model is in ready state (thread-safe)
// return false if model is ready
struct server_models_routes {
common_params params;
- json ui_settings = json::object(); // Primary: new name
- json webui_settings = json::object(); // Deprecated: use ui_settings (kept for compat)
+ json ui_settings = json::object(); // Primary: new name
+ json webui_settings = json::object(); // Deprecated: use ui_settings (kept for compat)
+ std::atomic<bool> stopping = false; // for graceful disconnecting SSE clients during shutdown
server_models models;
server_models_routes(const common_params & params, int argc, char ** argv)
: params(params), models(params, argc, argv) {
server_http_context::handler_t get_router_models;
server_http_context::handler_t post_router_models_load;
server_http_context::handler_t post_router_models_unload;
+ // management API
+ server_http_context::handler_t get_router_models_sse;
+ server_http_context::handler_t post_router_models;
+ server_http_context::handler_t del_router_models;
};
/**
overview / pkg / ggml / sources / llama.cpp / commitdiff