First commit

CMakeLists.txt

@@ -289,7 +289,7 @@ set(CMAKE_EXE_LINKER_FLAGS  "${CMAKE_EXE_LINKER_FLAGS} -ggdb2 -Wl,-rpath -Wl,${C
 # these changes are related to hardcoded path to include .h files
 set(debugOpt "-ggdb2 -DMALLOC_CHECK_=3 -fno-delete-null-pointer-checks")
 set(CMAKE_C_FLAGS_DEBUG "${debugOpt} -O0")
-set(CMAKE_C_FLAGS_RELWITHDEBINFO "${debugOpt} -O2")
+set(CMAKE_C_FLAGS_RELWITHDEBINFO "${debugOpt} -O3")
 set(CMAKE_C_FLAGS_RELEASE "-O3")
 
 # Enable assert() for RelWithDebInfo builds
@@ -2350,6 +2350,11 @@ add_executable(nr-softmodem
   ${nr_rrc_h}
   ${s1ap_h}
 #  ${OPENAIR_BIN_DIR}/messages_xml.h
+
+  ${OPENAIR_DIR}/common/utils/thread_pool/task_manager.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/notification_queue.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/seq_ring_task.c
+
   ${OPENAIR_DIR}/executables/nr-gnb.c
   ${OPENAIR_DIR}/executables/nr-ru.c
   ${OPENAIR_DIR}/executables/nr-softmodem.c
@@ -2514,6 +2519,13 @@ target_link_libraries(ldpctest PRIVATE asn1_nr_rrc asn1_lte_rrc)
 add_executable(nr_dlschsim
   ${OPENAIR1_DIR}/SIMULATION/NR_PHY/dlschsim.c
   ${OPENAIR_DIR}/common/utils/nr/nr_common.c
+
+  ${OPENAIR_DIR}/common/utils/thread_pool/task_manager.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/notification_queue.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/seq_ring_task.c
+
+
+
   ${OPENAIR_DIR}/executables/softmodem-common.c
   ${T_SOURCE}
   ${SHLIB_LOADER_SOURCES}
@@ -2527,6 +2539,11 @@ target_link_libraries(nr_dlschsim PRIVATE asn1_nr_rrc asn1_lte_rrc)
 add_executable(nr_pbchsim
   ${OPENAIR1_DIR}/SIMULATION/NR_PHY/pbchsim.c
   ${OPENAIR_DIR}/common/utils/nr/nr_common.c
+
+  ${OPENAIR_DIR}/common/utils/thread_pool/task_manager.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/notification_queue.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/seq_ring_task.c
+
   ${T_SOURCE}
   ${SHLIB_LOADER_SOURCES}
   )
@@ -2539,6 +2556,11 @@ target_link_libraries(nr_pbchsim PRIVATE asn1_nr_rrc asn1_lte_rrc)
 
 #PUCCH ---> Prashanth
 add_executable(nr_pucchsim
+
+  ${OPENAIR_DIR}/common/utils/thread_pool/task_manager.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/notification_queue.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/seq_ring_task.c
+
   ${OPENAIR1_DIR}/SIMULATION/NR_PHY/pucchsim.c
   ${OPENAIR_DIR}/common/utils/nr/nr_common.c
   ${OPENAIR_DIR}/executables/softmodem-common.c
@@ -2552,6 +2574,12 @@ target_link_libraries(nr_pucchsim PRIVATE
 target_link_libraries(nr_pucchsim PRIVATE asn1_nr_rrc asn1_lte_rrc)
 
 add_executable(nr_dlsim
+  ${OPENAIR_DIR}/common/utils/thread_pool/task_manager.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/notification_queue.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/seq_ring_task.c
+
+
+
   ${OPENAIR1_DIR}/SIMULATION/NR_PHY/dlsim.c
   ${OPENAIR_DIR}/common/utils/nr/nr_common.c
   ${OPENAIR_DIR}/executables/softmodem-common.c
@@ -2571,6 +2599,10 @@ target_link_libraries(nr_dlsim PRIVATE asn1_nr_rrc asn1_lte_rrc)
 target_compile_definitions(nr_dlsim PUBLIC -DPHYSICAL_SIMULATOR)
 
 add_executable(nr_prachsim
+  ${OPENAIR_DIR}/common/utils/thread_pool/task_manager.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/notification_queue.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/seq_ring_task.c
+
   ${OPENAIR1_DIR}/SIMULATION/NR_PHY/prachsim.c
   ${OPENAIR_DIR}/common/utils/nr/nr_common.c
   ${OPENAIR1_DIR}/SCHED_NR/phy_procedures_nr_common.c
@@ -2589,6 +2621,11 @@ target_link_libraries(nr_prachsim PRIVATE asn1_nr_rrc asn1_lte_rrc)
 add_executable(nr_ulschsim
   ${OPENAIR1_DIR}/SIMULATION/NR_PHY/ulschsim.c
   ${OPENAIR_DIR}/common/utils/nr/nr_common.c
+
+  ${OPENAIR_DIR}/common/utils/thread_pool/task_manager.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/notification_queue.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/seq_ring_task.c
+
   ${OPENAIR_DIR}/executables/softmodem-common.c
   ${PHY_INTERFACE_DIR}/queue_t.c
   ${T_SOURCE}
@@ -2604,6 +2641,13 @@ add_executable(nr_ulsim
   ${OPENAIR1_DIR}/SIMULATION/NR_PHY/ulsim.c
   ${OPENAIR_DIR}/common/utils/nr/nr_common.c
   ${OPENAIR_DIR}/executables/softmodem-common.c
+
+  ${OPENAIR_DIR}/common/utils/thread_pool/task_manager.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/notification_queue.c
+  ${OPENAIR_DIR}/common/utils/thread_pool/seq_ring_task.c
+
+
+
   ${OPENAIR2_DIR}/RRC/NAS/nas_config.c
   ${OPENAIR1_DIR}/SIMULATION/ETH_TRANSPORT/netlink_init.c
   ${NR_UE_RRC_DIR}/rrc_nsa.c

common/utils/thread_pool/main.c

@@ -0,0 +1,29 @@
+#include <assert.h>
+#include <stdlib.h>
+#include "task_manager.h"
+#include "../common_task.h"
+
+int main()
+{
+  task_manager_t man;
+  init_task_manager(&man, NUM_THREADS);
+
+  pair_t* arr = calloc(NUM_JOBS, sizeof(pair_t));
+  assert(arr != NULL);
+
+//  wake_and_spin_task_manager(&man);
+
+  for(int i = 0; i < NUM_JOBS; ++i){
+    arr[i] = fill_pair(i); 
+    task_t t = {.args = &arr[i], t.func = do_work};
+    async_task_manager(&man, t);
+  }
+
+//  stop_spin_task_manager(&man);
+  wait_all_spin_task_manager(&man); 
+  free(arr);
+  free_task_manager(&man, NULL);
+
+  return EXIT_SUCCESS;
+}
+

common/utils/thread_pool/notification_queue.c

@@ -0,0 +1,188 @@
+#include "notification_queue.h"
+
+#include <assert.h>
+#include <errno.h>
+#include <stdatomic.h>
+#include <string.h>
+
+void init_not_q(not_q_t* q)
+{
+  assert(q != NULL);
+
+  q->done = 0;
+  init_seq_ring_task(&q->r);
+
+  pthread_mutexattr_t attr = {0};
+#ifdef _DEBUG
+  int const rc_mtx = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
+  assert(rc_mtx == 0);
+#endif
+  int rc = pthread_mutex_init(&q->mtx, &attr);
+  assert(rc == 0 && "Error while creating the mtx");
+
+  pthread_condattr_t* c_attr = NULL; 
+  rc = pthread_cond_init(&q->cv, c_attr);
+  assert(rc == 0);
+  q->spin = false;
+}
+
+void free_not_q(not_q_t* q, void (*clean)(task_t*) )
+{
+  assert(q != NULL);
+  assert(q->done == 1);
+
+  free_seq_ring_task(&q->r, clean);
+
+  int rc = pthread_mutex_destroy(&q->mtx);
+  assert(rc == 0);
+
+  rc = pthread_cond_destroy(&q->cv);
+  assert(rc == 0);
+}
+
+bool try_push_not_q(not_q_t* q, task_t t)
+{
+  assert(q != NULL);
+  assert(q->done == 0 || q->done ==1);
+  assert(t.func != NULL);
+  assert(t.args != NULL);
+
+  if(pthread_mutex_trylock(&q->mtx ) != 0)
+    return false;
+
+  push_back_seq_ring_task(&q->r, t);
+
+  int rc = pthread_mutex_unlock(&q->mtx);
+  assert(rc == 0);
+
+  pthread_cond_signal(&q->cv);
+
+  return true;
+}
+
+void push_not_q(not_q_t* q, task_t t)
+{
+  assert(q != NULL);
+  assert(q->done == 0 || q->done ==1);
+  assert(t.func != NULL);
+  //assert(t.args != NULL);
+
+  int rc = pthread_mutex_lock(&q->mtx);
+  assert(rc == 0);
+
+  push_back_seq_ring_task(&q->r, t);
+
+  pthread_mutex_unlock(&q->mtx);
+
+  pthread_cond_signal(&q->cv);
+}
+
+ret_try_t try_pop_not_q(not_q_t* q)
+{
+  assert(q != NULL);
+
+  ret_try_t ret = {.success = false}; 
+
+  int rc = pthread_mutex_trylock(&q->mtx);
+  assert(rc == 0 || rc == EBUSY);
+
+  if(rc == EBUSY)
+    return ret;
+
+  assert(q->done == 0 || q->done ==1);
+
+  size_t sz = size_seq_ring_task(&q->r); 
+  if(sz == 0){
+    rc = pthread_mutex_unlock(&q->mtx);
+    assert(rc == 0);
+
+    return ret;
+  }
+
+  ret.t = pop_seq_ring_task(&q->r);
+
+  rc = pthread_mutex_unlock(&q->mtx);
+  assert(rc == 0);
+  ret.success = true; 
+
+  return ret;
+}
+
+bool pop_not_q(not_q_t* q, ret_try_t* out)
+{
+  assert(q != NULL);
+  assert(out != NULL);
+
+  int rc = pthread_mutex_lock(&q->mtx);
+  assert(rc == 0);
+  assert(q->done == 0 || q->done ==1);
+
+label:
+  while(size_seq_ring_task(&q->r) == 0 && q->done == 0 && q->spin == false)
+    pthread_cond_wait(&q->cv , &q->mtx);
+
+  if(q->done == 1){
+    int const rc = pthread_mutex_unlock(&q->mtx);
+    assert(rc == 0);
+    return false;
+  }
+
+  if(q->spin){
+    int rc = pthread_mutex_unlock(&q->mtx);
+    assert(rc == 0);
+
+   // Wait for lock to be released without generating cache misses
+    while (atomic_load_explicit(&q->spin, memory_order_relaxed)){
+      // Issue X86 PAUSE or ARM YIELD instruction to reduce contention between
+      // hyper-threads
+      pause_or_yield();
+    }
+
+    rc = pthread_mutex_lock(&q->mtx);
+    assert(rc == 0);
+  }
+
+  if(size_seq_ring_task(&q->r) == 0)
+    goto label;
+
+  out->t = pop_seq_ring_task(&q->r);
+
+  rc = pthread_mutex_unlock(&q->mtx);
+  assert(rc == 0);
+
+  return true;
+}
+
+void done_not_q(not_q_t* q)
+{
+  assert(q != NULL);
+
+  int rc = pthread_mutex_lock(&q->mtx);
+  assert(rc == 0);
+  
+  q->done = 1;
+
+  rc = pthread_mutex_unlock(&q->mtx);
+  assert(rc == 0);
+
+  rc = pthread_cond_signal(&q->cv);
+  assert(rc == 0);
+}
+
+void wake_spin_not_q(not_q_t* q)
+{
+  assert(q != NULL);
+
+  // spin until acquired i.e., q->spin == true
+  while(atomic_exchange_explicit(&q->spin, true, memory_order_acquire)); 
+
+  pthread_cond_signal(&q->cv);
+}
+
+void stop_spin_not_q(not_q_t* q)
+{
+  assert(q != NULL);
+
+  atomic_store_explicit(&q->spin, false, memory_order_release);
+}
+

common/utils/thread_pool/notification_queue.h

@@ -0,0 +1,44 @@
+#ifndef NOTIFICATION_QUEUE_THREAD_POOL_H
+#define NOTIFICATION_QUEUE_THREAD_POOL_H 
+
+#include "task.h"
+#include "seq_ring_task.h"
+
+#include <pthread.h>
+#include <stdbool.h>
+
+typedef struct {
+  pthread_mutex_t mtx;
+  pthread_cond_t cv;
+  seq_ring_task_t r;
+  int done;
+
+  _Atomic bool spin;
+} not_q_t;
+
+typedef struct{
+  task_t t;
+  bool success;
+} ret_try_t;
+
+void init_not_q(not_q_t* q);
+
+void free_not_q(not_q_t* q, void (*clean)(task_t*) );
+
+bool try_push_not_q(not_q_t* q, task_t t);
+
+void push_not_q(not_q_t* q, task_t t);
+
+ret_try_t try_pop_not_q(not_q_t* q);
+
+bool pop_not_q(not_q_t* q, ret_try_t* out);
+
+void done_not_q(not_q_t* q);
+
+void wake_spin_not_q(not_q_t* q);
+
+void stop_spin_not_q(not_q_t* q);
+
+
+#endif
+

common/utils/thread_pool/seq_ring_task.c

@@ -0,0 +1,121 @@
+/*
+MIT License
+
+Copyright (c) 2021 Mikel Irazabal
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+
+#include "seq_ring_task.h"
+
+#include <assert.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+
+// For working correctly, maintain the default elements to a multiple of 2
+#define DEFAULT_ELM 32
+
+inline static
+uint32_t mask(uint32_t cap, uint32_t val)
+{
+
+  return val & (cap-1);
+}
+
+static 
+bool full(seq_ring_task_t* r)
+{
+  return size_seq_ring_task(r) == r->cap -1;
+}
+
+static
+void enlarge_buffer(seq_ring_task_t* r)
+{
+  assert(r != NULL);
+  assert(full(r));
+
+  const uint32_t factor = 2;
+  task_t* tmp_buffer = calloc(r->cap * factor, sizeof(task_t) );
+  assert(tmp_buffer != NULL);
+
+  const uint32_t head_pos = mask(r->cap, r->head);
+  const uint32_t tail_pos = mask(r->cap, r->tail);
+
+  if(head_pos > tail_pos){
+    memcpy(tmp_buffer, r->array + tail_pos , (head_pos-tail_pos)*sizeof(task_t) );
+  } else {
+    memcpy(tmp_buffer, r->array + tail_pos, (r->cap-tail_pos)*sizeof(task_t));
+    memcpy(tmp_buffer + (r->cap-tail_pos), r->array, head_pos*sizeof(task_t));
+  }
+  r->cap *= factor;
+  free(r->array);
+  r->array = tmp_buffer;
+  r->tail = 0;
+  r->head = r->cap/2 - 1;
+}
+
+void init_seq_ring_task(seq_ring_task_t* r)
+{
+  assert(r != NULL);
+  task_t* tmp_buffer = calloc(DEFAULT_ELM, sizeof(task_t)); 
+  assert(tmp_buffer != NULL);
+  seq_ring_task_t tmp = {.array = tmp_buffer, .head = 0, .tail = 0, .cap = DEFAULT_ELM};
+  memcpy(r, &tmp, sizeof(seq_ring_task_t));
+}
+
+void free_seq_ring_task(seq_ring_task_t* r, seq_free_func fp)
+{
+  assert(r != NULL);
+  assert(fp == NULL);
+  free(r->array);
+}
+
+
+void push_back_seq_ring_task(seq_ring_task_t* r, task_t t)
+{
+  assert(r != NULL);
+
+  if(full(r))
+    enlarge_buffer(r);
+  
+  const uint32_t pos = mask(r->cap, r->head);
+  r->array[pos] = t;
+  r->head += 1;
+}
+
+task_t pop_seq_ring_task(seq_ring_task_t* r )
+{
+  assert(r != NULL);
+  assert(size_seq_ring_task(r) > 0);
+
+  const uint32_t pos = mask(r->cap, r->tail);
+  task_t t = r->array[pos];
+  r->tail += 1; 
+  return t;
+}
+
+size_t size_seq_ring_task(seq_ring_task_t* r)
+{
+  assert(r != NULL);
+
+  return r->head - r->tail;
+}
+

common/utils/thread_pool/seq_ring_task.h

@@ -0,0 +1,77 @@
+/*
+MIT License
+
+Copyright (c) 2021 Mikel Irazabal
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+
+#ifndef SEQ_CIRCULAR_BUFFER_TASK_H
+#define SEQ_CIRCULAR_BUFFER_TASK_H
+
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
+#include "task.h"
+
+typedef struct seq_ring_buf_s
+{
+//  const size_t elt_size;
+  task_t* array;
+
+  size_t cap;
+  uint32_t head;
+  uint32_t tail;
+} seq_ring_task_t;
+
+typedef void (*seq_free_func)(task_t*); 
+
+void init_seq_ring_task(seq_ring_task_t*);
+
+void free_seq_ring_task(seq_ring_task_t*, seq_free_func);
+
+void push_back_seq_ring_task(seq_ring_task_t* arr, task_t t);
+
+task_t pop_seq_ring_task(seq_ring_task_t* arr);
+
+size_t size_seq_ring_task(seq_ring_task_t* r);
+
+/*
+void seq_ring_erase(seq_ring_t*, void*, void*);
+
+void seq_ring_swap(seq_ring_t*, void*, void*);
+
+size_t seq_ring_size(seq_ring_t*);
+
+void* seq_ring_front(seq_ring_t*);
+
+void* seq_ring_next(seq_ring_t*, void*);
+
+void* seq_ring_end(seq_ring_t*);
+
+void* seq_ring_at(seq_ring_t* , uint32_t );
+
+int32_t seq_ring_dist(seq_ring_t*, void*, void*);
+
+bool seq_ring_equal(void*,void*);
+*/
+
+#endif
+

common/utils/thread_pool/task.h

@@ -0,0 +1,18 @@
+#ifndef TASK_WORK_STEALING_THREAD_POOL_H
+#define TASK_WORK_STEALING_THREAD_POOL_H 
+
+#if defined (__i386__) || defined(__x86_64__)
+  #define pause_or_yield  __builtin_ia32_pause
+#elif __aarch64__
+  #define pause_or_yield() asm volatile("yield" ::: "memory")
+#else
+    static_assert(0!=0, "Unknown CPU architecture");
+#endif
+
+typedef struct{
+  void* args;
+  void (*func)(void* args);
+} task_t;
+
+#endif
+

common/utils/thread_pool/task_manager.c

@@ -0,0 +1,192 @@
+#include "task_manager.h"
+
+#include <assert.h> 
+#include <stdlib.h>
+
+typedef struct{
+  task_manager_t* man;
+  int idx;
+} task_thread_args_t;
+
+static
+void* worker_thread(void* arg)
+{
+  assert(arg != NULL);
+
+  task_thread_args_t* args = (task_thread_args_t*)arg; 
+  int const idx = args->idx;
+  task_manager_t* man = args->man;
+
+  uint32_t const len = man->len_thr;
+  uint32_t const num_it = 2*(man->len_thr + idx); 
+
+  for(;;){
+
+    ret_try_t ret = {.success = false}; 
+
+    for(uint32_t i = idx; i < num_it; ++i){
+      ret = try_pop_not_q(&man->q_arr[i%len]);
+      if(ret.success == true){
+        break;
+      } 
+    }
+
+    if(ret.success == false && pop_not_q(&man->q_arr[idx], &ret) == false)
+      break;
+
+    //int64_t now = time_now_us();
+    //printf("Before func %ld id %lu \n", now , pthread_self() );
+    ret.t.func(ret.t.args); 
+    //printf("After func %ld id %lu elapsed %ld \n", time_now_us(), pthread_self(), time_now_us()-now );
+
+    atomic_fetch_sub(&man->num_task, 1); 
+
+    if(man->num_task < 1 && man->waiting != 0){
+
+        pthread_mutex_lock(&man->wait_mtx);
+       
+        man->waiting = 0;
+        pthread_cond_signal(&man->wait_cv); 
+        
+        pthread_mutex_unlock(&man->wait_mtx);
+    }
+
+  }
+
+  free(args);
+  return NULL;
+}
+
+void init_task_manager(task_manager_t* man, uint32_t num_threads)
+{
+  assert(man != NULL);
+  assert(num_threads > 0 && num_threads < 33 && "Do you have zero or more than 32 processors??");
+
+  man->q_arr = calloc(num_threads, sizeof(not_q_t));
+  assert(man->q_arr != NULL && "Memory exhausted");
+    
+  for(uint32_t i = 0; i < num_threads; ++i){
+    init_not_q(&man->q_arr[i]);   
+  }
+
+  man->index = 0;
+
+  // Waiting all
+  man->waiting = 0;
+
+  pthread_mutexattr_t attr = {0};
+#ifdef _DEBUG
+  int const rc_mtx = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
+  assert(rc_mtx == 0);
+#endif
+  int rc = pthread_mutex_init(&man->wait_mtx, &attr);
+  assert(rc == 0 && "Error while creating the mtx");
+
+  pthread_condattr_t* c_attr = NULL; 
+  rc = pthread_cond_init(&man->wait_cv, c_attr);
+  assert(rc == 0);
+
+  man->t_arr = calloc(num_threads, sizeof(pthread_t));
+  assert(man->t_arr != NULL && "Memory exhausted" );
+  man->len_thr = num_threads;
+
+  for(uint32_t i = 0; i < num_threads; ++i){
+    task_thread_args_t* args = malloc(sizeof(task_thread_args_t) ); 
+    args->idx = i;
+    args->man = man;
+
+    int rc = pthread_create(&man->t_arr[i], NULL, worker_thread, args);
+    assert(rc == 0);
+  }
+
+}
+
+void free_task_manager(task_manager_t* man, void (*clean)(task_t*))
+{
+  for(uint32_t i = 0; i < man->len_thr; ++i){
+    done_not_q(&man->q_arr[i]);
+  }
+
+  for(uint32_t i = 0; i < man->len_thr; ++i){
+    pthread_join(man->t_arr[i], NULL); 
+  }
+
+  for(uint32_t i = 0; i < man->len_thr; ++i){
+    free_not_q(&man->q_arr[i], clean); 
+  }
+
+  free(man->q_arr);
+
+  free(man->t_arr);
+
+  int rc = pthread_mutex_destroy(&man->wait_mtx);
+  assert(rc == 0);
+
+  rc = pthread_cond_destroy(&man->wait_cv);
+  assert(rc == 0);
+}
+
+void async_task_manager(task_manager_t* man, task_t t)
+{
+  assert(man != NULL);
+  assert(t.func != NULL);
+  //assert(t.args != NULL);
+
+  uint64_t const index = man->index++;
+  const int32_t len_thr = man->len_thr;
+  for(int32_t i = 0; i < len_thr ; ++i){
+    if(try_push_not_q(&man->q_arr[(i+index) % len_thr], t)){
+      atomic_fetch_add(&man->num_task, 1); 
+      return;
+    }
+  }
+
+  push_not_q(&man->q_arr[index%len_thr], t);
+  atomic_fetch_add(&man->num_task, 1); 
+}
+
+void wait_all_task_manager(task_manager_t* man)
+{
+  assert(man != NULL);
+
+  pthread_mutex_lock(&man->wait_mtx);
+  man->waiting = 1;
+
+  while(man->num_task > 0 && man->waiting != 0) 
+    pthread_cond_wait(&man->wait_cv, &man->wait_mtx);
+
+  pthread_mutex_unlock(&man->wait_mtx);
+}
+
+void wait_all_spin_task_manager(task_manager_t* man)
+{
+  assert(man != NULL);
+
+  // Wait without generating cache misses
+  while (atomic_load_explicit(&man->num_task, memory_order_relaxed)){
+    // Issue X86 PAUSE or ARM YIELD instruction to reduce contention between
+    // hyper-threads
+    pause_or_yield();
+  }
+}
+
+void wake_and_spin_task_manager(task_manager_t* man)
+{
+  assert(man != NULL);
+
+  const int32_t len_thr = man->len_thr;
+  for(int i = 0; i < len_thr; ++i){
+     wake_spin_not_q(&man->q_arr[i]);
+  }
+}
+
+void stop_spin_task_manager(task_manager_t* man)
+{
+  assert(man != NULL);
+
+  const int32_t len_thr = man->len_thr;
+  for(int i =0; i < len_thr; ++i){
+    stop_spin_not_q(&man->q_arr[i]);
+  }
+}
+

common/utils/thread_pool/task_manager.h

@@ -0,0 +1,47 @@
+#ifndef TASK_MANAGER_WORKING_STEALING_H
+#define TASK_MANAGER_WORKING_STEALING_H 
+
+//#define TASK_MANAGER
+
+#include "task.h"
+#include "notification_queue.h"
+
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdint.h>
+
+typedef struct {
+
+  pthread_t* t_arr;
+  size_t len_thr;
+  
+  atomic_uint_fast64_t index;
+
+  not_q_t* q_arr;
+
+  atomic_int_fast64_t num_task;
+
+  pthread_cond_t wait_cv; 
+  pthread_mutex_t wait_mtx;
+
+  _Atomic int waiting; // 1 cv, 2 spin
+                       //
+  //_Atomic bool waiting;
+} task_manager_t;
+
+void init_task_manager(task_manager_t* man, uint32_t num_threads);
+
+void free_task_manager(task_manager_t* man, void (*clean)(task_t* args) );
+
+void async_task_manager(task_manager_t* man, task_t t);
+
+void wait_all_task_manager(task_manager_t* man);
+
+void wait_all_spin_task_manager(task_manager_t* man);
+
+void wake_and_spin_task_manager(task_manager_t* man);
+
+void stop_spin_task_manager(task_manager_t* man);
+
+#endif
+

openair1/PHY/NR_TRANSPORT/nr_ulsch_decoding.c

@@ -46,6 +46,8 @@
 #include "defs.h"
 #include "common/utils/LOG/vcd_signal_dumper.h"
 #include "common/utils/LOG/log.h"
+#include "common/utils/thread_pool/task_manager.h"
+
 #include <syscall.h>
 //#define DEBUG_ULSCH_DECODING
 //#define gNB_DEBUG_TRACE
@@ -162,7 +164,9 @@ void clean_gNB_ulsch(NR_gNB_ULSCH_t *ulsch)
 #endif
 
 void nr_processULSegment(void* arg) {
+
   ldpcDecode_t *rdata = (ldpcDecode_t*) arg;
+
   PHY_VARS_gNB *phy_vars_gNB = rdata->gNB;
   NR_UL_gNB_HARQ_t *ulsch_harq = rdata->ulsch_harq;
   t_nrLDPC_dec_params *p_decoderParms = &rdata->decoderParms;
@@ -252,6 +256,15 @@ void nr_processULSegment(void* arg) {
     stop_meas(&phy_vars_gNB->ulsch_rate_unmatching_stats);
   }
 
+  // don't run LDPC decoding if some other thread had a failure 
+  // nr_rate_matching_ldpc_rx() must be called to feed d[r] in all cases
+  // so this test has to come after the call to nr_rate_matching_ldpc_rx()
+
+#ifdef TASK_MANAGER
+  if(*rdata->cancel_dec == 1)
+    return;
+#endif
+
   memset(ulsch_harq->c[r],0,Kr_bytes);
 
   if (ulsch_harq->C == 1) {
@@ -307,12 +320,21 @@ void nr_processULSegment(void* arg) {
       LOG_I(PHY,"CRC NOK\n");
 #endif
     rdata->decodeIterations = max_ldpc_iterations + 1;
+     /* set skip_ldpc_decoding to 1 to indicate to remaining threads that they shall not run LDPC decoding */
+#ifdef TASK_MANAGER
+    *rdata->cancel_dec = 1; 
+#endif
   }
 
   for (int m=0; m < Kr>>3; m ++) {
     ulsch_harq->c[r][m]= (uint8_t) llrProcBuf[m];
   }
 
+#ifdef TASK_MANAGER
+  if(phy_vars_gNB->ldpc_offload_flag == 0)
+    nr_postDecode(rdata->gNB, rdata);
+#endif
+
   //stop_meas(&phy_vars_gNB->ulsch_ldpc_decoding_stats);
 }
 
@@ -623,11 +645,25 @@ uint32_t nr_ulsch_decoding(PHY_VARS_gNB *phy_vars_gNB,
   else {
     dtx_det = 0;
 
+#ifdef TASK_MANAGER
+    ldpcDecode_t* arr = calloc(harq_process->C, sizeof(ldpcDecode_t));
+    assert(arr != NULL && "Memory exhausted");
+    int idx_arr = 0;
+   _Atomic int cancel_dec = 0;
+#endif
+
     for (int r = 0; r < harq_process->C; r++) {
       int E = nr_get_E(G, harq_process->C, Qm, n_layers, r);
+      
+#ifdef TASK_MANAGER
+    ldpcDecode_t* rdata = &arr[idx_arr];
+    ++idx_arr;
+    rdata->cancel_dec = &cancel_dec;
+#else
       union ldpcReqUnion id = {.s = {ulsch->rnti, frame, nr_tti_rx, 0, 0}};
       notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(ldpcDecode_t), id.p, &phy_vars_gNB->respDecode, &nr_processULSegment);
       ldpcDecode_t *rdata = (ldpcDecode_t *)NotifiedFifoData(req);
+#endif
       decParams.R = nr_get_R_ldpc_decoder(pusch_pdu->pusch_data.rv_index, E, decParams.BG, decParams.Z, &harq_process->llrLen, harq_process->round);
       rdata->gNB = phy_vars_gNB;
       rdata->ulsch_harq = harq_process;
@@ -647,13 +683,25 @@ uint32_t nr_ulsch_decoding(PHY_VARS_gNB *phy_vars_gNB,
       rdata->ulsch = ulsch;
       rdata->ulsch_id = ULSCH_id;
       rdata->tbslbrm = pusch_pdu->maintenance_parms_v3.tbSizeLbrmBytes;
+#ifdef TASK_MANAGER
+      task_t t = { .args = rdata, .func =  &nr_processULSegment };
+      async_task_manager(&phy_vars_gNB->man, t);
+#else
       pushTpool(&phy_vars_gNB->threadPool, req);
+#endif
       phy_vars_gNB->nbDecode++;
       LOG_D(PHY, "Added a block to decode, in pipe: %d\n", phy_vars_gNB->nbDecode);
       r_offset += E;
       offset += (Kr_bytes - (harq_process->F >> 3) - ((harq_process->C > 1) ? 3 : 0));
       //////////////////////////////////////////////////////////////////////////////////////////
     }
+
+#ifdef TASK_MANAGER
+     stop_spin_task_manager(&phy_vars_gNB->man);
+     wait_all_spin_task_manager(&phy_vars_gNB->man);
+     free(arr);
+#endif
+
   }
   return 1;
 }

openair1/PHY/defs_gNB.h

@@ -44,6 +44,8 @@
 #include "PHY/CODING/nrLDPC_decoder/nrLDPC_types.h"
 #include "executables/rt_profiling.h"
 
+#include "../../common/utils/thread_pool/task_manager.h"
+
 #include "nfapi_nr_interface_scf.h"
 
 #define MAX_NUM_RU_PER_gNB MAX_NUM_RU_PER_eNB
@@ -768,6 +770,11 @@ typedef struct PHY_VARS_gNB_s {
   void *scopeData;
   /// structure for analyzing high-level RT measurements
   rt_L1_profiling_t rt_L1_profiling; 
+
+#ifdef TASK_MANAGER
+  task_manager_t man;
+#endif
+
 } PHY_VARS_gNB;
 
 typedef struct LDPCDecode_s {
@@ -790,6 +797,11 @@ typedef struct LDPCDecode_s {
   int offset;
   int decodeIterations;
   uint32_t tbslbrm;
+
+#ifdef TASK_MANAGER
+  _Atomic int* cancel_dec; // Cancel Decoding. Not sure whether _Atomic is needed
+#endif
+
 } ldpcDecode_t;
 
 struct ldpcReqId {

openair1/SCHED_NR/phy_procedures_nr_gNB.c

@@ -39,12 +39,17 @@
 #include "executables/softmodem-common.h"
 #include "nfapi/oai_integration/vendor_ext.h"
 
+#include "common/utils/thread_pool/task_manager.h"
+
 #include "assertions.h"
 
 #include <time.h>
 
 #include "intertask_interface.h"
 
+#include "common/utils/thread_pool/task_manager.h"
+
+
 //#define DEBUG_RXDATA
 //#define SRS_IND_DEBUG
 
@@ -208,8 +213,15 @@ void phy_procedures_gNB_TX(processingData_L1tx_t *msgTx,
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_gNB_TX+offset,0);
 }
 
+#ifdef TASK_MANAGER
+void nr_postDecode(PHY_VARS_gNB *gNB, ldpcDecode_t *rdata){
+#else
 void nr_postDecode(PHY_VARS_gNB *gNB, notifiedFIFO_elt_t *req) {
+#endif
+
+#ifndef TASK_MANAGER
   ldpcDecode_t *rdata = (ldpcDecode_t*) NotifiedFifoData(req);
+#endif
   NR_UL_gNB_HARQ_t *ulsch_harq = rdata->ulsch_harq;
   NR_gNB_ULSCH_t *ulsch = rdata->ulsch;
   int r = rdata->segment_r;
@@ -227,6 +239,9 @@ void nr_postDecode(PHY_VARS_gNB *gNB, notifiedFIFO_elt_t *req) {
 
   } else {
     if ( rdata->nbSegments != ulsch_harq->processedSegments ) {
+#ifdef TASK_MANAGER
+  *rdata->cancel_dec = 1;
+#else
       int nb = abortTpoolJob(&gNB->threadPool, req->key);
       nb += abortNotifiedFIFOJob(&gNB->respDecode, req->key);
       gNB->nbDecode-=nb;
@@ -234,6 +249,7 @@ void nr_postDecode(PHY_VARS_gNB *gNB, notifiedFIFO_elt_t *req) {
       LOG_D(PHY, "ULSCH %d in error\n",rdata->ulsch_id);
       AssertFatal(ulsch_harq->processedSegments+nb == rdata->nbSegments,"processed: %d, aborted: %d, total %d\n",
 		  ulsch_harq->processedSegments, nb, rdata->nbSegments);
+#endif
       ulsch_harq->processedSegments=rdata->nbSegments;
     }
   }
@@ -378,6 +394,7 @@ void nr_ulsch_procedures(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, int ULSCH
                     slot_rx,
                     harq_pid,
                     G);
+#ifndef TASK_MANAGER
   if (enable_ldpc_offload ==0) {
     while (gNB->nbDecode > 0) {
       notifiedFIFO_elt_t *req = pullTpool(&gNB->respDecode, &gNB->threadPool);
@@ -387,6 +404,7 @@ void nr_ulsch_procedures(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, int ULSCH
       delNotifiedFIFO_elt(req);
     }
   } 
+#endif
   stop_meas(&gNB->ulsch_decoding_stats);
 }
 

openair1/SIMULATION/NR_PHY/ulsim.c

@@ -704,6 +704,10 @@ int main(int argc, char **argv)
   gNB->ofdm_offset_divisor = UINT_MAX;
   initNotifiedFIFO(&gNB->respDecode);
 
+#ifdef TASK_MANAGER
+  init_task_manager(&gNB->man, threadCnt);
+#endif
+
   initFloatingCoresTpool(threadCnt, &gNB->threadPool, false, "gNB-tpool");
   initNotifiedFIFO(&gNB->respDecode);
   initNotifiedFIFO(&gNB->L1_tx_free);
@@ -1393,6 +1397,10 @@ int main(int argc, char **argv)
                            0);
         }
 
+#ifdef TASK_MANAGER
+  wake_and_spin_task_manager(&gNB->man);
+#endif
+
         for (int aa = 0; aa < gNB->frame_parms.nb_antennas_rx; aa++)  {
           apply_nr_rotation_ul(&gNB->frame_parms,
                                gNB->common_vars.rxdataF[aa],