echodict/llama.cpp
version https://git-lfs.github.com/spec/v1 oid sha256:cfc44b7ba25614df70e6b65e3341cae0310163bd32fd31a6b928a542df433faf size 30786
0762
1#include "server-common.h"2#include "server-models.h"3 4#include "build-info.h"5#include "preset.h"6#include "download.h"7 8#include <cpp-httplib/httplib.h> // TODO: remove this once we use HTTP client from download.h9#include <sheredom/subprocess.h>10 11#include <functional>12#include <algorithm>13#include <thread>14#include <mutex>15#include <condition_variable>16#include <cstring>17#include <atomic>18#include <chrono>19#include <queue>20#include <filesystem>21#include <cstring>22 23#ifdef _WIN3224#include <winsock2.h>25#include <windows.h>26#else27#include <sys/socket.h>28#include <netinet/in.h>29#include <arpa/inet.h>30#include <unistd.h>31extern char **environ;32#endif33 34#if defined(__APPLE__) && defined(__MACH__)35// macOS: use _NSGetExecutablePath to get the executable path36#include <mach-o/dyld.h>37#include <limits.h>38#endif39 40#define DEFAULT_STOP_TIMEOUT 10 // seconds41 42#define CMD_ROUTER_TO_CHILD_EXIT "cmd_router_to_child:exit"43#define CMD_CHILD_TO_ROUTER_READY "cmd_child_to_router:ready" // also sent when waking up from sleep44#define CMD_CHILD_TO_ROUTER_SLEEP "cmd_child_to_router:sleep"45 46// address for child process, this is needed because router may run on 0.0.0.047// ref: https://github.com/ggml-org/llama.cpp/issues/1786248#define CHILD_ADDR "127.0.0.1"49 50static std::filesystem::path get_server_exec_path() {51#if defined(_WIN32)52 wchar_t buf[32768] = { 0 }; // Large buffer to handle long paths53 DWORD len = GetModuleFileNameW(nullptr, buf, _countof(buf));54 if (len == 0 || len >= _countof(buf)) {55 throw std::runtime_error("GetModuleFileNameW failed or path too long");56 }57 return std::filesystem::path(buf);58#elif defined(__APPLE__) && defined(__MACH__)59 char small_path[PATH_MAX];60 uint32_t size = sizeof(small_path);61 62 if (_NSGetExecutablePath(small_path, &size) == 0) {63 // resolve any symlinks to get absolute path64 try {65 return std::filesystem::canonical(std::filesystem::path(small_path));66 } catch (...) {67 return std::filesystem::path(small_path);68 }69 } else {70 // buffer was too small, allocate required size and call again71 std::vector<char> buf(size);72 if (_NSGetExecutablePath(buf.data(), &size) == 0) {73 try {74 return std::filesystem::canonical(std::filesystem::path(buf.data()));75 } catch (...) {76 return std::filesystem::path(buf.data());77 }78 }79 throw std::runtime_error("_NSGetExecutablePath failed after buffer resize");80 }81#else82 char path[FILENAME_MAX];83 ssize_t count = readlink("/proc/self/exe", path, FILENAME_MAX);84 if (count <= 0) {85 throw std::runtime_error("failed to resolve /proc/self/exe");86 }87 return std::filesystem::path(std::string(path, count));88#endif89}90 91static void unset_reserved_args(common_preset & preset, bool unset_model_args) {92 preset.unset_option("LLAMA_ARG_SSL_KEY_FILE");93 preset.unset_option("LLAMA_ARG_SSL_CERT_FILE");94 preset.unset_option("LLAMA_API_KEY");95 preset.unset_option("LLAMA_ARG_MODELS_DIR");96 preset.unset_option("LLAMA_ARG_MODELS_MAX");97 preset.unset_option("LLAMA_ARG_MODELS_PRESET");98 preset.unset_option("LLAMA_ARG_MODELS_AUTOLOAD");99 if (unset_model_args) {100 preset.unset_option("LLAMA_ARG_MODEL");101 preset.unset_option("LLAMA_ARG_MMPROJ");102 preset.unset_option("LLAMA_ARG_ALIAS");103 preset.unset_option("LLAMA_ARG_HF_REPO");104 }105}106 107#ifdef _WIN32108static std::string wide_to_utf8(const wchar_t * ws) {109 if (!ws || !*ws) {110 return {};111 }112 113 const int len = static_cast<int>(std::wcslen(ws));114 const int bytes = WideCharToMultiByte(CP_UTF8, 0, ws, len, nullptr, 0, nullptr, nullptr);115 if (bytes == 0) {116 return {};117 }118 119 std::string utf8(bytes, '\0');120 WideCharToMultiByte(CP_UTF8, 0, ws, len, utf8.data(), bytes, nullptr, nullptr);121 122 return utf8;123}124#endif125 126static std::vector<std::string> get_environment() {127 std::vector<std::string> env;128 129#ifdef _WIN32130 LPWCH env_block = GetEnvironmentStringsW();131 if (!env_block) {132 return env;133 }134 for (LPWCH e = env_block; *e; e += wcslen(e) + 1) {135 env.emplace_back(wide_to_utf8(e));136 }137 FreeEnvironmentStringsW(env_block);138#else139 if (environ == nullptr) {140 return env;141 }142 for (char ** e = environ; *e != nullptr; e++) {143 env.emplace_back(*e);144 }145#endif146 147 return env;148}149 150void server_model_meta::update_args(common_preset_context & ctx_preset, std::string bin_path) {151 // update params152 unset_reserved_args(preset, false);153 preset.set_option(ctx_preset, "LLAMA_ARG_HOST", CHILD_ADDR);154 preset.set_option(ctx_preset, "LLAMA_ARG_PORT", std::to_string(port));155 preset.set_option(ctx_preset, "LLAMA_ARG_ALIAS", name);156 // TODO: maybe validate preset before rendering ?157 // render args158 args = preset.to_args(bin_path);159}160 161//162// server_models163//164 165server_models::server_models(166 const common_params & params,167 int argc,168 char ** argv)169 : ctx_preset(LLAMA_EXAMPLE_SERVER),170 base_params(params),171 base_env(get_environment()),172 base_preset(ctx_preset.load_from_args(argc, argv)) {173 // clean up base preset174 unset_reserved_args(base_preset, true);175 // set binary path176 try {177 bin_path = get_server_exec_path().string();178 } catch (const std::exception & e) {179 bin_path = argv[0];180 LOG_WRN("failed to get server executable path: %s\n", e.what());181 LOG_WRN("using original argv[0] as fallback: %s\n", argv[0]);182 }183 load_models();184}185 186void server_models::add_model(server_model_meta && meta) {187 if (mapping.find(meta.name) != mapping.end()) {188 throw std::runtime_error(string_format("model '%s' appears multiple times", meta.name.c_str()));189 }190 191 // check model name does not conflict with existing aliases192 for (const auto & [key, inst] : mapping) {193 if (inst.meta.aliases.count(meta.name)) {194 throw std::runtime_error(string_format("model name '%s' conflicts with alias of model '%s'",195 meta.name.c_str(), key.c_str()));196 }197 }198 199 // parse aliases from preset's --alias option (comma-separated)200 std::string alias_str;201 if (meta.preset.get_option("LLAMA_ARG_ALIAS", alias_str) && !alias_str.empty()) {202 for (auto & alias : string_split<std::string>(alias_str, ',')) {203 alias = string_strip(alias);204 if (!alias.empty()) {205 meta.aliases.insert(alias);206 }207 }208 }209 210 // parse tags from preset's --tags option (comma-separated)211 std::string tags_str;212 if (meta.preset.get_option("LLAMA_ARG_TAGS", tags_str) && !tags_str.empty()) {213 for (auto & tag : string_split<std::string>(tags_str, ',')) {214 tag = string_strip(tag);215 if (!tag.empty()) {216 meta.tags.insert(tag);217 }218 }219 }220 221 // validate aliases do not conflict with existing names or aliases222 for (const auto & alias : meta.aliases) {223 if (mapping.find(alias) != mapping.end()) {224 throw std::runtime_error(string_format("alias '%s' for model '%s' conflicts with existing model name",225 alias.c_str(), meta.name.c_str()));226 }227 for (const auto & [key, inst] : mapping) {228 if (inst.meta.aliases.count(alias)) {229 throw std::runtime_error(string_format("alias '%s' for model '%s' conflicts with alias of model '%s'",230 alias.c_str(), meta.name.c_str(), key.c_str()));231 }232 }233 }234 235 meta.update_args(ctx_preset, bin_path); // render args236 std::string name = meta.name;237 mapping[name] = instance_t{238 /* subproc */ std::make_shared<subprocess_s>(),239 /* th */ std::thread(),240 /* meta */ std::move(meta)241 };242}243 244// TODO: allow refreshing cached model list245void server_models::load_models() {246 // loading models from 3 sources:247 // 1. cached models248 common_presets cached_models = ctx_preset.load_from_cache();249 SRV_INF("Loaded %zu cached model presets\n", cached_models.size());250 // 2. local models from --models-dir251 common_presets local_models;252 if (!base_params.models_dir.empty()) {253 local_models = ctx_preset.load_from_models_dir(base_params.models_dir);254 SRV_INF("Loaded %zu local model presets from %s\n", local_models.size(), base_params.models_dir.c_str());255 }256 // 3. custom-path models from presets257 common_preset global = {};258 common_presets custom_presets = {};259 if (!base_params.models_preset.empty()) {260 custom_presets = ctx_preset.load_from_ini(base_params.models_preset, global);261 SRV_INF("Loaded %zu custom model presets from %s\n", custom_presets.size(), base_params.models_preset.c_str());262 }263 264 // cascade, apply global preset first265 cached_models = ctx_preset.cascade(global, cached_models);266 local_models = ctx_preset.cascade(global, local_models);267 custom_presets = ctx_preset.cascade(global, custom_presets);268 269 // note: if a model exists in both cached and local, local takes precedence270 common_presets final_presets;271 for (const auto & [name, preset] : cached_models) {272 final_presets[name] = preset;273 }274 for (const auto & [name, preset] : local_models) {275 final_presets[name] = preset;276 }277 278 // process custom presets from INI279 for (const auto & [name, custom] : custom_presets) {280 if (final_presets.find(name) != final_presets.end()) {281 // apply custom config if exists282 common_preset & target = final_presets[name];283 target.merge(custom);284 } else {285 // otherwise add directly286 final_presets[name] = custom;287 }288 }289 290 // server base preset from CLI args take highest precedence291 for (auto & [name, preset] : final_presets) {292 preset.merge(base_preset);293 }294 295 // convert presets to server_model_meta and add to mapping296 for (const auto & preset : final_presets) {297 server_model_meta meta{298 /* preset */ preset.second,299 /* name */ preset.first,300 /* aliases */ {},301 /* tags */ {},302 /* port */ 0,303 /* status */ SERVER_MODEL_STATUS_UNLOADED,304 /* last_used */ 0,305 /* args */ std::vector<std::string>(),306 /* exit_code */ 0,307 /* stop_timeout */ DEFAULT_STOP_TIMEOUT,308 };309 add_model(std::move(meta));310 }311 312 // log available models313 {314 std::unordered_set<std::string> custom_names;315 for (const auto & [name, preset] : custom_presets) {316 custom_names.insert(name);317 }318 auto join_set = [](const std::set<std::string> & s) {319 std::string result;320 for (const auto & v : s) {321 if (!result.empty()) {322 result += ", ";323 }324 result += v;325 }326 return result;327 };328 329 SRV_INF("Available models (%zu) (*: custom preset)\n", mapping.size());330 for (const auto & [name, inst] : mapping) {331 bool has_custom = custom_names.find(name) != custom_names.end();332 std::string info;333 if (!inst.meta.aliases.empty()) {334 info += " (aliases: " + join_set(inst.meta.aliases) + ")";335 }336 if (!inst.meta.tags.empty()) {337 info += " [tags: " + join_set(inst.meta.tags) + "]";338 }339 SRV_INF(" %c %s%s\n", has_custom ? '*' : ' ', name.c_str(), info.c_str());340 }341 }342 343 // handle custom stop-timeout option344 for (auto & [name, inst] : mapping) {345 std::string val;346 if (inst.meta.preset.get_option(COMMON_ARG_PRESET_STOP_TIMEOUT, val)) {347 try {348 inst.meta.stop_timeout = std::stoi(val);349 } catch (...) {350 SRV_WRN("invalid stop-timeout value '%s' for model '%s', using default %d seconds\n",351 val.c_str(), name.c_str(), DEFAULT_STOP_TIMEOUT);352 inst.meta.stop_timeout = DEFAULT_STOP_TIMEOUT;353 }354 }355 }356 357 // load any autoload models358 std::vector<std::string> models_to_load;359 for (const auto & [name, inst] : mapping) {360 std::string val;361 if (inst.meta.preset.get_option(COMMON_ARG_PRESET_LOAD_ON_STARTUP, val)) {362 if (common_arg_utils::is_truthy(val)) {363 models_to_load.push_back(name);364 }365 }366 }367 if ((int)models_to_load.size() > base_params.models_max) {368 throw std::runtime_error(string_format(369 "number of models to load on startup (%zu) exceeds models_max (%d)",370 models_to_load.size(),371 base_params.models_max372 ));373 }374 for (const auto & name : models_to_load) {375 SRV_INF("(startup) loading model %s\n", name.c_str());376 load(name);377 }378}379 380void server_models::update_meta(const std::string & name, const server_model_meta & meta) {381 std::lock_guard<std::mutex> lk(mutex);382 auto it = mapping.find(name);383 if (it != mapping.end()) {384 it->second.meta = meta;385 }386 cv.notify_all(); // notify wait_until_loading_finished387}388 389bool server_models::has_model(const std::string & name) {390 std::lock_guard<std::mutex> lk(mutex);391 if (mapping.find(name) != mapping.end()) {392 return true;393 }394 for (const auto & [key, inst] : mapping) {395 if (inst.meta.aliases.count(name)) {396 return true;397 }398 }399 return false;400}401 402std::optional<server_model_meta> server_models::get_meta(const std::string & name) {403 std::lock_guard<std::mutex> lk(mutex);404 auto it = mapping.find(name);405 if (it != mapping.end()) {406 return it->second.meta;407 }408 for (const auto & [key, inst] : mapping) {409 if (inst.meta.aliases.count(name)) {410 return inst.meta;411 }412 }413 return std::nullopt;414}415 416static int get_free_port() {417#ifdef _WIN32418 WSADATA wsaData;419 if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) {420 return -1;421 }422 typedef SOCKET native_socket_t;423#define INVALID_SOCKET_VAL INVALID_SOCKET424#define CLOSE_SOCKET(s) closesocket(s)425#else426 typedef int native_socket_t;427#define INVALID_SOCKET_VAL -1428#define CLOSE_SOCKET(s) close(s)429#endif430 431 native_socket_t sock = socket(AF_INET, SOCK_STREAM, 0);432 if (sock == INVALID_SOCKET_VAL) {433#ifdef _WIN32434 WSACleanup();435#endif436 return -1;437 }438 439 struct sockaddr_in serv_addr;440 std::memset(&serv_addr, 0, sizeof(serv_addr));441 serv_addr.sin_family = AF_INET;442 serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);443 serv_addr.sin_port = htons(0);444 445 if (bind(sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) != 0) {446 CLOSE_SOCKET(sock);447#ifdef _WIN32448 WSACleanup();449#endif450 return -1;451 }452 453#ifdef _WIN32454 int namelen = sizeof(serv_addr);455#else456 socklen_t namelen = sizeof(serv_addr);457#endif458 if (getsockname(sock, (struct sockaddr*)&serv_addr, &namelen) != 0) {459 CLOSE_SOCKET(sock);460#ifdef _WIN32461 WSACleanup();462#endif463 return -1;464 }465 466 int port = ntohs(serv_addr.sin_port);467 468 CLOSE_SOCKET(sock);469#ifdef _WIN32470 WSACleanup();471#endif472 473 return port;474}475 476// helper to convert vector<string> to char **477// pointers are only valid as long as the original vector is valid478static std::vector<char *> to_char_ptr_array(const std::vector<std::string> & vec) {479 std::vector<char *> result;480 result.reserve(vec.size() + 1);481 for (const auto & s : vec) {482 result.push_back(const_cast<char*>(s.c_str()));483 }484 result.push_back(nullptr);485 return result;486}487 488std::vector<server_model_meta> server_models::get_all_meta() {489 std::lock_guard<std::mutex> lk(mutex);490 std::vector<server_model_meta> result;491 result.reserve(mapping.size());492 for (const auto & [name, inst] : mapping) {493 result.push_back(inst.meta);494 }495 return result;496}497 498void server_models::unload_lru() {499 if (base_params.models_max <= 0) {500 return; // no limit501 }502 // remove one of the servers if we passed the models_max (least recently used - LRU)503 std::string lru_model_name = "";504 int64_t lru_last_used = ggml_time_ms();505 size_t count_active = 0;506 {507 std::unique_lock<std::mutex> lk(mutex);508 for (const auto & m : mapping) {509 if (m.second.meta.is_running()) {510 count_active++;511 if (m.second.meta.last_used < lru_last_used) {512 lru_model_name = m.first;513 lru_last_used = m.second.meta.last_used;514 }515 }516 }517 }518 if (!lru_model_name.empty() && count_active >= (size_t)base_params.models_max) {519 SRV_INF("models_max limit reached, removing LRU name=%s\n", lru_model_name.c_str());520 unload(lru_model_name);521 // wait for unload to complete522 {523 std::unique_lock<std::mutex> lk(mutex);524 cv.wait(lk, [this, &lru_model_name]() {525 return mapping[lru_model_name].meta.status == SERVER_MODEL_STATUS_UNLOADED;526 });527 }528 }529}530 531void server_models::load(const std::string & name) {532 if (!has_model(name)) {533 throw std::runtime_error("model name=" + name + " is not found");534 }535 unload_lru();536 537 std::lock_guard<std::mutex> lk(mutex);538 539 auto meta = mapping[name].meta;540 if (meta.status != SERVER_MODEL_STATUS_UNLOADED) {541 SRV_INF("model %s is not ready\n", name.c_str());542 return;543 }544 545 // Re-check capacity under the lock to prevent concurrent loads from546 // exceeding models_max. Without this, the window between unload_lru()547 // releasing its lock and this lock_guard acquiring allows multiple548 // threads to each observe capacity and all proceed to load.549 if (base_params.models_max > 0) {550 size_t count_active = 0;551 for (const auto & m : mapping) {552 if (m.second.meta.is_running()) {553 count_active++;554 }555 }556 if (count_active >= (size_t)base_params.models_max) {557 throw std::runtime_error("model limit reached, try again later");558 }559 }560 561 // prepare new instance info562 instance_t inst;563 inst.meta = meta;564 inst.meta.port = get_free_port();565 inst.meta.status = SERVER_MODEL_STATUS_LOADING;566 inst.meta.last_used = ggml_time_ms();567 568 if (inst.meta.port <= 0) {569 throw std::runtime_error("failed to get a port number");570 }571 572 inst.subproc = std::make_shared<subprocess_s>();573 {574 SRV_INF("spawning server instance with name=%s on port %d\n", inst.meta.name.c_str(), inst.meta.port);575 576 inst.meta.update_args(ctx_preset, bin_path); // render args577 578 std::vector<std::string> child_args = inst.meta.args; // copy579 std::vector<std::string> child_env = base_env; // copy580 child_env.push_back("LLAMA_SERVER_ROUTER_PORT=" + std::to_string(base_params.port));581 582 SRV_INF("%s", "spawning server instance with args:\n");583 for (const auto & arg : child_args) {584 SRV_INF(" %s\n", arg.c_str());585 }586 inst.meta.args = child_args; // save for debugging587 588 std::vector<char *> argv = to_char_ptr_array(child_args);589 std::vector<char *> envp = to_char_ptr_array(child_env);590 591 // TODO @ngxson : maybe separate stdout and stderr in the future592 // so that we can use stdout for commands and stderr for logging593 int options = subprocess_option_no_window | subprocess_option_combined_stdout_stderr;594 int result = subprocess_create_ex(argv.data(), options, envp.data(), inst.subproc.get());595 if (result != 0) {596 throw std::runtime_error("failed to spawn server instance");597 }598 599 inst.stdin_file = subprocess_stdin(inst.subproc.get());600 }601 602 // start a thread to manage the child process603 // captured variables are guaranteed to be destroyed only after the thread is joined604 inst.th = std::thread([this, name, child_proc = inst.subproc, port = inst.meta.port, stop_timeout = inst.meta.stop_timeout]() {605 FILE * stdin_file = subprocess_stdin(child_proc.get());606 FILE * stdout_file = subprocess_stdout(child_proc.get()); // combined stdout/stderr607 608 std::thread log_thread([&]() {609 // read stdout/stderr and forward to main server log610 // also handle status report from child process611 if (stdout_file) {612 char buffer[4096];613 while (fgets(buffer, sizeof(buffer), stdout_file) != nullptr) {614 LOG("[%5d] %s", port, buffer);615 std::string str(buffer);616 if (string_starts_with(buffer, CMD_CHILD_TO_ROUTER_READY)) {617 this->update_status(name, SERVER_MODEL_STATUS_LOADED, 0);618 } else if (string_starts_with(buffer, CMD_CHILD_TO_ROUTER_SLEEP)) {619 this->update_status(name, SERVER_MODEL_STATUS_SLEEPING, 0);620 }621 }622 } else {623 SRV_ERR("failed to get stdout/stderr of child process for name=%s\n", name.c_str());624 }625 });626 627 std::thread stopping_thread([&]() {628 // thread to monitor stopping signal OR child crash629 auto is_stopping = [this, &name]() {630 return this->stopping_models.find(name) != this->stopping_models.end();631 };632 auto should_wake = [&]() {633 return is_stopping() || !subprocess_alive(child_proc.get());634 };635 {636 std::unique_lock<std::mutex> lk(this->mutex);637 this->cv_stop.wait(lk, should_wake);638 }639 // child may have already exited (e.g. crashed) — skip shutdown sequence640 if (!subprocess_alive(child_proc.get())) {641 return;642 }643 SRV_INF("stopping model instance name=%s\n", name.c_str());644 // send interrupt to child process645 fprintf(stdin_file, "%s\n", CMD_ROUTER_TO_CHILD_EXIT);646 fflush(stdin_file);647 // wait to stop gracefully or timeout648 int64_t start_time = ggml_time_ms();649 while (true) {650 std::unique_lock<std::mutex> lk(this->mutex);651 if (!is_stopping()) {652 return; // already stopped653 }654 int64_t elapsed = ggml_time_ms() - start_time;655 if (elapsed >= stop_timeout * 1000) {656 // timeout, force kill657 SRV_WRN("force-killing model instance name=%s after %d seconds timeout\n", name.c_str(), stop_timeout);658 subprocess_terminate(child_proc.get());659 return;660 }661 this->cv_stop.wait_for(lk, std::chrono::seconds(1));662 }663 });664 665 // we reach here when the child process exits666 // note: we cannot join() prior to this point because it will close stdin_file667 if (log_thread.joinable()) {668 log_thread.join();669 }670 671 // stop the timeout monitoring thread672 {673 std::lock_guard<std::mutex> lk(this->mutex);674 stopping_models.erase(name);675 cv_stop.notify_all();676 }677 if (stopping_thread.joinable()) {678 stopping_thread.join();679 }680 681 // get the exit code682 int exit_code = 0;683 subprocess_join(child_proc.get(), &exit_code);684 subprocess_destroy(child_proc.get());685 686 // update status and exit code687 this->update_status(name, SERVER_MODEL_STATUS_UNLOADED, exit_code);688 SRV_INF("instance name=%s exited with status %d\n", name.c_str(), exit_code);689 });690 691 // clean up old process/thread if exists692 {693 auto & old_instance = mapping[name];694 // old process should have exited already, but just in case, we clean it up here695 if (subprocess_alive(old_instance.subproc.get())) {696 SRV_WRN("old process for model name=%s is still alive, this is unexpected\n", name.c_str());697 subprocess_terminate(old_instance.subproc.get()); // force kill698 }699 if (old_instance.th.joinable()) {700 old_instance.th.join();701 }702 }703 704 mapping[name] = std::move(inst);705 cv.notify_all();706}707 708void server_models::unload(const std::string & name) {709 std::lock_guard<std::mutex> lk(mutex);710 auto it = mapping.find(name);711 if (it != mapping.end()) {712 if (it->second.meta.is_running()) {713 SRV_INF("stopping model instance name=%s\n", name.c_str());714 stopping_models.insert(name);715 cv_stop.notify_all();716 // status change will be handled by the managing thread717 } else {718 SRV_WRN("model instance name=%s is not running\n", name.c_str());719 }720 }721}722 723void server_models::unload_all() {724 std::vector<std::thread> to_join;725 {726 std::lock_guard<std::mutex> lk(mutex);727 for (auto & [name, inst] : mapping) {728 if (inst.meta.is_running()) {729 SRV_INF("stopping model instance name=%s\n", name.c_str());730 stopping_models.insert(name);731 cv_stop.notify_all();732 // status change will be handled by the managing thread733 }734 // moving the thread to join list to avoid deadlock735 to_join.push_back(std::move(inst.th));736 }737 }738 for (auto & th : to_join) {739 if (th.joinable()) {740 th.join();741 }742 }743}744 745void server_models::update_status(const std::string & name, server_model_status status, int exit_code) {746 std::unique_lock<std::mutex> lk(mutex);747 auto it = mapping.find(name);748 if (it != mapping.end()) {749 auto & meta = it->second.meta;750 meta.status = status;751 meta.exit_code = exit_code;752 }753 cv.notify_all();754}755 756void server_models::wait_until_loading_finished(const std::string & name) {757 std::unique_lock<std::mutex> lk(mutex);758 cv.wait(lk, [this, &name]() {759 auto it = mapping.find(name);760 if (it != mapping.end()) {761 return it->second.meta.status != SERVER_MODEL_STATUS_LOADING;762 }763 return false;764 });765}766 767bool server_models::ensure_model_ready(const std::string & name) {768 auto meta = get_meta(name);769 if (!meta.has_value()) {770 throw std::runtime_error("model name=" + name + " is not found");771 }772 if (meta->is_ready()) {773 return false; // ready for taking requests774 }775 if (meta->status == SERVER_MODEL_STATUS_SLEEPING) {776 return false; // child is sleeping but still running; new request will wake it up777 }778 if (meta->status == SERVER_MODEL_STATUS_UNLOADED) {779 SRV_INF("model name=%s is not loaded, loading...\n", name.c_str());780 load(name);781 }782 783 // wait for loading to complete784 SRV_INF("waiting until model name=%s is fully loaded...\n", name.c_str());785 wait_until_loading_finished(name);786 787 // check final status788 meta = get_meta(name);789 if (!meta.has_value() || meta->is_failed()) {790 throw std::runtime_error("model name=" + name + " failed to load");791 }792 793 return true;794}795 796server_http_res_ptr server_models::proxy_request(const server_http_req & req, const std::string & method, const std::string & name, bool update_last_used) {797 auto meta = get_meta(name);798 if (!meta.has_value()) {799 throw std::runtime_error("model name=" + name + " is not found");800 }801 if (!meta->is_running()) {802 throw std::invalid_argument("model name=" + name + " is not running");803 }804 if (update_last_used) {805 std::unique_lock<std::mutex> lk(mutex);806 mapping[name].meta.last_used = ggml_time_ms();807 }808 SRV_INF("proxying request to model %s on port %d\n", name.c_str(), meta->port);809 std::string proxy_path = req.path;810 if (!req.query_string.empty()) {811 proxy_path += '?' + req.query_string;812 }813 auto proxy = std::make_unique<server_http_proxy>(814 method,815 "http",816 CHILD_ADDR,817 meta->port,818 proxy_path,819 req.headers,820 req.body,821 req.should_stop,822 base_params.timeout_read,823 base_params.timeout_write824 );825 return proxy;826}827 828bool server_models::is_child_server() {829 const char * router_port = std::getenv("LLAMA_SERVER_ROUTER_PORT");830 return router_port != nullptr;831}832 833std::thread server_models::setup_child_server(const std::function<void(int)> & shutdown_handler) {834 // send a notification to the router server that a model instance is ready835 common_log_pause(common_log_main());836 fflush(stdout);837 fprintf(stdout, "%s\n", CMD_CHILD_TO_ROUTER_READY);838 fflush(stdout);839 common_log_resume(common_log_main());840 841 // setup thread for monitoring stdin842 return std::thread([shutdown_handler]() {843 // wait for EOF on stdin844 SRV_INF("%s", "child server monitoring thread started, waiting for EOF on stdin...\n");845 bool eof = false;846 while (true) {847 std::string line;848 if (!std::getline(std::cin, line)) {849 // EOF detected, that means the router server is unexpectedly exit or killed850 eof = true;851 break;852 }853 if (line.find(CMD_ROUTER_TO_CHILD_EXIT) != std::string::npos) {854 SRV_INF("%s", "exit command received, exiting...\n");855 shutdown_handler(0);856 break;857 }858 }859 if (eof) {860 SRV_INF("%s", "EOF on stdin detected, forcing shutdown...\n");861 exit(1);862 }863 });864}865 866void server_models::notify_router_sleeping_state(bool is_sleeping) {867 common_log_pause(common_log_main());868 fflush(stdout);869 fprintf(stdout, "%s\n", is_sleeping ? CMD_CHILD_TO_ROUTER_SLEEP : CMD_CHILD_TO_ROUTER_READY);870 fflush(stdout);871 common_log_resume(common_log_main());872}873 874 875//876// server_models_routes877//878 879static void res_ok(std::unique_ptr<server_http_res> & res, const json & response_data) {880 res->status = 200;881 res->data = safe_json_to_str(response_data);882}883 884static void res_err(std::unique_ptr<server_http_res> & res, const json & error_data) {885 res->status = json_value(error_data, "code", 500);886 res->data = safe_json_to_str({{ "error", error_data }});887}888 889static bool router_validate_model(std::string & name, server_models & models, bool models_autoload, std::unique_ptr<server_http_res> & res) {890 if (name.empty()) {891 res_err(res, format_error_response("model name is missing from the request", ERROR_TYPE_INVALID_REQUEST));892 return false;893 }894 auto meta = models.get_meta(name);895 if (!meta.has_value()) {896 res_err(res, format_error_response(string_format("model '%s' not found", name.c_str()), ERROR_TYPE_INVALID_REQUEST));897 return false;898 }899 // resolve alias to canonical model name900 name = meta->name;901 if (models_autoload) {902 models.ensure_model_ready(name);903 } else {904 if (!meta->is_running()) {905 res_err(res, format_error_response("model is not loaded", ERROR_TYPE_INVALID_REQUEST));906 return false;907 }908 }909 return true;910}911 912static bool is_autoload(const common_params & params, const server_http_req & req) {913 std::string autoload = req.get_param("autoload");914 if (autoload.empty()) {915 return params.models_autoload;916 } else {917 return autoload == "true" || autoload == "1";918 }919}920 921void server_models_routes::init_routes() {922 this->get_router_props = [this](const server_http_req & req) {923 std::string name = req.get_param("model");924 if (name.empty()) {925 // main instance926 auto res = std::make_unique<server_http_res>();927 res_ok(res, {928 // TODO: add support for this on web UI929 {"role", "router"},930 {"max_instances", params.models_max},931 {"models_autoload", params.models_autoload},932 // this is a dummy response to make sure webui doesn't break933 {"model_alias", "llama-server"},934 {"model_path", "none"},935 {"default_generation_settings", {936 {"params", json{}},937 {"n_ctx", 0},938 }},939 {"webui_settings", webui_settings},940 {"build_info", std::string(llama_build_info())},941 });942 return res;943 }944 return proxy_get(req);945 };946 947 this->proxy_get = [this](const server_http_req & req) {948 std::string method = "GET";949 std::string name = req.get_param("model");950 bool autoload = is_autoload(params, req);951 auto error_res = std::make_unique<server_http_res>();952 if (!router_validate_model(name, models, autoload, error_res)) {953 return error_res;954 }955 return models.proxy_request(req, method, name, false);956 };957 958 this->proxy_post = [this](const server_http_req & req) {959 std::string method = "POST";960 json body = json::parse(req.body);961 std::string name = json_value(body, "model", std::string());962 bool autoload = is_autoload(params, req);963 auto error_res = std::make_unique<server_http_res>();964 if (!router_validate_model(name, models, autoload, error_res)) {965 return error_res;966 }967 return models.proxy_request(req, method, name, true); // update last usage for POST request only968 };969 970 this->post_router_models_load = [this](const server_http_req & req) {971 auto res = std::make_unique<server_http_res>();972 json body = json::parse(req.body);973 std::string name = json_value(body, "model", std::string());974 auto meta = models.get_meta(name);975 if (!meta.has_value()) {976 res_err(res, format_error_response("model is not found", ERROR_TYPE_NOT_FOUND));977 return res;978 }979 if (meta->is_running()) {980 res_err(res, format_error_response("model is already running", ERROR_TYPE_INVALID_REQUEST));981 return res;982 }983 models.load(meta->name);984 res_ok(res, {{"success", true}});985 return res;986 };987 988 this->get_router_models = [this](const server_http_req &) {989 auto res = std::make_unique<server_http_res>();990 json models_json = json::array();991 auto all_models = models.get_all_meta();992 std::time_t t = std::time(0);993 for (const auto & meta : all_models) {994 json status {995 {"value", server_model_status_to_string(meta.status)},996 {"args", meta.args},997 };998 if (!meta.preset.name.empty()) {999 common_preset preset_copy = meta.preset;1000 unset_reserved_args(preset_copy, false);1001 preset_copy.unset_option("LLAMA_ARG_HOST");1002 preset_copy.unset_option("LLAMA_ARG_PORT");1003 preset_copy.unset_option("LLAMA_ARG_ALIAS");1004 preset_copy.unset_option("LLAMA_ARG_TAGS");1005 status["preset"] = preset_copy.to_ini();1006 }1007 if (meta.is_failed()) {1008 status["exit_code"] = meta.exit_code;1009 status["failed"] = true;1010 }1011 models_json.push_back(json {1012 {"id", meta.name},1013 {"aliases", meta.aliases},1014 {"tags", meta.tags},1015 {"object", "model"}, // for OAI-compat1016 {"owned_by", "llamacpp"}, // for OAI-compat1017 {"created", t}, // for OAI-compat1018 {"status", status},1019 // TODO: add other fields, may require reading GGUF metadata1020 });1021 }1022 res_ok(res, {1023 {"data", models_json},1024 {"object", "list"},1025 });1026 return res;1027 };1028 1029 this->post_router_models_unload = [this](const server_http_req & req) {1030 auto res = std::make_unique<server_http_res>();1031 json body = json::parse(req.body);1032 std::string name = json_value(body, "model", std::string());1033 auto model = models.get_meta(name);1034 if (!model.has_value()) {1035 res_err(res, format_error_response("model is not found", ERROR_TYPE_INVALID_REQUEST));1036 return res;1037 }1038 if (!model->is_running()) {1039 res_err(res, format_error_response("model is not running", ERROR_TYPE_INVALID_REQUEST));1040 return res;1041 }1042 models.unload(model->name);1043 res_ok(res, {{"success", true}});1044 return res;1045 };1046}1047 1048 1049 1050//1051// server_http_proxy1052//1053 1054// simple implementation of a pipe1055// used for streaming data between threads1056template<typename T>1057struct pipe_t {1058 std::mutex mutex;1059 std::condition_variable cv;1060 std::queue<T> queue;1061 std::atomic<bool> writer_closed{false};1062 std::atomic<bool> reader_closed{false};1063 void close_write() {1064 writer_closed.store(true, std::memory_order_relaxed);1065 cv.notify_all();1066 }1067 void close_read() {1068 reader_closed.store(true, std::memory_order_relaxed);1069 cv.notify_all();1070 }1071 bool read(T & output, const std::function<bool()> & should_stop) {1072 std::unique_lock<std::mutex> lk(mutex);1073 constexpr auto poll_interval = std::chrono::milliseconds(500);1074 while (true) {1075 if (!queue.empty()) {1076 output = std::move(queue.front());1077 queue.pop();1078 return true;1079 }1080 if (writer_closed.load()) {1081 return false; // clean EOF1082 }1083 if (should_stop()) {1084 close_read(); // signal broken pipe to writer1085 return false; // cancelled / reader no longer alive1086 }1087 cv.wait_for(lk, poll_interval);1088 }1089 }1090 bool write(T && data) {1091 std::lock_guard<std::mutex> lk(mutex);1092 if (reader_closed.load()) {1093 return false; // broken pipe1094 }1095 queue.push(std::move(data));1096 cv.notify_one();1097 return true;1098 }1099};1100 1101static std::string to_lower_copy(const std::string & value) {1102 std::string lowered(value.size(), '\0');1103 std::transform(value.begin(), value.end(), lowered.begin(), [](unsigned char c) { return std::tolower(c); });1104 return lowered;1105}1106 1107static bool should_strip_proxy_header(const std::string & header_name) {1108 // Headers that get duplicated when router forwards child responses1109 if (header_name == "server" ||1110 header_name == "transfer-encoding" ||1111 header_name == "content-length" || // quick fix for https://github.com/ggml-org/llama.cpp/issues/177101112 header_name == "keep-alive") {1113 return true;1114 }1115 1116 // Router injects CORS, child also sends them: duplicate1117 if (header_name.rfind("access-control-", 0) == 0) {1118 return true;1119 }1120 1121 return false;1122}1123 1124server_http_proxy::server_http_proxy(1125 const std::string & method,1126 const std::string & scheme,1127 const std::string & host,1128 int port,1129 const std::string & path,1130 const std::map<std::string, std::string> & headers,1131 const std::string & body,1132 const std::function<bool()> should_stop,1133 int32_t timeout_read,1134 int32_t timeout_write1135 ) {1136 // shared between reader and writer threads1137 auto cli = std::make_shared<httplib::ClientImpl>(host, port);1138 auto pipe = std::make_shared<pipe_t<msg_t>>();1139 1140 if (scheme == "https") {1141#ifdef CPPHTTPLIB_OPENSSL_SUPPORT1142 cli.reset(new httplib::SSLClient(host, port));1143#else1144 throw std::runtime_error("HTTPS requested but CPPHTTPLIB_OPENSSL_SUPPORT is not defined");1145#endif1146 }1147 1148 // setup Client1149 cli->set_follow_location(true);1150 cli->set_connection_timeout(timeout_read, 0); // use --timeout value instead of hardcoded 5 s1151 cli->set_write_timeout(timeout_read, 0); // reversed for cli (client) vs srv (server)1152 cli->set_read_timeout(timeout_write, 0);1153 this->status = 500; // to be overwritten upon response1154 this->cleanup = [pipe]() {1155 pipe->close_read();1156 pipe->close_write();1157 };1158 1159 // wire up the receive end of the pipe1160 this->next = [pipe, should_stop](std::string & out) -> bool {1161 msg_t msg;1162 bool has_next = pipe->read(msg, should_stop);1163 if (!msg.data.empty()) {1164 out = std::move(msg.data);1165 }1166 return has_next; // false if EOF or pipe broken1167 };1168 1169 // wire up the HTTP client1170 // note: do NOT capture `this` pointer, as it may be destroyed before the thread ends1171 httplib::ResponseHandler response_handler = [pipe, cli](const httplib::Response & response) {1172 msg_t msg;1173 msg.status = response.status;1174 for (const auto & [key, value] : response.headers) {1175 const auto lowered = to_lower_copy(key);1176 if (should_strip_proxy_header(lowered)) {1177 continue;1178 }1179 if (lowered == "content-type") {1180 msg.content_type = value;1181 continue;1182 }1183 msg.headers[key] = value;1184 }1185 return pipe->write(std::move(msg)); // send headers first1186 };1187 httplib::ContentReceiverWithProgress content_receiver = [pipe](const char * data, size_t data_length, size_t, size_t) {1188 // send data chunks1189 // returns false if pipe is closed / broken (signal to stop receiving)1190 return pipe->write({{}, 0, std::string(data, data_length), ""});1191 };1192 1193 // prepare the request to destination server1194 httplib::Request req;1195 {1196 req.method = method;1197 req.path = path;1198 for (const auto & [key, value] : headers) {1199 if (key == "Accept-Encoding") {1200 // disable Accept-Encoding to avoid compressed responses