minor refactor

Author: baranovmv

src/internal_modules/roc_core/target_posix/roc_core/thread.cpp

@@ -51,13 +51,13 @@ bool Thread::enable_realtime() {
     memset(&param, 0, sizeof(param));
     param.sched_priority = sched_get_priority_max(SCHED_RR);
 
+    roc_log(LogDebug, "thread: set realtime priority");
     if (int err = pthread_setschedparam(pthread_self(), SCHED_RR, &param)) {
-        roc_log(LogDebug,
+        roc_log(LogError,
                 "thread: can't set realtime priority: pthread_setschedparam(): %s",
                 errno_to_str(err).c_str());
         return false;
     }
-    roc_log(LogDebug, "thread: set realtime priority");
 
     return true;
 }

src/internal_modules/roc_rtcp/reporter.h

@@ -274,9 +274,9 @@ class Reporter : public core::NonCopyable<> {
             : core::RefCounted<Stream, core::PoolAllocation>(pool)
             , source_id(source_id)
             , has_remote_recv_report(false)
-            , remote_recv_rtt(arena, rtt_config, dumper)
+            , remote_recv_rtt(rtt_config, arena, dumper)
             , has_remote_send_report(false)
-            , remote_send_rtt(arena, rtt_config, dumper)
+            , remote_send_rtt(rtt_config, arena, dumper)
             , local_recv_report(NULL)
             , last_update(report_time)
             , last_local_sr(0)

src/internal_modules/roc_rtcp/rtt_estimator.cpp

@@ -12,17 +12,15 @@
 namespace roc {
 namespace rtcp {
 
-RttEstimator::RttEstimator(core::IArena& arena,
-                           const RttConfig& config,
-                           dbgio::CsvDumper* dumper)
+RttEstimator::RttEstimator(const RttConfig &config, core::IArena &arena, dbgio::CsvDumper *dumper)
     : config_(config)
     , metrics_()
     , has_metrics_(false)
     , first_report_ts_(0)
     , last_report_ts_(0)
     , dumper_(dumper)
-    , rtt_stats_(arena, 15, 0.5)
-    , clock_offset_stats_(arena, 100, 0.5) {
+    , rtt_stats_(arena, config.rtt_winlen, 0.5)
+    , clock_offset_stats_(arena, config.clock_offset_winlen, 0.5) {
 }
 
 bool RttEstimator::has_metrics() const {

src/internal_modules/roc_rtcp/rtt_estimator.h

@@ -26,8 +26,18 @@ struct RttConfig {
     //! All metrics below are computed for a sliding window of this length.
     core::nanoseconds_t interval_duration;
 
+    //! How many measurements of RTT is used
+    //! to find median value.
+    size_t rtt_winlen;
+
+    //! How many measurements of RTT is used
+    //! to find median value.
+    size_t clock_offset_winlen;
+
     RttConfig()
-        : interval_duration(core::Second * 5) {
+        : interval_duration(core::Second * 5)
+        , rtt_winlen(15)
+        , clock_offset_winlen(100) {
     }
 };
 
@@ -53,7 +63,7 @@ struct RttMetrics {
 class RttEstimator {
 public:
     //! Initialize.
-    RttEstimator(core::IArena& arena, const RttConfig& config, dbgio::CsvDumper* dumper);
+    RttEstimator(const RttConfig &config, core::IArena &arena, dbgio::CsvDumper *dumper);
 
     //! Check whether metrics are already available.
     bool has_metrics() const;

src/tests/roc_rtcp/test_communicator.cpp

@@ -438,15 +438,15 @@ void expect_recv_report(const RecvReport& report,
     }
 }
 
-packet::PacketPtr read_packet(packet::FifoQueue& source, core::nanoseconds_t ts = -1) {
+packet::PacketPtr read_packet(packet::FifoQueue& source, core::nanoseconds_t recv_ts = -1) {
     CHECK(source.size() != 0);
     packet::PacketPtr pp;
     LONGS_EQUAL(status::StatusOK, source.read(pp, packet::ModeFetch));
     CHECK(pp);
     CHECK(pp->rtcp());
     CHECK(pp->rtcp()->payload);
-    if (ts >= 0) {
-        pp->udp()->receive_timestamp = ts;
+    if (recv_ts >= 0) {
+        pp->udp()->receive_timestamp = recv_ts;
     }
     roc_log(LogTrace, "delivering rtcp packet");
     if (core::Logger::instance().get_level() >= LogTrace) {
@@ -3107,7 +3107,7 @@ TEST(communicator, report_to_address_sender) {
     CHECK_EQUAL(1, send_comm.total_destinations());
     CHECK_EQUAL(1, send_queue.size());
 
-    pp = read_packet(send_queue, send_time);
+    pp = read_packet(send_queue);
     expect_has_dest_address(pp, send_dest_addr);
     expect_has_orig_ssrc(pp, SendSsrc, true);
     expect_has_dest_ssrc(pp, Recv1Ssrc, true);
@@ -3279,7 +3279,7 @@ TEST(communicator, report_back_sender) {
     CHECK_EQUAL(1, send_comm.total_destinations());
     CHECK_EQUAL(1, send_queue.size());
 
-    pp = read_packet(send_queue, send_time);
+    pp = read_packet(send_queue);
     expect_has_dest_address(pp, recv1_addr);
     expect_has_orig_ssrc(pp, SendSsrc, true);
     expect_has_dest_ssrc(pp, Recv1Ssrc, true);
@@ -3320,13 +3320,13 @@ TEST(communicator, report_back_sender) {
     CHECK_EQUAL(2, send_comm.total_destinations());
     CHECK_EQUAL(2, send_queue.size());
 
-    pp = read_packet(send_queue, send_time);
+    pp = read_packet(send_queue);
     expect_has_dest_address(pp, recv1_addr);
     expect_has_orig_ssrc(pp, SendSsrc, true);
     expect_has_dest_ssrc(pp, Recv1Ssrc, true);
     expect_has_dest_ssrc(pp, Recv2Ssrc, false);
 
-    pp = read_packet(send_queue, send_time);
+    pp = read_packet(send_queue);
     expect_has_dest_address(pp, recv2_addr);
     expect_has_orig_ssrc(pp, SendSsrc, true);
     expect_has_dest_ssrc(pp, Recv1Ssrc, false);
@@ -3423,7 +3423,7 @@ TEST(communicator, report_back_receiver) {
     CHECK_EQUAL(1, recv_comm.total_destinations());
     CHECK_EQUAL(1, recv_queue.size());
 
-    pp = read_packet(recv_queue, send1_time);
+    pp = read_packet(recv_queue);
     expect_has_dest_address(pp, send1_addr);
     expect_has_orig_ssrc(pp, RecvSsrc, true);
     expect_has_dest_ssrc(pp, Send1Ssrc, true);
@@ -3469,13 +3469,13 @@ TEST(communicator, report_back_receiver) {
     CHECK_EQUAL(2, recv_comm.total_destinations());
     CHECK_EQUAL(2, recv_queue.size());
 
-    pp = read_packet(recv_queue, recv_time);
+    pp = read_packet(recv_queue);
     expect_has_dest_address(pp, send1_addr);
     expect_has_orig_ssrc(pp, RecvSsrc, true);
     expect_has_dest_ssrc(pp, Send1Ssrc, true);
     expect_has_dest_ssrc(pp, Send2Ssrc, false);
 
-    pp = read_packet(recv_queue, recv_time);
+    pp = read_packet(recv_queue);
     expect_has_dest_address(pp, send2_addr);
     expect_has_orig_ssrc(pp, RecvSsrc, true);
     expect_has_dest_ssrc(pp, Send1Ssrc, false);
@@ -3606,14 +3606,14 @@ TEST(communicator, report_back_combine_reports) {
     CHECK_EQUAL(2, recv_comm.total_destinations());
     CHECK_EQUAL(2, recv_queue.size());
 
-    pp = read_packet(recv_queue, send3_time);
+    pp = read_packet(recv_queue);
     expect_has_dest_address(pp, send1_addr);
     expect_has_orig_ssrc(pp, RecvSsrc, true);
     expect_has_dest_ssrc(pp, Send1Ssrc, true);
     expect_has_dest_ssrc(pp, Send2Ssrc, true);
     expect_has_dest_ssrc(pp, Send3Ssrc, false);
 
-    pp = read_packet(recv_queue, send3_time);
+    pp = read_packet(recv_queue);
     expect_has_dest_address(pp, send3_addr);
     expect_has_orig_ssrc(pp, RecvSsrc, true);
     expect_has_dest_ssrc(pp, Send1Ssrc, false);
@@ -3832,6 +3832,99 @@ TEST(communicator, rtt) {
     }
 }
 
+// Same, but there is a persistent clock drift between sender and receiver
+TEST(communicator, rtt_clock_drift) {
+    enum { SendSsrc = 11, RecvSsrc = 22, NumIters = 200 };
+
+    const char* SendCname = "send_cname";
+    const char* RecvCname = "recv_cname";
+
+    const core::nanoseconds_t SendStartTime = 10000000000000000;
+    const core::nanoseconds_t RecvStartTime = 30000000000000000;
+
+    const core::nanoseconds_t ReportInterval = 500 * core::Millisecond;
+    const core::nanoseconds_t Rtt = 200 * core::Millisecond;
+    const core::nanoseconds_t Drift = 1 * core::Millisecond;
+
+    Config config;
+    config.rtt.rtt_winlen = 1; // Disable smoothing
+    config.rtt.clock_offset_winlen = 1;
+
+    packet::FifoQueue send_queue;
+    MockParticipant send_part(SendCname, SendSsrc, Report_ToAddress);
+    Communicator send_comm(config, send_part, send_queue, composer, packet_factory, arena,
+                           NULL);
+    LONGS_EQUAL(status::StatusOK, send_comm.init_status());
+
+    packet::FifoQueue recv_queue;
+    MockParticipant recv_part(RecvCname, RecvSsrc, Report_Back);
+    Communicator recv_comm(config, recv_part, recv_queue, composer, packet_factory, arena,
+                           NULL);
+    LONGS_EQUAL(status::StatusOK, recv_comm.init_status());
+
+    core::nanoseconds_t send_time = SendStartTime;
+    core::nanoseconds_t recv_time = RecvStartTime;
+
+    for (int iter = 0; iter < NumIters; iter++) {
+        // Generate sender report
+        send_part.set_send_report(make_send_report(send_time, SendCname, SendSsrc, Seed));
+        LONGS_EQUAL(status::StatusOK, send_comm.generate_reports(send_time));
+        CHECK_EQUAL(1, send_queue.size());
+
+        advance_time(send_time, Rtt / 2);
+        advance_time(recv_time, Rtt / 2);
+
+        // Deliver sender report to receiver
+        if (iter != 0) {
+            recv_part.set_recv_report(
+                0, make_recv_report(recv_time, RecvCname, RecvSsrc, SendSsrc, Seed));
+        }
+        LONGS_EQUAL(status::StatusOK,
+                    recv_comm.process_packet(read_packet(send_queue, recv_time)));
+
+        {
+            // Check metrics on receiver
+            const SendReport report = recv_part.next_send_notification();
+            if (iter != 0) {
+                expect_timestamp("receiver rtt", Rtt, report.rtt, RttEpsilon + Drift);
+                expect_timestamp("receiver clock_offset",
+                                 SendStartTime + (Drift * iter) - RecvStartTime,
+                                 report.clock_offset, RttEpsilon + Drift);
+            }
+        }
+
+        advance_time(send_time, ReportInterval / 2 + Drift);
+        advance_time(recv_time, ReportInterval / 2);
+
+        // Generate receiver report
+        recv_part.set_recv_report(
+            0, make_recv_report(recv_time, RecvCname, RecvSsrc, SendSsrc, Seed));
+        LONGS_EQUAL(status::StatusOK, recv_comm.generate_reports(recv_time));
+        CHECK_EQUAL(1, recv_queue.size());
+
+        advance_time(send_time, Rtt / 2);
+        advance_time(recv_time, Rtt / 2);
+
+        // Deliver receiver report to sender
+        send_part.set_send_report(make_send_report(send_time, SendCname, SendSsrc, Seed));
+        LONGS_EQUAL(status::StatusOK,
+                    send_comm.process_packet(read_packet(recv_queue, send_time)));
+
+        {
+            // Check metrics on sender
+            const RecvReport report = send_part.next_recv_notification();
+
+            expect_timestamp("sender rtt", Rtt, report.rtt, RttEpsilon + Drift);
+            expect_timestamp("sender clock_offset",
+                             RecvStartTime - SendStartTime - (Drift * iter),
+                             report.clock_offset, RttEpsilon + Drift);
+        }
+
+        advance_time(send_time, ReportInterval / 2);
+        advance_time(recv_time, ReportInterval / 2);
+    }
+}
+
 // Same, but there is persistent network jitter
 TEST(communicator, rtt_network_jitter) {
     enum { SendSsrc = 11, RecvSsrc = 22, NumIters = 200 };
@@ -4269,7 +4362,7 @@ TEST(communicator, rtt_network_reordering) {
                 recv_reorder_countdown = ReorderBurst;
             }
             recv_reorder_countdown--;
-            packet::PacketPtr pp = read_packet(recv_queue, recv_time);
+            packet::PacketPtr pp = read_packet(recv_queue);
             recv_packet_stash.push_back(*pp);
         }