Felipe97/llama-cpp-compiled
01.1k
1#include "server-common.h"2#include "http.h"3#include "server-models.h"4#include "server-context.h"5#include "server-stream.h"6 7#include "build-info.h"8#include "preset.h"9#include "download.h"10#include "hf-cache.h"11#include "http.h"12#include "subproc.h"13 14#include <cpp-httplib/httplib.h> // TODO: remove this once we use HTTP client from download.h15#include <optional>16 17#include <functional>18#include <optional>19#include <algorithm>20#include <thread>21#include <mutex>22#include <condition_variable>23#include <cstring>24#include <cstdlib>25#include <atomic>26#include <chrono>27#include <queue>28#include <filesystem>29#include <random>30#include <sstream>31#include <cstring>32 33#ifndef _WIN3234extern char **environ;35#endif36 37#if defined(__APPLE__) && defined(__MACH__)38// macOS: use _NSGetExecutablePath to get the executable path39#include <mach-o/dyld.h>40#include <limits.h>41#endif42 43#define DEFAULT_STOP_TIMEOUT 10 // seconds44 45#define CMD_ROUTER_TO_CHILD_EXIT "cmd_router_to_child:exit"46#define CMD_CHILD_TO_ROUTER_STATE "cmd_child_to_router:state:" // followed by json string47 48// note: SIGPIPE is ignored by the server49static void request_child_exit(server_subproc & proc) {50 FILE * stdin_file = proc.sproc.stdin_file();51 if (stdin_file) {52 fprintf(stdin_file, "%s\n", CMD_ROUTER_TO_CHILD_EXIT);53 fflush(stdin_file);54 }55}56 57// address for child process, this is needed because router may run on 0.0.0.058// ref: https://github.com/ggml-org/llama.cpp/issues/1786259#define CHILD_ADDR "127.0.0.1"60 61// single-threaded, watching all child processes at once62struct server_monitor {63 server_monitor(server_models & models) : models(models) {64 th = std::thread([this]() { run(); });65 }66 67 ~server_monitor() {68 push({ cmd_t::QUIT, {}, "", 0, false });69 th.join();70 }71 72 // thread-safe73 void watch(const std::string & name, std::shared_ptr<server_subproc> proc, server_child_mode mode, int port) {74 child_t c;75 c.name = name;76 c.proc = std::move(proc);77 c.mode = mode;78 c.port = port;79 if (!c.proc->has_output()) {80 SRV_ERR("failed to get stdout/stderr of child process for name=%s\n", name.c_str());81 c.eof = true;82 }83 push({ cmd_t::WATCH, std::move(c), "", 0, false });84 }85 86 // thread-safe87 void stop(const std::string & name, int stop_timeout, bool send_exit) {88 push({ cmd_t::STOP, {}, name, stop_timeout, send_exit });89 }90 91private:92 struct child_t {93 std::string name;94 std::shared_ptr<server_subproc> proc;95 server_child_mode mode = SERVER_CHILD_MODE_NORMAL;96 int port = 0;97 std::string buf; // partial line98 bool eof = false; // output closed, waiting for the process to be reaped99 int64_t deadline = 0; // force-kill time in ms, 0 when no stop is pending100 };101 102 struct cmd_t {103 enum { WATCH, STOP, QUIT } type;104 child_t child;105 std::string name;106 int stop_timeout;107 bool send_exit;108 };109 110 void push(cmd_t && cmd) {111 {112 std::lock_guard<std::mutex> lk(mu);113 cmds.push_back(std::move(cmd));114 }115 waiter.wake();116 }117 118 // returns true if the loop should exit119 bool handle_commands() {120 std::deque<cmd_t> batch;121 {122 std::lock_guard<std::mutex> lk(mu);123 batch.swap(cmds);124 }125 for (auto & cmd : batch) {126 switch (cmd.type) {127 case cmd_t::WATCH:128 children.push_back(std::move(cmd.child));129 break;130 case cmd_t::STOP:131 // the newest child with this name is the one the registry knows132 for (auto it = children.rbegin(); it != children.rend(); ++it) {133 if (it->name != cmd.name) {134 continue;135 }136 if (cmd.send_exit && !it->eof) {137 request_child_exit(*it->proc);138 }139 it->deadline = ggml_time_ms() + (int64_t) cmd.stop_timeout * 1000;140 break;141 }142 break;143 case cmd_t::QUIT:144 return true;145 }146 }147 return false;148 }149 150 // read what the child wrote, forward complete lines151 void read_output(child_t & c) {152 char chunk[4096];153 while (!c.eof) {154 int n = c.proc->read_output(chunk, sizeof(chunk));155 if (n < 0) {156 c.eof = true;157 break;158 }159 if (n == 0) {160 break;161 }162 c.buf.append(chunk, (size_t) n);163 size_t start = 0;164 while (true) {165 size_t nl = c.buf.find('\n', start);166 if (nl == std::string::npos) {167 break;168 }169 std::string line = c.buf.substr(start, nl + 1 - start);170 start = nl + 1;171 on_line(c, line);172 }173 c.buf.erase(0, start);174 if (c.buf.size() > max_line) {175 c.buf.clear(); // a child that never writes a newline must not grow this without bound176 }177 }178 if (c.eof && !c.buf.empty()) {179 on_line(c, c.buf);180 c.buf.clear();181 }182 }183 184 void on_line(child_t & c, const std::string & line) {185 if (string_starts_with(line, CMD_CHILD_TO_ROUTER_STATE)) {186 LOG_DBG("[%5d] %s", c.port, line.c_str()); // prevent spamming the log187 models.handle_child_state(c.name, line);188 } else {189 LOG("[%5d] %s", c.port, line.c_str()); // forward log190 }191 }192 193 void run() {194 while (true) {195 if (handle_commands()) {196 return;197 }198 199 // wait for output, a wakeup, or the next deadline;200 // a child whose output closed is polled for its exit every 50 ms201 int64_t now = ggml_time_ms();202 int64_t timeout = -1;203 for (const auto & c : children) {204 if (c.eof) {205 timeout = timeout < 0 ? 50 : std::min<int64_t>(timeout, 50);206 }207 if (c.deadline) {208 int64_t d = std::max<int64_t>(0, c.deadline - now);209 timeout = timeout < 0 ? d : std::min(timeout, d);210 }211 }212 std::vector<server_subproc *> procs;213 std::vector<child_t *> owners;214 for (auto & c : children) {215 if (!c.eof) {216 procs.push_back(c.proc.get());217 owners.push_back(&c);218 }219 }220 std::vector<bool> ready;221 waiter.wait(procs, ready, timeout);222 for (size_t i = 0; i < owners.size(); i++) {223 if (ready[i]) {224 read_output(*owners[i]);225 }226 }227 228 // deadlines and exits229 now = ggml_time_ms();230 for (auto it = children.begin(); it != children.end();) {231 if (it->deadline && now >= it->deadline && !it->proc->stopped.load(std::memory_order_acquire)) {232 SRV_WRN("force-killing model instance name=%s after timeout\n", it->name.c_str());233 it->proc->terminate();234 it->deadline = 0;235 }236 if (it->eof && !it->proc->is_alive()) {237 int exit_code = it->proc->join();238 it->proc->stopped.store(true, std::memory_order_release);239 models.on_child_exit(it->name, it->proc, it->mode, exit_code);240 SRV_INF("instance name=%s exited with status %d\n", it->name.c_str(), exit_code);241 it = children.erase(it);242 } else {243 ++it;244 }245 }246 }247 }248 249 static constexpr size_t max_line = 1024 * 1024;250 251 server_models & models;252 std::mutex mu;253 std::deque<cmd_t> cmds;254 std::vector<child_t> children; // monitor thread only255 server_subproc::waiter waiter;256 std::thread th;257};258 259struct server_lru_sched {260 server_lru_sched(server_models & models) : models(models) {}261 262 bool has_capacity(std::unique_lock<std::mutex> & lk) {263 check_lock(lk);264 return models.base_params.models_max <= 0265 || count_running() < (size_t) models.base_params.models_max;266 }267 268 // returns "" if no model can be given up269 std::string pick_victim(std::unique_lock<std::mutex> & lk) {270 check_lock(lk);271 std::string victim;272 int64_t victim_last_used = 0;273 for (const auto & m : models.mapping) {274 // a busy model is mid-request, one still coming up has no request to finish275 if (m.second.req_count != 0 || !m.second.meta.is_ready_or_sleep()) {276 continue;277 }278 // already on its way out, or a queued request wants it279 if (models.stopping_models.count(m.first) || find(m.first)) {280 continue;281 }282 if (victim.empty() || m.second.meta.last_used < victim_last_used) {283 victim = m.first;284 victim_last_used = m.second.meta.last_used;285 }286 }287 return victim;288 }289 290 // requests wanting the same model share one entry, so they all need only one slot291 // and all get unblocked by the single load that entry performs292 void join(std::unique_lock<std::mutex> & lk, const std::string & model_id) {293 check_lock(lk);294 if (entry_t * e = find(model_id)) {295 e->n_waiters++;296 SRV_INF("request for name=%s joined the queue, %d waiting\n", model_id.c_str(), e->n_waiters);297 return;298 }299 queue.push_back({ model_id, 1, false });300 SRV_INF("models_max reached, request for name=%s queued at position %zu\n",301 model_id.c_str(), queue.size());302 }303 304 void leave(std::unique_lock<std::mutex> & lk, const std::string & model_id) {305 check_lock(lk);306 for (auto it = queue.begin(); it != queue.end(); ++it) {307 if (it->model_id == model_id) {308 if (--it->n_waiters <= 0) {309 queue.erase(it); // last one waiting for this model went away310 }311 return;312 }313 }314 }315 316 bool queue_empty(std::unique_lock<std::mutex> & lk) {317 check_lock(lk);318 return queue.empty();319 }320 321 // true if it is this model's turn to load, and nobody is loading it yet322 bool try_claim(std::unique_lock<std::mutex> & lk, const std::string & model_id) {323 check_lock(lk);324 if (queue.empty() || queue.front().model_id != model_id || queue.front().loading) {325 return false;326 }327 if (!has_capacity(lk)) {328 return false;329 }330 queue.front().loading = true;331 return true;332 }333 334 // on failure the entry is back in line; on success it stays until its waiters leave,335 // so the model coming up is never picked as a victim before they use it336 void claim_done(std::unique_lock<std::mutex> & lk, const std::string & model_id, bool ok) {337 check_lock(lk);338 if (ok) {339 return;340 }341 for (auto it = queue.begin(); it != queue.end(); ++it) {342 if (it->model_id == model_id) {343 it->loading = false;344 return;345 }346 }347 }348 349 // evict idle models while queued requests outnumber the slots that are free or being freed350 // caller must hold models.mutex; never blocks, so it is safe from any thread351 void tick(std::unique_lock<std::mutex> & lk) {352 check_lock(lk);353 if (models.base_params.models_max <= 0 || queue.empty()) {354 return;355 }356 int n_running = 0;357 int n_stopping = 0;358 for (const auto & m : models.mapping) {359 if (m.second.meta.is_running()) {360 n_running++;361 if (models.stopping_models.count(m.first)) {362 n_stopping++;363 }364 }365 }366 int n_needed = 0;367 int n_claimed = 0; // claimed the slot, but load() has not spawned yet368 for (const auto & e : queue) {369 if (!e.loading) {370 n_needed++;371 continue;372 }373 auto it = models.mapping.find(e.model_id);374 if (it != models.mapping.end() && !it->second.meta.is_running()) {375 n_claimed++;376 }377 }378 int n_free = models.base_params.models_max - n_running + n_stopping - n_claimed;379 while (n_free < n_needed) {380 std::string victim = pick_victim(lk);381 if (victim.empty()) {382 return; // all remaining models are busy, wait for a request to end383 }384 SRV_INF("evicting idle LRU name=%s for a queued request\n", victim.c_str());385 models.request_stop(victim);386 n_free++;387 }388 }389 390 private:391 struct entry_t {392 std::string model_id;393 int n_waiters; // requests waiting for this model394 bool loading; // one of the waiters is doing the load right now395 };396 397 entry_t * find(const std::string & model_id) {398 for (auto & e : queue) {399 if (e.model_id == model_id) {400 return &e;401 }402 }403 return nullptr;404 }405 406 void check_lock(std::unique_lock<std::mutex> & lk) {407 GGML_ASSERT(lk.owns_lock() && lk.mutex() == &models.mutex);408 }409 410 size_t count_running() {411 size_t count = 0;412 for (const auto & m : models.mapping) {413 if (m.second.meta.is_running()) {414 count++;415 }416 }417 return count;418 }419 420 server_models & models;421 std::deque<entry_t> queue;422};423 424// short loopback budget for the resumable stream router to child JSON calls (probe, lookup,425// delete). distinct from params.timeout_read/write which only applies to the generation proxy426static constexpr int STREAM_LOOKUP_TIMEOUT_MS = 250;427 428static std::filesystem::path get_server_exec_path() {429#if defined(_WIN32)430 wchar_t buf[32768] = { 0 }; // Large buffer to handle long paths431 DWORD len = GetModuleFileNameW(nullptr, buf, _countof(buf));432 if (len == 0 || len >= _countof(buf)) {433 throw std::runtime_error("GetModuleFileNameW failed or path too long");434 }435 return std::filesystem::path(buf);436#elif defined(__APPLE__) && defined(__MACH__)437 char small_path[PATH_MAX];438 uint32_t size = sizeof(small_path);439 440 if (_NSGetExecutablePath(small_path, &size) == 0) {441 // resolve any symlinks to get absolute path442 try {443 return std::filesystem::canonical(std::filesystem::path(small_path));444 } catch (...) {445 return std::filesystem::path(small_path);446 }447 } else {448 // buffer was too small, allocate required size and call again449 std::vector<char> buf(size);450 if (_NSGetExecutablePath(buf.data(), &size) == 0) {451 try {452 return std::filesystem::canonical(std::filesystem::path(buf.data()));453 } catch (...) {454 return std::filesystem::path(buf.data());455 }456 }457 throw std::runtime_error("_NSGetExecutablePath failed after buffer resize");458 }459#else460 char path[FILENAME_MAX];461 ssize_t count = readlink("/proc/self/exe", path, FILENAME_MAX);462 if (count <= 0) {463 throw std::runtime_error("failed to resolve /proc/self/exe");464 }465 return std::filesystem::path(std::string(path, count));466#endif467}468 469static void unset_reserved_args(common_preset & preset, bool unset_model_args) {470 preset.unset_option("LLAMA_ARG_SSL_KEY_FILE");471 preset.unset_option("LLAMA_ARG_SSL_CERT_FILE");472 preset.unset_option("LLAMA_API_KEY");473 preset.unset_option("LLAMA_ARG_MODELS_DIR");474 preset.unset_option("LLAMA_ARG_MODELS_MAX");475 preset.unset_option("LLAMA_ARG_MODELS_PRESET");476 preset.unset_option("LLAMA_ARG_MODELS_AUTOLOAD");477 if (unset_model_args) {478 preset.unset_option("LLAMA_ARG_MODEL");479 preset.unset_option("LLAMA_ARG_MMPROJ");480 preset.unset_option("LLAMA_ARG_ALIAS");481 preset.unset_option("LLAMA_ARG_HF_REPO");482 }483}484 485#ifdef _WIN32486static std::string wide_to_utf8(const wchar_t * ws) {487 if (!ws || !*ws) {488 return {};489 }490 491 const int len = static_cast<int>(std::wcslen(ws));492 const int bytes = WideCharToMultiByte(CP_UTF8, 0, ws, len, nullptr, 0, nullptr, nullptr);493 if (bytes == 0) {494 return {};495 }496 497 std::string utf8(bytes, '\0');498 WideCharToMultiByte(CP_UTF8, 0, ws, len, utf8.data(), bytes, nullptr, nullptr);499 500 return utf8;501}502#endif503 504static std::vector<std::string> get_environment() {505 std::vector<std::string> env;506 507#ifdef _WIN32508 LPWCH env_block = GetEnvironmentStringsW();509 if (!env_block) {510 return env;511 }512 for (LPWCH e = env_block; *e; e += wcslen(e) + 1) {513 env.emplace_back(wide_to_utf8(e));514 }515 FreeEnvironmentStringsW(env_block);516#else517 if (environ == nullptr) {518 return env;519 }520 for (char ** e = environ; *e != nullptr; e++) {521 env.emplace_back(*e);522 }523#endif524 525 return env;526}527 528void server_model_meta::update_args(common_preset_context & ctx_preset, std::string bin_path) {529 // update params530 unset_reserved_args(preset, false);531 preset.set_option(ctx_preset, "LLAMA_ARG_HOST", CHILD_ADDR);532 preset.set_option(ctx_preset, "LLAMA_ARG_PORT", std::to_string(port));533 preset.set_option(ctx_preset, "LLAMA_ARG_ALIAS", name);534 // TODO: maybe validate preset before rendering ?535 // render args536 args = preset.to_args(bin_path);537 538 // unified binary dispatches by subcommand, re-inject it right after the539 // binary path so the child starts as 'llama serve ...' not 'llama ...'540 const char * app_cmd = std::getenv("LLAMA_APP_CMD");541 if (app_cmd != nullptr && app_cmd[0] != '\0' && !bin_path.empty()) {542 args.insert(args.begin() + 1, app_cmd);543 }544}545 546void server_model_meta::update_caps() {547 try {548 common_params params;549 preset.apply_to_params(params, {550 "LLAMA_ARG_MODEL",551 "LLAMA_ARG_MODEL_URL",552 "LLAMA_ARG_MMPROJ",553 "LLAMA_ARG_MMPROJ_URL",554 "LLAMA_ARG_MMPROJ_AUTO",555 "LLAMA_ARG_HF_REPO",556 "LLAMA_ARG_HF_REPO_FILE",557 });558 params.offline = true;559 common_models_handler handler = common_models_handler_init(params, LLAMA_EXAMPLE_SERVER);560 common_models_handler_apply(handler, params); // note: this won't download the model because offline=true561 if (params.no_mmproj || params.mmproj.path.empty()) {562 multimodal = { false, false };563 } else {564 multimodal = mtmd_get_cap_from_file(params.mmproj.path.c_str());565 }566 } catch (const std::exception & e) {567 LOG_WRN("failed to initialize common_params for multimodal capability detection: %s\n", e.what());568 multimodal = { false, false };569 }570}571 572//573// server_models574//575 576server_models::server_models(577 const common_params & params,578 int argc,579 char ** argv)580 : ctx_preset(LLAMA_EXAMPLE_SERVER),581 base_params(params),582 base_env(get_environment()),583 base_preset(ctx_preset.load_from_args(argc, argv)),584 sched(std::make_unique<server_lru_sched>(*this)),585 monitor(std::make_unique<server_monitor>(*this)) {586 // clean up base preset587 unset_reserved_args(base_preset, true);588 // set binary path589 try {590 bin_path = get_server_exec_path().string();591 } catch (const std::exception & e) {592 bin_path = argv[0];593 LOG_WRN("failed to get server executable path: %s\n", e.what());594 LOG_WRN("using original argv[0] as fallback: %s\n", argv[0]);595 }596 load_models();597 debug_fake_timing = !common_get_env("LLAMA_SERVER_DEBUG_FAKE_TIMING").empty();598}599 600server_models::~server_models() = default;601 602void server_models::instance_t::request_exit() const {603 request_child_exit(*subproc);604}605 606void server_models::add_model(server_model_meta && meta) {607 if (mapping.find(meta.name) != mapping.end()) {608 throw std::runtime_error(string_format("model '%s' appears multiple times", meta.name.c_str()));609 }610 611 // check model name does not conflict with existing aliases612 for (const auto & [key, inst] : mapping) {613 if (inst.meta.aliases.count(meta.name)) {614 throw std::runtime_error(string_format("model name '%s' conflicts with alias of model '%s'",615 meta.name.c_str(), key.c_str()));616 }617 }618 619 // parse aliases from preset's --alias option (comma-separated)620 std::string alias_str;621 if (meta.preset.get_option("LLAMA_ARG_ALIAS", alias_str) && !alias_str.empty()) {622 for (auto & alias : string_split<std::string>(alias_str, ',')) {623 alias = string_strip(alias);624 if (!alias.empty()) {625 meta.aliases.insert(alias);626 }627 }628 }629 630 // parse tags from preset's --tags option (comma-separated)631 std::string tags_str;632 if (meta.preset.get_option("LLAMA_ARG_TAGS", tags_str) && !tags_str.empty()) {633 for (auto & tag : string_split<std::string>(tags_str, ',')) {634 tag = string_strip(tag);635 if (!tag.empty()) {636 meta.tags.insert(tag);637 }638 }639 }640 641 // validate aliases do not conflict with existing names or aliases642 for (const auto & alias : meta.aliases) {643 if (mapping.find(alias) != mapping.end()) {644 throw std::runtime_error(string_format("alias '%s' for model '%s' conflicts with existing model name",645 alias.c_str(), meta.name.c_str()));646 }647 for (const auto & [key, inst] : mapping) {648 if (inst.meta.aliases.count(alias)) {649 throw std::runtime_error(string_format("alias '%s' for model '%s' conflicts with alias of model '%s'",650 alias.c_str(), meta.name.c_str(), key.c_str()));651 }652 }653 }654 655 meta.update_args(ctx_preset, bin_path); // render args656 meta.update_caps();657 std::string name = meta.name;658 mapping[name] = instance_t{659 /* subproc */ std::make_shared<server_subproc>(),660 /* meta */ std::move(meta)661 };662}663 664void server_models::notify_sse(const std::string & event, const std::string & model_id, const json & data) {665 std::unique_ptr<server_task_result_router> result = std::make_unique<server_task_result_router>();666 result->data = {667 {"model", model_id},668 {"event", event},669 };670 if (!data.is_null()) {671 result->data["data"] = data;672 }673 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());674 sse.broadcast(std::move(result));675}676 677void server_models::load_models() {678 // Phase 1: load presets from all sources - pure I/O, no lock needed679 // 1. cached models680 common_presets cached_models = ctx_preset.load_from_cache();681 SRV_TRC("Loaded %zu cached model presets from %s\n", cached_models.size(), hf_cache::get_cache_path().c_str());682 // 2. local models from --models-dir683 common_presets local_models;684 if (!base_params.models_dir.empty()) {685 local_models = ctx_preset.load_from_models_dir(base_params.models_dir);686 SRV_TRC("Loaded %zu local model presets from %s\n", local_models.size(), base_params.models_dir.c_str());687 }688 // 3. custom-path models from presets689 common_preset global = {};690 common_presets custom_presets = {};691 if (!base_params.models_preset.empty()) {692 custom_presets = ctx_preset.load_from_ini(base_params.models_preset, global);693 SRV_TRC("Loaded %zu custom model presets from %s\n", custom_presets.size(), base_params.models_preset.c_str());694 }695 696 // cascade, apply global preset first697 cached_models = ctx_preset.cascade(global, cached_models);698 local_models = ctx_preset.cascade(global, local_models);699 custom_presets = ctx_preset.cascade(global, custom_presets);700 701 // note: if a model exists in both cached and local, local takes precedence702 common_presets final_presets;703 std::unordered_map<std::string, server_model_source> source_map;704 for (const auto & [name, preset] : cached_models) {705 final_presets[name] = preset;706 source_map[name] = SERVER_MODEL_SOURCE_CACHE;707 }708 for (const auto & [name, preset] : local_models) {709 final_presets[name] = preset;710 source_map[name] = SERVER_MODEL_SOURCE_MODELS_DIR;711 }712 for (const auto & [name, custom] : custom_presets) {713 if (final_presets.find(name) != final_presets.end()) {714 final_presets[name].merge(custom);715 } else {716 final_presets[name] = custom;717 }718 source_map[name] = SERVER_MODEL_SOURCE_PRESET;719 }720 721 // overlay router's own CLI args on top of every model preset so that722 // e.g. `llama-server --temp 0` is honoured by all child processes723 for (auto & [name, preset] : final_presets) {724 preset.merge(base_preset);725 }726 727 auto get_source = [&](const std::string & name) {728 return source_map.count(name) ? source_map.at(name) : SERVER_MODEL_SOURCE_PRESET;729 };730 731 // hide cache models whose resolved file is already used by a preset with dedup-cache-models enabled732 std::set<std::string> hidden_models;733 {734 std::set<std::string> preset_paths;735 for (const auto & [name, preset] : custom_presets) {736 std::string val;737 if (!preset.get_option(COMMON_ARG_PRESET_DEDUP_CACHE_MODELS, val) || !common_arg_utils::is_truthy(val)) {738 continue;739 }740 std::string hf_repo;741 if (!preset.get_option("LLAMA_ARG_HF_REPO", hf_repo) || hf_repo.empty()) {742 continue;743 }744 std::string hf_file;745 preset.get_option("LLAMA_ARG_HF_FILE", hf_file);746 std::string path = common_download_resolve_path(hf_repo, hf_file);747 if (!path.empty()) {748 preset_paths.insert(path);749 }750 }751 if (!preset_paths.empty()) {752 for (const auto & [name, preset] : cached_models) {753 if (get_source(name) != SERVER_MODEL_SOURCE_CACHE) {754 continue; // merged with another source, not a pure cache entry755 }756 std::string path = common_download_resolve_path(name);757 if (!path.empty() && preset_paths.count(path)) {758 SRV_INF("hiding cache model name=%s (deduplicated by a preset)\n", name.c_str());759 hidden_models.insert(name);760 }761 }762 }763 }764 765 // Helpers that read `mapping` - must be called while holding the lock.766 auto join_set = [](const std::set<std::string> & s) {767 std::string result;768 for (const auto & v : s) {769 if (!result.empty()) result += ", ";770 result += v;771 }772 return result;773 };774 auto log_available_models = [&]() {775 SRV_INF("Available models (%zu):\n", mapping.size());776 if (mapping.empty()) {777 SRV_INF("%s", " no models found on the system (visit https://llama.app/models for suggestions)\n");778 } else {779 for (const auto & [name, inst] : mapping) {780 const std::string source = server_model_source_to_string(inst.meta.source);781 782 std::string info;783 if (!inst.meta.aliases.empty()) info += " (aliases: " + join_set(inst.meta.aliases) + ")";784 if (!inst.meta.tags.empty()) info += " [tags: " + join_set(inst.meta.tags) + "]";785 786 SRV_INF(" [%10s] %s%s\n", source.c_str(), name.c_str(), info.c_str());787 }788 }789 };790 auto apply_stop_timeout = [&]() {791 for (auto & [name, inst] : mapping) {792 std::string val;793 if (inst.meta.preset.get_option(COMMON_ARG_PRESET_STOP_TIMEOUT, val)) {794 try {795 inst.meta.stop_timeout = std::stoi(val);796 } catch (...) {797 SRV_WRN("invalid stop-timeout value '%s' for model '%s', using default %d seconds\n",798 val.c_str(), name.c_str(), DEFAULT_STOP_TIMEOUT);799 inst.meta.stop_timeout = DEFAULT_STOP_TIMEOUT;800 }801 }802 }803 };804 auto apply_hidden = [&]() {805 for (auto & [name, inst] : mapping) {806 inst.meta.hidden = hidden_models.count(name) > 0;807 }808 };809 // update_args() injects HOST/PORT/ALIAS, so strip them before comparing presets810 auto preset_options_for_compare = [](common_preset p) {811 p.unset_option("LLAMA_ARG_HOST");812 p.unset_option("LLAMA_ARG_PORT");813 p.unset_option("LLAMA_ARG_ALIAS");814 return p.options;815 };816 817 // Phase 2: acquire the lock once for all mapping mutations.818 // We temporarily release it only when calling functions that acquire it internally (unload)819 std::unique_lock<std::mutex> lk(mutex);820 821 need_reload = false;822 bool is_first_load = mapping.empty();823 824 if (is_first_load) {825 // FIRST LOAD: add all models, then unlock for autoloading826 for (const auto & [name, preset] : final_presets) {827 server_model_meta meta{828 /* source */ get_source(name),829 /* preset */ preset,830 /* name */ name,831 /* aliases */ {},832 /* tags */ {},833 /* port */ 0,834 /* status */ SERVER_MODEL_STATUS_UNLOADED,835 /* last_used */ 0,836 /* args */ std::vector<std::string>(),837 /* loaded_info */ {},838 /* progress */ {},839 /* exit_code */ 0,840 /* stop_timeout */ DEFAULT_STOP_TIMEOUT,841 /* multimodal */ mtmd_caps{false, false},842 // /* need_download */ false,843 };844 add_model(std::move(meta));845 }846 apply_stop_timeout();847 apply_hidden();848 log_available_models();849 850 // skipped on reload, see startup_models851 if (startup_models.has_value()) {852 std::vector<std::string> models_to_load;853 for (const auto & [name, inst] : mapping) {854 std::string val;855 if (inst.meta.preset.get_option(COMMON_ARG_PRESET_LOAD_ON_STARTUP, val) && common_arg_utils::is_truthy(val)) {856 models_to_load.push_back(name);857 }858 }859 if ((int)models_to_load.size() > base_params.models_max) {860 throw std::runtime_error(string_format(861 "number of models to load on startup (%zu) exceeds models_max (%d)",862 models_to_load.size(), base_params.models_max));863 }864 865 // to be lazy-loaded after main() setup phase is completed866 startup_models = std::move(models_to_load);867 }868 869 lk.unlock();870 } else {871 // RELOAD: diff the new preset list against the current mapping and reconcile872 is_reloading = true;873 874 // find running models whose source was removed or whose preset changed875 std::vector<std::string> to_unload;876 for (const auto & [name, inst] : mapping) {877 if (!inst.meta.is_running()) continue;878 auto it = final_presets.find(name);879 if (it == final_presets.end()) {880 to_unload.push_back(name); // removed from source881 } else if (preset_options_for_compare(inst.meta.preset) != preset_options_for_compare(it->second)) {882 to_unload.push_back(name); // preset changed883 }884 }885 886 // unload() acquires the lock internally, so release before each call887 for (const auto & name : to_unload) {888 SRV_INF("(reload) unloading model name=%s (source updated or removed)\n", name.c_str());889 lk.unlock();890 unload(name);891 lk.lock();892 }893 894 // wait for all targeted models to reach UNLOADED; cv.wait handles unlock/relock895 cv.wait(lk, [&]() {896 for (const auto & name : to_unload) {897 auto it = mapping.find(name);898 if (it != mapping.end() && it->second.meta.is_running()) return false;899 }900 return true;901 });902 903 // erase models no longer in any source904 for (auto it = mapping.begin(); it != mapping.end(); ) {905 if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADING) {906 ++it; // download thread is still busy, skip907 } else if (it->second.meta.status == SERVER_MODEL_STATUS_DOWNLOADED) {908 // download finished, safe to erase909 it = mapping.erase(it);910 } else if (final_presets.find(it->first) == final_presets.end()) {911 SRV_INF("(reload) removing model name=%s (no longer in source)\n", it->first.c_str());912 it = mapping.erase(it);913 } else {914 ++it;915 }916 }917 918 // update presets for non-running models still in source919 for (auto & [name, inst] : mapping) {920 if (inst.meta.is_running()) continue;921 auto it = final_presets.find(name);922 if (it == final_presets.end()) continue; // erased above923 924 inst.meta.preset = it->second;925 926 // re-parse aliases, then validate against other models927 std::set<std::string> new_aliases;928 std::string alias_str;929 if (inst.meta.preset.get_option("LLAMA_ARG_ALIAS", alias_str) && !alias_str.empty()) {930 for (auto & alias : string_split<std::string>(alias_str, ',')) {931 alias = string_strip(alias);932 if (!alias.empty()) new_aliases.insert(alias);933 }934 }935 inst.meta.aliases.clear();936 for (const auto & alias : new_aliases) {937 bool conflict = false;938 for (const auto & [other_name, other_inst] : mapping) {939 if (other_name == name) continue;940 if (other_name == alias || other_inst.meta.aliases.count(alias)) {941 SRV_WRN("(reload) alias '%s' for model '%s' conflicts with model '%s', skipping\n",942 alias.c_str(), name.c_str(), other_name.c_str());943 conflict = true;944 break;945 }946 }947 if (!conflict) inst.meta.aliases.insert(alias);948 }949 950 // re-parse tags951 inst.meta.tags.clear();952 std::string tags_str;953 if (inst.meta.preset.get_option("LLAMA_ARG_TAGS", tags_str) && !tags_str.empty()) {954 for (auto & tag : string_split<std::string>(tags_str, ',')) {955 tag = string_strip(tag);956 if (!tag.empty()) inst.meta.tags.insert(tag);957 }958 }959 960 inst.meta.exit_code = 0; // clear failed state so the model can be reloaded961 inst.meta.update_args(ctx_preset, bin_path);962 inst.meta.update_caps();963 }964 965 // add models that are new in this reload, load-on-startup is not honored here since a966 // reload never spawns an instance967 for (const auto & [name, preset] : final_presets) {968 if (mapping.find(name) == mapping.end()) {969 server_model_meta meta{970 /* source */ get_source(name),971 /* preset */ preset,972 /* name */ name,973 /* aliases */ {},974 /* tags */ {},975 /* port */ 0,976 /* status */ SERVER_MODEL_STATUS_UNLOADED,977 /* last_used */ 0,978 /* args */ std::vector<std::string>(),979 /* loaded_info */ {},980 /* progress */ {},981 /* exit_code */ 0,982 /* stop_timeout */ DEFAULT_STOP_TIMEOUT,983 /* multimodal */ mtmd_caps{false, false},984 // /* need_download */ false,985 };986 add_model(std::move(meta));987 }988 }989 990 apply_stop_timeout();991 apply_hidden();992 993 // clear reload flag under the lock, this releases the load() calls waiting on !is_reloading994 is_reloading = false;995 cv.notify_all();996 997 log_available_models();998 999 lk.unlock();1000 1001 notify_sse("models_reload", "*");1002 }1003}1004 1005void server_models::load_startup_models() {1006 std::vector<std::string> to_load;1007 {1008 std::lock_guard<std::mutex> lk(mutex);1009 if (!startup_models.has_value()) {1010 return; // already drained1011 }1012 to_load = std::move(*startup_models);1013 startup_models.reset();1014 }1015 for (const auto & name : to_load) {1016 SRV_INF("(startup) loading model %s\n", name.c_str());1017 load(name);1018 }1019}1020 1021void server_models::update_meta(const std::string & name, const server_model_meta & meta) {1022 std::lock_guard<std::mutex> lk(mutex);1023 auto it = mapping.find(name);1024 if (it != mapping.end()) {1025 it->second.meta = meta;1026 }1027 cv.notify_all(); // notify wait_until_loading_finished1028}1029 1030bool server_models::has_model(const std::string & name) {1031 std::lock_guard<std::mutex> lk(mutex);1032 if (mapping.find(name) != mapping.end()) {1033 return true;1034 }1035 for (const auto & [key, inst] : mapping) {1036 if (inst.meta.aliases.count(name)) {1037 return true;1038 }1039 }1040 return false;1041}1042 1043std::optional<server_model_meta> server_models::get_meta(const std::string & name) {1044 std::unique_lock<std::mutex> lk(mutex);1045 if (need_reload) {1046 lk.unlock();1047 load_models();1048 lk.lock();1049 }1050 1051 auto it = mapping.find(name);1052 if (it != mapping.end()) {1053 return it->second.meta;1054 }1055 for (const auto & [key, inst] : mapping) {1056 if (inst.meta.aliases.count(name)) {1057 return inst.meta;1058 }1059 }1060 return std::nullopt;1061}1062 1063std::vector<server_model_meta> server_models::get_all_meta() {1064 std::unique_lock<std::mutex> lk(mutex);1065 if (need_reload) {1066 lk.unlock();1067 load_models();1068 lk.lock();1069 }1070 1071 std::vector<server_model_meta> result;1072 result.reserve(mapping.size());1073 for (const auto & [name, inst] : mapping) {1074 result.push_back(inst.meta);1075 }1076 return result;1077}1078 1079void server_models::unload_lru() {1080 if (base_params.models_max <= 0) {1081 return; // no limit1082 }1083 // remove one of the servers if we passed the models_max (least recently used - LRU)1084 std::string lru_model_name;1085 {1086 std::unique_lock<std::mutex> lk(mutex);1087 if (sched->has_capacity(lk)) {1088 return;1089 }1090 lru_model_name = sched->pick_victim(lk);1091 }1092 if (!lru_model_name.empty()) {1093 SRV_INF("models_max limit reached, removing LRU name=%s\n", lru_model_name.c_str());1094 unload(lru_model_name);1095 // wait for unload to complete1096 {1097 std::unique_lock<std::mutex> lk(mutex);1098 cv.wait(lk, [this, &lru_model_name]() {1099 return mapping[lru_model_name].meta.status == SERVER_MODEL_STATUS_UNLOADED;1100 });1101 }1102 }1103}1104 1105void server_models::load(const std::string & name) {1106 load(name, load_options{});1107}1108 1109void server_models::load(const std::string & name, const load_options & opts) {1110 if (debug_fake_timing) {1111 // do not hold the mutex here, other requests must keep making progress1112 std::this_thread::sleep_for(std::chrono::seconds(2));1113 }1114 1115 if (!opts.custom_meta.has_value()) {1116 if (!has_model(name)) {1117 throw std::runtime_error("model name=" + name + " is not found");1118 }1119 unload_lru();1120 }1121 1122 std::unique_lock<std::mutex> lk(mutex);1123 // edge case: block until any in-progress reload has finished so we always load1124 // against the freshest preset and a consistent mapping state1125 cv.wait(lk, [this]() { return !is_reloading; });1126 1127 auto meta = opts.custom_meta.has_value() ? *opts.custom_meta : mapping[name].meta;1128 if (meta.status != SERVER_MODEL_STATUS_UNLOADED) {1129 SRV_INF("model %s is not ready\n", name.c_str());1130 return;1131 }1132 1133 // Re-check capacity under the lock to prevent concurrent loads from1134 // exceeding models_max. Without this, the window between unload_lru()1135 // releasing its lock and this lock_guard acquiring allows multiple1136 // threads to each observe capacity and all proceed to load.1137 // Download workers do not use models_max slots.1138 if (opts.mode == SERVER_CHILD_MODE_NORMAL && base_params.models_max > 0) {1139 size_t count_active = 0;1140 for (const auto & m : mapping) {1141 if (m.second.meta.is_running()) {1142 count_active++;1143 }1144 }1145 if (count_active >= (size_t)base_params.models_max) {1146 throw std::runtime_error("model limit reached, try again later");1147 }1148 }1149 1150 // prepare new instance info1151 instance_t inst;1152 inst.meta = meta;1153 inst.meta.port = common_http_get_free_port();1154 inst.meta.status = SERVER_MODEL_STATUS_LOADING;1155 inst.meta.loaded_info = json{};1156 inst.meta.last_used = ggml_time_ms();1157 1158 if (inst.meta.port <= 0) {1159 throw std::runtime_error("failed to get a port number");1160 }1161 1162 inst.subproc = std::make_shared<server_subproc>();1163 {1164 SRV_INF("spawning server instance with name=%s on port %d\n", inst.meta.name.c_str(), inst.meta.port);1165 1166 inst.meta.update_args(ctx_preset, bin_path); // render args1167 1168 std::vector<std::string> child_args = inst.meta.args; // copy1169 std::vector<std::string> child_env = base_env; // copy1170 child_env.push_back("LLAMA_SERVER_ROUTER_PORT=" + std::to_string(base_params.port));1171 1172 if (opts.mode == SERVER_CHILD_MODE_DOWNLOAD) {1173 inst.meta.status = SERVER_MODEL_STATUS_DOWNLOADING;1174 child_env.push_back("LLAMA_SERVER_CHILD_MODE=download");1175 child_env.push_back("LLAMA_ARG_HF_REPO=" + name);1176 }1177 1178 SRV_INF("%s", "spawning server instance with args:\n");1179 for (const auto & arg : child_args) {1180 SRV_INF(" %s\n", arg.c_str());1181 }1182 inst.meta.args = child_args; // save for debugging1183 1184 // TODO @ngxson : maybe separate stdout and stderr in the future1185 // so that we can use stdout for commands and stderr for logging1186 int options = subprocess_option_no_window | subprocess_option_combined_stdout_stderr;1187 if (!inst.subproc->sproc.create(child_args, options, child_env)) {1188 throw std::runtime_error("failed to spawn server instance");1189 }1190 }1191 1192 // old process should have exited already, but just in case, we clean it up here1193 {1194 auto it = mapping.find(name);1195 if (it != mapping.end() && it->second.subproc && it->second.subproc->is_alive()) {1196 SRV_WRN("old process for model name=%s is still alive, this is unexpected\n", name.c_str());1197 it->second.subproc->terminate(); // force kill1198 }1199 }1200