move receiver thread to video_rxtx

sender thread is dispatched here as well so to be consistent

Moreover the thread init is at multiple places so it cannot be forgotten.

Author: MartinPulec

src/hd-rum-translator/hd-rum-decompress.cpp

@@ -88,7 +88,6 @@ struct state_transcoder_decompress final : public frame_recv_delegate {
 
         struct display *display = nullptr;
         struct control_state *control = nullptr;
-        thread         receiver_thread;
 
         void frame_arrived(struct video_frame *f, struct audio_frame *a) override;
 
@@ -228,7 +227,6 @@ void *hd_rum_decompress_init(struct module *parent, struct hd_rum_output_conf co
                 assert(s->video_rxtx);
 
                 s->worker_thread = thread(&state_transcoder_decompress::worker, s);
-                s->receiver_thread = thread(&video_rxtx::receiver_thread, s->video_rxtx);
                 display_run_new_thread(s->display);
 
                 if (capture_filter_init(parent, capture_filter, &s->capture_filter_state) != 0) {
@@ -246,8 +244,6 @@ void *hd_rum_decompress_init(struct module *parent, struct hd_rum_output_conf co
 void hd_rum_decompress_done(void *state) {
         auto *s = static_cast<state_transcoder_decompress *>(state);
 
-        s->receiver_thread.join();
-
         {
                 unique_lock<mutex> l(s->lock);
                 s->received_frame.push({});

src/main.cpp

@@ -1335,7 +1335,6 @@ int main(int argc, char *argv[])
 #endif
         ug_options opt{};
 
-        pthread_t receiver_thread_id = PTHREAD_NULL;
         pthread_t capture_thread_id  = PTHREAD_NULL;
 
         unsigned display_flags = 0;
@@ -1500,24 +1499,6 @@ int main(int argc, char *argv[])
                 goto cleanup;
         }
 
-        if ((opt.video.rxtx_mode & MODE_RECEIVER) != 0U) {
-                if (!uv.state_video_rxtx->supports_receiving()) {
-                        fprintf(
-                            stderr,
-                            "Selected RX/TX mode doesn't support receiving.\n");
-                        exit_uv(EXIT_FAILURE);
-                        goto cleanup;
-                }
-                // init module here so as it is capable of receiving messages
-                if (pthread_create(&receiver_thread_id, NULL,
-                                   video_rxtx::receiver_thread,
-                                   (void *) uv.state_video_rxtx) != 0) {
-                        perror("Unable to create display thread!\n");
-                        exit_uv(EXIT_FAILURE);
-                        goto cleanup;
-                }
-        }
-
         if ((opt.video.rxtx_mode & MODE_SENDER) != 0U) {
                 if (pthread_create(&capture_thread_id, NULL, capture_thread,
                                    (void *) &uv) != 0) {
@@ -1541,10 +1522,6 @@ int main(int argc, char *argv[])
         display_run_mainloop(uv.display_device);
 
 cleanup:
-        if (!pthread_equal(receiver_thread_id, PTHREAD_NULL)) {
-                pthread_join(receiver_thread_id, NULL);
-        }
-
         if (!pthread_equal(capture_thread_id, PTHREAD_NULL)) {
                 pthread_join(capture_thread_id, NULL);
         }

src/video_capture/ug_input.cpp

@@ -79,7 +79,6 @@ struct ug_input_state final : public frame_recv_delegate {
         struct display *display = nullptr;
 
         void frame_arrived(struct video_frame *f, struct audio_frame *a) override;
-        thread         receiver_thread;
         unique_ptr<struct video_rxtx> video_rxtx;
         struct state_audio *audio = nullptr;
 
@@ -204,7 +203,6 @@ static int vidcap_ug_input_init(const struct vidcap_params *cap_params, void **s
         }
         s->t0 = steady_clock::now();
 
-        s->receiver_thread = thread(&video_rxtx::receiver_thread, s->video_rxtx.get());
         display_run_new_thread(s->display);
 
         *state = s;
@@ -216,7 +214,6 @@ static void vidcap_ug_input_done(void *state)
         auto s = (ug_input_state *) state;
 
         audio_join(s->audio);
-        s->receiver_thread.join();
 
         display_put_frame(s->display, NULL, 0);
         display_join(s->display);

src/video_rxtx.cpp

@@ -119,8 +119,12 @@ video_rxtx::video_rxtx(const char                *protocol_name,
 
 video_rxtx::~video_rxtx() noexcept
 {
+        if (!pthread_equal(m_receiver_thread_id, PTHREAD_NULL)) {
+                pthread_join(m_receiver_thread_id, nullptr);
+        }
+
         join();
-        if (!m_poisoned && m_compression) {
+        if (!m_sender_poisoned && m_compression != nullptr) {
                 send(NULL);
                 compress_pop(m_compression);
         }
@@ -135,15 +139,20 @@ video_rxtx::~video_rxtx() noexcept
 void
 video_rxtx::join() noexcept
 {
-        if (m_joined) {
+        if (!pthread_equal(m_receiver_thread_id, PTHREAD_NULL)) {
+                pthread_join(m_receiver_thread_id, nullptr);
+                m_receiver_thread_id = PTHREAD_NULL;
+        }
+
+        if (m_sender_joined) {
                 return;
         }
         send(NULL); // pass poisoned pill
-        pthread_join(m_thread_id, NULL);
+        pthread_join(m_sender_thread_id, NULL);
         if (m_impl_funcs != nullptr && m_impl_funcs->join_sender != nullptr) {
                 m_impl_funcs->join_sender(m_impl_state);
         }
-        m_joined = true;
+        m_sender_joined = true;
 }
 
 const char *
@@ -162,11 +171,11 @@ video_rxtx::get_long_name(string const &short_name) noexcept
 void
 video_rxtx::send(shared_ptr<video_frame> frame) noexcept
 {
-        if (!frame && m_poisoned) {
+        if (!frame && m_sender_poisoned) {
                 return;
         }
         if (!frame) {
-                m_poisoned = true;
+                m_sender_poisoned = true;
         } else {
                 m_input_codec = frame->color_spec;
         }
@@ -257,10 +266,23 @@ video_rxtx::create(string const &proto, const struct vrxtx_params *params,
                 return nullptr;
         }
 
-        int rc = pthread_create(&ret->m_thread_id, nullptr, video_rxtx::sender_thread,
+        if ((params->rxtx_mode & MODE_RECEIVER) != 0U) {
+                if (ret->m_impl_funcs->receiver_routine == nullptr) {
+                        MSG(ERROR,
+                            "Selected RX/TX mode doesn't support receiving.\n");
+                        delete ret;
+                        return nullptr;
+                }
+                int rc = pthread_create(&ret->m_receiver_thread_id, nullptr,
+                                        ret->m_impl_funcs->receiver_routine,
+                                        ret->m_impl_state);
+                assert(rc == 0);
+        }
+
+        int rc = pthread_create(&ret->m_sender_thread_id, nullptr, video_rxtx::sender_thread,
                            (void *) ret);
         assert(rc == 0);
-        ret->m_joined = false;
+        ret->m_sender_joined = false;
 
         return ret;
 }

src/video_rxtx.hpp

@@ -102,6 +102,7 @@ struct video_rxtx_info {
         void *(*create)(const struct vrxtx_params *params,
                         const struct common_opts  *common);
         void (*done)(void *state);
+        /// this may be set optional if we had some receive-only mod
         void (*send_frame)(void *state, std::shared_ptr<video_frame>);
 
         // following callbacks are optional
@@ -118,15 +119,6 @@ struct video_rxtx {
         virtual ~video_rxtx() noexcept;
         void               send(std::shared_ptr<struct video_frame>) noexcept;
         static const char *get_long_name(std::string const &short_name) noexcept;
-        static void       *receiver_thread(void *arg) noexcept
-        {
-                video_rxtx *rxtx = static_cast<video_rxtx *>(arg);
-                return rxtx->m_impl_funcs->receiver_routine(rxtx->m_impl_state);
-        }
-        bool supports_receiving() noexcept
-        {
-                return m_impl_funcs->receiver_routine != nullptr;
-        }
         /**
          * If overridden, children must call also video_rxtx::join()
          */
@@ -160,12 +152,14 @@ struct video_rxtx {
         struct compress_state *m_compression = nullptr;
         pthread_mutex_t m_lock;
 
-        pthread_t m_thread_id = PTHREAD_NULL;
-        bool      m_poisoned  = false;
-        bool      m_joined    = true;
+        pthread_t m_sender_thread_id   = PTHREAD_NULL;
+        bool      m_sender_poisoned    = false;
+        bool      m_sender_joined      = true;
+        pthread_t m_receiver_thread_id = PTHREAD_NULL;
 
         video_desc       m_video_desc{};
         std::atomic<codec_t> m_input_codec{};
+
 };
 
 const char *vrxtx_get_compression(const char *video_protocol,