Added PDSCH coworkers

lib/include/srslte/phy/phch/pdsch.h

@@ -89,6 +89,8 @@ typedef struct SRSLTE_API {
 
   srslte_sch_t dl_sch;
 
+  void *coworker_ptr;
+
 } srslte_pdsch_t;
 
 
@@ -167,6 +169,8 @@ SRSLTE_API void srslte_pdsch_set_max_noi(srslte_pdsch_t *q,
 
 SRSLTE_API float srslte_pdsch_last_noi(srslte_pdsch_t *q);
 
+SRSLTE_API int srslte_pdsch_enable_coworker(srslte_pdsch_t *q);
+
 SRSLTE_API uint32_t srslte_pdsch_last_noi_cw(srslte_pdsch_t *q,
                                              uint32_t cw_idx);
 

lib/src/phy/phch/pdsch.c

@@ -32,6 +32,9 @@
 #include <srslte/phy/phch/pdsch_cfg.h>
 #include <srslte/srslte.h>
 
+#include <pthread.h>
+#include <semaphore.h>
+
 #include "prb_dl.h"
 #include "srslte/phy/phch/pdsch.h"
 #include "srslte/phy/utils/debug.h"
@@ -52,6 +55,38 @@ extern int indices[100000];
 extern int indices_ptr; 
 #endif
 
+
+
+typedef struct {
+  /* Thread identifier: they must set before thread creation */
+  pthread_t pthread;
+  uint32_t cw_idx;
+  uint32_t tb_idx;
+  void *pdsch_ptr;
+  bool *ack;
+
+  /* Configuration Encoder/Decoder: they must be set before posting start semaphore */
+  srslte_pdsch_cfg_t *cfg;
+  srslte_sch_t dl_sch;
+  uint16_t rnti;
+
+  /* Encoder/Decoder data pointers: they must be set before posting start semaphore  */
+  uint8_t *data;
+  void *softbuffer;
+
+  /* Execution status */
+  int ret_status;
+
+  /* Semaphores */
+  sem_t start;
+  sem_t finish;
+
+  /* Thread kill flag */
+  bool quit;
+} srslte_pdsch_coworker_t;
+
+static void *srslte_pdsch_decode_thread (void *arg);
+
 int srslte_pdsch_cp(srslte_pdsch_t *q, cf_t *input, cf_t *output, srslte_ra_dl_grant_t *grant, uint32_t lstart_grant, uint32_t nsubframe, bool put)
 {
   uint32_t s, n, l, lp, lstart, lend, nof_refs;
@@ -283,7 +318,26 @@ int srslte_pdsch_init_enb(srslte_pdsch_t *q, uint32_t max_prb)
   return pdsch_init(q, max_prb, false, 0);
 }
 
+static void srslte_pdsch_disable_coworker(srslte_pdsch_t *q) {
+  srslte_pdsch_coworker_t *h = (srslte_pdsch_coworker_t *) q->coworker_ptr;
+  if (h) {
+    /* Stop threads */
+    h->quit = true;
+    sem_post(&h->start);
+
+    pthread_join(h->pthread, NULL);
+    pthread_detach(h->pthread);
+
+    srslte_sch_free(&h->dl_sch);
+
+    free(h);
+
+    q->coworker_ptr = NULL;
+  }
+}
+
 void srslte_pdsch_free(srslte_pdsch_t *q) {
+  srslte_pdsch_disable_coworker(q);
 
   for (int i = 0; i < SRSLTE_MAX_CODEWORDS; i++) {
 
@@ -612,9 +666,9 @@ static int srslte_pdsch_codeword_encode(srslte_pdsch_t *q, srslte_pdsch_cfg_t *c
   return SRSLTE_SUCCESS;
 }
 
-static int srslte_pdsch_codeword_decode(srslte_pdsch_t *q, srslte_pdsch_cfg_t *cfg,
+static int srslte_pdsch_codeword_decode(srslte_pdsch_t *q, srslte_pdsch_cfg_t *cfg, srslte_sch_t *dl_sch,
-                                               srslte_softbuffer_rx_t *softbuffer, uint16_t rnti, uint8_t *data,
+                                        srslte_softbuffer_rx_t *softbuffer, uint16_t rnti, uint8_t *data,
-                                               uint32_t codeword_idx, uint32_t tb_idx, bool *ack) {
+                                        uint32_t codeword_idx, uint32_t tb_idx, bool *ack) {
   srslte_ra_nbits_t *nbits = &cfg->nbits[tb_idx];
   srslte_ra_mcs_t *mcs = &cfg->grant.mcs[tb_idx];
   uint32_t rv = cfg->rv[tb_idx];
@@ -673,7 +727,7 @@ static int srslte_pdsch_codeword_decode(srslte_pdsch_t *q, srslte_pdsch_cfg_t *c
     }
 
     /* Return  */
-    ret = srslte_dlsch_decode2(&q->dl_sch, cfg, softbuffer, q->e[codeword_idx], data, tb_idx);
+    ret = srslte_dlsch_decode2(dl_sch, cfg, softbuffer, q->e[codeword_idx], data, tb_idx);
 
     q->last_nof_iterations[codeword_idx] = srslte_sch_last_noi(&q->dl_sch);
 
@@ -694,6 +748,34 @@ static int srslte_pdsch_codeword_decode(srslte_pdsch_t *q, srslte_pdsch_cfg_t *c
   return ret;
 }
 
+static void *srslte_pdsch_decode_thread(void *arg) {
+  srslte_pdsch_coworker_t *q = (srslte_pdsch_coworker_t *) arg;
+
+  INFO("[PDSCH Coworker] waiting for data\n");
+
+  sem_wait(&q->start);
+  while (!q->quit) {
+    q->ret_status = srslte_pdsch_codeword_decode(q->pdsch_ptr,
+                                                 q->cfg,
+                                                 &q->dl_sch,
+                                                 q->softbuffer,
+                                                 q->rnti,
+                                                 q->data,
+                                                 q->cw_idx,
+                                                 q->tb_idx,
+                                                 q->ack);
+
+    /* Post finish semaphore */
+    sem_post(&q->finish);
+
+    /* Wait for next loop */
+    sem_wait(&q->start);
+  }
+
+  pthread_exit(NULL);
+  return q;
+}
+
 /** Decodes the PDSCH from the received symbols
  */
 int srslte_pdsch_decode(srslte_pdsch_t *q,
@@ -776,7 +858,33 @@ int srslte_pdsch_decode(srslte_pdsch_t *q,
       /* Decode only if transport block is enabled and the default ACK is not true */
       if (cfg->grant.tb_en[tb_idx]) {
         if (!acks[tb_idx]) {
-          int ret = srslte_pdsch_codeword_decode(q, cfg, softbuffers[tb_idx], rnti, data[tb_idx], cw_idx, tb_idx, &acks[tb_idx]);
+          int ret = SRSLTE_SUCCESS;
+          if (SRSLTE_RA_DL_GRANT_NOF_TB(&cfg->grant) > 1 && tb_idx == 0 && q->coworker_ptr) {
+            srslte_pdsch_coworker_t *h = (srslte_pdsch_coworker_t *) q->coworker_ptr;
+
+            h->pdsch_ptr = q;
+            h->cfg = cfg;
+            h->softbuffer = softbuffers[tb_idx];
+            h->rnti = rnti;
+            h->data = data[tb_idx];
+            h->cw_idx = cw_idx;
+            h->tb_idx = tb_idx;
+            h->ack = &acks[tb_idx];
+            h->dl_sch.max_iterations = q->dl_sch.max_iterations;
+
+            sem_post(&h->start);
+
+          } else {
+            ret = srslte_pdsch_codeword_decode(q,
+                                               cfg,
+                                               &q->dl_sch,
+                                               softbuffers[tb_idx],
+                                               rnti,
+                                               data[tb_idx],
+                                               cw_idx,
+                                               tb_idx,
+                                               &acks[tb_idx]);
+          }
 
           /* Check if there has been any execution error */
           if (ret) {
@@ -788,6 +896,14 @@ int srslte_pdsch_decode(srslte_pdsch_t *q,
       }
     }
 
+    if (SRSLTE_RA_DL_GRANT_NOF_TB(&cfg->grant) > 1 && q->coworker_ptr) {
+      srslte_pdsch_coworker_t *h = (srslte_pdsch_coworker_t *) q->coworker_ptr;
+      sem_wait(&h->finish);
+      if (h->ret_status) {
+        ERROR("PDSCH Coworker Decoder: Error decoding");
+      }
+    }
+
     pdsch_decode_debug(q, cfg, sf_symbols, ce);
 
     return SRSLTE_SUCCESS;
@@ -933,6 +1049,45 @@ float srslte_pdsch_last_noi(srslte_pdsch_t *q) {
   return srslte_pdsch_last_noi_cw(q, 0);
 }
 
+int srslte_pdsch_enable_coworker(srslte_pdsch_t *q) {
+  int ret = SRSLTE_SUCCESS;
+
+  if (!q->coworker_ptr) {
+    srslte_pdsch_coworker_t *h = calloc(sizeof(srslte_pdsch_coworker_t), 1);
+
+    if (!h) {
+      ERROR("Allocating coworker");
+      ret = SRSLTE_ERROR;
+      goto clean;
+    }
+    q->coworker_ptr = h;
+
+    if (srslte_sch_init(&h->dl_sch)) {
+      ERROR("Initiating DL SCH");
+      ret = SRSLTE_ERROR;
+      goto clean;
+    }
+
+    if (sem_init(&h->start, 0, 0)) {
+      ERROR("Creating semaphore");
+      ret = SRSLTE_ERROR;
+      goto clean;
+    }
+    if (sem_init(&h->finish, 0, 0)) {
+      ERROR("Creating semaphore");
+      ret = SRSLTE_ERROR;
+      goto clean;
+    }
+    pthread_create(&h->pthread, NULL, srslte_pdsch_decode_thread, (void *) h);
+  }
+
+  clean:
+  if (ret) {
+    srslte_pdsch_disable_coworker(q);
+  }
+  return ret;
+}
+
 uint32_t srslte_pdsch_last_noi_cw(srslte_pdsch_t *q, uint32_t cw_idx) {
   return q->last_nof_iterations[cw_idx];
 }

lib/src/phy/phch/test/pdsch_test.c

@@ -60,6 +60,7 @@ int rv_idx[SRSLTE_MAX_CODEWORDS] = {0, 1};
 uint16_t rnti = 1234;
 uint32_t nof_rx_antennas = 1;
 bool tb_cw_swap = false;
+bool enable_coworker = false;
 uint32_t pmi = 0;
 char *input_file = NULL; 
 
@@ -79,12 +80,13 @@ void usage(char *prog) {
   printf("\t-a nof_rx_antennas [Default %d]\n", nof_rx_antennas);
   printf("\t-p pmi (multiplex only)  [Default %d]\n", pmi);
   printf("\t-w Swap Transport Blocks\n");
+  printf("\t-j Enable PDSCH decoder coworker\n");
   printf("\t-v [set srslte_verbose to debug, default none]\n");
 }
 
 void parse_args(int argc, char **argv) {
   int opt;
-  while ((opt = getopt(argc, argv, "fmMcsrtRFpnawvx")) != -1) {
+  while ((opt = getopt(argc, argv, "fmMcsrtRFpnawvxj")) != -1) {
     switch(opt) {
     case 'f':
       input_file = argv[optind];
@@ -129,6 +131,9 @@ void parse_args(int argc, char **argv) {
     case 'w':
       tb_cw_swap = true;
       break;
+    case 'j':
+      enable_coworker = true;
+      break;
     case 'v':
       srslte_verbose++;
       break;
@@ -470,6 +475,10 @@ int main(int argc, char **argv) {
   int r=0;
   srslte_pdsch_set_max_noi(&pdsch_rx, 10);
 
+  if (enable_coworker) {
+    srslte_pdsch_enable_coworker(&pdsch_rx);
+  }
+
   gettimeofday(&t[1], NULL);
   for (k = 0; k < M; k++) {
 #ifdef DO_OFDM

srsue/hdr/phy/phch_worker.h

@@ -45,6 +45,7 @@ public:
   ~phch_worker();
   void  reset(); 
   void  set_common(phch_common *phy);
+  void  enable_pdsch_coworker();
   bool  init(uint32_t max_prb, srslte::log *log, srslte::log *log_phy_lib_h, chest_feedback_itf *chest_loop);
 
   bool  set_cell(srslte_cell_t cell);

srsue/hdr/phy/phy.h

@@ -153,8 +153,9 @@ private:
 
   bool     initiated;
   uint32_t nof_workers; 
-  
+  uint32_t nof_coworkers;
-  const static int MAX_WORKERS         = 4;
+
+  const static int MAX_WORKERS         = 3;
   const static int DEFAULT_WORKERS     = 2;
   
   const static int SF_RECV_THREAD_PRIO = 1;

srsue/src/phy/phch_worker.cc

@@ -103,6 +103,10 @@ void phch_worker::reset()
   rssi_read_cnt = 0;
 }
 
+void phch_worker::enable_pdsch_coworker() {
+  srslte_pdsch_enable_coworker(&ue_dl.pdsch);
+}
+
 void phch_worker::set_common(phch_common* phy_)
 {
   phy = phy_;   

srsue/src/phy/phy.cc

@@ -104,7 +104,7 @@ void phy::set_default_args(phy_args_t *args)
 
 bool phy::check_args(phy_args_t *args) 
 {
-  if (args->nof_phy_threads > 3) {
+  if (args->nof_phy_threads > MAX_WORKERS * 2) {
     log_h->console("Error in PHY args: nof_phy_threads must be 1, 2 or 3\n");
     return false; 
   }
@@ -139,6 +139,10 @@ bool phy::init(srslte::radio_multi* radio_handler, mac_interface_phy *mac, rrc_i
   }
 
   nof_workers = args->nof_phy_threads;
+  if (nof_workers > MAX_WORKERS) {
+    nof_coworkers = SRSLTE_MIN(nof_workers - MAX_WORKERS, MAX_WORKERS);
+    nof_workers = MAX_WORKERS;
+  }
   this->log_phy_lib_h = (srslte::log*) log_vec[nof_workers];
   srslte_phy_log_register_handler(this, srslte_phy_handler);
   
@@ -160,6 +164,10 @@ void phy::run_thread() {
     workers_pool.init_worker(i, &workers[i], WORKERS_THREAD_PRIO, args->worker_cpu_mask);
   }
 
+  for (uint32_t i=0;i<nof_coworkers;i++) {
+    workers[i].enable_pdsch_coworker();
+  }
+
   // Warning this must be initialized after all workers have been added to the pool
   sf_recv.init(radio_handler, mac, rrc, &prach_buffer, &workers_pool, &workers_common, log_h, log_phy_lib_h, args->nof_rx_ant, SF_RECV_THREAD_PRIO, args->sync_cpu_affinity);
 

srsue/src/ue.cc

@@ -55,6 +55,11 @@ ue::~ue()
 bool ue::init(all_args_t *args_) {
   args = args_;
 
+  int nof_phy_threads = args->expert.phy.nof_phy_threads;
+  if (nof_phy_threads > 3) {
+    nof_phy_threads = 3;
+  }
+
   if (!args->log.filename.compare("stdout")) {
     logger = &logger_stdout;
   } else {
@@ -66,7 +71,7 @@ bool ue::init(all_args_t *args_) {
 
   rf_log.init("RF  ", logger);
   // Create array of pointers to phy_logs
-  for (int i=0;i<args->expert.phy.nof_phy_threads;i++) {
+  for (int i=0;i<nof_phy_threads;i++) {
     srslte::log_filter *mylog = new srslte::log_filter;
     char tmp[16];
     sprintf(tmp, "PHY%d",i);
@@ -85,7 +90,7 @@ bool ue::init(all_args_t *args_) {
   // Init logs
   rf_log.set_level(srslte::LOG_LEVEL_INFO);
   rf_log.info("Starting UE\n");
-  for (int i=0;i<args->expert.phy.nof_phy_threads;i++) {
+  for (int i=0;i<nof_phy_threads;i++) {
     ((srslte::log_filter*) phy_log[i])->set_level(level(args->log.phy_level));
   }
   
@@ -95,7 +100,7 @@ bool ue::init(all_args_t *args_) {
   sprintf(tmp, "PHY_LIB");
   lib_log->init(tmp, logger, true);
   phy_log.push_back(lib_log);
-  ((srslte::log_filter*) phy_log[args->expert.phy.nof_phy_threads])->set_level(level(args->log.phy_lib_level));
+  ((srslte::log_filter*) phy_log[nof_phy_threads])->set_level(level(args->log.phy_lib_level));
  
   
   mac_log.set_level(level(args->log.mac_level));
@@ -106,7 +111,7 @@ bool ue::init(all_args_t *args_) {
   gw_log.set_level(level(args->log.gw_level));
   usim_log.set_level(level(args->log.usim_level));
 
-  for (int i=0;i<args->expert.phy.nof_phy_threads + 1;i++) {
+  for (int i=0;i<nof_phy_threads + 1;i++) {
     ((srslte::log_filter*) phy_log[i])->set_hex_limit(args->log.phy_hex_limit);
   }
   mac_log.set_hex_limit(args->log.mac_hex_limit);