Dynamic latency adjustment #688

Actual latency tuning code together
with related API and CLI parameters.

docs/sphinx/manuals/roc_recv.rst

@@ -28,8 +28,11 @@ Options
 --miface=MIFACE               IPv4 or IPv6 address of the network interface on which to join the multicast group
 --reuseaddr                   enable SO_REUSEADDR when binding sockets
 --target-latency=STRING       Target latency, TIME units
---io-latency=STRING           Playback target latency, TIME units
 --latency-tolerance=STRING    Maximum deviation from target latency, TIME units
+--start-latency=STRING        Target latency, TIME units
+--min-latency=STRING          Minimum allowed latency, TIME units
+--max-latency=STRING          Maximum allowed latency, TIME units
+--io-latency=STRING           Playback target latency, TIME units
 --no-play-timeout=STRING      No playback timeout, TIME units
 --choppy-play-timeout=STRING  Choppy playback timeout, TIME units
 --frame-len=TIME              Duration of the internal frames, TIME units

src/internal_modules/roc_audio/feedback_monitor.cpp

@@ -21,7 +21,7 @@ FeedbackMonitor::FeedbackMonitor(IFrameWriter& writer,
                                  const FeedbackConfig& feedback_config,
                                  const LatencyConfig& latency_config,
                                  const SampleSpec& sample_spec)
-    : tuner_(latency_config, sample_spec)
+    : tuner_(latency_config, sample_spec, NULL)
     , use_packetizer_(false)
     , has_feedback_(false)
     , last_feedback_ts_(0)

src/internal_modules/roc_audio/freq_estimator.cpp

@@ -9,6 +9,7 @@
 #include "roc_audio/freq_estimator.h"
 #include "roc_core/log.h"
 #include "roc_core/panic.h"
+#include "roc_core/time.h"
 
 namespace roc {
 namespace audio {
@@ -25,13 +26,15 @@ FreqEstimatorConfig make_config(FreqEstimatorProfile profile) {
         config.I = 1e-10;
         config.decimation_factor1 = fe_decim_factor_max;
         config.decimation_factor2 = 0;
+        config.stable_criteria = 0.1;
         break;
 
     case FreqEstimatorProfile_Gradual:
         config.P = 1e-6;
         config.I = 5e-9;
         config.decimation_factor1 = fe_decim_factor_max;
         config.decimation_factor2 = fe_decim_factor_max;
+        config.stable_criteria = 0.05;
         break;
     }
 
@@ -62,14 +65,18 @@ double dot_prod(const double* coeff,
 } // namespace
 
 FreqEstimator::FreqEstimator(FreqEstimatorProfile profile,
-                             packet::stream_timestamp_t target_latency)
+                             packet::stream_timestamp_t target_latency,
+                             core::CsvDumper* dumper)
     : config_(make_config(profile))
     , target_(target_latency)
     , dec1_ind_(0)
     , dec2_ind_(0)
     , samples_counter_(0)
     , accum_(0)
-    , coeff_(1) {
+    , coeff_(1)
+    , stable_(false)
+    , last_unstable_time_(core::timestamp(core::ClockMonotonic))
+    , dumper_(dumper) {
     roc_log(LogDebug, "freq estimator: initializing: P=%e I=%e dc1=%lu dc2=%lu",
             config_.P, config_.I, (unsigned long)config_.decimation_factor1,
             (unsigned long)config_.decimation_factor2);
@@ -105,6 +112,17 @@ void FreqEstimator::update(packet::stream_timestamp_t current) {
     double filtered;
 
     if (run_decimators_(current, filtered)) {
+        if (dumper_) {
+            core::CsvEntry e;
+            e.type = 'f';
+            e.n_fields = 5;
+            e.fields[0] = core::timestamp(core::ClockUnix);
+            e.fields[1] = filtered;
+            e.fields[2] = target_;
+            e.fields[3] = (filtered - target_) * config_.P;
+            e.fields[4] = accum_ * config_.I;
+            dumper_->write(e);
+        }
         coeff_ = run_controller_(filtered);
     }
 }
@@ -149,8 +167,63 @@ bool FreqEstimator::run_decimators_(packet::stream_timestamp_t current,
 double FreqEstimator::run_controller_(double current) {
     const double error = (current - target_);
 
-    accum_ = accum_ + error;
-    return 1 + config_.P * error + config_.I * accum_;
+    roc_log(LogTrace,
+            "Freq Estimator:"
+            " current latency error: %.0f",
+            error);
+
+    if (abs(error) > target_ * config_.stable_criteria && stable_) {
+        stable_ = false;
+        accum_ = 0;
+        last_unstable_time_ = core::timestamp(core::ClockMonotonic);
+        roc_log(LogDebug,
+                "Freq Estimator: "
+                " unstable, %0.f > %.0f / %0.f",
+                config_.stable_criteria, error, target_);
+    } else if (abs(error) < target_ * config_.stable_criteria && !stable_
+               && core::timestamp(core::ClockMonotonic) - last_unstable_time_
+                   > 15 * core::Second) {
+        stable_ = true;
+        roc_log(LogDebug,
+                "Freq Estimator: "
+                " stabilized");
+    }
+
+    double res = 0.;
+    // In stable state we are not using P term in order to avoid permanent variation
+    // of resampler control input.
+    if (stable_) {
+        accum_ = accum_ + error;
+        res += config_.I * accum_;
+    } else {
+        res += config_.P * error;
+    }
+    if (abs(res) > 1e-2) {
+        res = res / abs(res) * 1e-2;
+    }
+    res += 1.;
+
+    return res;
+}
+
+void FreqEstimator::update_target_latency(packet::stream_timestamp_t target_latency) {
+    target_ = (double)target_latency;
+}
+
+bool FreqEstimator::stable() const {
+    return stable_;
+}
+
+static const char* fe_profile_to_str(FreqEstimatorProfile profile) {
+    switch (profile) {
+    case FreqEstimatorProfile_Responsive:
+        return "responsive";
+
+    case FreqEstimatorProfile_Gradual:
+        return "gradual";
+    }
+
+    return "<invalid>";
 }
 
 } // namespace audio

src/internal_modules/roc_audio/freq_estimator.h

@@ -14,6 +14,7 @@
 
 #include "roc_audio/freq_estimator_decim.h"
 #include "roc_audio/sample.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/noncopyable.h"
 #include "roc_packet/units.h"
 
@@ -44,11 +45,16 @@ struct FreqEstimatorConfig {
     //! to fe_decim_factor_max. Could be zero to disable the second decimation stage.
     size_t decimation_factor2;
 
+    //! Within this range we consider the FreqEstimator is stable.
+    //! stable_criteria > error / target;
+    double stable_criteria;
+
     FreqEstimatorConfig()
         : P(0)
         , I(0)
         , decimation_factor1(0)
-        , decimation_factor2(0) {
+        , decimation_factor2(0)
+        , stable_criteria(0.1) {
     }
 };
 
@@ -67,20 +73,27 @@ class FreqEstimator : public core::NonCopyable<> {
     //!  - @p profile defines configuration preset.
     //!  - @p target_latency defines latency we want to archive.
     FreqEstimator(FreqEstimatorProfile profile,
-                  packet::stream_timestamp_t target_latency);
+                  packet::stream_timestamp_t target_latency,
+                  roc::core::CsvDumper* dumper);
 
     //! Get current frequecy coefficient.
     float freq_coeff() const;
 
     //! Compute new value of frequency coefficient.
     void update(packet::stream_timestamp_t current_latency);
 
+    //! Update target latency.
+    void update_target_latency(packet::stream_timestamp_t target_latency);
+
+    //! If FreqEstimator has stabilized.
+    bool stable() const;
+
 private:
     bool run_decimators_(packet::stream_timestamp_t current, double& filtered);
     double run_controller_(double current);
 
     const FreqEstimatorConfig config_;
-    const double target_; // Target latency.
+    double target_; // Target latency.
 
     double dec1_casc_buff_[fe_decim_len];
     size_t dec1_ind_;
@@ -92,6 +105,12 @@ class FreqEstimator : public core::NonCopyable<> {
     double accum_;           // Integrator value.
 
     double coeff_; // Current frequency coefficient value.
+
+    bool stable_; // True if FreqEstimator has stabilized.
+    // Last time when FreqEstimator was out of range.
+    core::nanoseconds_t last_unstable_time_;
+
+    core::CsvDumper* dumper_;
 };
 
 } // namespace audio

src/internal_modules/roc_audio/latency_monitor.cpp

@@ -25,8 +25,9 @@ LatencyMonitor::LatencyMonitor(IFrameReader& frame_reader,
                                ResamplerReader* resampler,
                                const LatencyConfig& config,
                                const SampleSpec& packet_sample_spec,
-                               const SampleSpec& frame_sample_spec)
-    : tuner_(config, frame_sample_spec)
+                               const SampleSpec& frame_sample_spec,
+                               core::CsvDumper* dumper)
+    : tuner_(config, frame_sample_spec, dumper)
     , frame_reader_(frame_reader)
     , incoming_queue_(incoming_queue)
     , depacketizer_(depacketizer)
@@ -104,6 +105,12 @@ bool LatencyMonitor::reclock(const core::nanoseconds_t playback_timestamp) {
 }
 
 bool LatencyMonitor::pre_process_(const Frame& frame) {
+    if (fec_reader_) {
+        latency_metrics_.fec_block_duration =
+            packet_sample_spec_.stream_timestamp_2_ns(fec_reader_->max_block_duration());
+    } else {
+        latency_metrics_.fec_block_duration = 0;
+    }
     tuner_.write_metrics(latency_metrics_, link_metrics_);
     if (!tuner_.update_stream()) {
         // TODO(gh-183): forward status code

src/internal_modules/roc_audio/latency_monitor.h

@@ -19,16 +19,106 @@
 #include "roc_audio/resampler_reader.h"
 #include "roc_audio/sample_spec.h"
 #include "roc_core/attributes.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/noncopyable.h"
 #include "roc_core/optional.h"
 #include "roc_core/time.h"
 #include "roc_fec/reader.h"
 #include "roc_packet/sorted_queue.h"
 #include "roc_packet/units.h"
+#include "roc_rtp/link_meter.h"
 
 namespace roc {
 namespace audio {
 
+//! Parameters for latency monitor.
+struct LatencyMonitorConfig {
+    //! Enable FreqEstimator.
+    bool fe_enable;
+
+    //! FreqEstimator profile.
+    FreqEstimatorProfile fe_profile;
+
+    //! FreqEstimator update interval, nanoseconds.
+    //! How often to run FreqEstimator and update Resampler scaling.
+    core::nanoseconds_t fe_update_interval;
+
+    //! Maximum allowed deviation from target latency, nanoseconds.
+    //! If the latency goes out of bounds, the session is terminated.
+    core::nanoseconds_t latency_tolerance;
+
+    //! Maximum allowed deviation of freq_coeff from 1.0.
+    //! If the scaling goes out of bounds, it is trimmed.
+    //! For example, 0.01 allows freq_coeff values in range [0.99; 1.01].
+    float scaling_tolerance;
+
+    //! Automatically tune target latency within tolarance range so as to
+    //!
+    //! increase it when:
+    //! * jitter grows
+    //! * FEC start being sent or its block length grows
+    //!
+    //! or decrease it when jitter and FEC block length allows to.
+    bool auto_tune_target_latency;
+
+    LatencyMonitorConfig()
+        : fe_enable(true)
+        , fe_profile(FreqEstimatorProfile_Responsive)
+        , fe_update_interval(5 * core::Millisecond)
+        , latency_tolerance(0)
+        , scaling_tolerance(0.005f)
+        , auto_tune_target_latency(false) {
+    }
+
+    //! Automatically deduce FreqEstimator profile from target latency.
+    void deduce_fe_profile(const core::nanoseconds_t target_latency) {
+        fe_profile = target_latency < 30 * core::Millisecond
+            // prefer responsive profile on low latencies, because gradual profile
+            // won't do it at all
+            ? FreqEstimatorProfile_Responsive
+            // prefer gradual profile for higher latencies, because it can handle
+            // higher network jitter
+            : FreqEstimatorProfile_Gradual;
+    }
+
+    //! Automatically deduce latency_tolerance from target_latency.
+    void deduce_latency_tolerance(core::nanoseconds_t target_latency) {
+        // this formula returns target_latency * N, where N starts with larger
+        // number and approaches 0.5 as target_latency grows
+        // examples:
+        //  target=1ms -> tolerance=8ms (x8)
+        //  target=10ms -> tolerance=20ms (x2)
+        //  target=200ms -> tolerance=200ms (x1)
+        //  target=2000ms -> tolerance=1444ms (x0.722)
+        if (target_latency < core::Millisecond) {
+            target_latency = core::Millisecond;
+        }
+        latency_tolerance = core::nanoseconds_t(
+            target_latency
+            * (std::log((200 * core::Millisecond) * 2) / std::log(target_latency * 2)));
+    }
+};
+
+//! Metrics of latency monitor.
+struct LatencyMonitorMetrics {
+    //! Estimated NIQ latency.
+    //! NIQ = network incoming queue.
+    //! Defines how many samples are buffered in receiver packet queue and
+    //! receiver pipeline before depacketizer (packet part of pipeline).
+    core::nanoseconds_t niq_latency;
+
+    //! Estimated E2E latency.
+    //! E2E = end-to-end.
+    //! Defines how much time passed between frame entered sender pipeline
+    //! (when it is captured) and leaved received pipeline (when it is played).
+    core::nanoseconds_t e2e_latency;
+
+    LatencyMonitorMetrics()
+        : niq_latency(0)
+        , e2e_latency(0) {
+    }
+};
+
 //! Latency monitor.
 //!
 //! @b Features
@@ -68,7 +158,8 @@ class LatencyMonitor : public IFrameReader, public core::NonCopyable<> {
                    ResamplerReader* resampler,
                    const LatencyConfig& config,
                    const SampleSpec& packet_sample_spec,
-                   const SampleSpec& frame_sample_spec);
+                   const SampleSpec& frame_sample_spec,
+                   core::CsvDumper* dumper);
 
     //! Check if the object was initialized successfully.
     bool is_valid() const;