server: (router) do not evict busy models (#26567)

The reference implementation has two mutually exclusive prompt layouts.
In non streaming mode the prefill carries the whole utterance text plus
tts_eos summed with codec_pad, and the trailing text hidden collapses to
a single tts_pad row. In streaming mode the prefill carries only the
first text token and the trailing rows stream the rest of the text
followed by tts_eos.

The pipeline built the non streaming prefill but the streaming overlay,
so the talker saw the utterance a second time during generation and read
it twice before emitting codec_eos.

The overlay is now the single tts_pad row that matches the prefill.

tools/mtmd/mtmd-helper-gen.cpp

@@ -112,7 +112,6 @@ public:
         c2w_state.clear();
         audio_pcm.clear();
         overlay.clear();
-        overlay_idx = 0;
         h_state_buf.clear();
         out_buf.clear();
         prompt_embd_buf.clear();
@@ -205,11 +204,9 @@ public:
         top_p = inp->top_p > 0 ? inp->top_p : 1.0f;
         out_type = inp->out_type;
 
-        // the text stream keeps flowing during generation: after frame k, the input adds
-        // trailing text row k on top of the codes embedding, then tts_eos, then tts_pad
-        for (int i = 3; i < n_ids - 5; i++) overlay.push_back(row(ids[(size_t) i]));
-        overlay.push_back(row(tts_eos));
-        overlay.push_back(row(tts_pad));
+        // the prompt above holds the whole text stream up to tts_eos, so every generated
+        // frame adds tts_pad on top of the codes embedding
+        overlay = row(tts_pad);
 
         return 0;
     }
@@ -265,9 +262,7 @@ public:
         }
 
         std::vector<float> fb(out.embd, out.embd + n_embd);
-        const auto & ov = overlay[std::min(overlay_idx, overlay.size() - 1)];
-        for (int i = 0; i < n_embd; i++) fb[(size_t) i] += ov[(size_t) i];
-        overlay_idx++;
+        for (int i = 0; i < n_embd; i++) fb[(size_t) i] += overlay[(size_t) i];
 
         const int n_pos_per_embd = mrope ? 4 : 1;
         decode_embd_batch batch_embd(fb.data(), 1, n_pos_per_embd, n_embd);
@@ -437,8 +432,7 @@ private:
     std::vector<int32_t> codes_buf;
     std::vector<uint8_t> c2w_state;
     std::vector<float>   audio_pcm;
-    std::vector<std::vector<float>> overlay;
-    size_t overlay_idx = 0;
+    std::vector<float> overlay;
     std::vector<float> h_state_buf;
     mtmd_helper_gen_audio_outtype out_type = MTMD_HELPER_GEN_AUDIO_OUTTYPE_WAV;
     std::vector<char> out_buf;

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<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;
@@ -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<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;
 }
 
@@ -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();
     };

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<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;
@@ -343,7 +347,6 @@ struct server_models_routes {
  */
 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,
@@ -357,11 +360,15 @@ public:
                       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;