diff --git a/tools/server/server-models.cpp b/tools/server/server-models.cpp index 2fd9519c10..b4ef05b926 100644 --- a/tools/server/server-models.cpp +++ b/tools/server/server-models.cpp @@ -721,7 +721,9 @@ void server_models::unload_lru() { for (const auto & m : mapping) { if (m.second.meta.is_running()) { count_active++; - if (m.second.meta.last_used < lru_last_used) { + // do not evict busy one + bool is_model_idle = m.second.req_count == 0 && m.second.meta.is_ready_or_sleep(); + if (is_model_idle && m.second.meta.last_used < lru_last_used) { lru_model_name = m.first; lru_last_used = m.second.meta.last_used; } @@ -739,6 +741,7 @@ void server_models::unload_lru() { }); } } + // TODO @ngxson : if no idle model is found, queue the load request } void server_models::load(const std::string & name) { @@ -1180,9 +1183,12 @@ server_http_res_ptr server_models::proxy_request(const server_http_req & req, co if (!meta->is_running()) { throw std::invalid_argument("model name=" + name + " is not running"); } - if (update_last_used) { + { std::unique_lock lk(mutex); - mapping[name].meta.last_used = ggml_time_ms(); + if (update_last_used) { + mapping[name].meta.last_used = ggml_time_ms(); + } + mapping[name].req_count++; } SRV_INF("proxying request to model %s on port %d\n", name.c_str(), meta->port); std::string proxy_path = req.path; @@ -1198,13 +1204,22 @@ server_http_res_ptr server_models::proxy_request(const server_http_req & req, co req.headers, req.body, req.files, - // a detached request belongs to a replay session that outlives the client socket: - // it reaches the child even when the downstream died during the load wait, the - // session buffer is the recipient and DELETE remains the stop - detached ? std::function([]() { return false; }) : req.should_stop, + // a detached request belongs to a replay session + detached + ? std::function([]() { return false; }) + : req.should_stop, base_params.timeout_read, base_params.timeout_write ); + + proxy->cleanup = [this, name]() { + std::unique_lock lk(mutex); + auto it = mapping.find(name); + if (it != mapping.end() && it->second.req_count > 0) { + it->second.req_count--; + } + }; + return proxy; } @@ -2064,7 +2079,7 @@ server_http_proxy::server_http_proxy( cli->set_write_timeout(timeout_read, 0); // reversed for cli (client) vs srv (server) cli->set_read_timeout(timeout_write, 0); this->status = 500; // to be overwritten upon response - this->cleanup = [pipe]() { + this->cleanup_pipes = [pipe]() { pipe->close_read(); pipe->close_write(); }; diff --git a/tools/server/server-models.h b/tools/server/server-models.h index 614798186c..1c6123cb1e 100644 --- a/tools/server/server-models.h +++ b/tools/server/server-models.h @@ -84,7 +84,6 @@ struct server_model_meta { int exit_code = 0; // exit code of the model instance process (only valid if status == FAILED) int stop_timeout = 0; // seconds to wait before force-killing the model instance during shutdown mtmd_caps multimodal; // multimodal capabilities - // bool need_download = false; // whether the model needs to be downloaded before loading // TODO @ngxson: implement this bool is_ready() const { return status == SERVER_MODEL_STATUS_LOADED; @@ -94,6 +93,10 @@ struct server_model_meta { return status == SERVER_MODEL_STATUS_LOADED || status == SERVER_MODEL_STATUS_LOADING || status == SERVER_MODEL_STATUS_SLEEPING; } + bool is_ready_or_sleep() const { + return status == SERVER_MODEL_STATUS_LOADED || status == SERVER_MODEL_STATUS_SLEEPING; + } + bool is_failed() const { return status == SERVER_MODEL_STATUS_UNLOADED && exit_code != 0; } @@ -113,6 +116,7 @@ private: std::shared_ptr subproc; // shared between main thread and monitoring thread std::thread th; server_model_meta meta; + int req_count = 0; // number of active proxy requests }; std::mutex mutex; @@ -343,7 +347,6 @@ struct server_models_routes { */ struct server_http_proxy : server_http_res { std::function cleanup = nullptr; -public: server_http_proxy(const std::string & method, const std::string & scheme, const std::string & host, @@ -357,11 +360,15 @@ public: int32_t timeout_write ); ~server_http_proxy() { + if (cleanup_pipes) { + cleanup_pipes(); + } if (cleanup) { cleanup(); } } private: + std::function cleanup_pipes = nullptr; std::thread thread; struct msg_t { std::map headers;