Hold worker thread until baseband is transmitted

lib/include/srsran/interfaces/phy_common_interface.h

@@ -23,6 +23,35 @@ namespace srsran {
  */
 class phy_common_interface
 {
+private:
+  std::mutex              tx_mutex;        ///< Protect Tx attributes
+  std::condition_variable tx_cvar;         ///< Tx condition variable
+  bool                    tx_hold = false; ///< Hold threads until the signal is transmitted
+
+protected:
+  /**
+   * @brief Waits for the last worker to call `last_worker()` to prevent that the current SF worker is released and
+   * overwrites the transmit signal prior transmission
+   */
+  void wait_last_worker()
+  {
+    std::unique_lock<std::mutex> lock(tx_mutex);
+    tx_hold = true;
+    while (tx_hold) {
+      tx_cvar.wait(lock);
+    }
+  }
+
+  /**
+   * @brief Notifies the last SF worker transmitted the baseband and all the workers waiting are released
+   */
+  void last_worker()
+  {
+    std::unique_lock<std::mutex> lock(tx_mutex);
+    tx_hold = false;
+    tx_cvar.notify_all();
+  }
+
 public:
   /**
    * @brief Describes a worker context

srsenb/src/phy/phy_common.cc

@@ -119,6 +119,12 @@ void phy_common::worker_end(const worker_context_t& w_ctx, const bool& tx_enable
   if (not w_ctx.last) {
     // Release semaphore and let next worker to get in
     semaphore.release();
+
+    // Wait for the last worker to finish
+    if (tx_enable) {
+      wait_last_worker();
+    }
+
     return;
   }
 
@@ -136,6 +142,9 @@ void phy_common::worker_end(const worker_context_t& w_ctx, const bool& tx_enable
   // Reset transmit buffer
   tx_buffer = {};
 
+  // Notify this is the last worker
+  last_worker();
+
   // Allow next TTI to transmit
   semaphore.release();
 }

srsue/src/phy/phy_common.cc

@@ -539,15 +539,21 @@ void phy_common::worker_end(const worker_context_t& w_ctx, const bool& tx_enable
   // For each channel set or combine baseband
   if (tx_enable) {
     tx_buffer.set_combine(buffer);
-  }
 
-  // Flag transmit enabled
+    // Flag transmit enabled
-  tx_enabled = true;
+    tx_enabled = true;
+  }
 
   // If the current worker is not the last one, skip transmission
   if (not w_ctx.last) {
     // Release semaphore and let next worker to get in
     semaphore.release();
+
+    // If this worker transmitted, hold the worker until last SF worker finishes
+    if (tx_enable) {
+      wait_last_worker();
+    }
+
     return;
   }
 
@@ -594,6 +600,9 @@ void phy_common::worker_end(const worker_context_t& w_ctx, const bool& tx_enable
     }
   }
 
+  // Notify that last SF worker finished
+  last_worker();
+
   // Allow next TTI to transmit
   semaphore.release();
 }