roc_recv: --real-time cli parameter

NetworkLoop and IoPump got realtime
priority as a parameter, and if it's
not 0, attempt to elevate their threads
priorities.

Author: baranovmv

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

@@ -46,12 +46,12 @@ 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: set realtime priority");
+    roc_log(LogInfo, "thread: set realtime priority");
     if (int err = pthread_setschedparam(pthread_self(), SCHED_RR, &param)) {
         roc_log(LogError,
                 "thread: can't set realtime priority: pthread_setschedparam(): %s",

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_ATTR_NODISCARD static bool enable_realtime();
+    ROC_ATTR_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_loop.cpp

@@ -104,10 +104,13 @@ NetworkLoop::Tasks::ResolveEndpointAddress::get_address() const {
 
 NetworkLoop::NetworkLoop(core::IPool& packet_pool,
                          core::IPool& buffer_pool,
+                         const int realtime_prio,
                          core::IArena& arena)
     : packet_factory_(packet_pool, buffer_pool)
+    , realtime_prio_(realtime_prio)
     , arena_(arena)
     , started_(false)
+    , thr_init_cond_(thr_init_mutex_)
     , loop_initialized_(false)
     , stop_sem_initialized_(false)
     , task_sem_initialized_(false)
@@ -140,17 +143,15 @@ NetworkLoop::NetworkLoop(core::IPool& packet_pool,
     task_sem_.data = this;
     task_sem_initialized_ = true;
 
-    if (!enable_realtime()) {
-        roc_log(LogInfo,
-                "network loop: can't set realtime priority of network thread. May need "
-                "to be root");
-    }
     if (!(started_ = Thread::start())) {
         init_status_ = status::StatusErrThread;
         return;
     }
 
-    init_status_ = status::StatusOK;
+    while (init_status_ == status::NoStatus) {
+        core::Mutex::Lock lock(thr_init_mutex_);
+        thr_init_cond_.wait();
+    }
 }
 
 NetworkLoop::~NetworkLoop() {
@@ -285,6 +286,21 @@ void NetworkLoop::handle_resolved(ResolverRequest& req) {
 
 void NetworkLoop::run() {
     roc_log(LogDebug, "network loop: starting event loop");
+    if (realtime_prio_ > 0 && !enable_realtime(realtime_prio_)) {
+        core::Mutex::Lock lock(thr_init_mutex_);
+
+        roc_log(LogError,
+                "network loop: can't set realtime priority of network thread. May need "
+                "to be root");
+        init_status_ = status::StatusFailedRealtime;
+        thr_init_cond_.signal();
+    } else {
+        core::Mutex::Lock lock(thr_init_mutex_);
+
+        roc_log(LogDebug, "network loop: elevated realtime priority");
+        init_status_ = status::StatusOK;
+        thr_init_cond_.signal();
+    }
 
     int err = uv_run(&loop_, UV_RUN_DEFAULT);
     if (err != 0) {

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.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"
@@ -54,6 +55,8 @@ class NetworkLoop : private ITerminateHandler,
     //! Opaque port handle.
     typedef struct PortHandle* PortHandle;
 
+    enum { DEFAULT_PRIORITY = 0 };
+
     //! Subclasses for specific tasks.
     class Tasks {
     public:
@@ -189,7 +192,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(core::IPool& packet_pool,
+                core::IPool& buffer_pool,
+                const int realtime_prio,
+                core::IArena& arena);
 
     //! Destroy. Stop all receivers and senders.
     //! @remarks
@@ -248,10 +254,14 @@ class NetworkLoop : private ITerminateHandler,
     void task_resolve_endpoint_address_(NetworkTask&);
 
     packet::PacketFactory packet_factory_;
+    const uint8_t realtime_prio_;
     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

@@ -23,7 +23,7 @@ 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_(packet_pool_, packet_buffer_pool_, (int)config.realtime_prio, arena_)
     , control_loop_(network_loop_, arena_)
     , init_status_(status::NoStatus) {
     roc_log(LogDebug, "context: initializing");

src/internal_modules/roc_node/context.h

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

src/internal_modules/roc_rtcp/rtt_estimator.cpp

@@ -12,7 +12,9 @@
 namespace roc {
 namespace rtcp {
 
-RttEstimator::RttEstimator(const RttConfig &config, core::IArena &arena, dbgio::CsvDumper *dumper)
+RttEstimator::RttEstimator(const RttConfig& config,
+                           core::IArena& arena,
+                           dbgio::CsvDumper* dumper)
     : config_(config)
     , metrics_()
     , has_metrics_(false)

src/internal_modules/roc_rtcp/rtt_estimator.h

@@ -63,7 +63,7 @@ struct RttMetrics {
 class RttEstimator {
 public:
     //! Initialize.
-    RttEstimator(const RttConfig &config, core::IArena &arena, dbgio::CsvDumper *dumper);
+    RttEstimator(const RttConfig& config, core::IArena& arena, dbgio::CsvDumper* dumper);
 
     //! Check whether metrics are already available.
     bool has_metrics() const;

src/internal_modules/roc_sndio/io_pump.cpp

@@ -9,6 +9,7 @@
 #include "roc_sndio/io_pump.h"
 #include "roc_audio/sample_spec_to_str.h"
 #include "roc_core/log.h"
+#include "roc_core/thread.h"
 #include "roc_status/code_to_str.h"
 
 namespace roc {
@@ -67,11 +68,18 @@ status::StatusCode IoPump::init_status() const {
     return init_status_;
 }
 
-status::StatusCode IoPump::run() {
+status::StatusCode IoPump::run(const int realtime_priority) {
     roc_log(LogDebug, "io pump: starting main loop");
-
     status::StatusCode code = status::NoStatus;
 
+    if (realtime_priority > 0 && !core::Thread::enable_realtime(realtime_priority)) {
+        roc_log(LogError, "io pump: can't set realtime priority. May need to be root");
+        code = status::StatusFailedRealtime;
+        return code;
+    } else {
+        roc_log(LogDebug, "io pump: elevated realtime priority");
+    }
+
     for (;;) {
         // Transfer one frame from source to sink.
         if ((code = next_()) != status::StatusOK) {

src/internal_modules/roc_sndio/io_pump.h

@@ -59,7 +59,7 @@ class IoPump : public core::NonCopyable<> {
     //! @remarks
     //!  Run until the stop() is called or, if oneshot mode is enabled,
     //!  the source becomes inactive.
-    ROC_ATTR_NODISCARD status::StatusCode run();
+    ROC_ATTR_NODISCARD status::StatusCode run(const int realtime_priority = 0);
 
     //! Stop the pump.
     //! @remarks