Preparation for E2E latency tuning

src/internal_modules/roc_audio/latency_monitor.cpp

@@ -34,6 +34,7 @@ LatencyMonitor::LatencyMonitor(IFrameReader& frame_reader,
     , fec_reader_(fec_reader)
     , resampler_(resampler)
     , enable_scaling_(latency_config.tuner_profile != audio::LatencyTunerProfile_Intact)
+    , current_time_(0)
     , capture_ts_(0)
     , packet_sample_spec_(packet_sample_spec)
     , frame_sample_spec_(frame_sample_spec)
@@ -86,6 +87,14 @@ status::StatusCode LatencyMonitor::read(Frame& frame,
     return status::StatusOK;
 }
 
+void LatencyMonitor::refresh(const core::nanoseconds_t current_time) {
+    roc_panic_if(init_status_ != status::StatusOK);
+
+    // this method is called before reading next frame
+    // now we know current time and can use it to compute metrics
+    current_time_ = current_time;
+}
+
 void LatencyMonitor::reclock(const core::nanoseconds_t playback_timestamp) {
     roc_panic_if(init_status_ != status::StatusOK);
 
@@ -146,10 +155,8 @@ void LatencyMonitor::compute_niq_latency_() {
 
     // compute delay since last packet
     const core::nanoseconds_t rts = latest_packet->receive_timestamp();
-    const core::nanoseconds_t now = core::timestamp(core::ClockUnix);
-
-    if (rts > 0 && rts < now) {
-        latency_metrics_.niq_stalling = now - rts;
+    if (rts > 0 && rts < current_time_) {
+        latency_metrics_.niq_stalling = current_time_ - rts;
     }
 }
 

src/internal_modules/roc_audio/latency_monitor.h

@@ -85,6 +85,13 @@ class LatencyMonitor : public IFrameReader, public core::NonCopyable<> {
     virtual ROC_NODISCARD status::StatusCode
     read(Frame& frame, packet::stream_timestamp_t duration, FrameReadMode mode);
 
+    //! Report current time.
+    //! @remarks
+    //!  Pipeline invokes this method before reading each frame.
+    //!  Until it's invoked for the first time, metrics that need current
+    //!  time (like niq_stalling) are not computed.
+    void refresh(core::nanoseconds_t current_time);
+
     //! Report playback timestamp of last frame returned by read.
     //! @remarks
     //!  Pipeline invokes this method after adding last frame to
@@ -117,6 +124,7 @@ class LatencyMonitor : public IFrameReader, public core::NonCopyable<> {
     ResamplerReader* resampler_;
     const bool enable_scaling_;
 
+    core::nanoseconds_t current_time_;
     core::nanoseconds_t capture_ts_;
 
     const SampleSpec packet_sample_spec_;

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

@@ -46,13 +46,15 @@ uint64_t Thread::get_tid() {
 #endif
 }
 
-bool Thread::enable_realtime() {
+bool Thread::enable_realtime(const int sched_prio) {
     sched_param param;
     memset(&param, 0, sizeof(param));
-    param.sched_priority = sched_get_priority_max(SCHED_RR);
+    param.sched_priority = sched_prio;
+
+    roc_log(LogDebug, "thread: setting realtime priority to %d", sched_prio);
 
     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;

src/internal_modules/roc_core/target_posix/roc_core/thread.h

@@ -33,7 +33,7 @@ class Thread : public NonCopyable<Thread> {
     static uint64_t get_tid();
 
     //! Raise current thread priority to realtime.
-    ROC_NODISCARD static bool enable_realtime();
+    ROC_NODISCARD static bool enable_realtime(const int sched_prio);
 
     //! Check if thread was started and can be joined.
     //! @returns

src/internal_modules/roc_netio/target_libuv/roc_netio/network_config.h

@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2026 Roc Streaming authors
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+//! @file roc_netio/target_libuv/roc_netio/network_config.h
+//! @brief Network loop config.
+
+#ifndef ROC_NETIO_NETWORK_CONFIG_H_
+#define ROC_NETIO_NETWORK_CONFIG_H_
+
+#include "roc_core/stddefs.h"
+
+namespace roc {
+namespace netio {
+
+//! Network loop config.
+struct NetworkConfig {
+    //! Priority of the network thread.
+    //! @remarks
+    //!  If non-zero, network thread is switched to real-time scheduling policy
+    //!  with this priority. Should be in range [0; 99]. Requires privileges.
+    int realtime_prio;
+
+    //! Initialize.
+    NetworkConfig()
+        : realtime_prio(0) {
+    }
+};
+
+} // namespace netio
+} // namespace roc
+
+#endif // ROC_NETIO_NETWORK_CONFIG_H_

src/internal_modules/roc_netio/target_libuv/roc_netio/network_loop.cpp

@@ -110,12 +110,15 @@ NetworkLoop::Tasks::ResolveEndpointAddress::get_address() const {
     return resolve_req_.resolved_address;
 }
 
-NetworkLoop::NetworkLoop(core::IPool& packet_pool,
+NetworkLoop::NetworkLoop(const NetworkConfig& net_config,
+                         core::IPool& packet_pool,
                          core::IPool& buffer_pool,
                          core::IArena& arena)
-    : packet_factory_(packet_pool, buffer_pool)
+    : net_config_(net_config)
+    , packet_factory_(packet_pool, buffer_pool)
     , arena_(arena)
     , started_(false)
+    , thr_init_cond_(thr_init_mutex_)
     , loop_initialized_(false)
     , stop_sem_initialized_(false)
     , task_sem_initialized_(false)
@@ -148,12 +151,19 @@ NetworkLoop::NetworkLoop(core::IPool& packet_pool,
     task_sem_.data = this;
     task_sem_initialized_ = true;
 
+    roc_panic_if(init_status_ != status::NoStatus);
+
     if (!(started_ = Thread::start())) {
         init_status_ = status::StatusErrThread;
         return;
     }
 
-    init_status_ = status::StatusOK;
+    {
+        core::Mutex::Lock lock(thr_init_mutex_);
+        while (init_status_ == status::NoStatus) {
+            thr_init_cond_.wait();
+        }
+    }
 }
 
 NetworkLoop::~NetworkLoop() {
@@ -289,6 +299,22 @@ void NetworkLoop::handle_resolved(ResolverRequest& req) {
 void NetworkLoop::run() {
     roc_log(LogDebug, "network loop: starting event loop");
 
+    if (net_config_.realtime_prio > 0
+        && !Thread::enable_realtime(net_config_.realtime_prio)) {
+        core::Mutex::Lock lock(thr_init_mutex_);
+
+        roc_log(LogError,
+                "network loop: can't elevate realtime priority, may need to be root");
+        init_status_ = status::StatusErrThread;
+        thr_init_cond_.signal();
+    } else {
+        core::Mutex::Lock lock(thr_init_mutex_);
+
+        roc_log(LogNote, "network loop: elevated realtime priority");
+        init_status_ = status::StatusOK;
+        thr_init_cond_.signal();
+    }
+
     int err = uv_run(&loop_, UV_RUN_DEFAULT);
     if (err != 0) {
         roc_log(LogInfo, "network loop: uv_run() returned non-zero");

src/internal_modules/roc_netio/target_libuv/roc_netio/network_loop.h

@@ -17,6 +17,7 @@
 #include "roc_address/socket_addr.h"
 #include "roc_core/atomic_int.h"
 #include "roc_core/attributes.h"
+#include "roc_core/cond.h"
 #include "roc_core/iarena.h"
 #include "roc_core/ipool.h"
 #include "roc_core/list.h"
@@ -32,6 +33,7 @@
 #include "roc_netio/iconn_handler.h"
 #include "roc_netio/inetwork_task_completer.h"
 #include "roc_netio/iterminate_handler.h"
+#include "roc_netio/network_config.h"
 #include "roc_netio/network_task.h"
 #include "roc_netio/resolver.h"
 #include "roc_netio/tcp_connection_port.h"
@@ -196,7 +198,10 @@ class NetworkLoop : private ITerminateHandler,
     //! Initialize.
     //! @remarks
     //!  Start background thread if the object was successfully constructed.
-    NetworkLoop(core::IPool& packet_pool, core::IPool& buffer_pool, core::IArena& arena);
+    NetworkLoop(const NetworkConfig& net_config,
+                core::IPool& packet_pool,
+                core::IPool& buffer_pool,
+                core::IArena& arena);
 
     //! Destroy. Stop all receivers and senders.
     //! @remarks
@@ -255,11 +260,15 @@ class NetworkLoop : private ITerminateHandler,
     void task_remove_port_(NetworkTask&);
     void task_resolve_endpoint_address_(NetworkTask&);
 
+    const NetworkConfig net_config_;
     packet::PacketFactory packet_factory_;
     core::IArena& arena_;
 
     bool started_;
 
+    core::Mutex thr_init_mutex_;
+    core::Cond thr_init_cond_;
+
     uv_loop_t loop_;
     bool loop_initialized_;
 

src/internal_modules/roc_node/context.cpp

@@ -13,6 +13,16 @@
 namespace roc {
 namespace node {
 
+namespace {
+
+netio::NetworkConfig make_network_config(const ContextConfig& config) {
+    netio::NetworkConfig net_config;
+    net_config.realtime_prio = config.realtime_prio;
+    return net_config;
+}
+
+} // namespace
+
 Context::Context(const ContextConfig& config, core::IArena& arena)
     : arena_(arena)
     , packet_pool_("packet_pool", arena_)
@@ -23,7 +33,8 @@ Context::Context(const ContextConfig& config, core::IArena& arena)
           "frame_buffer_pool", arena_, sizeof(core::Buffer) + config.max_frame_size)
     , processor_map_(arena_)
     , encoding_map_(arena_)
-    , network_loop_(packet_pool_, packet_buffer_pool_, arena_)
+    , network_loop_(
+          make_network_config(config), packet_pool_, packet_buffer_pool_, arena_)
     , control_loop_(network_loop_, arena_)
     , init_status_(status::NoStatus) {
     roc_log(LogDebug, "context: initializing");

src/internal_modules/roc_node/context.h

@@ -34,9 +34,13 @@ struct ContextConfig {
     //! Maximum size in bytes of an audio frame.
     size_t max_frame_size;
 
+    //! Set real-time priority. Requires root privileges.
+    int realtime_prio;
+
     ContextConfig()
         : max_packet_size(2048)
-        , max_frame_size(4096) {
+        , max_frame_size(4096)
+        , realtime_prio(0) {
     }
 };
 

src/internal_modules/roc_node/receiver_decoder.cpp

@@ -169,6 +169,8 @@ status::StatusCode ReceiverDecoder::write_packet(address::Interface iface,
     roc_panic_if(!bytes);
     roc_panic_if(n_bytes == 0);
 
+    const core::nanoseconds_t receive_ts = core::timestamp(core::ClockUnix);
+
     if (n_bytes > packet_factory_.packet_buffer_size()) {
         roc_log(LogError,
                 "receiver decoder node:"
@@ -195,6 +197,7 @@ status::StatusCode ReceiverDecoder::write_packet(address::Interface iface,
     }
 
     packet->add_flags(packet::Packet::FlagUDP);
+    packet->udp()->receive_timestamp = receive_ts;
     packet->set_buffer(buffer);
 
     packet::IWriter* writer = endpoint_writers_[iface];

src/internal_modules/roc_node/sender_encoder.cpp

@@ -226,6 +226,8 @@ SenderEncoder::write_packet(address::Interface iface, const void* bytes, size_t
     roc_panic_if(!bytes);
     roc_panic_if(n_bytes == 0);
 
+    const core::nanoseconds_t receive_ts = core::timestamp(core::ClockUnix);
+
     if (n_bytes > packet_factory_.packet_buffer_size()) {
         roc_log(LogError,
                 "sender encoder node:"
@@ -252,6 +254,7 @@ SenderEncoder::write_packet(address::Interface iface, const void* bytes, size_t
     }
 
     packet->add_flags(packet::Packet::FlagUDP);
+    packet->udp()->receive_timestamp = receive_ts;
     packet->set_buffer(buffer);
 
     packet::IWriter* writer = endpoint_writers_[iface];

src/internal_modules/roc_pipeline/receiver_endpoint.cpp

@@ -224,14 +224,19 @@ status::StatusCode ReceiverEndpoint::handle_packet_(const packet::PacketPtr& pac
                                                     core::nanoseconds_t current_time) {
     status::StatusCode code = status::NoStatus;
 
+    // Apparently the packet is not from network, set its TS manually.
+    if (packet->udp() && packet->udp()->receive_timestamp == 0 && current_time != 0) {
+        packet->udp()->receive_timestamp = current_time;
+    }
+
     if ((code = parser_->parse(*packet, packet->buffer())) != status::StatusOK) {
         roc_log(LogDebug,
                 "receiver endpoint: dropping bad packet: can't parse: status=%s",
                 status::code_to_str(code));
         return status::StatusOK;
     }
 
-    code = session_group_.route_packet(packet, current_time);
+    code = session_group_.route_packet(packet);
 
     if (code == status::StatusNoRoute) {
         roc_log(LogDebug, "receiver endpoint: dropping bad packet: can't route");

src/internal_modules/roc_pipeline/receiver_session.cpp

@@ -296,6 +296,8 @@ status::StatusCode ReceiverSession::refresh(core::nanoseconds_t current_time,
         return fail_status_;
     }
 
+    latency_monitor_->refresh(current_time);
+
     return status::StatusOK;
 }
 

src/internal_modules/roc_pipeline/receiver_session_group.cpp

@@ -71,7 +71,7 @@ ReceiverSessionGroup::create_control_pipeline(ReceiverEndpoint* control_endpoint
     // and later when we call generate_packets() or process_packets().
     rtcp_communicator_.reset(new (rtcp_communicator_) rtcp::Communicator(
         source_config_.common.rtcp, *this, *control_endpoint->outbound_writer(),
-        *control_endpoint->outbound_composer(), packet_factory_, arena_));
+        *control_endpoint->outbound_composer(), packet_factory_, arena_, dumper_));
 
     const status::StatusCode code = rtcp_communicator_->init_status();
     if (code != status::StatusOK) {
@@ -140,12 +140,11 @@ void ReceiverSessionGroup::reclock_sessions(core::nanoseconds_t playback_time) {
     }
 }
 
-status::StatusCode ReceiverSessionGroup::route_packet(const packet::PacketPtr& packet,
-                                                      core::nanoseconds_t current_time) {
+status::StatusCode ReceiverSessionGroup::route_packet(const packet::PacketPtr& packet) {
     roc_panic_if(init_status_ != status::StatusOK);
 
     if (packet->has_flags(packet::Packet::FlagControl)) {
-        return route_control_packet_(packet, current_time);
+        return route_control_packet_(packet);
     }
 
     return route_transport_packet_(packet);
@@ -344,15 +343,14 @@ ReceiverSessionGroup::route_transport_packet_(const packet::PacketPtr& packet) {
 }
 
 status::StatusCode
-ReceiverSessionGroup::route_control_packet_(const packet::PacketPtr& packet,
-                                            core::nanoseconds_t current_time) {
+ReceiverSessionGroup::route_control_packet_(const packet::PacketPtr& packet) {
     if (!rtcp_communicator_) {
         roc_panic("session group: rtcp communicator is null");
     }
 
     // This will invoke IParticipant methods implemented by us,
     // in particular notify_recv_stream() and maybe halt_recv_stream().
-    return rtcp_communicator_->process_packet(packet, current_time);
+    return rtcp_communicator_->process_packet(packet);
 }
 
 bool ReceiverSessionGroup::can_create_session_(const packet::PacketPtr& packet) {

src/internal_modules/roc_pipeline/receiver_session_group.h

@@ -90,8 +90,7 @@ class ReceiverSessionGroup : public core::NonCopyable<>, private rtcp::IParticip
     void reclock_sessions(core::nanoseconds_t playback_time);
 
     //! Route packet to session.
-    ROC_NODISCARD status::StatusCode route_packet(const packet::PacketPtr& packet,
-                                                  core::nanoseconds_t current_time);
+    ROC_NODISCARD status::StatusCode route_packet(const packet::PacketPtr& packet);
 
     //! Get number of sessions in group.
     size_t num_sessions() const;
@@ -129,8 +128,7 @@ class ReceiverSessionGroup : public core::NonCopyable<>, private rtcp::IParticip
     virtual void halt_recv_stream(packet::stream_source_t send_source_id);
 
     status::StatusCode route_transport_packet_(const packet::PacketPtr& packet);
-    status::StatusCode route_control_packet_(const packet::PacketPtr& packet,
-                                             core::nanoseconds_t current_time);
+    status::StatusCode route_control_packet_(const packet::PacketPtr& packet);
 
     bool can_create_session_(const packet::PacketPtr& packet);