mirrors/gst-plugins-rs
mirrors/gst-plugins-rs
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git synced 2024-11-15 14:52:04 +00:00
Code Issues Projects Releases Wiki Activity

livesync: Wait for the end timestamp of the previous buffer before looking at queue

Browse source 
Previously livesync was waiting for the start timestamp of the current
buffer after looking at the queue and right before pushing it
downstream. This meant that it generally looked too early at the queue
and especially that upstream had to provide the next buffer already at
the start timestamp of the previous one.

Instead, now wait before looking at the queue and wait for the end
timestamp of the previous buffer. Once the previous buffer has expired,
a new buffer really needs to be available or otherwise a filler buffer
has to be pushed downstream.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1266>
This commit is contained in:
Sebastian Dröge 2023-06-20 10:36:51 +03:00
parent 51173cfdc8
commit 69d4ecc3be
1 changed files with 36 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
utils/livesync/src/livesync/imp.rs
View file

@ -1030,6 +1030,42 @@ impl LiveSync {
}
state.srcresult?;
if let Some(out_timestamp) = state.out_timestamp {
let sync_ts = out_timestamp.end;
let element = self.obj();
let base_time = element.base_time().ok_or_else(|| {
gst::error!(CAT, imp: self, "Missing base time");
gst::FlowError::Flushing
})?;
let clock = element.clock().ok_or_else(|| {
gst::error!(CAT, imp: self, "Missing clock");
gst::FlowError::Flushing
})?;
let clock_id = clock.new_single_shot_id(base_time + sync_ts);
state.clock_id = Some(clock_id.clone());
gst::trace!(
CAT,
imp: self,
"Waiting for clock to reach {}",
clock_id.time(),
);
let (res, _) = MutexGuard::unlocked(&mut state, || clock_id.wait());
gst::trace!(CAT, imp: self, "Clock returned {res:?}",);
if res == Err(gst::ClockError::Unscheduled) {
return Err(gst::FlowError::Flushing);
}
state.srcresult?;
state.clock_id = None;
}
let in_item = state.queue.pop_front();
gst::trace!(CAT, imp: self, "Unqueueing {:?}", in_item);
@ -1186,40 +1222,6 @@ impl LiveSync {
state.out_segment = Some(segment);
}
{
let element = self.obj();
let base_time = element.base_time().ok_or_else(|| {
gst::error!(CAT, imp: self, "Missing base time");
gst::FlowError::Flushing
})?;
let clock = element.clock().ok_or_else(|| {
gst::error!(CAT, imp: self, "Missing clock");
gst::FlowError::Flushing
})?;
let clock_id = clock.new_single_shot_id(base_time + sync_ts);
state.clock_id = Some(clock_id.clone());
gst::trace!(
CAT,
imp: self,
"Waiting for clock to reach {}",
clock_id.time(),
);
let (res, _) = MutexGuard::unlocked(&mut state, || clock_id.wait());
gst::trace!(CAT, imp: self, "Clock returned {res:?}",);
if res == Err(gst::ClockError::Unscheduled) {
return Err(gst::FlowError::Flushing);
}
state.srcresult?;
state.clock_id = None;
}
state.num_out += 1;
if duplicate {
state.num_duplicate += 1;