mtmd: refactor video subproc handling (#24316)

* mtmd: refactor video subproc handling

* Update tools/mtmd/mtmd-helper.cpp

Co-authored-by: Mikko Juola <mikjuo@gmail.com>

---------

Co-authored-by: Mikko Juola <mikjuo@gmail.com>
This commit is contained in:
Xuan-Son Nguyen
2026-06-09 13:15:12 +03:00
committed by GitHub
co-authored by Mikko Juola
parent 1e912561dd
commit 9682e351b8
+65 -46
View File
@@ -617,10 +617,52 @@ struct mtmd_helper_video {
float fps_target = 0.0f; float fps_target = 0.0f;
mtmd_helper_video_info info = {}; mtmd_helper_video_info info = {};
// RAII wrapper for managing subprocess
struct subprocess_handle {
struct subprocess_s proc = {}; struct subprocess_s proc = {};
bool proc_alive = false; bool alive = false;
std::thread feeder;
subprocess_handle() = default;
subprocess_handle(const subprocess_handle &) = delete;
subprocess_handle & operator=(const subprocess_handle &) = delete;
~subprocess_handle() { stop(); }
void stop() {
if (alive) {
subprocess_terminate(&proc);
}
// join before destroy: feeder holds a FILE* from subprocess_stdin;
// subprocess_destroy closes it, so the thread must finish first
if (feeder.joinable()) {
feeder.join();
}
if (alive) {
subprocess_destroy(&proc);
alive = false;
}
}
FILE * stdout_pipe() {
return subprocess_stdout(&proc);
}
// buf is tied to lifetime of mtmd_helper_video, so it's guaranteed to outlive the feeder thread
void start_feeder(const std::vector<uint8_t> & buf) {
feeder = std::thread([this, &buf]() {
FILE * f = subprocess_stdin(&proc);
if (!f) {
return;
}
fwrite(buf.data(), 1, buf.size(), f);
fclose(f);
proc.stdin_file = nullptr; // prevent double-close in subprocess_destroy
});
}
};
subprocess_handle sp;
int32_t current_frame = 0; int32_t current_frame = 0;
std::thread feeder_thread;
std::string prompt_start = "Video:"; std::string prompt_start = "Video:";
int32_t timestamp_interval_ms = 5000; // emit a timestamp text every N ms (0 = disabled) int32_t timestamp_interval_ms = 5000; // emit a timestamp text every N ms (0 = disabled)
@@ -630,19 +672,8 @@ struct mtmd_helper_video {
std::string pending_text; // text queued to be returned before the next frame std::string pending_text; // text queued to be returned before the next frame
bool start_emitted = false; bool start_emitted = false;
bool is_buf_input() const { return !input_buf.empty(); } bool is_buf_input() const {
return !input_buf.empty();
// must run in a separate thread alongside stdout reading to avoid pipe deadlock
void feed_stdin(struct subprocess_s * sp) {
FILE * f = subprocess_stdin(sp);
if (!f) {
LOG_DBG("%s: subprocess has no stdin pipe\n", __func__);
return;
}
LOG_DBG("%s: feeding %zu bytes to stdin\n", __func__, input_buf.size());
size_t written = fwrite(input_buf.data(), 1, input_buf.size(), f);
LOG_DBG("%s: wrote %zu bytes, closing stdin\n", __func__, written);
fclose(f);
} }
bool probe(float fps_target_arg) { bool probe(float fps_target_arg) {
@@ -661,17 +692,17 @@ struct mtmd_helper_video {
for (size_t i = 0; cmd[i]; i++) { LOG_DBG(" %s", cmd[i]); } for (size_t i = 0; cmd[i]; i++) { LOG_DBG(" %s", cmd[i]); }
LOG_DBG("\n"); LOG_DBG("\n");
struct subprocess_s fprobe; subprocess_handle probe_sp;
if (subprocess_create(cmd, if (subprocess_create(cmd,
subprocess_option_search_user_path | subprocess_option_inherit_environment, subprocess_option_search_user_path | subprocess_option_inherit_environment,
&fprobe) != 0) { &probe_sp.proc) != 0) {
LOG_ERR("%s: failed to launch ffprobe\n", __func__); LOG_ERR("%s: failed to launch ffprobe\n", __func__);
return false; return false;
} }
probe_sp.alive = true;
std::thread probe_feeder;
if (is_buf_input()) { if (is_buf_input()) {
probe_feeder = std::thread([this, &fprobe]() { feed_stdin(&fprobe); }); probe_sp.start_feeder(input_buf);
} }
uint32_t width = 0; uint32_t width = 0;
@@ -680,7 +711,7 @@ struct mtmd_helper_video {
float duration = -1.0f; float duration = -1.0f;
int32_t n_frames_orig = -1; int32_t n_frames_orig = -1;
char line[256]; char line[256];
FILE * fp = subprocess_stdout(&fprobe); FILE * fp = probe_sp.stdout_pipe();
while (fgets(line, sizeof(line), fp)) { while (fgets(line, sizeof(line), fp)) {
char * eq = strchr(line, '='); char * eq = strchr(line, '=');
@@ -704,13 +735,7 @@ struct mtmd_helper_video {
} }
} }
if (probe_feeder.joinable()) { probe_sp.stop();
probe_feeder.join();
}
int ret_code;
subprocess_join(&fprobe, &ret_code);
subprocess_destroy(&fprobe);
if (width == 0 || height == 0 || orig_fps <= 0.0f) { if (width == 0 || height == 0 || orig_fps <= 0.0f) {
return false; return false;
@@ -745,6 +770,7 @@ struct mtmd_helper_video {
cmd.push_back(seek_buf); cmd.push_back(seek_buf);
} }
cmd.push_back("-nostdin");
cmd.push_back("-i"); cmd.push_back("-i");
// cache:pipe:0 wraps stdin with a seekable in-memory cache, letting ffmpeg seek // cache:pipe:0 wraps stdin with a seekable in-memory cache, letting ffmpeg seek
// backwards for container headers (e.g. MP4 moov atom at end of file) // backwards for container headers (e.g. MP4 moov atom at end of file)
@@ -781,34 +807,27 @@ struct mtmd_helper_video {
int ret = subprocess_create( int ret = subprocess_create(
cmd.data(), cmd.data(),
subprocess_option_search_user_path | subprocess_option_inherit_environment, subprocess_option_search_user_path | subprocess_option_inherit_environment,
&proc); &sp.proc);
proc_alive = (ret == 0); sp.alive = (ret == 0);
LOG_DBG("%s: subprocess_create ret=%d proc_alive=%d\n", __func__, ret, (int)proc_alive); LOG_DBG("%s: subprocess_create ret=%d proc_alive=%d\n", __func__, ret, (int)sp.alive);
if (proc_alive && is_buf_input()) { if (sp.alive && is_buf_input()) {
LOG_DBG("%s: starting feeder thread for %zu-byte buffer\n", __func__, input_buf.size()); LOG_DBG("%s: starting feeder thread for %zu-byte buffer\n", __func__, input_buf.size());
feeder_thread = std::thread([this]() { feed_stdin(&proc); }); sp.start_feeder(input_buf);
} }
return proc_alive; return sp.alive;
} }
void stop_ffmpeg() { void stop_ffmpeg() {
if (proc_alive) { sp.stop();
subprocess_terminate(&proc);
subprocess_destroy(&proc);
proc_alive = false;
}
if (feeder_thread.joinable()) {
feeder_thread.join();
}
} }
mtmd_bitmap * read_next_frame() { mtmd_bitmap * read_next_frame() {
if (!proc_alive) return nullptr; if (!sp.alive) return nullptr;
FILE * fp = subprocess_stdout(&proc); FILE * fp = sp.stdout_pipe();
const size_t frame_size = (size_t)info.width * info.height * 3; const size_t frame_size = (size_t)info.width * info.height * 3;
LOG_DBG("%s: reading frame %d, expecting %zu bytes (%ux%u)\n", LOG_DBG("%s: reading frame %d, expecting %zu bytes (%ux%u)\n",
__func__, current_frame, frame_size, info.width, info.height); __func__, current_frame, frame_size, info.width, info.height);
@@ -820,7 +839,7 @@ struct mtmd_helper_video {
// clean EOF only if no bytes read yet; partial frame is an error // clean EOF only if no bytes read yet; partial frame is an error
LOG_DBG("%s: fread returned 0 after %zu/%zu bytes (ferror=%d)\n", LOG_DBG("%s: fread returned 0 after %zu/%zu bytes (ferror=%d)\n",
__func__, total_read, frame_size, ferror(fp)); __func__, total_read, frame_size, ferror(fp));
proc_alive = false; sp.alive = false;
return nullptr; return nullptr;
} }
total_read += n; total_read += n;
@@ -842,9 +861,9 @@ struct mtmd_helper_video {
} }
LOG_DBG("%s: proc_alive=%d start_emitted=%d current_frame=%d\n", LOG_DBG("%s: proc_alive=%d start_emitted=%d current_frame=%d\n",
__func__, (int)proc_alive, (int)start_emitted, current_frame); __func__, (int)sp.alive, (int)start_emitted, current_frame);
if (!proc_alive) { if (!sp.alive) {
return (current_frame == 0) ? -2 : -1; return (current_frame == 0) ? -2 : -1;
} }