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;
}
});
}
}
+ // TODO @ngxson : if no idle model is found, queue the load request
}
void server_models::load(const std::string & name) {
if (!meta->is_running()) {
throw std::invalid_argument("model name=" + name + " is not running");
}
- if (update_last_used) {
+ {
std::unique_lock<std::mutex> 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;
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<bool()>([]() { return false; }) : req.should_stop,
+ // a detached request belongs to a replay session
+ detached
+ ? std::function<bool()>([]() { return false; })
+ : req.should_stop,
base_params.timeout_read,
base_params.timeout_write
);
+
+ proxy->cleanup = [this, name]() {
+ std::unique_lock<std::mutex> lk(mutex);
+ auto it = mapping.find(name);
+ if (it != mapping.end() && it->second.req_count > 0) {
+ it->second.req_count--;
+ }
+ };
+
return 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();
};
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;
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;
}
std::shared_ptr<server_subproc> 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;
*/
struct server_http_proxy : server_http_res {
std::function<void()> cleanup = nullptr;
-public:
server_http_proxy(const std::string & method,
const std::string & scheme,
const std::string & host,
int32_t timeout_write
);
~server_http_proxy() {
+ if (cleanup_pipes) {
+ cleanup_pipes();
+ }
if (cleanup) {
cleanup();
}
}
private:
+ std::function<void()> cleanup_pipes = nullptr;
std::thread thread;
struct msg_t {
std::map<std::string, std::string> headers;