Skip to content

Commit 8426177

Browse files
Merge pull request #2643 from CapSoftware/fix/oop-muxer-flush-packets
fix(recording): flush every packet to the out-of-process muxer
2 parents 053c121 + f855094 commit 8426177

5 files changed

Lines changed: 281 additions & 55 deletions

File tree

‎crates/enc-ffmpeg/src/remux.rs‎

Lines changed: 179 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,76 @@ pub fn concatenate_video_fragments(fragments: &[PathBuf], output: &Path) -> Resu
8080
result
8181
}
8282

83+
/// Joins files cut from one recording timeline, such as the output of each
84+
/// muxer process when one was replaced mid-recording. Each file keeps its own
85+
/// start time, so video after a gap stays at its capture time instead of
86+
/// closing the gap; the last frame before a gap is held across it.
87+
pub fn concatenate_video_fragments_on_timeline(
88+
fragments: &[PathBuf],
89+
output: &Path,
90+
) -> Result<(), RemuxError> {
91+
if fragments.is_empty() {
92+
return Err(RemuxError::NoFragments);
93+
}
94+
95+
for fragment in fragments {
96+
if !fragment.exists() {
97+
return Err(RemuxError::FragmentNotFound(fragment.clone()));
98+
}
99+
}
100+
101+
let spans = fragments
102+
.iter()
103+
.map(|fragment| media_span(fragment))
104+
.collect::<Result<Vec<_>, _>>()?;
105+
let concat_list_path = output.with_extension("concat.txt");
106+
let entries = fragments
107+
.iter()
108+
.enumerate()
109+
.map(|(index, fragment)| {
110+
let entry = concat_fragment_entry(fragment, &concat_list_path)?;
111+
Ok(match timeline_duration(&spans, index) {
112+
Some(duration) => format!("{entry}duration {duration}us\n"),
113+
None => entry,
114+
})
115+
})
116+
.collect::<Result<Vec<_>, RemuxError>>()?;
117+
std::fs::write(&concat_list_path, entries.concat())?;
118+
119+
let result = concatenate_with_concat_demuxer(&concat_list_path, output);
120+
121+
let _ = std::fs::remove_file(&concat_list_path);
122+
123+
result
124+
}
125+
126+
#[derive(Clone, Copy, Debug)]
127+
struct MediaSpan {
128+
start: i64,
129+
duration: i64,
130+
}
131+
132+
fn media_span(path: &Path) -> Result<Option<MediaSpan>, RemuxError> {
133+
let input = ffmpeg::format::input(path)?;
134+
let (start, duration) = (container_start_time(&input), input.duration());
135+
Ok((start != ffmpeg::ffi::AV_NOPTS_VALUE && duration > 0)
136+
.then_some(MediaSpan { start, duration }))
137+
}
138+
139+
fn container_start_time(input: &avformat::context::Input) -> i64 {
140+
unsafe { (*input.as_ptr()).start_time }
141+
}
142+
143+
/// How far into the joined timeline the file after `index` starts, in
144+
/// microseconds: the gap between the two start times, but never less than the
145+
/// file's own duration, so overlapping files still play end to end. `None`
146+
/// leaves the concat demuxer to use the file's probed duration.
147+
fn timeline_duration(spans: &[Option<MediaSpan>], index: usize) -> Option<i64> {
148+
let current = spans.get(index).copied().flatten()?;
149+
let next = spans.get(index + 1).copied().flatten()?;
150+
Some((next.start - current.start).max(current.duration))
151+
}
152+
83153
fn concat_fragment_entry(fragment: &Path, concat_list: &Path) -> Result<String, RemuxError> {
84154
let fragment = std::path::absolute(fragment)?;
85155
let concat_list = std::path::absolute(concat_list)?;
@@ -1374,7 +1444,8 @@ mod tests {
13741444
concatenate_audio_to_ogg, concatenate_m4s_segments_with_init,
13751445
concatenate_m4s_segments_with_init_validated,
13761446
concatenate_m4s_segments_with_init_validated_with_reader, concatenate_video_fragments,
1377-
get_video_capture_fps, get_video_fps, probe_video_pts_ladder, write_concat_list,
1447+
concatenate_video_fragments_on_timeline, get_video_capture_fps, get_video_fps,
1448+
probe_video_pts_ladder, write_concat_list,
13781449
};
13791450
use std::{
13801451
path::{Path, PathBuf},
@@ -1794,6 +1865,113 @@ mod tests {
17941865
assert!(!output.with_extension("concat.txt").exists());
17951866
}
17961867

1868+
fn first_video_pts_after(path: &Path, frames_before: usize) -> f64 {
1869+
let (time_base, packets) = video_packet_contents(path);
1870+
let mut pts: Vec<i64> = packets.iter().filter_map(|packet| packet.pts).collect();
1871+
pts.sort_unstable();
1872+
pts[frames_before] as f64 * f64::from(time_base)
1873+
}
1874+
1875+
fn timeline_inputs(parent: &Path, second_start_frame: u64) -> [PathBuf; 2] {
1876+
let frames = |range: std::ops::Range<u64>| -> Vec<Duration> {
1877+
range
1878+
.map(|frame| Duration::from_nanos(frame * 1_000_000_000 / 30))
1879+
.collect()
1880+
};
1881+
let first_dir = parent.join("first");
1882+
let second_dir = parent.join("second");
1883+
std::fs::create_dir_all(&first_dir).unwrap();
1884+
std::fs::create_dir_all(&second_dir).unwrap();
1885+
let second = second_dir.join("shifted.mp4");
1886+
shift_test_mp4(
1887+
&encode_test_mp4(&second_dir, &frames(0..12)),
1888+
&second,
1889+
second_start_frame as f64 / 30.0,
1890+
);
1891+
[encode_test_mp4(&first_dir, &frames(0..12)), second]
1892+
}
1893+
1894+
/// A copy of `input` whose packets start `offset` seconds later, like a
1895+
/// replacement muxer's output on the recording's shared timeline.
1896+
fn shift_test_mp4(input: &Path, output: &Path, offset: f64) {
1897+
let mut ictx = ffmpeg::format::input(input).unwrap();
1898+
let mut octx = ffmpeg::format::output(output).unwrap();
1899+
let stream = ictx.streams().best(ffmpeg::media::Type::Video).unwrap();
1900+
let (index, time_base) = (stream.index(), stream.time_base());
1901+
let mut output_stream = octx.add_stream(None).unwrap();
1902+
output_stream.set_parameters(stream.parameters());
1903+
unsafe {
1904+
(*output_stream.as_mut_ptr())
1905+
.codecpar
1906+
.as_mut()
1907+
.unwrap()
1908+
.codec_tag = 0;
1909+
}
1910+
octx.write_header().unwrap();
1911+
let output_time_base = octx.stream(0).unwrap().time_base();
1912+
let shift = (offset * f64::from(time_base.denominator()) / f64::from(time_base.numerator()))
1913+
.round() as i64;
1914+
for (stream, mut packet) in ictx.packets() {
1915+
if stream.index() != index {
1916+
continue;
1917+
}
1918+
packet.set_pts(packet.pts().map(|pts| pts + shift));
1919+
packet.set_dts(packet.dts().map(|dts| dts + shift));
1920+
packet.rescale_ts(time_base, output_time_base);
1921+
packet.set_stream(0);
1922+
packet.set_position(-1);
1923+
packet.write_interleaved(&mut octx).unwrap();
1924+
}
1925+
octx.write_trailer().unwrap();
1926+
}
1927+
1928+
#[test]
1929+
fn timeline_concat_keeps_a_later_file_at_its_start_time() {
1930+
let directory = tempfile::tempdir().unwrap();
1931+
let inputs = timeline_inputs(directory.path(), 30);
1932+
let start = |path: &Path| {
1933+
super::container_start_time(&ffmpeg::format::input(path).unwrap()) as f64
1934+
/ f64::from(ffmpeg::ffi::AV_TIME_BASE)
1935+
};
1936+
assert!(
1937+
(start(&inputs[1]) - 1.0).abs() < 0.002,
1938+
"{}",
1939+
start(&inputs[1])
1940+
);
1941+
1942+
let joined = directory.path().join("timeline.mp4");
1943+
concatenate_video_fragments_on_timeline(&inputs, &joined).unwrap();
1944+
let second_at = first_video_pts_after(&joined, 12);
1945+
assert!(
1946+
(second_at - 1.0).abs() < 0.002,
1947+
"second file starts at {second_at}"
1948+
);
1949+
assert_eq!(video_packet_contents(&joined).1.len(), 24);
1950+
assert!(!joined.with_extension("concat.txt").exists());
1951+
1952+
let end_to_end = directory.path().join("end-to-end.mp4");
1953+
concatenate_video_fragments(&inputs, &end_to_end).unwrap();
1954+
assert!(first_video_pts_after(&end_to_end, 12) < 0.5);
1955+
}
1956+
1957+
#[test]
1958+
fn timeline_concat_plays_overlapping_files_end_to_end() {
1959+
let directory = tempfile::tempdir().unwrap();
1960+
let inputs = timeline_inputs(directory.path(), 6);
1961+
1962+
let joined = directory.path().join("timeline.mp4");
1963+
concatenate_video_fragments_on_timeline(&inputs, &joined).unwrap();
1964+
let packets = video_packet_contents(&joined).1;
1965+
assert_eq!(packets.len(), 24);
1966+
let dts: Vec<i64> = packets.iter().filter_map(|packet| packet.dts).collect();
1967+
assert!(dts.windows(2).all(|pair| pair[0] < pair[1]), "{dts:?}");
1968+
let second_at = first_video_pts_after(&joined, 12);
1969+
assert!(
1970+
second_at >= 0.39,
1971+
"overlap played over the first file: {second_at}"
1972+
);
1973+
}
1974+
17971975
#[test]
17981976
fn concat_audio_and_video_with_quoted_parent_directories() {
17991977
assert_concat_roundtrip("recording 'quoted' λ");

‎crates/recording/src/output_pipeline/oop_muxer.rs‎

Lines changed: 58 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -365,7 +365,9 @@ impl MuxerSubprocess {
365365
data,
366366
};
367367

368-
match write_packet_frame(stdin, &packet) {
368+
// Flush every packet: anything left in the buffer is lost if the
369+
// recorder dies, and a dead child only shows up as a write error.
370+
match write_packet_frame(stdin, &packet).and_then(|()| Ok(stdin.flush()?)) {
369371
Ok(()) => {
370372
self.packets_written.fetch_add(1, Ordering::Relaxed);
371373
Ok(())
@@ -871,6 +873,61 @@ mod tests {
871873
(directory, subprocess)
872874
}
873875

876+
fn script_subprocess(script: &str) -> (tempfile::TempDir, MuxerSubprocess, PathBuf) {
877+
let directory = tempfile::tempdir().unwrap();
878+
let binary = directory.path().join("muxer");
879+
std::fs::write(&binary, format!("#!/bin/sh\n{script}\n")).unwrap();
880+
std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o700)).unwrap();
881+
let config = MuxerSubprocessConfig {
882+
output_directory: directory.path().join("video"),
883+
init_segment_name: "init.mp4".into(),
884+
media_segment_pattern: "segment_$Number%03d$.m4s".into(),
885+
video_init: None,
886+
audio_init: None,
887+
};
888+
let received = directory.path().join("received");
889+
let subprocess = MuxerSubprocess::spawn(binary, config, None).unwrap();
890+
(directory, subprocess, received)
891+
}
892+
893+
#[test]
894+
fn every_packet_reaches_the_child_before_the_next_one_is_written() {
895+
let (directory, mut subprocess, received) =
896+
script_subprocess("exec cat > \"$(dirname \"$0\")/received\"");
897+
let size = |path: &Path| std::fs::metadata(path).map(|m| m.len()).unwrap_or(0);
898+
let deadline = Instant::now() + Duration::from_secs(5);
899+
while size(&received) == 0 && Instant::now() < deadline {
900+
std::thread::sleep(Duration::from_millis(5));
901+
}
902+
let after_start = size(&received);
903+
assert!(after_start > 0, "init and start frames never arrived");
904+
subprocess
905+
.write_video_packet(0, 0, 1, true, &[7; 1000])
906+
.unwrap();
907+
let deadline = Instant::now() + Duration::from_secs(5);
908+
while size(&received) < after_start + 1000 && Instant::now() < deadline {
909+
std::thread::sleep(Duration::from_millis(5));
910+
}
911+
assert!(
912+
size(&received) >= after_start + 1000,
913+
"a written packet stayed in the recorder's buffer"
914+
);
915+
drop(subprocess);
916+
drop(directory);
917+
}
918+
919+
#[test]
920+
fn a_dead_child_is_reported_on_the_next_packet() {
921+
let (_directory, mut subprocess, _) = script_subprocess("sleep 0.3; exit 3");
922+
let child = subprocess.child.as_mut().unwrap();
923+
let status = child.wait().unwrap();
924+
assert_eq!(status.code(), Some(3));
925+
assert!(matches!(
926+
subprocess.write_video_packet(0, 0, 1, true, &[7; 1000]),
927+
Err(MuxerSubprocessError::Crashed(_))
928+
));
929+
}
930+
874931
#[test]
875932
fn disk_full_stays_terminal_and_finish_confirms_shutdown() {
876933
let (_directory, mut subprocess) = exited_subprocess(EXIT_DISK_FULL, 2);

‎crates/recording/src/recovery.rs‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,10 @@ use cap_enc_ffmpeg::RelocatableSource;
88
use cap_enc_ffmpeg::fragmented_mp4::{read_fragment_prefix, tail_is_complete};
99
use cap_enc_ffmpeg::remux::{
1010
concatenate_audio_to_ogg, concatenate_m4s_segments_with_init,
11-
concatenate_m4s_segments_with_init_validated, concatenate_video_fragments, get_media_duration,
12-
get_video_base_fps, merge_video_audio, probe_media_valid, probe_video_can_decode,
13-
probe_video_seek_points, remux_file,
11+
concatenate_m4s_segments_with_init_validated, concatenate_video_fragments,
12+
concatenate_video_fragments_on_timeline, get_media_duration, get_video_base_fps,
13+
merge_video_audio, probe_media_valid, probe_video_can_decode, probe_video_seek_points,
14+
remux_file,
1415
};
1516
use cap_project::{
1617
AudioMeta, Cursors, MultipleSegment, MultipleSegments, ProjectConfiguration, RecordingMeta,
@@ -2336,7 +2337,8 @@ impl RecoveryManager {
23362337
std::fs::rename(&group_outputs[0], output)
23372338
.map_err(|e| RecoveryError::VideoConcat(cap_enc_ffmpeg::remux::RemuxError::Io(e)))
23382339
} else {
2339-
concatenate_video_fragments(&group_outputs, output).map_err(RecoveryError::VideoConcat)
2340+
concatenate_video_fragments_on_timeline(&group_outputs, output)
2341+
.map_err(RecoveryError::VideoConcat)
23402342
};
23412343

23422344
for tmp in &temp_paths {

‎crates/recording/tests/oop_muxer.rs‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -217,6 +217,43 @@ fn subprocess_exits_after_finish_or_abort_without_waiting_for_stdin_eof() {
217217
}
218218
}
219219

220+
#[cfg(unix)]
221+
#[test]
222+
fn finish_waits_for_a_muxer_that_takes_more_than_five_seconds_to_exit() {
223+
use cap_muxer_protocol::{Frame, read_frame};
224+
use std::io::Cursor;
225+
use std::os::unix::fs::PermissionsExt;
226+
use std::time::{Duration, Instant};
227+
228+
let directory = TempDir::new().unwrap();
229+
let binary = directory.path().join("slow-exit-muxer");
230+
std::fs::write(&binary, "#!/bin/sh\ncat > \"$0.received\"\nsleep 6\n").unwrap();
231+
std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o700)).unwrap();
232+
let config = minimal_video_config(&directory.path().join("video"), Vec::new());
233+
let mut subprocess = MuxerSubprocess::spawn(binary, config, None).unwrap();
234+
let payload = vec![42; 128 * 1024];
235+
subprocess
236+
.write_video_packet(0, 0, 1, true, &payload)
237+
.unwrap();
238+
let started = Instant::now();
239+
let report = subprocess.finish().unwrap();
240+
assert_eq!(report.exit_code, Some(0));
241+
assert!(started.elapsed() >= Duration::from_secs(5));
242+
243+
let received = std::fs::read(directory.path().join("slow-exit-muxer.received")).unwrap();
244+
let mut reader = Cursor::new(received.as_slice());
245+
assert!(matches!(
246+
read_frame(&mut reader).unwrap(),
247+
Frame::InitVideo(_)
248+
));
249+
assert!(matches!(read_frame(&mut reader).unwrap(), Frame::Start(_)));
250+
let Frame::Packet(packet) = read_frame(&mut reader).unwrap() else {
251+
panic!("expected the video packet");
252+
};
253+
assert_eq!(packet.data, payload);
254+
assert!(matches!(read_frame(&mut reader).unwrap(), Frame::Finish));
255+
}
256+
220257
#[cfg(unix)]
221258
#[test]
222259
fn finish_preserves_packets_when_a_slow_muxer_takes_more_than_five_seconds() {
@@ -232,10 +269,10 @@ fn finish_preserves_packets_when_a_slow_muxer_takes_more_than_five_seconds() {
232269
let config = minimal_video_config(&directory.path().join("video"), Vec::new());
233270
let mut subprocess = MuxerSubprocess::spawn(binary, config, None).unwrap();
234271
let payload = vec![42; 128 * 1024];
272+
let started = Instant::now();
235273
subprocess
236274
.write_video_packet(0, 0, 1, true, &payload)
237275
.unwrap();
238-
let started = Instant::now();
239276
let report = subprocess.finish().unwrap();
240277
assert_eq!(report.exit_code, Some(0));
241278
assert!(started.elapsed() >= Duration::from_secs(5));

0 commit comments

Comments
 (0)