diff --git a/tools/mtmd/mtmd.cpp b/tools/mtmd/mtmd.cpp index 9a1663015f..7c825c0dc4 100644 --- a/tools/mtmd/mtmd.cpp +++ b/tools/mtmd/mtmd.cpp @@ -3096,7 +3096,8 @@ struct mtmd_audio_stream { std::vector out; // rows of the last step std::vector tok_embd; - int64_t n_rows = 0; // rows produced so far + int64_t n_rows = 0; // rows produced so far + int64_t n_rows_flushed = 0; // rows produced once the audio up to the last flush is consumed bool pending_flush = false; bool ended = false; @@ -3209,6 +3210,8 @@ int32_t mtmd_audio_stream_flush(mtmd_audio_stream * s, bool end) { // align to a whole row, then add the silence the model expects after the audio const int64_t n_align = (s->samples_per_tok - s->n_samples % s->samples_per_tok) % s->samples_per_tok; mtmd_audio_stream_add_silence(s, n_align + (int64_t) s->n_right_pad * s->samples_per_tok); + // one row per samples_per_tok, the left padding included + s->n_rows_flushed = s->n_samples / s->samples_per_tok; s->pending_flush = true; s->ended = end; return 0; @@ -3358,3 +3361,11 @@ int32_t mtmd_audio_stream_step(mtmd_audio_stream * s, llama_token last_token) { const float * mtmd_audio_stream_get_output_embd(const mtmd_audio_stream * s) { return s->out.data(); } + +int64_t mtmd_audio_stream_get_n_rows(const mtmd_audio_stream * s) { + return s->n_rows; +} + +int64_t mtmd_audio_stream_get_n_rows_flushed(const mtmd_audio_stream * s) { + return s->n_rows_flushed; +} diff --git a/tools/mtmd/mtmd.h b/tools/mtmd/mtmd.h index f248ed7390..5af73a0f40 100644 --- a/tools/mtmd/mtmd.h +++ b/tools/mtmd/mtmd.h @@ -412,6 +412,13 @@ MTMD_API int32_t mtmd_audio_stream_step(mtmd_audio_stream * stream, llama_token // rows of the last step, the reading size (in bytes) is n_rows * llama_model_n_embd_inp(model) * sizeof(float) MTMD_API const float * mtmd_audio_stream_get_output_embd(const mtmd_audio_stream * stream); +// rows produced so far, the prefix rows included +MTMD_API int64_t mtmd_audio_stream_get_n_rows(const mtmd_audio_stream * stream); + +// rows the stream produces up to the end of the last flush +// once get_n_rows() reaches it, every row of the flushed audio has been produced +MTMD_API int64_t mtmd_audio_stream_get_n_rows_flushed(const mtmd_audio_stream * stream); + ///////////////////////////////////////// // EXPERIMENTAL API for audio generation, subjected to breaking changes diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 9c7894574e..a5cba1d71d 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -25,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -399,7 +400,7 @@ struct server_slot { // realtime transcription, see SERVER_TASK_TYPE_TRANSCRIBE_RT mtmd::audio_stream_ptr audio_stream; bool audio_wait = false; // the stream needs more audio before it can produce rows - bool audio_commit_pending = false; // a non-final commit's tail is still draining + std::deque audio_commit_rows; // end row of each non-final commit not yet completed server_slot_stats stats; @@ -457,7 +458,7 @@ struct server_slot { audio_stream.reset(); audio_wait = false; - audio_commit_pending = false; + audio_commit_rows.clear(); inp_embd.clear(); } @@ -3073,9 +3074,11 @@ private: if (task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_APPEND) { mtmd_audio_stream_push(stream, task.params.control_audio.data(), task.params.control_audio.size()); } else { - mtmd_audio_stream_flush(stream, task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_END); - // a non-final commit completes its item only once the flushed tail drains - slot->audio_commit_pending = task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_COMMIT; + const int32_t ret = mtmd_audio_stream_flush(stream, task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_END); + // a non-final commit completes its item once the rows of its audio are decoded + if (ret == 0 && task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_COMMIT) { + slot->audio_commit_rows.push_back(mtmd_audio_stream_get_n_rows_flushed(stream)); + } } slot->audio_wait = false; res->success = true; @@ -3406,6 +3409,22 @@ private: return true; // rows from a previous step, not decoded yet } + // every produced row is decoded and its delta queued, the commits ending there are complete + const int64_t n_rows = mtmd_audio_stream_get_n_rows(slot.audio_stream.get()); + while (!slot.audio_commit_rows.empty() && slot.audio_commit_rows.front() <= n_rows) { + slot.audio_commit_rows.pop_front(); + auto res = std::make_unique(); + res->id = slot.task->id; + res->index = slot.task->index; + res->n_decoded = slot.stats.n_gen; + res->n_prompt_tokens = slot.task->n_tokens(); + res->res_type = slot.task->params.res_type; + res->oaicompat_model = slot.task->params.oaicompat_model; + res->oaicompat_cmpl_id = slot.task->params.oaicompat_cmpl_id; + res->audio_commit_drained = true; + queue_results.send(std::move(res)); + } + const int32_t n = mtmd_audio_stream_step(slot.audio_stream.get(), slot.sampled); if (n > 0) { const float * rows = mtmd_audio_stream_get_output_embd(slot.audio_stream.get()); @@ -3415,20 +3434,6 @@ private: } if (n == 0) { slot.audio_wait = true; - if (slot.audio_commit_pending) { - // committed audio fully decoded; its deltas are already queued ahead of this marker - slot.audio_commit_pending = false; - auto res = std::make_unique(); - res->id = slot.task->id; - res->index = slot.task->index; - res->n_decoded = slot.stats.n_gen; - res->n_prompt_tokens = slot.task->n_tokens(); - res->res_type = slot.task->params.res_type; - res->oaicompat_model = slot.task->params.oaicompat_model; - res->oaicompat_cmpl_id = slot.task->params.oaicompat_cmpl_id; - res->audio_commit_drained = true; - queue_results.send(std::move(res)); - } return false; } if (n == -2) { @@ -6129,7 +6134,7 @@ void server_routes::init_routes() { if (!post_control(SERVER_CONTROL_ACTION_AUDIO_COMMIT, {})) { break; } - // completed is deferred until the flushed tail drains (audio_commit_drained below) + // completed is sent once the rows of the committed audio are decoded (audio_commit_drained below) std::lock_guard lock(mutex_item); send_event(json { {"type", "input_audio_buffer.committed"},