+void JitterHistory::register_metrics(const vector<pair<string, string>> &labels)
+{
+ global_metrics.add("input_underestimated_jitter_frames", labels, &metric_input_underestimated_jitter_frames);
+ global_metrics.add("input_estimated_max_jitter_seconds", labels, &metric_input_estimated_max_jitter_seconds, Metrics::TYPE_GAUGE);
+}
+
+void JitterHistory::unregister_metrics(const vector<pair<string, string>> &labels)
+{
+ global_metrics.remove("input_underestimated_jitter_frames", labels);
+ global_metrics.remove("input_estimated_max_jitter_seconds", labels);
+}
+
+void JitterHistory::frame_arrived(steady_clock::time_point now, int64_t frame_duration, size_t dropped_frames)
+{
+ if (expected_timestamp > steady_clock::time_point::min()) {
+ expected_timestamp += dropped_frames * nanoseconds(frame_duration * 1000000000 / TIMEBASE);
+ double jitter_seconds = fabs(duration<double>(expected_timestamp - now).count());
+ history.push_back(orders.insert(jitter_seconds));
+ if (jitter_seconds > estimate_max_jitter()) {
+ ++metric_input_underestimated_jitter_frames;
+ }
+
+ metric_input_estimated_max_jitter_seconds = estimate_max_jitter();
+
+ if (history.size() > history_length) {
+ orders.erase(history.front());
+ history.pop_front();
+ }
+ assert(history.size() <= history_length);
+ }
+ expected_timestamp = now + nanoseconds(frame_duration * 1000000000 / TIMEBASE);
+}
+
+double JitterHistory::estimate_max_jitter() const
+{
+ if (orders.empty()) {
+ return 0.0;
+ }
+ size_t elem_idx = lrint((orders.size() - 1) * percentile);
+ if (percentile <= 0.5) {
+ return *next(orders.begin(), elem_idx) * multiplier;
+ } else {
+ return *prev(orders.end(), orders.size() - elem_idx) * multiplier;
+ }
+}
+
+void QueueLengthPolicy::register_metrics(const vector<pair<string, string>> &labels)
+{
+ global_metrics.add("input_queue_safe_length_frames", labels, &metric_input_queue_safe_length_frames, Metrics::TYPE_GAUGE);
+}
+
+void QueueLengthPolicy::unregister_metrics(const vector<pair<string, string>> &labels)
+{
+ global_metrics.remove("input_queue_safe_length_frames", labels);
+}
+
+void QueueLengthPolicy::update_policy(steady_clock::time_point now,
+ steady_clock::time_point expected_next_frame,
+ int64_t input_frame_duration,
+ int64_t master_frame_duration,
+ double max_input_card_jitter_seconds,
+ double max_master_card_jitter_seconds)
+{
+ double input_frame_duration_seconds = input_frame_duration / double(TIMEBASE);
+ double master_frame_duration_seconds = master_frame_duration / double(TIMEBASE);
+
+ // Figure out when we can expect the next frame for this card, assuming
+ // worst-case jitter (ie., the frame is maximally late).
+ double seconds_until_next_frame = max(duration<double>(expected_next_frame - now).count() + max_input_card_jitter_seconds, 0.0);
+
+ // How many times are the master card expected to tick in that time?
+ // We assume the master clock has worst-case jitter but not any rate
+ // discrepancy, ie., it ticks as early as possible every time, but not
+ // cumulatively.
+ double frames_needed = (seconds_until_next_frame + max_master_card_jitter_seconds) / master_frame_duration_seconds;
+
+ // As a special case, if the master card ticks faster than the input card,
+ // we expect the queue to drain by itself even without dropping. But if
+ // the difference is small (e.g. 60 Hz master and 59.94 input), it would
+ // go slowly enough that the effect wouldn't really be appreciable.
+ // We account for this by looking at the situation five frames ahead,
+ // assuming everything else is the same.
+ double frames_allowed;
+ if (master_frame_duration < input_frame_duration) {
+ frames_allowed = frames_needed + 5 * (input_frame_duration_seconds - master_frame_duration_seconds) / master_frame_duration_seconds;
+ } else {
+ frames_allowed = frames_needed;
+ }
+
+ safe_queue_length = max<int>(floor(frames_allowed), 0);
+ metric_input_queue_safe_length_frames = safe_queue_length;
+}
+