Merge pull request #254 from softwareradiosystems/tx_mutex_fix

Change TX mutex to semaphores (mutex implementation was violating loc…

srsenb/hdr/phy/phch_common.h

@@ -28,6 +28,7 @@
 #define SRSENB_PHCH_COMMON_H
 
 #include <map>
+#include <semaphore.h>
 #include "srslte/interfaces/enb_interfaces.h"
 #include "srslte/interfaces/enb_metrics_interface.h"
 #include "srslte/common/gen_mch_tables.h"
@@ -36,6 +37,7 @@
 #include "srslte/common/thread_pool.h"
 #include "srslte/radio/radio.h"
 #include <string.h>
+
 namespace srsenb {
 
 typedef struct {
@@ -74,28 +76,15 @@ class phch_common
 public:
  
 
-  phch_common(uint32_t max_mutex_) : tx_mutex(max_mutex_) {
+  phch_common(uint32_t nof_workers);
-    nof_mutex = 0;
+  ~phch_common();
-    max_mutex = max_mutex_; 
+
-    params.max_prach_offset_us = 20;
+  void set_nof_workers(uint32_t nof_workers);
-    radio = NULL;
-    mac = NULL;
-    is_first_tx = false;
-    is_first_of_burst = false;
-    pdsch_p_b = 0;
-    nof_workers = 0;
-    bzero(&pusch_cfg, sizeof(pusch_cfg));
-    bzero(&hopping_cfg, sizeof(hopping_cfg));
-    bzero(&pucch_cfg, sizeof(pucch_cfg));
-    bzero(&ul_grants, sizeof(ul_grants));
-  }
 
   bool init(srslte_cell_t *cell, srslte::radio *radio_handler, mac_interface_phy *mac);  
   void reset(); 
   void stop();
   
-  void set_nof_mutex(uint32_t nof_mutex); 
-
   void worker_end(uint32_t tx_mutex_cnt, cf_t *buffer[SRSLTE_MAX_PORTS], uint32_t nof_samples, srslte_timestamp_t tx_time);
 
   // Common objects
@@ -148,13 +137,12 @@ public:
   
 
 private:
-  std::vector<pthread_mutex_t>    tx_mutex; 
+  std::vector<sem_t>    tx_sem;
   bool            is_first_tx;
   bool            is_first_of_burst; 
 
   uint32_t        nof_workers;
-  uint32_t        nof_mutex;
+  uint32_t        max_workers;
-  uint32_t        max_mutex;
 
   pthread_mutex_t user_mutex;
   

srsenb/hdr/phy/phch_worker.h

@@ -46,7 +46,7 @@ public:
   void  reset();
   
   cf_t *get_buffer_rx(uint32_t antenna_idx);
-  void set_time(uint32_t tti, uint32_t tx_mutex_cnt, srslte_timestamp_t tx_time);
+  void set_time(uint32_t tti, uint32_t tx_worker_cnt, srslte_timestamp_t tx_time);
   
   int  add_rnti(uint16_t rnti);
   void rem_rnti(uint16_t rnti);
@@ -93,8 +93,9 @@ private:
   cf_t          *signal_buffer_rx[SRSLTE_MAX_PORTS];
   cf_t          *signal_buffer_tx[SRSLTE_MAX_PORTS];
   uint32_t       tti_rx, tti_tx_dl, tti_tx_ul;
-  uint32_t       sf_rx, sf_tx, tx_mutex_cnt;
+  uint32_t       sf_rx, sf_tx;
   uint32_t       t_rx, t_tx_dl, t_tx_ul;
+  uint32_t       tx_worker_cnt;
   srslte_enb_dl_t enb_dl;
   srslte_enb_ul_t enb_ul;
   srslte_softbuffer_tx_t temp_mbsfn_softbuffer;

srsenb/hdr/phy/txrx.h

@@ -50,8 +50,6 @@ public:
             uint32_t prio);
   void stop();
     
-  const static int MUTEX_X_WORKER = 4; 
-  
 private:
     
   void run_thread(); 
@@ -62,12 +60,12 @@ private:
   prach_worker         *prach; 
   phch_common          *worker_com;
 
-  uint32_t tx_mutex_cnt; 
-  uint32_t nof_tx_mutex; 
-  
   // Main system TTI counter   
   uint32_t tti;
 
+  uint32_t tx_worker_cnt;
+  uint32_t nof_workers;
+  
   bool running; 
 };
 

srsenb/src/phy/phch_common.cc

@@ -41,9 +41,35 @@ using namespace std;
 
 namespace srsenb {
 
-void phch_common::set_nof_mutex(uint32_t nof_mutex_) {
+phch_common::phch_common(uint32_t max_workers) : tx_sem(max_workers)
-  nof_mutex = nof_mutex_; 
+{
-  assert(nof_mutex <= max_mutex);
+  this->nof_workers = nof_workers;
+  params.max_prach_offset_us = 20;
+  radio = NULL;
+  mac = NULL;
+  is_first_tx = false;
+  is_first_of_burst = false;
+  pdsch_p_b = 0;
+  this->max_workers = max_workers;
+  bzero(&pusch_cfg, sizeof(pusch_cfg));
+  bzero(&hopping_cfg, sizeof(hopping_cfg));
+  bzero(&pucch_cfg, sizeof(pucch_cfg));
+  bzero(&ul_grants, sizeof(ul_grants));
+
+  for (uint32_t i=0;i<max_workers;i++) {
+    sem_init(&tx_sem[i], 0, 0); // All semaphores start blocked
+  }
+}
+
+phch_common::~phch_common() {
+  for (uint32_t i=0;i<max_workers;i++) {
+    sem_destroy(&tx_sem[i]);
+  }
+}
+
+void phch_common::set_nof_workers(uint32_t nof_workers)
+{
+  this->nof_workers = nof_workers;
 }
 
 void phch_common::reset() {
@@ -61,35 +87,42 @@ bool phch_common::init(srslte_cell_t *cell_, srslte::radio* radio_h_, mac_interf
   
   is_first_of_burst = true; 
   is_first_tx = true; 
-  for (uint32_t i=0;i<max_mutex;i++) {
-    pthread_mutex_init(&tx_mutex[i], NULL);
-  }
   reset();
   return true; 
 }
 
 void phch_common::stop() {
-  for (uint32_t i=0;i<nof_mutex;i++) {
+  for (uint32_t i=0;i<max_workers;i++) {
-    pthread_mutex_trylock(&tx_mutex[i]);
+    sem_post(&tx_sem[i]);
-    pthread_mutex_unlock(&tx_mutex[i]);
   }
 }
 
-void phch_common::worker_end(uint32_t tx_mutex_cnt, cf_t* buffer[SRSLTE_MAX_PORTS], uint32_t nof_samples, srslte_timestamp_t tx_time)
+/* The transmission of UL subframes must be in sequence. The correct sequence is guaranteed by a chain of N semaphores,
+ * one per TTI%nof_workers. Each threads waits for the semaphore for the current thread and after transmission allows
+ * next TTI to be transmitted
+ *
+ * Each worker uses this function to indicate that all processing is done and data is ready for transmission or
+ * there is no transmission at all (tx_enable). In that case, the end of burst message will be sent to the radio
+ */
+void phch_common::worker_end(uint32_t tti, cf_t* buffer[SRSLTE_MAX_PORTS], uint32_t nof_samples, srslte_timestamp_t tx_time)
 {
 
-  // Wait previous TTIs to be transmitted 
+  // This variable is not protected but it is very unlikely that 2 threads arrive here simultaneously since at the beginning
+  // there is no workload and threads are separated by 1 ms
   if (is_first_tx) {
     is_first_tx = false;
-  } else {
+    // Allow my own transmission if I'm the first to transmit
-    pthread_mutex_lock(&tx_mutex[tx_mutex_cnt%nof_mutex]);
+    sem_post(&tx_sem[tti%nof_workers]);
   }
 
-  radio->set_tti(tx_mutex_cnt);
+  // Wait for the green light to transmit in the current TTI
+  sem_wait(&tx_sem[tti%nof_workers]);
+
+  radio->set_tti(tti);
   radio->tx((void **) buffer, nof_samples, tx_time);
 
-  // Trigger next transmission 
+  // Allow next TTI to transmit
-  pthread_mutex_unlock(&tx_mutex[(tx_mutex_cnt+1)%nof_mutex]);
+  sem_post(&tx_sem[(tti+1)%nof_workers]);
 
   // Trigger MAC clock
   mac->tti_clock();

srsenb/src/phy/phch_worker.cc

@@ -206,7 +206,7 @@ cf_t* phch_worker::get_buffer_rx(uint32_t antenna_idx)
   return signal_buffer_rx[antenna_idx];
 }
 
-void phch_worker::set_time(uint32_t tti_, uint32_t tx_mutex_cnt_, srslte_timestamp_t tx_time_)
+void phch_worker::set_time(uint32_t tti_, uint32_t tx_worker_cnt_, srslte_timestamp_t tx_time_)
 {
   tti_rx       = tti_; 
   tti_tx_dl    = TTI_TX(tti_rx);
@@ -219,7 +219,7 @@ void phch_worker::set_time(uint32_t tti_, uint32_t tx_mutex_cnt_, srslte_timesta
   t_rx         = TTIMOD(tti_rx);
   t_tx_ul      = TTIMOD(tti_tx_ul);
 
-  tx_mutex_cnt = tx_mutex_cnt_;
+  tx_worker_cnt = tx_worker_cnt_;
   memcpy(&tx_time, &tx_time_, sizeof(srslte_timestamp_t));
 }
 
@@ -483,7 +483,7 @@ void phch_worker::work_imp()
   pthread_mutex_unlock(&mutex);
 
   Debug("Sending to radio\n");
-  phy->worker_end(tx_mutex_cnt, signal_buffer_tx, SRSLTE_SF_LEN_PRB(phy->cell.nof_prb), tx_time);
+  phy->worker_end(tx_worker_cnt, signal_buffer_tx, SRSLTE_SF_LEN_PRB(phy->cell.nof_prb), tx_time);
 
   is_worker_running = false;
 

srsenb/src/phy/phy.cc

@@ -48,7 +48,7 @@ namespace srsenb {
 
 phy::phy() : workers_pool(MAX_WORKERS), 
              workers(MAX_WORKERS), 
-             workers_common(txrx::MUTEX_X_WORKER*MAX_WORKERS),
+             workers_common(MAX_WORKERS),
              nof_workers(0)
 {
   radio_handler = NULL;
@@ -141,10 +141,10 @@ bool phy::init(phy_args_t *args,
 void phy::stop()
 {  
   tx_rx.stop();  
-  workers_common.stop();
   for (uint32_t i=0;i<nof_workers;i++) {
     workers[i].stop();
   }
+  workers_common.stop();
   workers_pool.stop();
   prach.stop();
 }

srsenb/src/phy/txrx.cc

@@ -42,7 +42,7 @@ using namespace std;
 
 namespace srsenb {
 
-txrx::txrx() : tx_mutex_cnt(0), nof_tx_mutex(0), tti(0) {
+txrx::txrx() : tx_worker_cnt(0), nof_workers(0), tti(0) {
   running = false;   
   radio_h = NULL; 
   log_h   = NULL; 
@@ -58,11 +58,11 @@ bool txrx::init(srslte::radio* radio_h_, srslte::thread_pool* workers_pool_, phc
   workers_pool = workers_pool_;
   worker_com   = worker_com_;
   prach        = prach_; 
-  tx_mutex_cnt = 0; 
+  tx_worker_cnt = 0; 
   running      = true; 
   
-  nof_tx_mutex = MUTEX_X_WORKER*workers_pool->get_nof_workers();
+  nof_workers = workers_pool->get_nof_workers();
-  worker_com->set_nof_mutex(nof_tx_mutex);
+  worker_com->set_nof_workers(nof_workers);
     
   start(prio_);
   return true; 
@@ -126,12 +126,12 @@ void txrx::run_thread()
       srslte_timestamp_add(&tx_time, 0, HARQ_DELAY_MS*1e-3);
       
       Debug("Settting TTI=%d, tx_mutex=%d, tx_time=%ld:%f to worker %d\n", 
-            tti, tx_mutex_cnt, 
+            tti, tx_worker_cnt, 
             tx_time.full_secs, tx_time.frac_secs,
             worker->get_id());
       
-      worker->set_time(tti, tx_mutex_cnt, tx_time);
+      worker->set_time(tti, tx_worker_cnt, tx_time);
-      tx_mutex_cnt = (tx_mutex_cnt+1)%nof_tx_mutex;
+      tx_worker_cnt = (tx_worker_cnt+1)%nof_workers;
       
       // Trigger phy worker execution
       workers_pool->start_worker(worker);       

srsue/hdr/phy/phch_common.h

@@ -33,6 +33,7 @@
 #include <pthread.h>
 #include <string.h>
 #include <vector>
+#include <semaphore.h>
 #include "srslte/srslte.h"
 #include "srslte/interfaces/ue_interfaces.h"
 #include "srslte/radio/radio.h"
@@ -116,7 +117,8 @@ typedef struct {
     uint8_t last_ri;
     uint8_t last_pmi;
 
-    phch_common(uint32_t max_mutex = 3);
+    phch_common(uint32_t max_workers);
+    ~phch_common();
     void init(phy_interface_rrc::phy_cfg_t *config,
               phy_args_t  *args,
               srslte::log *_log,
@@ -144,8 +146,7 @@ typedef struct {
 
     void worker_end(uint32_t tti, bool tx_enable, cf_t *buffer, uint32_t nof_samples, srslte_timestamp_t tx_time);
 
-    void set_nof_mutex(uint32_t nof_mutex);
+    void set_nof_workers(uint32_t nof_workers);
-
     bool sr_enabled;
     int  sr_last_tx_tti;
 
@@ -179,7 +180,9 @@ typedef struct {
 
     
     
-    std::vector<pthread_mutex_t>    tx_mutex;
+    std::vector<sem_t>    tx_sem;
+    uint32_t              nof_workers;
+    uint32_t              max_workers;
 
     bool               is_first_of_burst;
     srslte::radio      *radio_h;
@@ -208,10 +211,6 @@ typedef struct {
     
     bool            is_first_tx;
 
-    uint32_t        nof_workers;
-    uint32_t        nof_mutex;
-    uint32_t        max_mutex;
-
     srslte_cell_t   cell;
 
     dl_metrics_t    dl_metrics;

srsue/hdr/phy/phch_recv.h

@@ -83,7 +83,6 @@ public:
   void    force_freq(float dl_freq, float ul_freq);
 
   // Other functions
-  const static int MUTEX_X_WORKER = 4;
   double set_rx_gain(double gain);
   int radio_recv_fnc(cf_t *data[SRSLTE_MAX_PORTS], uint32_t nsamples, srslte_timestamp_t *rx_time);
   int scell_recv_fnc(cf_t *data[SRSLTE_MAX_PORTS], uint32_t nsamples, srslte_timestamp_t *rx_time);
@@ -439,8 +438,8 @@ private:
   uint32_t      tti;
   bool          do_agc;
 
-  uint32_t      nof_tx_mutex;
+  uint32_t      tx_worker_cnt;
-  uint32_t      tx_mutex_cnt;
+  uint32_t      nof_workers;
 
   float         ul_dl_factor;
   int           current_earfcn;

srsue/hdr/phy/phch_worker.h

@@ -52,7 +52,7 @@ public:
 
   /* Functions used by main PHY thread */
   cf_t* get_buffer(uint32_t antenna_idx);
-  void  set_tti(uint32_t tti, uint32_t tx_tti); 
+  void  set_tti(uint32_t tti, uint32_t tx_worker_cnt);
   void  set_tx_time(srslte_timestamp_t tx_time, uint32_t next_offset);
   void  set_prach(cf_t *prach_ptr, float prach_power);
   void  set_cfo(float cfo);

srsue/src/phy/phch_common.cc

@@ -39,15 +39,14 @@ namespace srsue {
 
 cf_t zeros[50000];
 
-phch_common::phch_common(uint32_t max_mutex_) : tx_mutex(max_mutex_)
+phch_common::phch_common(uint32_t max_workers) : tx_sem(max_workers)
 {
   config    = NULL; 
   args      = NULL; 
   log_h     = NULL; 
   radio_h   = NULL; 
   mac       = NULL;
-  max_mutex = max_mutex_;
+  this->max_workers = max_workers;
-  nof_mutex = 0;
   rx_gain_offset = 0;
   last_ri = 0;
   last_pmi = 0;
@@ -65,11 +64,9 @@ phch_common::phch_common(uint32_t max_mutex_) : tx_mutex(max_mutex_)
 
   bzero(zeros, 50000*sizeof(cf_t));
 
-  // FIXME: This is an ugly fix to avoid the TX filters to empty
+  for (uint32_t i=0;i<max_workers;i++) {
-  /*
+    sem_init(&tx_sem[i], 0, 0); // All semaphores start blocked
-  for (int i=0;i<50000;i++) {
+  }
-    zeros[i] = 0.01*cexpf(((float) i/50000)*0.1*_Complex_I);
-  }*/
 
   reset();
 
@@ -77,6 +74,19 @@ phch_common::phch_common(uint32_t max_mutex_) : tx_mutex(max_mutex_)
   mcch_configured  = false;
 }
 
+phch_common::~phch_common() {
+  for (uint32_t i=0;i<max_workers;i++) {
+    sem_post(&tx_sem[i]);
+  }
+  for (uint32_t i=0;i<max_workers;i++) {
+    sem_destroy(&tx_sem[i]);
+  }
+}
+
+void phch_common::set_nof_workers(uint32_t nof_workers) {
+  this->nof_workers = nof_workers;
+}
+  
 void phch_common::init(phy_interface_rrc::phy_cfg_t *_config, phy_args_t *_args, srslte::log *_log, srslte::radio *_radio, rrc_interface_phy *_rrc, mac_interface_phy *_mac)
 {
   log_h     = _log; 
@@ -87,15 +97,6 @@ void phch_common::init(phy_interface_rrc::phy_cfg_t *_config, phy_args_t *_args,
   args      = _args; 
   is_first_tx = true; 
   sr_last_tx_tti = -1;
-  
-  for (uint32_t i=0;i<nof_mutex;i++) {
-    pthread_mutex_init(&tx_mutex[i], NULL);
-  }
-}
-
-void phch_common::set_nof_mutex(uint32_t nof_mutex_) {
-  nof_mutex = nof_mutex_; 
-  assert(nof_mutex <= max_mutex);
 }
 
 bool phch_common::ul_rnti_active(uint32_t tti) {
@@ -231,22 +232,29 @@ bool phch_common::is_any_pending_ack() {
   return false;
 }
 
-/* The transmisison of UL subframes must be in sequence. Each worker uses this function to indicate
+/* The transmission of UL subframes must be in sequence. The correct sequence is guaranteed by a chain of N semaphores,
- * that all processing is done and data is ready for transmission or there is no transmission at all (tx_enable). 
+ * one per TTI%max_workers. Each threads waits for the semaphore for the current thread and after transmission allows
- * In that case, the end of burst message will be send to the radio 
+ * next TTI to be transmitted
+ *
+ * Each worker uses this function to indicate that all processing is done and data is ready for transmission or
+ * there is no transmission at all (tx_enable). In that case, the end of burst message will be sent to the radio
  */
 void phch_common::worker_end(uint32_t tti, bool tx_enable, 
                                    cf_t *buffer, uint32_t nof_samples, 
                                    srslte_timestamp_t tx_time)
 {
 
-  // Wait previous TTIs to be transmitted 
+  // This variable is not protected but it is very unlikely that 2 threads arrive here simultaneously since at the beginning
+  // there is no workload and threads are separated by 1 ms
   if (is_first_tx) {
     is_first_tx = false;
-  } else {
+    // Allow my own transmission if I'm the first to transmit
-    pthread_mutex_lock(&tx_mutex[tti%nof_mutex]);
+    sem_post(&tx_sem[tti%nof_workers]);
   }
 
+  // Wait for the green light to transmit in the current TTI
+  sem_wait(&tx_sem[tti%nof_workers]);
+
   radio_h->set_tti(tti);
   if (tx_enable) {
     radio_h->tx_single(buffer, nof_samples, tx_time);
@@ -263,8 +271,9 @@ void phch_common::worker_end(uint32_t tti, bool tx_enable,
       }
     }
   }
-  // Trigger next transmission 
+
-  pthread_mutex_unlock(&tx_mutex[(tti+1)%nof_mutex]);
+  // Allow next TTI to transmit
+  sem_post(&tx_sem[(tti+1)%nof_workers]);
 }    
 
 

srsue/src/phy/phch_recv.cc

@@ -81,8 +81,8 @@ void phch_recv::init(srslte::radio_multi *_radio_handler, mac_interface_phy *_ma
     return;
   }
 
-  nof_tx_mutex = MUTEX_X_WORKER * workers_pool->get_nof_workers();
+  nof_workers = workers_pool->get_nof_workers();
-  worker_com->set_nof_mutex(nof_tx_mutex);
+  worker_com->set_nof_workers(nof_workers);
 
   // Initialize cell searcher
   search_p.init(sf_buffer, log_h, nof_rx_antennas, this);
@@ -128,7 +128,7 @@ void phch_recv::reset()
   radio_overflow_return = false;
   in_sync_cnt = 0;
   out_of_sync_cnt = 0;
-  tx_mutex_cnt = 0;
+  tx_worker_cnt = 0;
   time_adv_sec = 0;
   next_offset  = 0;
   srate_mode = SRATE_NONE;
@@ -454,13 +454,13 @@ void phch_recv::run_thread()
 
               worker->set_prach(prach_ptr?&prach_ptr[prach_sf_cnt*SRSLTE_SF_LEN_PRB(cell.nof_prb)]:NULL, prach_power);
               worker->set_cfo(get_tx_cfo());
-              worker->set_tti(tti, tx_mutex_cnt);
+              worker->set_tti(tti, tx_worker_cnt);
               worker->set_tx_time(tx_time, next_offset);
               next_offset  = 0;
               if (next_time_adv_sec != time_adv_sec) {
                 time_adv_sec = next_time_adv_sec;
               }
-              tx_mutex_cnt = (tx_mutex_cnt+1) % nof_tx_mutex;
+              tx_worker_cnt = (tx_worker_cnt+1) % nof_workers;
 
               // Advance/reset prach subframe pointer
               if (prach_ptr) {

srsue/src/phy/phch_worker.cc

@@ -193,10 +193,10 @@ cf_t* phch_worker::get_buffer(uint32_t antenna_idx)
   return signal_buffer[antenna_idx]; 
 }
 
-void phch_worker::set_tti(uint32_t tti_, uint32_t tx_tti_)
+void phch_worker::set_tti(uint32_t tti_, uint32_t tx_worker_cnt)
 {
   tti    = tti_;
-  tx_tti = tx_tti_;
+  tx_tti = tx_worker_cnt;
   log_h->step(tti);
   if (log_phy_lib_h) {
     log_phy_lib_h->step(tti);

srsue/src/phy/phy.cc

@@ -52,7 +52,7 @@ namespace srsue {
 
 phy::phy() : workers_pool(MAX_WORKERS), 
              workers(MAX_WORKERS), 
-             workers_common(phch_recv::MUTEX_X_WORKER*MAX_WORKERS),nof_coworkers(0)
+             workers_common(MAX_WORKERS),nof_coworkers(0)
 {
 }