Skip to content

Commit 9682e35

Browse files
ngxsonNoeda
andauthored
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>
1 parent 1e91256 commit 9682e35

1 file changed

Lines changed: 66 additions & 47 deletions

File tree

tools/mtmd/mtmd-helper.cpp

Lines changed: 66 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -617,10 +617,52 @@ struct mtmd_helper_video {
617617
float fps_target = 0.0f;
618618
mtmd_helper_video_info info = {};
619619

620-
struct subprocess_s proc = {};
621-
bool proc_alive = false;
620+
// RAII wrapper for managing subprocess
621+
struct subprocess_handle {
622+
struct subprocess_s proc = {};
623+
bool alive = false;
624+
std::thread feeder;
625+
626+
subprocess_handle() = default;
627+
subprocess_handle(const subprocess_handle &) = delete;
628+
subprocess_handle & operator=(const subprocess_handle &) = delete;
629+
~subprocess_handle() { stop(); }
630+
631+
void stop() {
632+
if (alive) {
633+
subprocess_terminate(&proc);
634+
}
635+
// join before destroy: feeder holds a FILE* from subprocess_stdin;
636+
// subprocess_destroy closes it, so the thread must finish first
637+
if (feeder.joinable()) {
638+
feeder.join();
639+
}
640+
if (alive) {
641+
subprocess_destroy(&proc);
642+
alive = false;
643+
}
644+
}
645+
646+
FILE * stdout_pipe() {
647+
return subprocess_stdout(&proc);
648+
}
649+
650+
// buf is tied to lifetime of mtmd_helper_video, so it's guaranteed to outlive the feeder thread
651+
void start_feeder(const std::vector<uint8_t> & buf) {
652+
feeder = std::thread([this, &buf]() {
653+
FILE * f = subprocess_stdin(&proc);
654+
if (!f) {
655+
return;
656+
}
657+
fwrite(buf.data(), 1, buf.size(), f);
658+
fclose(f);
659+
proc.stdin_file = nullptr; // prevent double-close in subprocess_destroy
660+
});
661+
}
662+
};
663+
664+
subprocess_handle sp;
622665
int32_t current_frame = 0;
623-
std::thread feeder_thread;
624666

625667
std::string prompt_start = "Video:";
626668
int32_t timestamp_interval_ms = 5000; // emit a timestamp text every N ms (0 = disabled)
@@ -630,19 +672,8 @@ struct mtmd_helper_video {
630672
std::string pending_text; // text queued to be returned before the next frame
631673
bool start_emitted = false;
632674

633-
bool is_buf_input() const { return !input_buf.empty(); }
634-
635-
// must run in a separate thread alongside stdout reading to avoid pipe deadlock
636-
void feed_stdin(struct subprocess_s * sp) {
637-
FILE * f = subprocess_stdin(sp);
638-
if (!f) {
639-
LOG_DBG("%s: subprocess has no stdin pipe\n", __func__);
640-
return;
641-
}
642-
LOG_DBG("%s: feeding %zu bytes to stdin\n", __func__, input_buf.size());
643-
size_t written = fwrite(input_buf.data(), 1, input_buf.size(), f);
644-
LOG_DBG("%s: wrote %zu bytes, closing stdin\n", __func__, written);
645-
fclose(f);
675+
bool is_buf_input() const {
676+
return !input_buf.empty();
646677
}
647678

648679
bool probe(float fps_target_arg) {
@@ -661,17 +692,17 @@ struct mtmd_helper_video {
661692
for (size_t i = 0; cmd[i]; i++) { LOG_DBG(" %s", cmd[i]); }
662693
LOG_DBG("\n");
663694

664-
struct subprocess_s fprobe;
695+
subprocess_handle probe_sp;
665696
if (subprocess_create(cmd,
666697
subprocess_option_search_user_path | subprocess_option_inherit_environment,
667-
&fprobe) != 0) {
698+
&probe_sp.proc) != 0) {
668699
LOG_ERR("%s: failed to launch ffprobe\n", __func__);
669700
return false;
670701
}
702+
probe_sp.alive = true;
671703

672-
std::thread probe_feeder;
673704
if (is_buf_input()) {
674-
probe_feeder = std::thread([this, &fprobe]() { feed_stdin(&fprobe); });
705+
probe_sp.start_feeder(input_buf);
675706
}
676707

677708
uint32_t width = 0;
@@ -680,7 +711,7 @@ struct mtmd_helper_video {
680711
float duration = -1.0f;
681712
int32_t n_frames_orig = -1;
682713
char line[256];
683-
FILE * fp = subprocess_stdout(&fprobe);
714+
FILE * fp = probe_sp.stdout_pipe();
684715

685716
while (fgets(line, sizeof(line), fp)) {
686717
char * eq = strchr(line, '=');
@@ -704,13 +735,7 @@ struct mtmd_helper_video {
704735
}
705736
}
706737

707-
if (probe_feeder.joinable()) {
708-
probe_feeder.join();
709-
}
710-
711-
int ret_code;
712-
subprocess_join(&fprobe, &ret_code);
713-
subprocess_destroy(&fprobe);
738+
probe_sp.stop();
714739

715740
if (width == 0 || height == 0 || orig_fps <= 0.0f) {
716741
return false;
@@ -745,6 +770,7 @@ struct mtmd_helper_video {
745770
cmd.push_back(seek_buf);
746771
}
747772

773+
cmd.push_back("-nostdin");
748774
cmd.push_back("-i");
749775
// cache:pipe:0 wraps stdin with a seekable in-memory cache, letting ffmpeg seek
750776
// backwards for container headers (e.g. MP4 moov atom at end of file)
@@ -781,34 +807,27 @@ struct mtmd_helper_video {
781807
int ret = subprocess_create(
782808
cmd.data(),
783809
subprocess_option_search_user_path | subprocess_option_inherit_environment,
784-
&proc);
810+
&sp.proc);
785811

786-
proc_alive = (ret == 0);
787-
LOG_DBG("%s: subprocess_create ret=%d proc_alive=%d\n", __func__, ret, (int)proc_alive);
812+
sp.alive = (ret == 0);
813+
LOG_DBG("%s: subprocess_create ret=%d proc_alive=%d\n", __func__, ret, (int)sp.alive);
788814

789-
if (proc_alive && is_buf_input()) {
815+
if (sp.alive && is_buf_input()) {
790816
LOG_DBG("%s: starting feeder thread for %zu-byte buffer\n", __func__, input_buf.size());
791-
feeder_thread = std::thread([this]() { feed_stdin(&proc); });
817+
sp.start_feeder(input_buf);
792818
}
793819

794-
return proc_alive;
820+
return sp.alive;
795821
}
796822

797823
void stop_ffmpeg() {
798-
if (proc_alive) {
799-
subprocess_terminate(&proc);
800-
subprocess_destroy(&proc);
801-
proc_alive = false;
802-
}
803-
if (feeder_thread.joinable()) {
804-
feeder_thread.join();
805-
}
824+
sp.stop();
806825
}
807826

808827
mtmd_bitmap * read_next_frame() {
809-
if (!proc_alive) return nullptr;
828+
if (!sp.alive) return nullptr;
810829

811-
FILE * fp = subprocess_stdout(&proc);
830+
FILE * fp = sp.stdout_pipe();
812831
const size_t frame_size = (size_t)info.width * info.height * 3;
813832
LOG_DBG("%s: reading frame %d, expecting %zu bytes (%ux%u)\n",
814833
__func__, current_frame, frame_size, info.width, info.height);
@@ -820,7 +839,7 @@ struct mtmd_helper_video {
820839
// clean EOF only if no bytes read yet; partial frame is an error
821840
LOG_DBG("%s: fread returned 0 after %zu/%zu bytes (ferror=%d)\n",
822841
__func__, total_read, frame_size, ferror(fp));
823-
proc_alive = false;
842+
sp.alive = false;
824843
return nullptr;
825844
}
826845
total_read += n;
@@ -842,9 +861,9 @@ struct mtmd_helper_video {
842861
}
843862

844863
LOG_DBG("%s: proc_alive=%d start_emitted=%d current_frame=%d\n",
845-
__func__, (int)proc_alive, (int)start_emitted, current_frame);
864+
__func__, (int)sp.alive, (int)start_emitted, current_frame);
846865

847-
if (!proc_alive) {
866+
if (!sp.alive) {
848867
return (current_frame == 0) ? -2 : -1;
849868
}
850869

0 commit comments

Comments
 (0)