Skip to content

Commit

Permalink
ndisrc: Improve handling of broken sources with regards to timestamping
Browse files Browse the repository at this point in the history 
 - NDI HX Camera Android in the past used 1ns instead of 100ns as unit
   for timecodes/timestamps.
 - NDI HX Camera iOS uses 0 for all timecodes and the same non-zero
   value for all audio timestamps

Detect such situations and try to compensate for them. Also add a new
"auto" timestamping mode that prefers to use timecodes and otherwise
falls back to timestamps or receive times.

Fixes teltek/gst-plugin-ndi#79
  • Loading branch information
@sdroege
sdroege committed Oct 12, 2022
1 parent a3c7528 commit 9c10ba8
Show file tree
Hide file tree
Showing 3 changed files with 173 additions and 47 deletions.
12 changes: 7 additions & 5 deletions net/ndi/src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,16 +24,18 @@ use once_cell::sync::Lazy;
#[repr(u32)]
#[enum_type(name = "GstNdiTimestampMode")]
pub enum TimestampMode {
#[enum_value(name = "Auto", nick = "auto")]
Auto = 0,
#[enum_value(name = "Receive Time / Timecode", nick = "receive-time-vs-timecode")]
ReceiveTimeTimecode = 0,
ReceiveTimeTimecode = 1,
#[enum_value(name = "Receive Time / Timestamp", nick = "receive-time-vs-timestamp")]
ReceiveTimeTimestamp = 1,
ReceiveTimeTimestamp = 2,
#[enum_value(name = "NDI Timecode", nick = "timecode")]
Timecode = 2,
Timecode = 3,
#[enum_value(name = "NDI Timestamp", nick = "timestamp")]
Timestamp = 3,
Timestamp = 4,
#[enum_value(name = "Receive Time", nick = "receive-time")]
ReceiveTime = 4,
ReceiveTime = 5,
}

#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Hash, Clone, Copy, glib::Enum)]
Expand Down
2 changes: 1 addition & 1 deletion net/ndi/src/ndisrc/imp.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ impl Default for Settings {
max_queue_length: 10,
bandwidth: ndisys::NDIlib_recv_bandwidth_highest,
color_format: RecvColorFormat::UyvyBgra,
timestamp_mode: TimestampMode::ReceiveTimeTimecode,
timestamp_mode: TimestampMode::Auto,
}
}
}
Expand Down
206 changes: 165 additions & 41 deletions net/ndi/src/receiver.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
)
});

#[derive(Clone)]
pub struct Receiver(Arc<ReceiverInner>);

#[derive(Debug, PartialEq, Eq)]
Expand Down Expand Up @@ -226,6 +225,7 @@ struct ReceiverQueueInner {
timeout: bool,
}

const PREFILL_WINDOW_LENGTH: usize = 12;
const WINDOW_LENGTH: u64 = 512;
const WINDOW_DURATION: u64 = 2_000_000_000;

Expand All @@ -240,6 +240,11 @@ struct ObservationsInner {
skew: i64,
filling: bool,
window_size: usize,

// Remote/local times for workaround around fundamentally wrong slopes
// This is not reset below and has a bigger window.
times: VecDeque<(u64, u64)>,
slope_correction: (u64, u64),
}

impl Default for ObservationsInner {
Expand All @@ -252,10 +257,24 @@ impl Default for ObservationsInner {
skew: 0,
filling: true,
window_size: 0,
times: VecDeque::new(),
slope_correction: (1, 1),
}
}
}

impl ObservationsInner {
fn reset(&mut self) {
self.base_local_time = None;
self.base_remote_time = None;
self.deltas = VecDeque::new();
self.min_delta = 0;
self.skew = 0;
self.filling = true;
self.window_size = 0;
}
}

impl Observations {
// Based on the algorithm used in GStreamer's rtpjitterbuffer, which comes from
// Fober, Orlarey and Letz, 2005, "Real Time Clock Skew Estimation over Network Delays":
Expand All @@ -266,22 +285,41 @@ impl Observations {
remote_time: Option<gst::ClockTime>,
local_time: gst::ClockTime,
duration: Option<gst::ClockTime>,
) -> (gst::ClockTime, Option<gst::ClockTime>, bool) {
let remote_time = match remote_time {
None => return (local_time, duration, false),
Some(remote_time) => remote_time.nseconds(),
};
) -> Option<(gst::ClockTime, Option<gst::ClockTime>, bool)> {
let remote_time = remote_time?.nseconds();
let local_time = local_time.nseconds();

let mut inner = self.0.borrow_mut();

gst_trace!(
CAT,
obj: element,
"Local time {}, remote time {}",
"Local time {}, remote time {}, slope correct {}/{}",
gst::ClockTime::from_nseconds(local_time),
gst::ClockTime::from_nseconds(remote_time),
inner.slope_correction.0,
inner.slope_correction.1,
);

let mut inner = self.0.borrow_mut();
inner.times.push_back((remote_time, local_time));
while inner
.times
.back()
.unwrap()
.1
.saturating_sub(inner.times.front().unwrap().1)
> WINDOW_DURATION
{
let _ = inner.times.pop_front();
}

// Static remote times
if inner.slope_correction.1 == 0 {
return None;
}

let remote_time =
remote_time.mul_div_round(inner.slope_correction.0, inner.slope_correction.1)?;

let (base_remote_time, base_local_time) =
match (inner.base_remote_time, inner.base_local_time) {
Expand All @@ -297,10 +335,102 @@ impl Observations {
inner.base_remote_time = Some(remote_time);
inner.base_local_time = Some(local_time);

return (gst::ClockTime::from_nseconds(local_time), duration, true);
return Some((gst::ClockTime::from_nseconds(local_time), duration, true));
}
};

if inner.times.len() < PREFILL_WINDOW_LENGTH {
return Some((gst::ClockTime::from_nseconds(local_time), duration, false));
}

// Check if the slope is simply wrong and try correcting
{
let local_diff = inner
.times
.back()
.unwrap()
.1
.saturating_sub(inner.times.front().unwrap().1);
let remote_diff = inner
.times
.back()
.unwrap()
.0
.saturating_sub(inner.times.front().unwrap().0);

if remote_diff == 0 {
inner.reset();
inner.base_remote_time = Some(remote_time);
inner.base_local_time = Some(local_time);

// Static remote times
inner.slope_correction = (0, 0);
return None;
} else {
let slope = local_diff as f64 / remote_diff as f64;
let scaled_slope =
slope * (inner.slope_correction.1 as f64) / (inner.slope_correction.0 as f64);

// Check for some obviously wrong slopes and try to correct for that
if !(0.5..1.5).contains(&scaled_slope) {
gst_warning!(
CAT,
obj: element,
"Too small/big slope {}, resetting",
scaled_slope
);

let discont = !inner.deltas.is_empty();
inner.reset();

if (0.0005..0.0015).contains(&slope) {
// Remote unit was actually 0.1ns
inner.slope_correction = (1, 1000);
} else if (0.005..0.015).contains(&slope) {
// Remote unit was actually 1ns
inner.slope_correction = (1, 100);
} else if (0.05..0.15).contains(&slope) {
// Remote unit was actually 10ns
inner.slope_correction = (1, 10);
} else if (5.0..15.0).contains(&slope) {
// Remote unit was actually 1us
inner.slope_correction = (10, 1);
} else if (50.0..150.0).contains(&slope) {
// Remote unit was actually 10us
inner.slope_correction = (100, 1);
} else if (50.0..150.0).contains(&slope) {
// Remote unit was actually 100us
inner.slope_correction = (1000, 1);
} else if (50.0..150.0).contains(&slope) {
// Remote unit was actually 1ms
inner.slope_correction = (10000, 1);
} else {
inner.slope_correction = (1, 1);
}

let remote_time = inner
.times
.back()
.unwrap()
.0
.mul_div_round(inner.slope_correction.0, inner.slope_correction.1)?;
gst_debug!(
CAT,
obj: element,
"Initializing base time: local {}, remote {}, slope correction {}/{}",
gst::ClockTime::from_nseconds(local_time),
gst::ClockTime::from_nseconds(remote_time),
inner.slope_correction.0,
inner.slope_correction.1,
);
inner.base_remote_time = Some(remote_time);
inner.base_local_time = Some(local_time);

return Some((gst::ClockTime::from_nseconds(local_time), duration, discont));
}
}
}

let remote_diff = remote_time.saturating_sub(base_remote_time);
let local_diff = local_time.saturating_sub(base_local_time);
let delta = (local_diff as i64) - (remote_diff as i64);
Expand All @@ -314,33 +444,6 @@ impl Observations {
delta,
);

if remote_diff > 0 && local_diff > 0 {
let slope = (local_diff as f64) / (remote_diff as f64);
if !(0.8..1.2).contains(&slope) {
gst_warning!(
CAT,
obj: element,
"Too small/big slope {}, resetting",
slope
);

let discont = !inner.deltas.is_empty();
*inner = ObservationsInner::default();

gst_debug!(
CAT,
obj: element,
"Initializing base time: local {}, remote {}",
gst::ClockTime::from_nseconds(local_time),
gst::ClockTime::from_nseconds(remote_time),
);
inner.base_remote_time = Some(remote_time);
inner.base_local_time = Some(local_time);

return (gst::ClockTime::from_nseconds(local_time), duration, discont);
}
}

if (delta > inner.skew && delta - inner.skew > 1_000_000_000)
|| (delta < inner.skew && inner.skew - delta > 1_000_000_000)
{
Expand All @@ -353,7 +456,6 @@ impl Observations {
);

let discont = !inner.deltas.is_empty();
*inner = ObservationsInner::default();

gst_debug!(
CAT,
Expand All @@ -362,10 +464,12 @@ impl Observations {
gst::ClockTime::from_nseconds(local_time),
gst::ClockTime::from_nseconds(remote_time),
);

inner.reset();
inner.base_remote_time = Some(remote_time);
inner.base_local_time = Some(local_time);

return (gst::ClockTime::from_nseconds(local_time), duration, discont);
return Some((gst::ClockTime::from_nseconds(local_time), duration, discont));
}

if inner.filling {
Expand Down Expand Up @@ -421,7 +525,7 @@ impl Observations {
gst::ClockTime::from_nseconds(out_time)
);

(gst::ClockTime::from_nseconds(out_time), duration, false)
Some((gst::ClockTime::from_nseconds(out_time), duration, false))
}
}

Expand Down Expand Up @@ -820,8 +924,23 @@ impl Receiver {
);

let (pts, duration, discont) = match self.0.timestamp_mode {
TimestampMode::ReceiveTimeTimecode => res_timecode,
TimestampMode::ReceiveTimeTimestamp => res_timestamp,
TimestampMode::ReceiveTimeTimecode => match res_timecode {
Some((pts, duration, discont)) => (pts, duration, discont),
None => {
gst_warning!(CAT, obj: element, "Can't calculate timestamp");
(receive_time, duration, false)
}
},
TimestampMode::ReceiveTimeTimestamp => match res_timestamp {
Some((pts, duration, discont)) => (pts, duration, discont),
None => {
if timestamp.is_some() {
gst_warning!(CAT, obj: element, "Can't calculate timestamp");
}

(receive_time, duration, false)
}
},
TimestampMode::Timecode => (timecode, duration, false),
TimestampMode::Timestamp if timestamp.is_none() => (receive_time, duration, false),
TimestampMode::Timestamp => {
Expand All @@ -840,6 +959,11 @@ impl Receiver {
}
}
TimestampMode::ReceiveTime => (receive_time, duration, false),
TimestampMode::Auto => {
res_timecode
.or(res_timestamp)
.unwrap_or((receive_time, duration, false))
}
};

gst_log!(
Expand Down

0 comments on commit 9c10ba8

Please sign in to comment.