diff --git a/common/arg.cpp b/common/arg.cpp index 996ea75fef..63e342776d 100644 --- a/common/arg.cpp +++ b/common/arg.cpp @@ -3308,9 +3308,18 @@ common_params_context common_params_parser_init(common_params & params, llama_ex ).set_examples({LLAMA_EXAMPLE_EMBEDDING})); add_opt(common_arg( {"--host"}, "HOST", - string_format("ip address to listen, or bind to an UNIX socket if the address ends with .sock (default: %s)", params.hostname.c_str()), + string_format("IP addresses to listen on, comma-separated, or UNIX socket paths ending in .sock; with multiple TCP addresses, :: binds IPv6 only; overlapping addresses result in undefined behavior (default: %s)", params.hostnames[0].c_str()), [](common_params & params, const std::string & value) { - params.hostname = value; + params.hostnames.clear(); + for (auto & host : parse_csv_row(value)) { + host = string_strip(host); + if (!host.empty()) { + params.hostnames.push_back(host); + } + } + if (params.hostnames.empty()) { + throw std::invalid_argument("--host requires at least one address"); + } } ).set_examples({LLAMA_EXAMPLE_SERVER}).set_env("LLAMA_ARG_HOST")); add_opt(common_arg( diff --git a/common/common.h b/common/common.h index 63d0badd0f..7afc266acd 100644 --- a/common/common.h +++ b/common/common.h @@ -631,10 +631,10 @@ struct common_params { int32_t checkpoint_min_step = 8192; // minimum spacing between context checkpoints int32_t cache_ram_mib = 8192; // -1 = no limit, 0 - disable, 1 = 1 MiB, etc. - std::string hostname = "127.0.0.1"; std::string public_path = ""; // NOLINT std::string api_prefix = ""; // NOLINT std::string chat_template = ""; // NOLINT + std::vector hostnames = {"127.0.0.1"}; bool use_jinja = true; // NOLINT // server CORS params diff --git a/tools/server/README.md b/tools/server/README.md index 0ee8df291b..e665904ba2 100644 --- a/tools/server/README.md +++ b/tools/server/README.md @@ -189,7 +189,7 @@ For the full list of features, please refer to [server's changelog](https://gith | `-a, --alias STRING` | set model name aliases, comma-separated (to be used by API)
(env: LLAMA_ARG_ALIAS) | | `--tags STRING` | set model tags, comma-separated (informational, not used for routing)
(env: LLAMA_ARG_TAGS) | | `--embd-normalize N` | normalisation for embeddings (default: 2) (-1=none, 0=max absolute int16, 1=taxicab, 2=euclidean, >2=p-norm) | -| `--host HOST` | ip address to listen, or bind to an UNIX socket if the address ends with .sock (default: 127.0.0.1)
(env: LLAMA_ARG_HOST) | +| `--host HOST` | IP addresses to listen on, comma-separated, or UNIX socket paths ending in .sock; with multiple TCP addresses, :: binds IPv6 only; overlapping addresses result in undefined behavior (default: 127.0.0.1)
(env: LLAMA_ARG_HOST) | | `--port PORT` | port to listen (default: 8080)
(env: LLAMA_ARG_PORT) | | `--reuse-port` | allow multiple sockets to bind to the same port (default: disabled)
(env: LLAMA_ARG_REUSE_PORT) | | `--path PATH` | path to serve static files from (default: )
(env: LLAMA_ARG_STATIC_PATH) | diff --git a/tools/server/server-http.cpp b/tools/server/server-http.cpp index 2ec137aa07..e46a1f6f10 100644 --- a/tools/server/server-http.cpp +++ b/tools/server/server-http.cpp @@ -18,14 +18,37 @@ class server_http_context::Impl { public: - std::unique_ptr srv; + std::vector> servers; + std::vector hosts; + std::vector threads; // one thread per listener + std::unique_ptr pool; // single pool shared among all listeners + int n_threads_http = 0; +}; + +class server_http_task_queue : public httplib::TaskQueue { + httplib::ThreadPool & pool; +public: + explicit server_http_task_queue(httplib::ThreadPool & pool) : pool(pool) {} + bool enqueue(std::function fn) override { return pool.enqueue(std::move(fn)); } + // note: must call join() to drain the pool + void shutdown() override { /* no-op */ } }; server_http_context::server_http_context() : pimpl(std::make_unique()) {} -server_http_context::~server_http_context() = default; +server_http_context::~server_http_context() { + // just in case any exit paths that forget to call join() + try { + stop(); + join(); + } catch (const std::exception & e) { + SRV_ERR("failed to stop HTTP server: %s\n", e.what()); + } catch (...) { + SRV_ERR("%s", "failed to stop HTTP server\n"); + } +} static void log_server_request(const httplib::Request & req, const httplib::Response & res) { // skip logging requests that are regularly sent, to avoid log spam @@ -90,7 +113,6 @@ bool server_http_context::init(const common_params & params) { path_prefix = params.api_prefix; port = params.port; - hostname = params.hostname; if (gcp.enabled) { SRV_TRC("Google Cloud Platform compat: health route = %s, predict route = %s, port = %d\n", gcp.path_health.c_str(), gcp.path_predict.c_str(), gcp.port); @@ -102,7 +124,39 @@ bool server_http_context::init(const common_params & params) { port = gcp.port; } - auto & srv = pimpl->srv; + pimpl->hosts = params.hostnames; + size_t n_tcp_hosts = 0; + for (const auto & host : pimpl->hosts) { + if (!string_ends_with(host, ".sock")) { + n_tcp_hosts++; + } + } + if (port == 0 && n_tcp_hosts > 1) { + SRV_ERR("%s", "--port 0 is not supported with multiple TCP addresses\n"); + return false; + } + for (size_t i = 0; i < pimpl->hosts.size(); ++i) { + pimpl->servers.emplace_back(); + if (!init_listener(params)) { + return false; + } + // with multiple TCP addresses, [::] must not also claim 0.0.0.0 + if (n_tcp_hosts > 1) { + pimpl->servers.back()->set_ipv6_v6only(true); + } + } + + pimpl->n_threads_http = params.n_threads_http; + if (pimpl->n_threads_http < 1) { + // +4 threads for monitoring, health and MCP. + pimpl->n_threads_http = std::max(params.n_parallel + 4, static_cast(std::thread::hardware_concurrency() - 1)); + } + SRV_TRC("using %d threads for HTTP server\n", pimpl->n_threads_http); + return true; +} + +bool server_http_context::init_listener(const common_params & params) { + auto & srv = pimpl->servers.back(); #ifdef CPPHTTPLIB_OPENSSL_SUPPORT if (!params.ssl_file_key.empty() && !params.ssl_file_cert.empty()) { @@ -306,18 +360,8 @@ bool server_http_context::init(const common_params & params) { return httplib::Server::HandlerResponse::Unhandled; }); - auto n_threads_http = params.n_threads_http; - if (n_threads_http < 1) { - // +4 threads for monitoring, health and some threads reserved for MCP and other tasks in the future - n_threads_http = std::max(params.n_parallel + 4, static_cast(std::thread::hardware_concurrency() - 1)); - } - SRV_TRC("using %d threads for HTTP server\n", n_threads_http); - srv->new_task_queue = [n_threads_http] { - // spawn n_threads_http fixed thread (always alive), while allow up to 1024 max possible additional threads - // when n_threads_http is used, server will create new "dynamic" threads that will be destroyed after processing each request - // ref: https://github.com/yhirose/cpp-httplib/pull/2368 - const auto max_threads = static_cast(n_threads_http + 1024); - return new httplib::ThreadPool(n_threads_http, max_threads); + srv->new_task_queue = [this] { + return new server_http_task_queue(*pimpl->pool); }; // @@ -432,47 +476,76 @@ bool server_http_context::init(const common_params & params) { bool server_http_context::start() { // Bind and listen - const auto & srv = pimpl->srv; - auto was_bound = false; - auto is_sock = false; - if (string_ends_with(std::string(hostname), ".sock")) { - is_sock = true; - SRV_TRC("%s", "setting address family to AF_UNIX\n"); - srv->set_address_family(AF_UNIX); - // bind_to_port requires a second arg, any value other than 0 should - // simply get ignored - was_bound = srv->bind_to_port(hostname, 8080); - } else { - SRV_TRC("%s", "binding port with default address family\n"); - // bind HTTP listen port - if (port == 0) { - const auto bound_port = srv->bind_to_any_port(hostname); - was_bound = (bound_port >= 0); + listening_addresses.clear(); + for (size_t i = 0; i < pimpl->servers.size(); ++i) { + const auto & srv = pimpl->servers[i]; + const auto & host = pimpl->hosts[i]; + const bool is_sock = string_ends_with(host, ".sock"); + bool was_bound; + if (is_sock) { + SRV_TRC("%s", "setting address family to AF_UNIX\n"); + srv->set_address_family(AF_UNIX); + // AF_UNIX ignores the port, but bind_to_port requires a nonzero value. + was_bound = srv->bind_to_port(host, 8080); + } else if (port == 0) { + const auto bound_port = srv->bind_to_any_port(host); + was_bound = bound_port >= 0; if (was_bound) { port = bound_port; } } else { - was_bound = srv->bind_to_port(hostname, port); + was_bound = srv->bind_to_port(host, port); + } + if (!was_bound) { + SRV_ERR("couldn't bind HTTP server socket, hostname: %s, port: %d\n", host.c_str(), port); + stop(); + listening_addresses.clear(); + return false; + } + listening_addresses.push_back(is_sock ? string_format("unix://%s", host.c_str()) + : string_format("%s://%s:%d", is_ssl ? "https" : "http", common_http_format_host(host).c_str(), port)); + } + + // n_threads_http fixed threads (always alive), plus up to 1024 dynamic threads destroyed after each request + // ref: https://github.com/yhirose/cpp-httplib/pull/2368 + pimpl->pool = std::make_unique(pimpl->n_threads_http, pimpl->n_threads_http + 1024); + for (size_t i = 0; i < pimpl->servers.size(); ++i) { + const auto & srv = pimpl->servers[i]; + pimpl->threads.emplace_back([srv = srv.get(), addr = listening_addresses[i]] { + if (!srv->listen_after_bind()) { + SRV_ERR("listener on %s stopped unexpectedly\n", addr.c_str()); + } + }); + srv->wait_until_ready(); + if (!srv->is_running()) { + SRV_ERR("couldn't start HTTP listener on %s\n", listening_addresses[i].c_str()); + stop(); + join(); + listening_addresses.clear(); + return false; } } - - if (!was_bound) { - SRV_ERR("couldn't bind HTTP server socket, hostname: %s, port: %d\n", hostname.c_str(), port); - return false; - } - - // run the HTTP server in a thread - thread = std::thread([this] { pimpl->srv->listen_after_bind(); }); - srv->wait_until_ready(); - - listening_address = is_sock ? string_format("unix://%s", hostname.c_str()) - : string_format("%s://%s:%d", is_ssl ? "https" : "http", common_http_format_host(hostname).c_str(), port); return true; } void server_http_context::stop() const { - if (pimpl->srv) { - pimpl->srv->stop(); + for (const auto & srv : pimpl->servers) { + if (srv) { + srv->stop(); + } + } +} + +void server_http_context::join() { + for (auto & thread : pimpl->threads) { + if (thread.joinable()) { + thread.join(); + } + } + // Queued requests still refer to their servers until the workers finish. + if (pimpl->pool) { + pimpl->pool->shutdown(); + pimpl->pool.reset(); } } @@ -584,7 +657,7 @@ static void process_handler_response(server_http_req_ptr && request, server_http void server_http_context::get(const std::string & path, const server_http_context::handler_t & handler) const { handlers.emplace(path, handler); - pimpl->srv->Get(path_prefix + path, [handler](const httplib::Request & req, httplib::Response & res) { + auto callback = [handler](const httplib::Request & req, httplib::Response & res) { server_http_req_ptr request = std::make_unique(server_http_req{ get_params(req), get_headers(req), @@ -596,12 +669,16 @@ void server_http_context::get(const std::string & path, const server_http_contex }); server_http_res_ptr response = handler(*request); process_handler_response(std::move(request), response, res); - }); + }; + const std::string full_path = path_prefix + path; + for (const auto & srv : pimpl->servers) { + srv->Get(full_path, callback); + } } void server_http_context::post(const std::string & path, const server_http_context::handler_t & handler) const { handlers.emplace(path, handler); - pimpl->srv->Post(path_prefix + path, [handler](const httplib::Request & req, httplib::Response & res) { + auto callback = [handler](const httplib::Request & req, httplib::Response & res) { std::string body = req.body; std::map files; @@ -643,12 +720,16 @@ void server_http_context::post(const std::string & path, const server_http_conte }); server_http_res_ptr response = handler(*request); process_handler_response(std::move(request), response, res); - }); + }; + const std::string full_path = path_prefix + path; + for (const auto & srv : pimpl->servers) { + srv->Post(full_path, callback); + } } void server_http_context::del(const std::string & path, const server_http_context::handler_t & handler) const { handlers.emplace(path, handler); - pimpl->srv->Delete(path_prefix + path, [handler](const httplib::Request & req, httplib::Response & res) { + auto callback = [handler](const httplib::Request & req, httplib::Response & res) { server_http_req_ptr request = std::make_unique(server_http_req{ get_params(req), get_headers(req), @@ -660,7 +741,11 @@ void server_http_context::del(const std::string & path, const server_http_contex }); server_http_res_ptr response = handler(*request); process_handler_response(std::move(request), response, res); - }); + }; + const std::string full_path = path_prefix + path; + for (const auto & srv : pimpl->servers) { + srv->Delete(full_path, callback); + } } // diff --git a/tools/server/server-http.h b/tools/server/server-http.h index 032b08d0d2..4554b20f4a 100644 --- a/tools/server/server-http.h +++ b/tools/server/server-http.h @@ -68,7 +68,6 @@ struct server_http_context { class Impl; std::unique_ptr pimpl; - std::thread thread; // server thread std::atomic is_ready = false; // note: the handler should never throw exceptions @@ -76,7 +75,6 @@ struct server_http_context { mutable std::unordered_map handlers; std::string path_prefix; - std::string hostname; int port = 8080; bool is_ssl = false; @@ -86,6 +84,7 @@ struct server_http_context { bool init(const common_params & params); bool start(); void stop() const; + void join(); void get(const std::string & path, const handler_t & handler) const; void post(const std::string & path, const handler_t & handler) const; @@ -96,5 +95,8 @@ struct server_http_context { void register_gcp_compat() const; // for debugging - std::string listening_address; + std::vector listening_addresses; + +private: + bool init_listener(const common_params & params); }; diff --git a/tools/server/server.cpp b/tools/server/server.cpp index 1167c0aea4..049bdcebb3 100644 --- a/tools/server/server.cpp +++ b/tools/server/server.cpp @@ -111,7 +111,9 @@ int llama_server(int argc, char ** argv) { llama_backend_init(); llama_numa_init(params.numa); - return llama_server(params, argc, argv); + const int result = llama_server(params, argc, argv); + common_log_flush(common_log_main()); + return result; } int llama_server(common_params & params, int argc, char ** argv) { @@ -183,12 +185,6 @@ int llama_server(common_params & params, int argc, char ** argv) { // struct that contains llama context and inference server_context ctx_server; - server_http_context ctx_http; - if (!ctx_http.init(params)) { - SRV_ERR("%s", "failed to initialize HTTP server\n"); - return 1; - } - // // Router // @@ -199,6 +195,13 @@ int llama_server(common_params & params, int argc, char ** argv) { server_tools tools; std::optional models_routes{}; + + server_http_context ctx_http; + if (!ctx_http.init(params)) { + SRV_ERR("%s", "failed to initialize HTTP server\n"); + return 1; + } + if (is_router_server) { // setup server instances manager try { @@ -438,9 +441,7 @@ int llama_server(common_params & params, int argc, char ** argv) { } catch (const std::exception & e) { SRV_ERR("failed to load models on startup: %s\n", e.what()); ctx_http.stop(); - if (ctx_http.thread.joinable()) { - ctx_http.thread.join(); - } + ctx_http.join(); clean_up(); return 1; } @@ -473,9 +474,7 @@ int llama_server(common_params & params, int argc, char ** argv) { if (!ctx_server.load_model(params)) { clean_up(); - if (ctx_http.thread.joinable()) { - ctx_http.thread.join(); - } + ctx_http.join(); SRV_ERR("%s", "exiting due to model loading error\n"); return 1; } @@ -509,11 +508,15 @@ int llama_server(common_params & params, int argc, char ** argv) { #endif } - SRV_INF("listening on %s\n", ctx_http.listening_address.c_str()); + bool uses_default_port = false; + for (const auto & address : ctx_http.listening_addresses) { + SRV_INF("listening on %s\n", address.c_str()); + uses_default_port |= string_ends_with(address, ":8080"); + } // TODO: remove this in the future // check the string to also handle the .sock case - if (string_ends_with(ctx_http.listening_address, ":8080")) { + if (uses_default_port) { SRV_WRN("%s", "notice: server default port will be changed to :9931 in a future release (ref: https://github.com/ggml-org/llama.cpp/pull/26508)\n"); } @@ -523,9 +526,7 @@ int llama_server(common_params & params, int argc, char ** argv) { SRV_WRN("%s", " please only use presets that you can trust! Unknown presets may be unsafe\n"); } - if (ctx_http.thread.joinable()) { - ctx_http.thread.join(); // keep the main thread alive - } + ctx_http.join(); // keep the main thread alive // when the HTTP server stops, clean up and exit clean_up(); @@ -541,9 +542,7 @@ int llama_server(common_params & params, int argc, char ** argv) { ctx_server.start_loop(); clean_up(); - if (ctx_http.thread.joinable()) { - ctx_http.thread.join(); - } + ctx_http.join(); if (monitor_thread.joinable()) { monitor_thread.join(); } diff --git a/tools/server/tests/unit/test_basic.py b/tools/server/tests/unit/test_basic.py index 285726abf4..b9e9f84f63 100644 --- a/tools/server/tests/unit/test_basic.py +++ b/tools/server/tests/unit/test_basic.py @@ -1,5 +1,6 @@ import pytest import requests +import socket from utils import * server = ServerPreset.tinyllama2() @@ -18,6 +19,37 @@ def test_server_start_simple(): assert res.status_code == 200 +def test_server_multiple_addresses(monkeypatch): + # The CLI value replaces the environment value, including an unavailable address. + monkeypatch.setenv("LLAMA_ARG_HOST", "192.0.2.1") + try: + with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as probe: + probe.bind(("::1", 0)) + except OSError: + pytest.skip("IPv6 loopback is unavailable") # ty: ignore[too-many-positional-arguments] + + server.server_host = "127.0.0.1,::1" + server.api_key = "test-multiple-addresses" + server.start() + + def check_address(host): + res = server.make_request("GET", "/health", host=host) + assert res.status_code == 200 + res = server.make_request("POST", "/v1/completions", data={}, host=host) + assert res.status_code == 401 + events = list(server.make_stream_request("POST", "/v1/completions", data={ + "prompt": "Once upon a time", + "max_tokens": 8, + "stream": True, + }, headers={"Authorization": f"Bearer {server.api_key}"}, host=host)) + assert len(events) > 1 + return True + + # parallel_function_calls swallows exceptions, a failed check leaves None in the results + results = parallel_function_calls([(check_address, (host,)) for host in ["127.0.0.1", "[::1]"]]) + assert all(results) + + def test_server_props(): global server server.start() diff --git a/tools/server/tests/utils.py b/tools/server/tests/utils.py index 826aef2d5b..3a50ae5c3f 100644 --- a/tools/server/tests/utils.py +++ b/tools/server/tests/utils.py @@ -155,8 +155,6 @@ class ServerProcess: else: server_path = "../../../build/bin/llama-server" server_args = [ - "--host", - self.server_host, "--port", self.server_port, "--temp", @@ -164,6 +162,7 @@ class ServerProcess: "--seed", self.seed, ] + server_args.extend(["--host", self.server_host]) if self.offline: server_args.append("--offline") if self.model_file: @@ -365,6 +364,11 @@ class ServerProcess: if hasattr(self, '_log') and self._log != sys.stdout: self._log.close() + def make_url(self, path: str, host: str | None = None) -> str: + if host is None: + host = self.server_host.split(",")[0].strip() + return f"http://{host}:{self.server_port}{path}" + def make_request( self, method: str, @@ -372,8 +376,9 @@ class ServerProcess: data: dict | Any | None = None, headers: dict | None = None, timeout: float | None = DEFAULT_REQUEST_TIMEOUT, + host: str | None = None, ) -> ServerResponse: - url = f"http://{self.server_host}:{self.server_port}{path}" + url = self.make_url(path, host) parse_body = False if method == "GET": response = requests.get(url, headers=headers, timeout=timeout) @@ -407,8 +412,9 @@ class ServerProcess: path: str, data: dict | None = None, headers: dict | None = None, + host: str | None = None, ) -> Iterator[dict]: - url = f"http://{self.server_host}:{self.server_port}{path}" + url = self.make_url(path, host) if method == "POST": response = requests.post(url, headers=headers, json=data, stream=True) else: