New threadpool

- removed some whitespace noise.
 - cleaned up CMakeLists.txt
 - reintegrated task_manager.c into thread-pool.c
 - cleaned up some unnecessary code

Co-authored-by: Cedric Roux <cedric.roux@eurecom.fr>
Co-authored-by: Mikel Irazabal <mikel.irazabal@openairinterface.com>

common/utils/actor/actor.h

@@ -21,7 +21,7 @@
 
 #ifndef ACTOR_H
 #define ACTOR_H
-#include "thread-pool.h"
+#include "notified_fifo.h"
 
 #define INIT_ACTOR(ptr, name, core_affinity) init_actor((Actor_t *)ptr, name, core_affinity);
 

common/utils/telnetsrv/telnetsrv.c

@@ -53,7 +53,6 @@
 #include <sys/resource.h>
 #include "common/utils/load_module_shlib.h"
 #include "common/config/config_userapi.h"
-#include "common/utils/threadPool/thread-pool.h"
 #include "executables/softmodem-common.h"
 #include <readline/history.h>
 #include "common/oai_version.h"

common/utils/threadPool/CMakeLists.txt

@@ -1,6 +1,6 @@
 if (ENABLE_TESTS)
   add_subdirectory(test)
 endif()
-add_library(thread-pool thread-pool.c)
+add_library(thread-pool thread-pool.c task_ans.c)
 target_link_libraries(thread-pool PRIVATE utils)
 target_include_directories(thread-pool PUBLIC .)

common/utils/threadPool/bounded_notified_fifo.h

@@ -0,0 +1,245 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+
+#ifndef BOUNDED_NOTIFIED_FIFO_H
+#define BOUNDED_NOTIFIED_FIFO_H
+
+#include "assertions.h"
+#include <stdint.h>
+#include <memory.h>
+#include "task.h"
+#include <pthread.h>
+#include "pthread_utils.h"
+
+// For working correctly, maintain the default elements to a 2^N e.g., 2^5=32
+#define DEFAULT_ELM 256
+
+typedef struct seq_ring_buf_s {
+  task_t* array;
+  size_t cap;
+  uint32_t head;
+  uint32_t tail;
+  _Atomic uint64_t sz;
+} seq_ring_task_t;
+
+
+static size_t size_seq_ring_task(seq_ring_task_t* r)
+{
+  DevAssert(r != NULL);
+
+  return r->head - r->tail;
+}
+
+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)
+{
+  DevAssert(r != NULL);
+  DevAssert(full(r));
+
+  const uint32_t factor = 2;
+  task_t* tmp_buffer = calloc(r->cap * factor, sizeof(task_t));
+  DevAssert(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;
+}
+
+static void init_seq_ring_task(seq_ring_task_t* r)
+{
+  DevAssert(r != NULL);
+  task_t* tmp_buffer = calloc(DEFAULT_ELM, sizeof(task_t));
+  DevAssert(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));
+  r->sz = 0;
+}
+
+static void free_seq_ring_task(seq_ring_task_t* r)
+{
+  DevAssert(r != NULL);
+  free(r->array);
+}
+
+static void push_back_seq_ring_task(seq_ring_task_t* r, task_t t)
+{
+  DevAssert(r != NULL);
+
+  if (full(r))
+    enlarge_buffer(r);
+
+  const uint32_t pos = mask(r->cap, r->head);
+  r->array[pos] = t;
+  r->head += 1;
+  r->sz += 1;
+}
+
+static task_t pop_seq_ring_task(seq_ring_task_t* r)
+{
+  DevAssert(r != NULL);
+  DevAssert(size_seq_ring_task(r) > 0);
+
+  const uint32_t pos = mask(r->cap, r->tail);
+  task_t t = r->array[pos];
+  r->tail += 1;
+  r->sz -= 1;
+  return t;
+}
+
+#undef DEFAULT_ELM
+
+typedef struct {
+  pthread_mutex_t mtx;
+  pthread_cond_t cv;
+  seq_ring_task_t r;
+  size_t idx;
+} not_q_t;
+
+typedef struct {
+  task_t t;
+  bool success;
+} ret_try_t;
+
+static void init_not_q(not_q_t* q, size_t idx)
+{
+  DevAssert(q != NULL);
+
+  q->idx = idx;
+
+  init_seq_ring_task(&q->r);
+
+  mutexinit(q->mtx);
+  condinit(q->cv);
+}
+
+static void free_not_q(not_q_t* q)
+{
+  DevAssert(q != NULL);
+
+  free_seq_ring_task(&q->r);
+
+  mutexdestroy(q->mtx);
+  conddestroy(q->cv);
+}
+
+static bool try_push_not_q(not_q_t* q, task_t t)
+{
+  DevAssert(q != NULL);
+
+
+  if (mutextrylock(q->mtx) != 0)
+    return false;
+
+  push_back_seq_ring_task(&q->r, t);
+
+  const size_t sz = size_seq_ring_task(&q->r);
+  DevAssert(sz > 0);
+
+  mutexunlock(q->mtx);
+
+  condsignal(q->cv);
+
+  return true;
+}
+
+static void push_not_q(not_q_t* q, task_t t)
+{
+  DevAssert(q != NULL);
+  DevAssert(t.func != NULL);
+
+  mutexlock(q->mtx);
+
+  push_back_seq_ring_task(&q->r, t);
+
+  DevAssert(size_seq_ring_task(&q->r) > 0);
+
+  mutexunlock(q->mtx);
+
+  condsignal(q->cv);
+}
+
+static ret_try_t try_pop_not_q(not_q_t* q)
+{
+  DevAssert(q != NULL);
+
+  ret_try_t ret = {.success = false};
+
+  int rc = mutextrylock(q->mtx);
+  DevAssert(rc == 0 || rc == EBUSY);
+
+  if (rc == EBUSY)
+    return ret;
+
+  size_t sz = size_seq_ring_task(&q->r);
+  if (sz == 0) {
+    mutexunlock(q->mtx);
+
+    return ret;
+  }
+
+  DevAssert(sz > 0);
+  ret.t = pop_seq_ring_task(&q->r);
+
+  mutexunlock(q->mtx);
+  ret.success = true;
+
+  return ret;
+}
+
+static bool pop_not_q(not_q_t* q, ret_try_t* out)
+{
+  DevAssert(q != NULL);
+  DevAssert(out != NULL);
+
+  mutexlock(q->mtx);
+
+  while (size_seq_ring_task(&q->r) == 0) {
+    condwait(q->cv, q->mtx);
+  }
+
+  out->t = pop_seq_ring_task(&q->r);
+
+  mutexunlock(q->mtx);
+
+  return true;
+}
+
+#endif

common/utils/threadPool/notified_fifo.h

@@ -0,0 +1,223 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+
+#ifndef NOTIFIED_FIFO_H
+#define NOTIFIED_FIFO_H
+#include "pthread_utils.h"
+#include <stdint.h>
+#include <pthread.h>
+#include "time_meas.h"
+#include <memory.h>
+#include <stdalign.h>
+#include "assertions.h"
+
+/// @brief Element on the notifiedFifo_t
+/// next: internal FIFO chain, do not set it
+/// key: a long int that the client can use to identify a job or a group of messages
+/// ResponseFifo: if the client defines a response FIFO, the job will be posted back after processing
+/// processingFunc: any function (of type void processingFunc(void *)) that a worker will process
+/// msgData: the data passed to `processingFunc`. It can be added automatically, or you can set it to a buffer you are managing
+/// malloced: a boolean that enables internal free in the case of no return FIFO or abort feature
+typedef struct notifiedFIFO_elt_s {
+  struct notifiedFIFO_elt_s *next;
+  uint64_t key; // To filter out elements
+  struct notifiedFIFO_s *reponseFifo;
+  void (*processingFunc)(void *);
+  bool malloced;
+  oai_cputime_t creationTime;
+  oai_cputime_t startProcessingTime;
+  oai_cputime_t endProcessingTime;
+  oai_cputime_t returnTime;
+  // use alignas(32) to align msgData to 32b
+  // user data behind it will be aligned to 32b as well
+  // important! this needs to be the last member in the struct
+  alignas(32) void *msgData;
+} notifiedFIFO_elt_t;
+
+typedef struct notifiedFIFO_s {
+  notifiedFIFO_elt_t *outF;
+  notifiedFIFO_elt_t *inF;
+  pthread_mutex_t lockF;
+  pthread_cond_t notifF;
+  bool abortFIFO; // if set, the FIFO always returns NULL -> abort condition
+} notifiedFIFO_t;
+
+/// @brief Creates a new job.
+/// @param size The data part of the job will have this size
+/// @param key the job can be identified
+/// @param reponseFifo response fifo
+/// @param processingFunc function to call
+/// @return new notifiedFIFO_elt_t element with extra memory allocated at the end equal to size.
+static inline notifiedFIFO_elt_t *newNotifiedFIFO_elt(int size,
+                                                      uint64_t key,
+                                                      notifiedFIFO_t *reponseFifo,
+                                                      void (*processingFunc)(void *))
+{
+  notifiedFIFO_elt_t *ret = (notifiedFIFO_elt_t *)memalign(32, sizeof(notifiedFIFO_elt_t) + size);
+  AssertFatal(NULL != ret, "out of memory\n");
+  ret->next = NULL;
+  ret->key = key;
+  ret->reponseFifo = reponseFifo;
+  ret->processingFunc = processingFunc;
+  // We set user data piece aligend 32 bytes to be able to process it with SIMD
+  // msgData is aligned to 32bytes, so everything after will be as well
+  ret->msgData = ((uint8_t *)ret) + sizeof(notifiedFIFO_elt_t);
+  ret->malloced = true;
+  return ret;
+}
+
+/// @brief Get pointer to the data carried by notifiedFIFO_elt_t
+/// @param elt
+/// @return void pointer to the data allocated with the message
+static inline void *NotifiedFifoData(notifiedFIFO_elt_t *elt)
+{
+  return elt->msgData;
+}
+
+/// @brief Delete a notifiedFIFO_elt_t if its allocated
+/// @param elt
+static inline void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt)
+{
+  if (elt->malloced) {
+    elt->malloced = false;
+    free(elt);
+  }
+}
+
+static inline void initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)
+{
+  nf->inF = NULL;
+  nf->outF = NULL;
+  nf->abortFIFO = false;
+}
+static inline void initNotifiedFIFO(notifiedFIFO_t *nf)
+{
+  mutexinit(nf->lockF);
+  condinit(nf->notifF);
+  initNotifiedFIFO_nothreadSafe(nf);
+  // No delete function: the creator has only to free the memory
+}
+
+static inline void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)
+{
+  msg->next = NULL;
+
+  if (nf->outF == NULL)
+    nf->outF = msg;
+
+  if (nf->inF != NULL)
+    nf->inF->next = msg;
+
+  nf->inF = msg;
+}
+
+static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)
+{
+  mutexlock(nf->lockF);
+  if (!nf->abortFIFO) {
+    pushNotifiedFIFO_nothreadSafe(nf, msg);
+    condsignal(nf->notifF);
+  }
+  mutexunlock(nf->lockF);
+}
+
+static inline notifiedFIFO_elt_t *pullNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)
+{
+  if (nf->outF == NULL)
+    return NULL;
+  if (nf->abortFIFO)
+    return NULL;
+
+  notifiedFIFO_elt_t *ret = nf->outF;
+
+  AssertFatal(nf->outF != nf->outF->next, "Circular list in thread pool: push several times the same buffer is forbidden\n");
+
+  nf->outF = nf->outF->next;
+
+  if (nf->outF == NULL)
+    nf->inF = NULL;
+
+  return ret;
+}
+
+static inline notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf)
+{
+  mutexlock(nf->lockF);
+  notifiedFIFO_elt_t *ret = NULL;
+
+  while ((ret = pullNotifiedFIFO_nothreadSafe(nf)) == NULL && !nf->abortFIFO)
+    condwait(nf->notifF, nf->lockF);
+
+  mutexunlock(nf->lockF);
+  return ret;
+}
+
+static inline notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf)
+{
+  int tmp = mutextrylock(nf->lockF);
+
+  if (tmp != 0)
+    return NULL;
+
+  if (nf->abortFIFO) {
+    mutexunlock(nf->lockF);
+    return NULL;
+  }
+
+  notifiedFIFO_elt_t *ret = pullNotifiedFIFO_nothreadSafe(nf);
+  mutexunlock(nf->lockF);
+  return ret;
+}
+
+static inline time_stats_t exec_time_stats_NotifiedFIFO(const notifiedFIFO_elt_t *elt)
+{
+  time_stats_t ts = {0};
+  if (elt->startProcessingTime == 0 && elt->endProcessingTime == 0)
+    return ts; /* no measurements done */
+  ts.in = elt->startProcessingTime;
+  ts.diff = elt->endProcessingTime - ts.in;
+  ts.p_time = ts.diff;
+  ts.diff_square = ts.diff * ts.diff;
+  ts.max = ts.diff;
+  ts.trials = 1;
+  return ts;
+}
+
+// This functions aborts all messages in the queue, and marks the queue as
+// "aborted", such that every call to it will return NULL
+static inline void abortNotifiedFIFO(notifiedFIFO_t *nf)
+{
+  mutexlock(nf->lockF);
+  nf->abortFIFO = true;
+  notifiedFIFO_elt_t **elt = &nf->outF;
+  while (*elt != NULL) {
+    notifiedFIFO_elt_t *p = *elt;
+    *elt = (*elt)->next;
+    delNotifiedFIFO_elt(p);
+  }
+
+  if (nf->outF == NULL)
+    nf->inF = NULL;
+  condbroadcast(nf->notifF);
+  mutexunlock(nf->lockF);
+}
+
+#endif

common/utils/threadPool/pthread_utils.h

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+
+#ifndef PTHREAD_UTILS_H
+#define PTHREAD_UTILS_H
+
+#include <pthread.h>
+#define mutexinit(mutex)   {int ret=pthread_mutex_init(&mutex,NULL); \
+                            AssertFatal(ret==0,"ret=%d\n",ret);}
+#define condinit(signal)   {int ret=pthread_cond_init(&signal,NULL); \
+                            AssertFatal(ret==0,"ret=%d\n",ret);}
+#define mutexlock(mutex)   {int ret=pthread_mutex_lock(&mutex); \
+                            AssertFatal(ret==0,"ret=%d\n",ret);}
+#define mutextrylock(mutex)   pthread_mutex_trylock(&mutex)
+#define mutexunlock(mutex) {int ret=pthread_mutex_unlock(&mutex); \
+                            AssertFatal(ret==0,"ret=%d\n",ret);}
+#define condwait(condition, mutex) {int ret=pthread_cond_wait(&condition, &mutex); \
+                                    AssertFatal(ret==0,"ret=%d\n",ret);}
+#define condbroadcast(signal) {int ret=pthread_cond_broadcast(&signal); \
+                               AssertFatal(ret==0,"ret=%d\n",ret);}
+#define condsignal(signal)    {int ret=pthread_cond_signal(&signal); \
+                               AssertFatal(ret==0,"ret=%d\n",ret);}
+#define mutexdestroy(mutex) { int ret = pthread_mutex_destroy(&mutex);\
+                              AssertFatal(ret==0,"ret=%d\n",ret);}
+#define conddestroy(condition) { int ret = pthread_cond_destroy(&condition);\
+                                 AssertFatal(ret==0,"ret=%d\n",ret);}
+#endif

common/utils/threadPool/task.h

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+
+#ifndef TASK_WORK_STEALING_THREAD_POOL_H
+#define TASK_WORK_STEALING_THREAD_POOL_H
+
+typedef struct {
+  void* args;
+  void (*func)(void* args);
+} task_t;
+
+#endif

common/utils/threadPool/task_ans.c

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+
+#include "task_ans.h"
+#include "assertions.h"
+#include <limits.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <time.h>
+
+void completed_task_ans(task_ans_t* task)
+{
+  DevAssert(task != NULL);
+
+  if (atomic_load_explicit(&task->status, memory_order_acquire) != 0)
+    AssertFatal(0, "Task already finished?");
+
+  atomic_store_explicit(&task->status, 1, memory_order_release);
+}
+
+void join_task_ans(task_ans_t* arr, size_t len)
+{
+  DevAssert(len < INT_MAX);
+  DevAssert(arr != NULL);
+
+  // Spin lock inspired by:
+  // The Art of Writing Efficient Programs:
+  // An advanced programmer's guide to efficient hardware utilization
+  // and compiler optimizations using C++ examples
+  const struct timespec ns = {0, 1};
+  uint64_t i = 0;
+  int j = len - 1;
+  for (; j != -1; i++) {
+    for (; j != -1; --j) {
+      int const task_completed = 1;
+      if (atomic_load_explicit(&arr[j].status, memory_order_acquire) != task_completed)
+        break;
+    }
+    if (i % 8 == 0) {
+      nanosleep(&ns, NULL);
+    }
+  }
+}

common/utils/threadPool/task_ans.h

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+
+#ifndef TASK_ANSWER_THREAD_POOL_H
+#define TASK_ANSWER_THREAD_POOL_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#ifndef __cplusplus
+#include <stdalign.h>
+#include <stdatomic.h>
+#else
+#include <atomic>
+#define _Atomic(X) std::atomic<X>
+#define _Alignas(X) alignas(X)
+#endif
+
+#include <stddef.h>
+#include <stdint.h>
+
+#if defined(__i386__) || defined(__x86_64__)
+#define LEVEL1_DCACHE_LINESIZE 64
+#elif defined(__aarch64__)
+// This is not always true for ARM
+// in linux, you can obtain the size at runtime using sysconf (_SC_LEVEL1_DCACHE_LINESIZE)
+// or from the bash with the command $ getconf LEVEL1_DCACHE_LINESIZE
+// in c++ using std::hardware_destructive_interference_size
+#define LEVEL1_DCACHE_LINESIZE 64
+#else
+#error Unknown CPU architecture
+#endif
+
+typedef struct {
+  // Avoid false sharing
+  _Alignas(LEVEL1_DCACHE_LINESIZE) _Atomic(int) status;
+} task_ans_t;
+
+typedef struct {
+  uint8_t* buf;
+  size_t len;
+  size_t cap; // capacity
+  task_ans_t* ans;
+} thread_info_tm_t;
+
+
+void join_task_ans(task_ans_t* arr, size_t len);
+
+void completed_task_ans(task_ans_t* task);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

common/utils/threadPool/test/test_thread-pool.c

@@ -27,8 +27,10 @@
 #include <unistd.h>
 #include <ctype.h>
 #include <sys/sysinfo.h>
-#include <threadPool/thread-pool.h>
+#include "thread-pool.h"
 #include "log.h"
+#include "task_ans.h"
+#include "notified_fifo.h"
 
 void displayList(notifiedFIFO_t *nf)
 {
@@ -47,6 +49,7 @@ struct testData {
   int id;
   int sleepTime;
   char txt[50];
+  task_ans_t* task_ans;
 };
 
 void processing(void *arg)
@@ -57,6 +60,7 @@ void processing(void *arg)
   in->sleepTime = rand() % 1000;
   usleep(in->sleepTime);
   // printf("done: %d, %s, in thr %ld\n",in->id, in->txt,pthread_self() );
+  completed_task_ans(in->task_ans);
 }
 
 int main()
@@ -102,8 +106,6 @@ int main()
   tpool_t pool;
   char params[] = "1,2,3,4,5";
   initTpool(params, &pool, true);
-  notifiedFIFO_t worker_back;
-  initNotifiedFIFO(&worker_back);
 
   //sleep(1);
   int cumulProcessTime = 0;
@@ -112,21 +114,22 @@ int main()
   int nb_jobs = 4;
   for (int i = 0; i < 1000; i++) {
     int parall = nb_jobs;
+    task_ans_t task_ans[parall];
+    memset(task_ans, 0, sizeof(task_ans));
+    struct testData test_data[parall];
+    memset(test_data, 0, sizeof(test_data));
     for (int j = 0; j < parall; j++) {
-      notifiedFIFO_elt_t *work = newNotifiedFIFO_elt(sizeof(struct testData), i, &worker_back, processing);
-      struct testData *x = (struct testData *)NotifiedFifoData(work);
+      task_t task = {.args = &test_data[j], .func = processing};
+      struct testData *x = (struct testData *)task.args;
       x->id = i;
-      pushTpool(&pool, work);
+      x->task_ans = &task_ans[j];
+      pushTpool(&pool, task);
     }
+    join_task_ans(task_ans, parall);
     int sleepmax = 0;
-    while (parall) {
-      tmp = pullTpool(&worker_back, &pool);
-      if (tmp) {
-        parall--;
-        struct testData *dd = NotifiedFifoData(tmp);
-        if (dd->sleepTime > sleepmax)
-          sleepmax = dd->sleepTime;
-        delNotifiedFIFO_elt(tmp);
+    for (int j = 0; j < parall; j++) {
+      if (test_data[j].sleepTime > sleepmax) {
+        sleepmax = test_data[j].sleepTime;
       }
     }
     cumulProcessTime += sleepmax;

common/utils/threadPool/thread-pool.c

@@ -1,163 +1,192 @@
 /*
-* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
-* contributor license agreements.  See the NOTICE file distributed with
-* this work for additional information regarding copyright ownership.
-* The OpenAirInterface Software Alliance licenses this file to You under
-* the OAI Public License, Version 1.1  (the "License"); you may not use this file
-* except in compliance with the License.
-* You may obtain a copy of the License at
-*
-*      http://www.openairinterface.org/?page_id=698
-*
-* Author and copyright: Laurent Thomas, open-cells.com
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License.
-*-------------------------------------------------------------------------------
-* For more information about the OpenAirInterface (OAI) Software Alliance:
-*      contact@openairinterface.org
-*/
-
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
 
 #define _GNU_SOURCE
 #include <sys/stat.h>
 #include <fcntl.h>
 #include <string.h>
 #include <ctype.h>
-#include <threadPool/thread-pool.h>
+#include "thread-pool.h"
+#include "bounded_notified_fifo.h"
+#include <sys/sysinfo.h>
 
+typedef struct {
+  tpool_t* tpool;
+  int idx;
+} task_thread_args_t;
 
-static inline  notifiedFIFO_elt_t *pullNotifiedFifoRemember( notifiedFIFO_t *nf, struct one_thread *thr) {
-  mutexlock(nf->lockF);
-
-  while(!nf->outF && !thr->terminate)
-    condwait(nf->notifF, nf->lockF);
-
-  if (thr->terminate) {
-    mutexunlock(nf->lockF);
-    return NULL;
+void pushTpool(tpool_t* tpool, task_t task)
+{
+  DevAssert(tpool != NULL);
+  if (tpool->len_thr == 0) {
+    task.func(task.args);
+    return;
   }
 
-  notifiedFIFO_elt_t *ret=nf->outF;
-  nf->outF=nf->outF->next;
+  size_t const index = tpool->index++;
+  size_t const len_thr = tpool->len_thr;
 
-  if (nf->outF==NULL)
-    nf->inF=NULL;
+  not_q_t* q_arr = (not_q_t*)tpool->q_arr;
 
-  // For abort feature
-  thr->runningOnKey=ret->key;
-  thr->dropJob = false;
-  mutexunlock(nf->lockF);
-  return ret;
+  for (size_t i = 0; i < len_thr; ++i) {
+    if (try_push_not_q(&q_arr[(i + index) % len_thr], task)) {
+      return;
+    }
+  }
+
+  push_not_q(&q_arr[index % len_thr], task);
 }
 
-void *one_thread(void *arg) {
-  struct  one_thread *myThread=(struct  one_thread *) arg;
-  struct  thread_pool *tp=myThread->pool;
+static void* worker_thread(void* arg)
+{
+  DevAssert(arg != NULL);
 
-  // Infinite loop to process requests
-  do {
-    notifiedFIFO_elt_t *elt=pullNotifiedFifoRemember(&tp->incomingFifo, myThread);
-    if (elt == NULL) {
-      AssertFatal(myThread->terminate, "pullNotifiedFifoRemember() returned NULL although thread not aborted\n");
+  task_thread_args_t* args = (task_thread_args_t*)arg;
+  int const idx = args->idx;
+  tpool_t* tpool = args->tpool;
+
+  uint32_t const len = tpool->len_thr;
+  uint32_t const num_it = 2 * (tpool->len_thr + idx);
+
+  not_q_t* q_arr = (not_q_t*)tpool->q_arr;
+
+  init_not_q(&q_arr[idx], idx);
+  // Synchronize all threads
+  pthread_barrier_wait(&tpool->barrier);
+
+  for (;;) {
+    ret_try_t ret = {.success = false};
+
+    for (uint32_t i = idx; i < num_it; ++i) {
+      ret = try_pop_not_q(&q_arr[i % len]);
+      if (ret.success == true)
+        break;
+    }
+
+    if (ret.success == false) {
+      if (pop_not_q(&q_arr[idx], &ret) == false)
+        break;
+    }
+
+    if (ret.t.func == NULL && ret.t.args == NULL) {
+      pushTpool(tpool, (task_t){.args = NULL, .func = NULL});
       break;
     }
+    ret.t.func(ret.t.args);
+  }
 
-    if (tp->measurePerf) elt->startProcessingTime=rdtsc_oai();
-
-    elt->processingFunc(NotifiedFifoData(elt));
-
-    if (tp->measurePerf) elt->endProcessingTime=rdtsc_oai();
-
-    if (elt->reponseFifo) {
-      // Check if the job is still alive, else it has been aborted
-      mutexlock(tp->incomingFifo.lockF);
-
-      if (myThread->dropJob)
-        delNotifiedFIFO_elt(elt);
-      else
-        pushNotifiedFIFO(elt->reponseFifo, elt);
-      myThread->runningOnKey=-1;
-      mutexunlock(tp->incomingFifo.lockF);
-    }
-    else
-      delNotifiedFIFO_elt(elt);
-  } while (!myThread->terminate);
+  free(args);
   return NULL;
 }
 
-void initNamedTpool(char *params,tpool_t *pool, bool performanceMeas, char *name) {
-  memset(pool,0,sizeof(*pool));
-  char *measr=getenv("OAI_THREADPOOLMEASUREMENTS");
-  pool->measurePerf=performanceMeas;
-  // force measurement if the output is defined
-  pool->measurePerf |= measr!=NULL;
+void initNamedTpool(char* params, tpool_t* tpool, bool performanceMeas, char* name)
+{
+  (void)performanceMeas;
 
-  if (measr) {
-    mkfifo(measr,0666);
-    AssertFatal(-1 != (pool->dummyKeepReadingTraceFd=
-                         open(measr, O_RDONLY| O_NONBLOCK)),"");
-    AssertFatal(-1 != (pool->traceFd=
-                         open(measr, O_WRONLY|O_APPEND|O_NOATIME|O_NONBLOCK)),"");
-  } else
-    pool->traceFd=-1;
+  DevAssert(tpool != NULL);
+  memset(tpool, 0, sizeof(*tpool));
 
-  pool->activated=true;
-  initNotifiedFIFO(&pool->incomingFifo);
-  char *saveptr, * curptr;
-  char *parms_cpy=strdup(params);
-  pool->nbThreads=0;
-  curptr=strtok_r(parms_cpy,",",&saveptr);
-  struct one_thread * ptr;
-  char *tname = (name == NULL ? "Tpool" : name);
-  while ( curptr!=NULL ) {
-    int c=toupper(curptr[0]);
+  char* tname = (name == NULL ? "Tpool" : name);
+  char *saveptr, *curptr;
+  char* parms_cpy = strdup(params);
+  curptr = strtok_r(parms_cpy, ",", &saveptr);
+  int core_id[128] = {0};
+  int num_workers = 0;
+  while (curptr != NULL) {
+    int c = toupper(curptr[0]);
 
     switch (c) {
-
       case 'N':
-        pool->activated=false;
         break;
 
       default:
-        ptr=pool->allthreads;
-        pool->allthreads=(struct one_thread *)malloc(sizeof(struct one_thread));
-        pool->allthreads->next=ptr;
-        pool->allthreads->coreID=atoi(curptr);
-        pool->allthreads->id=pool->nbThreads;
-        pool->allthreads->pool=pool;
-        pool->allthreads->dropJob = false;
-        pool->allthreads->terminate = false;
-        //Configure the thread scheduler policy for Linux
-        // set the thread name for debugging
-        sprintf(pool->allthreads->name,"%s%d_%d",tname,pool->nbThreads,pool->allthreads->coreID);
-        // we set the maximum priority for thread pool threads (which is close
-        // but not equal to Linux maximum). See also the corresponding commit
-        // message; initially introduced for O-RAN 7.2 fronthaul split
-        threadCreate(&pool->allthreads->threadID, one_thread, (void *)pool->allthreads,
-                     pool->allthreads->name, pool->allthreads->coreID, OAI_PRIORITY_RT_MAX);
-        pool->nbThreads++;
+        core_id[num_workers++] = atoi(curptr);
     }
 
-    curptr=strtok_r(NULL,",",&saveptr);
+    curptr = strtok_r(NULL, ",", &saveptr);
   }
   free(parms_cpy);
-  if (pool->activated && pool->nbThreads==0) {
-    printf("No servers created in the thread pool, exit\n");
-    exit(1);
+
+  if (num_workers) {
+    tpool->q_arr = calloc(num_workers, sizeof(not_q_t));
+    AssertFatal(tpool->q_arr != NULL, "Memory exhausted");
+
+    tpool->t_arr = calloc(num_workers, sizeof(pthread_t));
+    AssertFatal(tpool->t_arr != NULL, "Memory exhausted");
   }
+  tpool->len_thr = num_workers;
+
+  tpool->index = 0;
+
+  const pthread_barrierattr_t* barrier_attr = NULL;
+  int rc = pthread_barrier_init(&tpool->barrier, barrier_attr, num_workers + 1);
+  DevAssert(rc == 0);
+
+  for (size_t i = 0; i < num_workers; ++i) {
+    task_thread_args_t* args = malloc(sizeof(task_thread_args_t));
+    AssertFatal(args != NULL, "Memory exhausted");
+    args->idx = i;
+    args->tpool = tpool;
+    char name[64];
+    sprintf(name, "%s%ld_%d", tname, i, core_id[i]);
+    threadCreate(&tpool->t_arr[i], worker_thread, args, name, core_id[i], OAI_PRIORITY_RT_MAX);
+  }
+
+  // Syncronize thread pool threads. All the threads started
+  pthread_barrier_wait(&tpool->barrier);
 }
 
-void initFloatingCoresTpool(int nbThreads,tpool_t *pool, bool performanceMeas, char *name) {
+void initFloatingCoresTpool(int nbThreads, tpool_t* pool, bool performanceMeas, char* name)
+{
   char threads[1024] = "n";
   if (nbThreads) {
-    strcpy(threads,"-1");
-    for (int i=1; i < nbThreads; i++)
-      strncat(threads,",-1", sizeof(threads)-1);
+    strcpy(threads, "-1");
+    for (int i = 1; i < nbThreads; i++)
+      strncat(threads, ",-1", sizeof(threads) - 1);
   }
-  threads[sizeof(threads)-1]=0;
+  threads[sizeof(threads) - 1] = 0;
   initNamedTpool(threads, pool, performanceMeas, name);
 }
+
+
+void abortTpool(tpool_t* tpool)
+{
+  if (tpool->len_thr > 0) {
+    not_q_t* q_arr = (not_q_t*)tpool->q_arr;
+
+    pushTpool(tpool, (task_t){.args = NULL, .func = NULL});
+
+    for (uint32_t i = 0; i < tpool->len_thr; ++i) {
+      int rc = pthread_join(tpool->t_arr[i], NULL);
+      DevAssert(rc == 0);
+    }
+
+    for (uint32_t i = 0; i < tpool->len_thr; ++i) {
+      free_not_q(&q_arr[i]);
+    }
+
+    free(tpool->q_arr);
+    free(tpool->t_arr);
+  }
+
+  int rc = pthread_barrier_destroy(&tpool->barrier);
+  DevAssert(rc == 0);
+}

common/utils/threadPool/thread-pool.h

@@ -1,26 +1,23 @@
 /*
-* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
-* contributor license agreements.  See the NOTICE file distributed with
-* this work for additional information regarding copyright ownership.
-* The OpenAirInterface Software Alliance licenses this file to You under
-* the OAI Public License, Version 1.1  (the "License"); you may not use this file
-* except in compliance with the License.
-* You may obtain a copy of the License at
-*
-*      http://www.openairinterface.org/?page_id=698
-*
-* Author and copyright: Laurent Thomas, open-cells.com
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License.
-*-------------------------------------------------------------------------------
-* For more information about the OpenAirInterface (OAI) Software Alliance:
-*      contact@openairinterface.org
-*/
-
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
 
 #ifndef THREAD_POOL_H
 #define THREAD_POOL_H
@@ -35,316 +32,54 @@
 #include "assertions.h"
 #include "common/utils/time_meas.h"
 #include "common/utils/system.h"
+#include "task.h"
+#include "pthread_utils.h"
 
 #ifdef DEBUG
   #define THREADINIT   PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP
 #else
   #define THREADINIT   PTHREAD_MUTEX_INITIALIZER
 #endif
-#define mutexinit(mutex)   {int ret=pthread_mutex_init(&mutex,NULL); \
-                            AssertFatal(ret==0,"ret=%d\n",ret);}
-#define condinit(signal)   {int ret=pthread_cond_init(&signal,NULL); \
-                            AssertFatal(ret==0,"ret=%d\n",ret);}
-#define mutexlock(mutex)   {int ret=pthread_mutex_lock(&mutex); \
-                            AssertFatal(ret==0,"ret=%d\n",ret);}
-#define mutextrylock(mutex)   pthread_mutex_trylock(&mutex)
-#define mutexunlock(mutex) {int ret=pthread_mutex_unlock(&mutex); \
-                            AssertFatal(ret==0,"ret=%d\n",ret);}
-#define condwait(condition, mutex) {int ret=pthread_cond_wait(&condition, &mutex); \
-                                    AssertFatal(ret==0,"ret=%d\n",ret);}
-#define condbroadcast(signal) {int ret=pthread_cond_broadcast(&signal); \
-                               AssertFatal(ret==0,"ret=%d\n",ret);}
-#define condsignal(signal)    {int ret=pthread_cond_signal(&signal); \
-                               AssertFatal(ret==0,"ret=%d\n",ret);}
-#define tpool_nbthreads(tpool)   (tpool.nbThreads)
-typedef struct notifiedFIFO_elt_s {
-  struct notifiedFIFO_elt_s *next;
-  uint64_t key; //To filter out elements
-  struct notifiedFIFO_s *reponseFifo;
-  void (*processingFunc)(void *);
-  bool malloced;
-  oai_cputime_t creationTime;
-  oai_cputime_t startProcessingTime;
-  oai_cputime_t endProcessingTime;
-  oai_cputime_t returnTime;
-  // use alignas(32) to align msgData to 32b
-  // user data behind it will be aligned to 32b as well
-  // important! this needs to be the last member in the struct
-  alignas(32) void *msgData;
-}  notifiedFIFO_elt_t;
 
-typedef struct notifiedFIFO_s {
-  notifiedFIFO_elt_t *outF;
-  notifiedFIFO_elt_t *inF;
-  pthread_mutex_t lockF;
-  pthread_cond_t  notifF;
-  bool abortFIFO; // if set, the FIFO always returns NULL -> abort condition
-} notifiedFIFO_t;
+typedef struct {
+  pthread_t* t_arr;
+  size_t len_thr;
 
-// You can use this allocator or use any piece of memory
-static inline notifiedFIFO_elt_t *newNotifiedFIFO_elt(int size,
-    uint64_t key,
-    notifiedFIFO_t *reponseFifo,
-    void (*processingFunc)(void *)) {
-  notifiedFIFO_elt_t *ret = (notifiedFIFO_elt_t *)memalign(32, sizeof(notifiedFIFO_elt_t) + size);
-  AssertFatal(NULL != ret, "out of memory\n");
-  ret->next=NULL;
-  ret->key=key;
-  ret->reponseFifo=reponseFifo;
-  ret->processingFunc=processingFunc;
-  // We set user data piece aligend 32 bytes to be able to process it with SIMD
-  // msgData is aligned to 32bytes, so everything after will be as well
-  ret->msgData = ((uint8_t *)ret) + sizeof(notifiedFIFO_elt_t);
-  ret->malloced=true;
-  return ret;
-}
+  _Atomic(uint64_t) index;
 
-static inline void *NotifiedFifoData(notifiedFIFO_elt_t *elt) {
-  return elt->msgData;
-}
+  void* q_arr;
 
-static inline void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt) {
-  if (elt->malloced) {
-    elt->malloced = false;
-    free(elt);
-  }
-  /* it is allowed to call delNotifiedFIFO_elt when the memory is managed by
-   * the caller */
-}
-
-static inline void initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf) {
-  nf->inF=NULL;
-  nf->outF=NULL;
-  nf->abortFIFO = false;
-}
-static inline void initNotifiedFIFO(notifiedFIFO_t *nf) {
-  mutexinit(nf->lockF);
-  condinit (nf->notifF);
-  initNotifiedFIFO_nothreadSafe(nf);
-  // No delete function: the creator has only to free the memory
-}
-
-static inline void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) {
-  msg->next=NULL;
-
-  if (nf->outF == NULL)
-    nf->outF = msg;
-
-  if (nf->inF != NULL)
-    nf->inF->next = msg;
-
-  nf->inF = msg;
-}
-
-static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) {
-  mutexlock(nf->lockF);
-  if (!nf->abortFIFO) {
-    pushNotifiedFIFO_nothreadSafe(nf,msg);
-    condsignal(nf->notifF);
-  }
-  mutexunlock(nf->lockF);
-}
-
-static inline  notifiedFIFO_elt_t *pullNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf) {
-  if (nf->outF == NULL)
-    return NULL;
-  if (nf->abortFIFO)
-    return NULL;
-
-  notifiedFIFO_elt_t *ret=nf->outF;
-
-  AssertFatal(nf->outF != nf->outF->next,"Circular list in thread pool: push several times the same buffer is forbidden\n");
-
-  nf->outF=nf->outF->next;
-
-  if (nf->outF==NULL)
-    nf->inF=NULL;
-
-  return ret;
-}
-
-static inline  notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf) {
-  mutexlock(nf->lockF);
-  notifiedFIFO_elt_t *ret = NULL;
-
-  while((ret=pullNotifiedFIFO_nothreadSafe(nf)) == NULL && !nf->abortFIFO)
-    condwait(nf->notifF, nf->lockF);
-
-  mutexunlock(nf->lockF);
-  return ret;
-}
-
-static inline  notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf) {
-  int tmp=mutextrylock(nf->lockF);
-
-  if (tmp != 0 )
-    return NULL;
-
-  if (nf->abortFIFO) {
-    mutexunlock(nf->lockF);
-    return NULL;
-  }
-
-  notifiedFIFO_elt_t *ret=pullNotifiedFIFO_nothreadSafe(nf);
-  mutexunlock(nf->lockF);
-  return ret;
-}
-
-static inline time_stats_t exec_time_stats_NotifiedFIFO(const notifiedFIFO_elt_t* elt)
-{
-  time_stats_t ts = {0};
-  if (elt->startProcessingTime == 0 && elt->endProcessingTime == 0)
-    return ts; /* no measurements done */
-  ts.in = elt->startProcessingTime;
-  ts.diff = elt->endProcessingTime - ts.in;
-  ts.p_time = ts.diff;
-  ts.diff_square = ts.diff * ts.diff;
-  ts.max = ts.diff;
-  ts.trials = 1;
-  return ts;
-}
-
-// This functions aborts all messages in the queue, and marks the queue as
-// "aborted", such that every call to it will return NULL
-static inline void abortNotifiedFIFO(notifiedFIFO_t *nf) {
-  mutexlock(nf->lockF);
-  nf->abortFIFO = true;
-  notifiedFIFO_elt_t **elt = &nf->outF;
-  while(*elt != NULL) {
-    notifiedFIFO_elt_t *p = *elt;
-    *elt = (*elt)->next;
-    delNotifiedFIFO_elt(p);
-  }
-
-  if (nf->outF == NULL)
-    nf->inF = NULL;
-  condbroadcast(nf->notifF);
-  mutexunlock(nf->lockF);
-}
-
-struct one_thread {
-  pthread_t  threadID;
-  int id;
-  int coreID;
-  char name[256];
-  uint64_t runningOnKey;
-  bool dropJob;
-  bool terminate;
-  struct thread_pool *pool;
-  struct one_thread *next;
-};
-
-typedef struct thread_pool {
-  bool activated;
-  bool measurePerf;
-  int traceFd;
-  int dummyKeepReadingTraceFd;
-  uint64_t cpuCyclesMicroSec;
-  int nbThreads;
-  notifiedFIFO_t incomingFifo;
-  struct one_thread *allthreads;
+  pthread_barrier_t barrier;
 } tpool_t;
 
-static inline void pushTpool(tpool_t *t, notifiedFIFO_elt_t *msg) {
-  if (t->measurePerf) msg->creationTime=rdtsc_oai();
+/// @brief Push job to threadpool. May run task inline in case there are no worker threads
+///        defined for threadpool
+/// @param tpool threadpool to use
+/// @param task task description
+void pushTpool(tpool_t *tpool, task_t task);
 
-  if ( t->activated)
-    pushNotifiedFIFO(&t->incomingFifo, msg);
-  else {
-    if (t->measurePerf)
-      msg->startProcessingTime=rdtsc_oai();
+/// @brief Abort tpool, stop all threads and return
+/// @param t 
+void abortTpool(tpool_t *t);
 
-    msg->processingFunc(NotifiedFifoData(msg));
-
-    if (t->measurePerf)
-      msg->endProcessingTime=rdtsc_oai();
-
-    if (msg->reponseFifo)
-      pushNotifiedFIFO(msg->reponseFifo, msg);
-    else
-      delNotifiedFIFO_elt(msg);
-  }
-}
-
-static inline notifiedFIFO_elt_t *pullTpool(notifiedFIFO_t *responseFifo, tpool_t *t) {
-  notifiedFIFO_elt_t *msg= pullNotifiedFIFO(responseFifo);
-  if (msg == NULL)
-    return NULL;
-  AssertFatal(t->traceFd != 0, "Thread pool used while not initialized");
-  if (t->measurePerf)
-    msg->returnTime=rdtsc_oai();
-
-  if (t->traceFd > 0) {
-    ssize_t b = write(t->traceFd, msg, sizeof(*msg));
-    AssertFatal(b == sizeof(*msg), "error in write(): %d, %s\n", errno, strerror(errno));
-  }
-
-  return msg;
-}
-
-static inline notifiedFIFO_elt_t *tryPullTpool(notifiedFIFO_t *responseFifo, tpool_t *t) {
-  notifiedFIFO_elt_t *msg= pollNotifiedFIFO(responseFifo);
-  AssertFatal(t->traceFd != 0, "Thread pool used while not initialized");
-  if (msg == NULL)
-    return NULL;
-
-  if (t->measurePerf)
-    msg->returnTime=rdtsc_oai();
-
-  if (t->traceFd > 0) {
-    ssize_t b = write(t->traceFd, msg, sizeof(*msg));
-    AssertFatal(b == sizeof(*msg), "error in write(): %d, %s\n", errno, strerror(errno));
-  }
-
-  return msg;
-}
-
-static inline int abortTpool(tpool_t *t) {
-  int nbRemoved=0;
-  /* disables threading: if a message comes in now, we cannot have a race below
-   * as each thread will simply execute the message itself */
-  t->activated = false;
-  notifiedFIFO_t *nf=&t->incomingFifo;
-  mutexlock(nf->lockF);
-  nf->abortFIFO = true;
-  notifiedFIFO_elt_t **start=&nf->outF;
-
-  /* mark threads to abort them */
-  struct one_thread *thread = t->allthreads;
-  while (thread != NULL) {
-    thread->dropJob = true;
-    thread->terminate = true;
-    nbRemoved++;
-    thread = thread->next;
-  }
-
-  /* clear FIFOs */
-  while(*start!=NULL) {
-    notifiedFIFO_elt_t **request=start;
-    *start=(*start)->next;
-    delNotifiedFIFO_elt(*request);
-    *request = NULL;
-    nbRemoved++;
-  }
-
-  if (t->incomingFifo.outF==NULL)
-    t->incomingFifo.inF=NULL;
-
-  condbroadcast(t->incomingFifo.notifF);
-  mutexunlock(nf->lockF);
-
-  /* join threads that are still runing */
-  thread = t->allthreads;
-  while (thread != NULL) {
-    pthread_cancel(thread->threadID);
-    struct one_thread *next = thread->next;
-    free(thread);
-    thread = next;
-  }
-
-  return nbRemoved;
-}
+/// @brief Initialize a threadPool.
+/// @param params A string in a form of "<int>,<int>,<int>,...,<int>" or "n"
+///               if params is "n" threadpool is disabled and all functions are called inline.
+///               else if params is a list of comma separated integers, the number of threads
+///               is equal to the list length and each thread is pinned to a core indicated by
+///               the integer value. If -1 is specified thread is not pinned to any core.
+/// @param pool Theadpool
+/// @param performanceMeas unused
+/// @param name Name of the theadpool, will be used as pthread names for the pool worker threads
 void initNamedTpool(char *params,tpool_t *pool, bool performanceMeas, char *name);
+
+/// @brief Initialize a threadpool with floating core worker threads only
+/// @param nbThreads number of floating core worker threads
+/// @param pool Threadpool
+/// @param performanceMeas unused 
+/// @param name Name of the theadpool, will be used as pthread names for the pool worker threads
 void initFloatingCoresTpool(int nbThreads,tpool_t *pool, bool performanceMeas, char *name);
+
+/// Convenience macro
 #define  initTpool(PARAMPTR,TPOOLPTR, MEASURFLAG) initNamedTpool(PARAMPTR,TPOOLPTR, MEASURFLAG, NULL)
 #endif

common/utils/threadPool/thread-pool.md

@@ -1,198 +1,7 @@
 # Thread pool
 
-The **thread pool** is a working server, made of a set of worker threads that can be mapped on CPU cores.  Each thread pool has an **input queue** ("**FIFO**"), from which its workers pick **jobs** (FIFO element) to execute.  When a job is done, the worker sends a message to an output queue, if it has been defined.
+The **thread pool** is a working server, made of a set of worker threads that can be mapped on CPU cores.
 
 All the thread pool functions are thread safe. The functions executed by worker threads are provided by the thread pool client, so the client has to handle the concurrency/parallel execution of his functions.
 
-## license
-    Author:
-      Laurent Thomas, Open cells project
-      The owner share this piece code to Openairsoftware alliance as per OSA license terms
-      With contributions from Robert Schmidt, OSA
-
-# Jobs
-
-A job is a message of type `notifiedFIFO_elt_t`:
-
-* `next`: internal FIFO chain, do not set it
-* `key`: a `long int` that the client can use to identify a job or a group of messages
-* `ResponseFifo`: if the client defines a response FIFO, the job will be posted back after processing
-* `processingFunc`: any function (of type `void processingFunc(void *)`) that a worker will process
-* `msgData`: the data passed to `processingFunc`. It can be added automatically, or you can set it to a buffer you are managing
-* `malloced`: a boolean that enables internal free in the case of no return FIFO or abort feature
-
-The job messages can be created and destroyed with `newNotifiedFIFO_elt()` and `delNotifiedFIFO_elt()`, or managed by the client:
-
-* `newNotifiedFIFO_elt()`: Creates a job, that will later be used in queues/FIFO
-* `delNotifiedFIFO_elt()`: Deletes a job
-* `NotifiedFifoData()`: gives a pointer to the beginning of free usage memory in a job (you can put any data there, up to 'size' parameter you passed to `newNotifiedFIFO_elt()`)
-
-These 3 calls are not mandatory, you can also use your own memory to save the `malloc()`/`free()` that are behind these calls.
-
-## API details
-
-### `notifiedFIFO_elt_t *newNotifiedFIFO_elt(int size, uint64_t key, notifiedFIFO_t *reponseFifo, void (*processingFunc)(void *))`
-
-Creates a new job. The data part of the job will have size `size`, the job can be identified via `key`. `reponseFifo` is a FIFO queue to which the job should be pushed after being processed by a thread pool (see below for details) using function `processingFunc()`.
-
-The function returns a pointer to an allocated job.
-
-### `void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt)`
-
-Free the memory allocated for a job `elt`.
-
-### `void *NotifiedFifoData(notifiedFIFO_elt_t *elt)`
-
-Returns a pointer (`void *`) to the data segment of an allocated job.
-
-# Queues of jobs (`FIFO`)
-
-Queues can be used to enqueue messages/jobs, of type `notifiedFIFO_t`.
-
-* `initNotifiedFIFO()`: Create a queue
-* No delete function is required, the creator has only to free the data of type `notifiedFIFO_t`
-* `pushNotifiedFIFO()`: Add a job to a queue
-* `pullNotifiedFIFO()`: Pull a job from a queue. This call is blocking until a job arrived.
-* `pollNotifiedFIFO()`: Pull a job from a queue. This call is not blocking, so it returns always very shortly
-* `abortNotifiedFIFO()`: Aborts a FIFO, such that it will always return `NULL`
-
-Note that in 99.9% of cases, `pull()` is better than `poll()`.
-
-The above push/pull/poll functions are built on not-thread-safe versions of these functions that are described below.
-
-## Low level: fast and simple, but not thread-safe
-
-* `initNotifiedFIFO_nothreadSafe()`: Create a queue
-* `pushNotifiedFIFO_nothreadSafe()`: Add a element in a queue
-* `pullNotifiedFIFO_nothreadSafe()`: Pull a element from a queue
-
-As these queues are not thread safe, there is NO blocking mechanism, neither `pull()` versus `poll()` calls
-
-There is no delete for a message queue: you only have to abandon the memory you allocated to call `initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)`
-
-If you malloced the memory under `nf` parameter, you have to free it.
-
-## FIFO queue abort
-
-It is possible to *abort* the FIFO queue. In this case, any pulling/polling of jobs in the queue will return `NULL`. This can be used when stopping a program, to release multiple queues that might depend on each other (a proper design would try to avoid such dependencies, though).
-
-## API details
-
-### `void initNotifiedFIFO(notifiedFIFO_t *nf)`
-### `void initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)`
-
-Initializes a FIFO queue pointed to by `nf` ("notified FIFO").
-
-### `void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)`
-### `void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)`
-
-Pushes a job `msg` to the FIFO queue `nf`.
-
-### `notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf)`
-
-Pulls a job from a FIFO queue `nf`. The returned pointer points to the job.  This call is blocking, i.e., a caller will be blocked until there is a message.
-
-If the FIFO has been aborted, the returned pointer will be `NULL`. In other words, if the call to this functions returns `NULL`, the FIFO has been deactivated.
-
-### `notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf)`
-### `notifiedFIFO_elt_t *pullNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)`
-
-Like `pullNotifiedFIFO()`, but non-blocking: they check if the queue `nf` contains a job, and return it. Otherwise, they return `NULL`.  Note the slight difference in naming: the thread-safe function is called *poll*, not *pull*.
-
-Note that unlike for `pullNotifiedFIFO()`, returning `NULL` does not inform whether the queue has been aborted; the caller should manually check the `abortFIFO` flag of `nf` in this case.
-
-### `void abortNotifiedFIFO(notifiedFIFO_t *nf)`
-
-Aborts the entire FIFO queue `nf`: all jobs will be dropped, and the FIFO is marked as aborted, such that a call to `pullNotifiedFIFO()` returns `NULL`.
-
-Returns the total number of jobs that were waiting or under execution.
-
-# Thread pools
-
-## Initialization
-
-The clients can create one or more thread pools with
-
-* `initTpool(char *params, tpool_t *pool, bool performanceMeas)` or
-* `initNamedTpool(char *params,tpool_t *pool, bool performanceMeas , char *name)`
-
-The threads are governed by the Linux real time scheduler, and their name is set automatically to `Tpool<thread index>_<core id>` if `initTpool()` is used or to `<name><thread index>_<core id>` when `initNamedTpool()` is used.
-
-## Adding jobs
-
-The client create their jobs messages as a `notifiedFIFO_elt_t`. They then push it with `pushTpool()` which internally calls `push_notifiedFIFO()`.
-
-If they need a return value (e.g., result of a computation), they have to create response queues with `init_notifiedFIFO()` and set this FIFO pointer in the `notifiedFIFO_elt_t` before pushing the job. This way, multiple result queues can be used in one thread pool.
-
-## Abort
-
-`abortTpool()` kills all jobs in the Tpool, and terminates the pool.
-
-## API details
-
-Each thread pool (there can be several in the same process) should be initialized once using one of the two API's:
-
-### `void initNamedTpool(char *params, tpool_t *pool, bool performanceMeas, char *name)`
-
-### `void initTpool(char *params, tpool_t *pool, bool performanceMeas)`
-
-
-The `params` string describes a list of cores, separated by "," that run a worker thread.  If the core exists on the CPU, the thread pool initialization sets the affinity between this thread and the related core. The following options are available:
-
-* `N`: if an `N` is in the parameter list, no threads are created, and the threadpool will be disabled (jobs will be treated by each calling thread independently).
-* A number that represent a valid CPU core on the target CPU, i.e., if there are `M` cores, a number within `[0,M-1]`: A thread is created and pinned to the core (with set affinity)
-* `-1`: The thread will not be mapped onto a specific core (a "floating" thread)
-
-Example: `-1,-1,-1`: the thread pool is made of 3 floating threads.
-Be careful with a fixed allocation: it is hard to be more clever than Linux kernel!
-
-`pool` is a pointer to the thread pool object.
-
-`performanceMeas` is a flag to enable measurements (see also below).
-
-`name` is used to build the thread names. 
-
-### `void pushTpool(tpool_t *t, notifiedFIFO_elt_t *msg)`
-
-Adds a job for processing in the thread pool.
-
-The job data you can set are, inside `msg`:
-
-* `key`: an arbitrary key to find a job in a response queue.
-* `reponseFifo`: if non-`NULL`, the message will be sent back on this queue when the job is done. If `NULL`, the thread pool automatically frees the job when it is done.
-* `processingFunc`: the function to execute for this job.
-
-The `processingFunc` prototype is `void <func>(void *memory)`. The data part of the job (the pointer returned by `NotifiedFifoData(msg)`) is passed to the function. The job data will be updated by `processingFunc`, and the job will be pushed into return queue (the parameter `reponseFifo`).
-
-### `notifiedFIFO_elt_t *pullTpool(notifiedFIFO_t *responseFifo, tpool_t *t)`
-
-Collects a job result in a return queue that has been defined in `msg` when calling `pushTpool()`, and that has been updated by `processingFunc()`. Returns the corresponding job pointer of type `notifiedFIFO_elt_t *`.
-
-Multiple return queues might be useful. Consider the following example in the eNB: I created one single thread pool (because it depends mainly on CPU hardware), but i use two return queues: one for turbo encode, one for turbo decode. Thus, when performing Turbo encoding/decoding, jobs are pushed into a single thread pool, but will end up in different result queues, such that DL/UL processing can be separated more easily.
-
-### `notifiedFIFO_elt_t *tryPullTpool(notifiedFIFO_t *responseFifo, tpool_t *t)`
-
-The same as `pullTpool()` in a non-blocking fashion (an alternative name would have been `pollTpool()`).
-
-### `int abortTpool(tpool_t *t)`
-
-Aborts the complete Tpool: cancel every work in the input queue, marks to drop existing jobs in processing, and terminates all worker threads. It is afterwards still possible to call functions such as `pushTpool()`, but each calling thread will execute the job itself.
-
-Returns the total number of jobs that were aborted, i.e., waiting for execution or being executed.
-
-## Performance measurements
-
-A performance measurement is integrated: the pool will automacillay fill timestamps if you set the environement variable `OAI_THREADPOOLMEASUREMENTS` to a valid file name.  The following measurements will be written to Linux pipe on a per-job basis:
-
-* `creationTime`: time the request is push to the pool;
-* `startProcessingTime`: time a worker start to run on the job
-* `endProcessingTime`: time the worker finished the job
-* `returnTime`: time the client reads the result
-
-The `measurement_display` tool to read the Linux pipe and display it in ASCII is provided.
-In the cmake build directory, type `make/ninja measurement_display`. Use as
-follows:
-```
-sudo OAI_THREADPOOLMEASUREMENTS=tpool.meas ./nr-softmodem ...
-./measurement_display tpool.meas
-```
+See header file doxygen comments for detailed description of the API

common/utils/time_meas.c

@@ -27,7 +27,7 @@
 #include "assertions.h"
 #include <pthread.h>
 #include "common/config/config_userapi.h"
-#include <common/utils/threadPool/thread-pool.h>
+#include "common/utils/threadPool/notified_fifo.h"
 // global var for openair performance profiler
 int cpu_meas_enabled = 0;
 double cpu_freq_GHz  __attribute__ ((aligned(32)));

executables/nr-softmodem-common.h

@@ -28,7 +28,6 @@
 #undef MALLOC
 #include "assertions.h"
 #include "PHY/types.h"
-#include <threadPool/thread-pool.h>
 
 /* help strings definition for command line options, used in CMDLINE_XXX_DESC macros and printed when -h option is used */
 #define CONFIG_HLP_RFCFGF        "Configuration file for front-end (e.g. LMS7002M)\n"

executables/nr-ue.c

@@ -40,6 +40,7 @@
 #include "openair1/PHY/TOOLS/phy_scope_interface.h"
 #include "PHY/MODULATION/nr_modulation.h"
 #include "instrumentation.h"
+#include "common/utils/threadPool/notified_fifo.h"
 
 /*
  *  NR SLOT PROCESSING SEQUENCE
@@ -101,8 +102,10 @@
 static void *NRUE_phy_stub_standalone_pnf_task(void *arg);
 
 static void start_process_slot_tx(void* arg) {
-  notifiedFIFO_elt_t *newTx = arg;
-  pushTpool(&(get_nrUE_params()->Tpool), newTx);
+  task_t task;
+  task.args = arg;
+  task.func = processSlotTX;
+  pushTpool(&(get_nrUE_params()->Tpool), task);
 }
 
 static size_t dump_L1_UE_meas_stats(PHY_VARS_NR_UE *ue, char *output, size_t max_len)
@@ -487,7 +490,7 @@ static void RU_write(nr_rxtx_thread_data_t *rxtxD, bool sl_tx_action)
 void processSlotTX(void *arg)
 {
   TracyCZone(ctx, true);
-  nr_rxtx_thread_data_t *rxtxD = (nr_rxtx_thread_data_t *) arg;
+  nr_rxtx_thread_data_t *rxtxD = arg;
   const UE_nr_rxtx_proc_t *proc = &rxtxD->proc;
   PHY_VARS_NR_UE *UE = rxtxD->UE;
   nr_phy_data_tx_t phy_data = {0};
@@ -550,6 +553,7 @@ void processSlotTX(void *arg)
   int next_slot = (proc->nr_slot_tx + 1) % slots_per_frame;
   dynamic_barrier_join(&UE->process_slot_tx_barriers[next_slot]);
   RU_write(rxtxD, sl_tx_action);
+  free(rxtxD);
   TracyCZoneEnd(ctx);
 }
 
@@ -823,7 +827,7 @@ void *UE_thread(void *arg)
 
   while (!oai_exit) {
     if (syncRunning) {
-      notifiedFIFO_elt_t *res=tryPullTpool(&nf,&(get_nrUE_params()->Tpool));
+      notifiedFIFO_elt_t *res = pollNotifiedFIFO(&nf);
 
       if (res) {
         syncRunning = false;
@@ -1003,7 +1007,7 @@ void *UE_thread(void *arg)
       nr_ue_rrc_timer_trigger(UE->Mod_id, curMsg.proc.frame_tx, curMsg.proc.gNB_id);
 
     // RX slot processing. We launch and forget.
-    notifiedFIFO_elt_t *newRx = newNotifiedFIFO_elt(sizeof(nr_rxtx_thread_data_t), curMsg.proc.nr_slot_rx, NULL, UE_dl_processing);
+    notifiedFIFO_elt_t *newRx = newNotifiedFIFO_elt(sizeof(nr_rxtx_thread_data_t), curMsg.proc.nr_slot_tx, NULL, UE_dl_processing);
     nr_rxtx_thread_data_t *curMsgRx = (nr_rxtx_thread_data_t *)NotifiedFifoData(newRx);
     *curMsgRx = (nr_rxtx_thread_data_t){.proc = curMsg.proc, .UE = UE};
     int ret = UE_dl_preprocessing(UE, &curMsgRx->proc, tx_wait_for_dlsch, &curMsgRx->phy_data, &stats_printed);
@@ -1013,8 +1017,7 @@ void *UE_thread(void *arg)
 
     // Start TX slot processing here. It runs in parallel with RX slot processing
     // in current code, DURATION_RX_TO_TX constant is the limit to get UL data to encode from a RX slot
-    notifiedFIFO_elt_t *newTx = newNotifiedFIFO_elt(sizeof(nr_rxtx_thread_data_t), curMsg.proc.nr_slot_tx, NULL, processSlotTX);
-    nr_rxtx_thread_data_t *curMsgTx = (nr_rxtx_thread_data_t *)NotifiedFifoData(newTx);
+    nr_rxtx_thread_data_t *curMsgTx = calloc(1, sizeof(*curMsgTx));
     curMsgTx->proc = curMsg.proc;
     curMsgTx->writeBlockSize = writeBlockSize;
     curMsgTx->proc.timestamp_tx = writeTimestamp;
@@ -1026,7 +1029,7 @@ void *UE_thread(void *arg)
     dynamic_barrier_update(&UE->process_slot_tx_barriers[slot],
                            tx_wait_for_dlsch[slot] + sync_to_previous_thread,
                            start_process_slot_tx,
-                           newTx);
+                           curMsgTx);
     stream_status = STREAM_STATUS_SYNCED;
     tx_wait_for_dlsch[slot] = 0;
   }

executables/nr-uesoftmodem.c

@@ -33,7 +33,6 @@
 #include "SCHED_NR_UE/defs.h"
 #include "common/ran_context.h"
 #include "common/config/config_userapi.h"
-//#include "common/utils/threadPool/thread-pool.h"
 #include "common/utils/load_module_shlib.h"
 //#undef FRAME_LENGTH_COMPLEX_SAMPLES //there are two conflicting definitions, so we better make sure we don't use it at all
 #include "common/utils/nr/nr_common.h"

executables/nr-uesoftmodem.h

@@ -2,6 +2,7 @@
 #define NR_UESOFTMODEM_H
 #include <executables/nr-softmodem-common.h>
 #include <executables/softmodem-common.h>
+#include "common/utils/threadPool/thread-pool.h"
 #include "PHY/defs_nr_UE.h"
 
 #define  CONFIG_HLP_IF_FREQ                "IF frequency for RF, if needed\n"
@@ -78,7 +79,7 @@ typedef struct {
   uint64_t optmask; // mask to store boolean config options
   uint32_t ofdm_offset_divisor; // Divisor for sample offset computation for each OFDM symbol
   int max_ldpc_iterations; // number of maximum LDPC iterations
-  tpool_t Tpool; // thread pool
+  tpool_t Tpool;
   int UE_scan_carrier;
   int UE_fo_compensation;
   uint64_t if_freq;

openair1/PHY/CODING/nrLDPC_defs.h

@@ -64,6 +64,7 @@ typedef struct {
   nrLDPC_params_per_cb_t perCB[NR_LDPC_MAX_NUM_CB];
   // Redundancy version index
   uint8_t rv;
+  task_ans_t *ans;
 } encoder_implemparams_t;
 
 typedef int32_t(LDPC_initfunc_t)(void);

openair1/PHY/LTE_TRANSPORT/dlsch_coding.c

@@ -42,7 +42,6 @@
 #include "common/utils/LOG/log.h"
 #include "executables/lte-softmodem.h"
 #include <syscall.h>
-#include <common/utils/threadPool/thread-pool.h>
 
 //#define DEBUG_DLSCH_CODING
 //#define DEBUG_DLSCH_FREE 1
@@ -284,6 +283,9 @@ static void TPencode(void * arg) {
 			  rdata->r,
 			  hadlsch->nb_rb);
     stop_meas(rdata->rm_stats);
+
+    // Task completed in parallel
+    completed_task_ans(rdata->ans);
 }
 
 int dlsch_encoding(PHY_VARS_eNB *eNB,
@@ -321,8 +323,6 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
 	    num_pdcch_symbols,
 	    frame,subframe,beamforming_mode);
 
-  int nbEncode = 0;
-
   //  if (hadlsch->Ndi == 1) {  // this is a new packet
   if (hadlsch->round == 0) {  // this is a new packet
     // Add 24-bit crc (polynomial A) to payload
@@ -349,13 +349,14 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
       return(-1);
   }
 
-  notifiedFIFO_t respEncode;
-  initNotifiedFIFO(&respEncode);
-  for (int r=0, r_offset=0; r<hadlsch->C; r++) {
-    
-    union turboReqUnion id= {.s={dlsch->rnti,frame,subframe,r,0}};
-    notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(turboEncode_t), id.p, &respEncode, TPencode);
-    turboEncode_t * rdata=(turboEncode_t *) NotifiedFifoData(req);
+  turboEncode_t arr[hadlsch->C];
+  task_ans_t ans[hadlsch->C];
+  memset(ans, 0, hadlsch->C * sizeof(task_ans_t));
+
+  for (int r = 0, r_offset = 0; r < hadlsch->C; r++) {
+    turboEncode_t *rdata = &arr[r];
+    rdata->ans = &ans[r];
+
     rdata->input=hadlsch->c[r];
     rdata->Kr_bytes= ( r<hadlsch->Cminus ? hadlsch->Kminus : hadlsch->Kplus) >>3;
     rdata->filler=(r==0) ? hadlsch->F : 0;
@@ -369,8 +370,8 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
     rdata->r_offset=r_offset;
     rdata->G=G;
 
-    pushTpool(proc->threadPool, req);
-    nbEncode++;
+    task_t t = {.func = TPencode, .args = rdata};
+    pushTpool(proc->threadPool, t);
 
     int Qm=hadlsch->Qm;
     int C=hadlsch->C;
@@ -382,14 +383,9 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
     else
       r_offset += Nl*Qm * ((GpmodC==0?0:1) + (Gp/C));
   }
-  // Wait all other threads finish to process
-  while (nbEncode) {
-    notifiedFIFO_elt_t *res = pullTpool(&respEncode, proc->threadPool);
-    if (res == NULL)
-      break; // Tpool has been stopped
-    delNotifiedFIFO_elt(res);
-    nbEncode--;
-  }
+
+  join_task_ans(ans, hadlsch->C);
+
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_ENB_DLSCH_ENCODING, VCD_FUNCTION_OUT);
   return(0);
 }
@@ -454,14 +450,14 @@ int dlsch_encoding_fembms_pmch(PHY_VARS_eNB *eNB,
                          &hadlsch->F)<0)
       return(-1);
   }
-  int nbEncode = 0;
-  notifiedFIFO_t respEncode;
-  initNotifiedFIFO(&respEncode);
-  for (int r=0, r_offset=0; r<hadlsch->C; r++) {
-    
-    union turboReqUnion id= {.s={dlsch->rnti,frame,subframe,r,0}};
-    notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(turboEncode_t), id.p, &respEncode, TPencode);
-    turboEncode_t * rdata=(turboEncode_t *) NotifiedFifoData(req);
+  turboEncode_t arr[hadlsch->C];
+  task_ans_t ans[hadlsch->C];
+  memset(ans, 0, hadlsch->C * sizeof(task_ans_t));
+
+  for (int r = 0, r_offset = 0; r < hadlsch->C; r++) {
+    turboEncode_t *rdata = &arr[r];
+    rdata->ans = &ans[r];
+
     rdata->input=hadlsch->c[r];
     rdata->Kr_bytes= ( r<hadlsch->Cminus ? hadlsch->Kminus : hadlsch->Kplus) >>3;
     rdata->filler=(r==0) ? hadlsch->F : 0;
@@ -475,8 +471,8 @@ int dlsch_encoding_fembms_pmch(PHY_VARS_eNB *eNB,
     rdata->r_offset=r_offset;
     rdata->G=G;
 
-    pushTpool(proc->threadPool, req);
-    nbEncode++;
+    task_t t = {.func = TPencode, .args = rdata};
+    pushTpool(proc->threadPool, t);
 
     int Qm=hadlsch->Qm;
     int C=hadlsch->C;
@@ -488,14 +484,9 @@ int dlsch_encoding_fembms_pmch(PHY_VARS_eNB *eNB,
     else
       r_offset += Nl*Qm * ((GpmodC==0?0:1) + (Gp/C));
   }
-  // Wait all other threads finish to process
-  while (nbEncode) {
-    notifiedFIFO_elt_t *res = pullTpool(&respEncode, proc->threadPool);
-    if (res == NULL)
-      break; // Tpool has been stopped
-    delNotifiedFIFO_elt(res);
-    nbEncode--;
-  }
+
+  join_task_ans(ans, hadlsch->C);
+
   return(0);
 }
 

openair1/PHY/LTE_TRANSPORT/transport_proto.h

@@ -36,6 +36,7 @@
 #include "nfapi_interface.h"
 #include "transport_common_proto.h"
 
+
 // Functions below implement 36-211 and 36-212
 
 /** @addtogroup _PHY_TRANSPORT_
@@ -520,12 +521,13 @@ void rx_ulsch(PHY_VARS_eNB *eNB,
   @param llr8_flag If 1, indicate that the 8-bit turbo decoder should be used
   @returns 0 on success
 */
-unsigned int  ulsch_decoding(PHY_VARS_eNB *phy_vars_eNB,
-                             L1_rxtx_proc_t *proc,
-                             uint8_t UE_id,
-                             uint8_t control_only_flag,
-                             uint8_t Nbundled,
-                             uint8_t llr8_flag);
+unsigned int ulsch_decoding(PHY_VARS_eNB *phy_vars_eNB,
+                            L1_rxtx_proc_t *proc,
+                            uint8_t UE_id,
+                            uint8_t control_only_flag,
+                            uint8_t Nbundled,
+                            uint8_t llr8_flag,
+                            thread_info_tm_t *t_info);
 
 void generate_phich_top(PHY_VARS_eNB *phy_vars_eNB,
                         L1_rxtx_proc_t *proc,

openair1/PHY/LTE_TRANSPORT/ulsch_decoding.c

@@ -239,8 +239,10 @@ void processULSegment(void * arg) {
                                    1,
                                    r,
                                    &E)==-1) {
-      LOG_E(PHY,"ulsch_decoding.c: Problem in rate matching\n");
-      return;
+    // Task completed in parallel
+    completed_task_ans(rdata->ans);
+    LOG_E(PHY, "ulsch_decoding.c: Problem in rate matching\n");
+    return;
    }
    stop_meas(&eNB->ulsch_rate_unmatching_stats);
    int max_Ncb = 3*ulsch_harq->RTC[r]*32 ;
@@ -284,6 +286,9 @@ void processULSegment(void * arg) {
                                             &eNB->ulsch_tc_intl1_stats,
                                             &eNB->ulsch_tc_intl2_stats,
                                             &ulsch_harq->abort_decode);
+
+  // Task completed in parallel
+  completed_task_ans(rdata->ans);
 }
 
 /*!
@@ -295,7 +300,12 @@ void processULSegment(void * arg) {
   @returns 0 on success
 */
 
-static int ulsch_decoding_data(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int UE_id, int harq_pid, int llr8_flag)
+static int ulsch_decoding_data(PHY_VARS_eNB *eNB,
+                               L1_rxtx_proc_t *proc,
+                               int UE_id,
+                               int harq_pid,
+                               int llr8_flag,
+                               thread_info_tm_t *t_info)
 {
   unsigned int r_offset=0;
   int offset = 0;
@@ -330,13 +340,11 @@ static int ulsch_decoding_data(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int UE_i
       E = ulsch_harq->Qm * (Gp/ulsch_harq->C);
     else
       E = ulsch_harq->Qm * ((GpmodC==0?0:1) + (Gp/ulsch_harq->C));
-    
-    union turboReqUnion id= {.s={ulsch->rnti,proc->frame_rx,proc->subframe_rx,0,0}};
-    notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(turboDecode_t),
-                                                id.p,
-                                                proc->respDecode,
-                                                processULSegment);
-    turboDecode_t * rdata=(turboDecode_t *) NotifiedFifoData(req);
+
+    turboDecode_t *rdata = &((turboDecode_t *)t_info->buf)[t_info->len];
+    DevAssert(t_info->len < t_info->cap);
+    rdata->ans = &t_info->ans[t_info->len];
+    t_info->len += 1;
 
     rdata->eNB=eNB;
     rdata->frame=proc->frame_rx;
@@ -355,7 +363,10 @@ static int ulsch_decoding_data(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int UE_i
     rdata->function=td;
     int Fbytes=(r==0) ? rdata->Fbits>>3 : 0;
     int sz=Kr_bytes - Fbytes - ((ulsch_harq->C>1)?3:0);
-    pushTpool(proc->threadPool,req);
+
+    task_t t = {.func = &processULSegment, .args = rdata};
+    pushTpool(proc->threadPool, t);
+
     proc->nbDecode++;
     LOG_D(PHY,"Added a block to decode, in pipe: %d\n",proc->nbDecode);
     r_offset+=E;
@@ -390,12 +401,13 @@ static inline unsigned int lte_gold_unscram(unsigned int *x1, unsigned int *x2,
   //  printf("n=%d : c %x\n",n,x1^x2);
 }
 
-unsigned int  ulsch_decoding(PHY_VARS_eNB *eNB,
-                             L1_rxtx_proc_t *proc,
-                             uint8_t UE_id,
-                             uint8_t control_only_flag,
-                             uint8_t Nbundled,
-                             uint8_t llr8_flag)
+unsigned int ulsch_decoding(PHY_VARS_eNB *eNB,
+                            L1_rxtx_proc_t *proc,
+                            uint8_t UE_id,
+                            uint8_t control_only_flag,
+                            uint8_t Nbundled,
+                            uint8_t llr8_flag,
+                            thread_info_tm_t *t_info)
 {
   int16_t *ulsch_llr = eNB->pusch_vars[UE_id]->llr;
   LTE_DL_FRAME_PARMS *frame_parms = &eNB->frame_parms;
@@ -1082,7 +1094,7 @@ unsigned int  ulsch_decoding(PHY_VARS_eNB *eNB,
 
   LOG_D(PHY,"frame %d subframe %d O_ACK:%d o_ACK[]=%d:%d:%d:%d\n",frame,subframe,ulsch_harq->O_ACK,ulsch_harq->o_ACK[0],ulsch_harq->o_ACK[1],ulsch_harq->o_ACK[2],ulsch_harq->o_ACK[3]);
   // Do ULSCH Decoding for data portion
-  ret = ulsch_decoding_data(eNB, proc, UE_id, harq_pid, llr8_flag);
+  ret = ulsch_decoding_data(eNB, proc, UE_id, harq_pid, llr8_flag, t_info);
   return(ret);
 }
 

openair1/PHY/NR_ESTIMATION/nr_ul_channel_estimation.c

@@ -409,6 +409,7 @@ static void nr_pusch_antenna_processing(void *arg)
     *(rdata->noise_amp2) = noise_amp2;
     *(rdata->nest_count) = nest_count;
   }
+  completed_task_ans(rdata->ans);
 }
 
 
@@ -503,14 +504,13 @@ int nr_pusch_channel_estimation(PHY_VARS_gNB *gNB,
   start_meas(&gNB->pusch_channel_estimation_antenna_processing_stats);
   int numAntennas = gNB->dmrs_num_antennas_per_thread;
   int num_jobs = CEILIDIV(gNB->frame_parms.nb_antennas_rx, numAntennas);
+  puschAntennaProc_t rdatas[num_jobs];
+  memset(rdatas, 0, sizeof(rdatas));
+  task_ans_t ans[num_jobs];
+  memset(ans, 0, sizeof(ans));
   for (int job_id = 0; job_id < num_jobs; job_id++) {
-    union puschAntennaReqUnion id = {.s = {ul_id, 0}};
-    id.p = 1 + job_id * numAntennas;
-    notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(puschAntennaProc_t),
-                                                  id.p,
-                                                  &respPuschAarx,
-                                                  &nr_pusch_antenna_processing); // create a job for Tpool
-    puschAntennaProc_t *rdata = (puschAntennaProc_t *)NotifiedFifoData(req); // data for the job
+    puschAntennaProc_t *rdata = &rdatas[job_id];
+    task_t task = {.func = nr_pusch_antenna_processing, .args = rdata};
 
     // Local init in the current loop
     rdata->Ns = Ns;
@@ -531,22 +531,18 @@ int nr_pusch_channel_estimation(PHY_VARS_gNB *gNB,
     rdata->pusch_vars = &gNB->pusch_vars[ul_id];
     rdata->chest_freq = gNB->chest_freq;
     rdata->rxdataF = gNB->common_vars.rxdataF;
+    rdata->ans = &ans[job_id];
     // Call the nr_pusch_antenna_processing function
     if (job_id == num_jobs - 1) {
       // Run the last job inline
       nr_pusch_antenna_processing(rdata);
-      delNotifiedFIFO_elt(req);
     } else {
-      pushTpool(&gNB->threadPool, req);
+      pushTpool(&gNB->threadPool, task);
     }
     LOG_D(PHY, "Added Antenna (count %d/%d) to process, in pipe\n", job_id, num_jobs);
   } // Antenna Loop
 
-  while (num_jobs - 1 > 0) {
-    notifiedFIFO_elt_t *req = pullTpool(&respPuschAarx, &gNB->threadPool);
-    num_jobs--;
-    delNotifiedFIFO_elt(req);
-  }
+  join_task_ans(ans, num_jobs - 1);
 
   stop_meas(&gNB->pusch_channel_estimation_antenna_processing_stats);
   for (int aarx = 0; aarx < gNB->frame_parms.nb_antennas_rx; aarx++) {

openair1/PHY/NR_TRANSPORT/nr_dlsch_coding.c

@@ -117,7 +117,7 @@ NR_gNB_DLSCH_t new_gNB_dlsch(NR_DL_FRAME_PARMS *frame_parms, uint16_t N_RB)
   return(dlsch);
 }
 
-void ldpc8blocks(void *p)
+static void ldpc8blocks(void *p)
 {
   encoder_implemparams_t *impp=(encoder_implemparams_t *) p;
   NR_DL_gNB_HARQ_t *harq = (NR_DL_gNB_HARQ_t *)impp->harq;
@@ -232,6 +232,9 @@ void ldpc8blocks(void *p)
 #endif
     r_offset += E;
   }
+
+  // Task running in // completed
+  completed_task_ans(impp->ans);
 }
 
 int nr_dlsch_encoding(PHY_VARS_gNB *gNB,
@@ -369,24 +372,23 @@ int nr_dlsch_encoding(PHY_VARS_gNB *gNB,
     }
     ldpc_interface_offload.LDPCencoder(harq->c, &impp.output, &impp);
   } else {
-    notifiedFIFO_t nf;
-    initNotifiedFIFO(&nf);
-    int nbJobs = 0;
-    for (int j = 0; j < (impp.n_segments / 8 + ((impp.n_segments & 7) == 0 ? 0 : 1)); j++) {
-      notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(impp), j, &nf, ldpc8blocks);
-      encoder_implemparams_t *perJobImpp = (encoder_implemparams_t *)NotifiedFifoData(req);
+    size_t const n_seg = (impp.n_segments / 8 + ((impp.n_segments & 7) == 0 ? 0 : 1));
+
+    encoder_implemparams_t arr[n_seg];
+    task_ans_t ans[n_seg];
+    memset(ans, 0, n_seg * sizeof(task_ans_t));
+
+    for (int j = 0; j < n_seg; j++) {
+      encoder_implemparams_t *perJobImpp = &arr[j];
       *perJobImpp = impp;
       perJobImpp->macro_num = j;
-      pushTpool(&gNB->threadPool, req);
-      nbJobs++;
-    }
-    while (nbJobs) {
-      notifiedFIFO_elt_t *req = pullTpool(&nf, &gNB->threadPool);
-      if (req == NULL)
-        break; // Tpool has been stopped
-      delNotifiedFIFO_elt(req);
-      nbJobs--;
+      perJobImpp->ans = &ans[j];
+
+      task_t t = {.func = ldpc8blocks, .args = perJobImpp};
+      pushTpool(&gNB->threadPool, t);
     }
+
+    join_task_ans(ans, n_seg);
   }
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_gNB_DLSCH_ENCODING, VCD_FUNCTION_OUT);
   return 0;

openair1/PHY/NR_TRANSPORT/nr_ulsch.h

@@ -62,7 +62,8 @@ int nr_ulsch_decoding(PHY_VARS_gNB *phy_vars_gNB,
                       uint32_t frame,
                       uint8_t nr_tti_rx,
                       uint8_t harq_pid,
-                      uint32_t G);
+                      uint32_t G,
+                      thread_info_tm_t *t_info);
 
 /*! \brief Perform PUSCH unscrambling. TS 38.211 V15.4.0 subclause 6.3.1.1
   @param llr, Pointer to llr bits

openair1/PHY/NR_TRANSPORT/nr_ulsch_decoding.c

@@ -50,6 +50,10 @@
 //#define DEBUG_ULSCH_DECODING
 //#define gNB_DEBUG_TRACE
 
+#include <stdint.h>
+#include <time.h>
+#include <stdalign.h>
+
 #define OAI_UL_LDPC_MAX_NUM_LLR 27000//26112 // NR_LDPC_NCOL_BG1*NR_LDPC_ZMAX = 68*384
 //#define DEBUG_CRC
 #ifdef DEBUG_CRC
@@ -181,6 +185,9 @@ static void nr_processULSegment(void *arg)
     LOG_E(PHY, "ulsch_decoding.c: Problem in rate_matching\n");
     rdata->decodeIterations = max_ldpc_iterations + 1;
     set_abort(&ulsch_harq->abort_decode, true);
+
+    // Task completed
+    completed_task_ans(rdata->ans);
     return;
   }
 
@@ -220,6 +227,9 @@ static void nr_processULSegment(void *arg)
 
   if (rdata->decodeIterations <= p_decoderParms->numMaxIter)
     memcpy(ulsch_harq->c[r],llrProcBuf,  Kr>>3);
+
+  // Task completed
+  completed_task_ans(rdata->ans);
 }
 
 int decode_offload(PHY_VARS_gNB *phy_vars_gNB,
@@ -320,7 +330,8 @@ int nr_ulsch_decoding(PHY_VARS_gNB *phy_vars_gNB,
                       uint32_t frame,
                       uint8_t nr_tti_rx,
                       uint8_t harq_pid,
-                      uint32_t G)
+                      uint32_t G,
+                      thread_info_tm_t *t_info)
 {
   if (!ulsch_llr) {
     LOG_E(PHY, "ulsch_decoding.c: NULL ulsch_llr pointer\n");
@@ -430,9 +441,12 @@ int nr_ulsch_decoding(PHY_VARS_gNB *phy_vars_gNB,
   set_abort(&harq_process->abort_decode, false);
   for (int r = 0; r < harq_process->C; r++) {
     int E = nr_get_E(G, harq_process->C, Qm, n_layers, r);
-    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);
+
+    ldpcDecode_t *rdata = &((ldpcDecode_t *)t_info->buf)[t_info->len];
+    DevAssert(t_info->len < t_info->cap);
+    rdata->ans = &t_info->ans[t_info->len];
+    t_info->len += 1;
+
     decParams.R = nr_get_R_ldpc_decoder(pusch_pdu->pusch_data.rv_index,
                                         E,
                                         decParams.BG,
@@ -457,7 +471,10 @@ int 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;
-    pushTpool(&phy_vars_gNB->threadPool, req);
+
+    task_t t = {.func = &nr_processULSegment, .args = rdata};
+    pushTpool(&phy_vars_gNB->threadPool, t);
+
     LOG_D(PHY, "Added a block to decode, in pipe: %d\n", r);
     r_offset += E;
     offset += ((harq_process->K >> 3) - (harq_process->F >> 3) - ((harq_process->C > 1) ? 3 : 0));

openair1/PHY/NR_TRANSPORT/nr_ulsch_demodulation.c

@@ -1179,6 +1179,7 @@ typedef struct puschSymbolProc_s {
   int16_t *scramblingSequence;
   uint32_t nvar;
   int beam_nb;
+  task_ans_t *ans;
 } puschSymbolProc_t;
 
 static void nr_pusch_symbol_processing(void *arg)
@@ -1227,6 +1228,9 @@ static void nr_pusch_symbol_processing(void *arg)
     for (int i = 0; i < end; i++)
       llr16[i] = llr_ptr[i] * s[i];
   }
+
+  // Task running in // completed
+  completed_task_ans(rdata->ans);
 }
 
 static uint32_t average_u32(const uint32_t *x, uint16_t size)
@@ -1483,8 +1487,13 @@ int nr_rx_pusch_tp(PHY_VARS_gNB *gNB,
 
   start_meas(&gNB->rx_pusch_symbol_processing_stats);
   int numSymbols = gNB->num_pusch_symbols_per_thread;
-
   int total_res = 0;
+  int const loop_iter = rel15_ul->nr_of_symbols / numSymbols;
+  puschSymbolProc_t arr[loop_iter];
+  task_ans_t arr_ans[loop_iter];
+
+  memset(arr_ans, 0, loop_iter * sizeof(task_ans_t));
+  int sz_arr = 0;
   for(uint8_t symbol = rel15_ul->start_symbol_index; symbol < end_symbol; symbol += numSymbols) {
     int res_per_task = 0;
     for (int s = 0; s < numSymbols; s++) { 
@@ -1496,10 +1505,9 @@ int nr_rx_pusch_tp(PHY_VARS_gNB *gNB,
     }
     total_res += res_per_task;
     if (res_per_task > 0) {
-      union puschSymbolReqUnion id = {.s={ulsch_id,frame,slot,0}};
-      id.p=1+symbol;
-      notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(puschSymbolProc_t), id.p, &gNB->respPuschSymb, &nr_pusch_symbol_processing); // create a job for Tpool
-      puschSymbolProc_t *rdata = (puschSymbolProc_t*)NotifiedFifoData(req); // data for the job
+      puschSymbolProc_t *rdata = &arr[sz_arr];
+      rdata->ans = &arr_ans[sz_arr];
+      ++sz_arr;
 
       rdata->gNB = gNB;
       rdata->frame_parms = frame_parms;
@@ -1517,7 +1525,8 @@ int nr_rx_pusch_tp(PHY_VARS_gNB *gNB,
       if (rel15_ul->pdu_bit_map & PUSCH_PDU_BITMAP_PUSCH_PTRS) {
         nr_pusch_symbol_processing(rdata);
       } else {
-        pushTpool(&gNB->threadPool, req);
+        task_t t = {.func = &nr_pusch_symbol_processing, .args = rdata};
+        pushTpool(&gNB->threadPool, t);
         nbSymb++;
       }
 
@@ -1525,12 +1534,9 @@ int nr_rx_pusch_tp(PHY_VARS_gNB *gNB,
     }
   } // symbol loop
 
-  while (nbSymb) {
-    notifiedFIFO_elt_t *req = pullTpool(&gNB->respPuschSymb, &gNB->threadPool);
-    nbSymb--;
-    delNotifiedFIFO_elt(req);
+  if (nbSymb > 0) {
+    join_task_ans(arr_ans, sz_arr);
   }
-
   stop_meas(&gNB->rx_pusch_symbol_processing_stats);
 
   // Copy the data to the scope. This cannot be performed in one call to gNBscopeCopy because the data is not contiguous in the

openair1/PHY/NR_UE_TRANSPORT/nr_dlsch_decoding.c

@@ -70,15 +70,13 @@ void nr_dlsch_unscrambling(int16_t *llr, uint32_t size, uint8_t q, uint32_t Nid,
 }
 
 static bool nr_ue_postDecode(PHY_VARS_NR_UE *phy_vars_ue,
-                             notifiedFIFO_elt_t *req,
-                             notifiedFIFO_t *nf_p,
-                             const bool last,
+                             ldpcDecode_ue_t *rdata,
+                             bool last,
                              int b_size,
                              uint8_t b[b_size],
                              int *num_seg_ok,
                              const UE_nr_rxtx_proc_t *proc)
 {
-  ldpcDecode_ue_t *rdata = (ldpcDecode_ue_t*) NotifiedFifoData(req);
   NR_DL_UE_HARQ_t *harq_process = rdata->harq_process;
   NR_UE_DLSCH_t *dlsch = (NR_UE_DLSCH_t *) rdata->dlsch;
   int r = rdata->segment_r;
@@ -257,6 +255,7 @@ static void nr_processDLSegment(void *arg)
       memcpy(harq_process->c[r], LDPCoutput, Kr >> 3);
     stop_meas(&rdata->ts_ldpc_decode);
   }
+  completed_task_ans(rdata->ans);
 }
 
 uint32_t nr_dlsch_decoding(PHY_VARS_NR_UE *phy_vars_ue,
@@ -369,16 +368,17 @@ uint32_t nr_dlsch_decoding(PHY_VARS_NR_UE *phy_vars_ue,
   Kr = harq_process->K;
   Kr_bytes = Kr>>3;
   offset = 0;
-  notifiedFIFO_t nf;
-  initNotifiedFIFO(&nf);
+
+  ldpcDecode_ue_t arr[harq_process->C];
+  task_ans_t ans[harq_process->C];
+  memset(ans, 0, harq_process->C * sizeof(task_ans_t));
   set_abort(&harq_process->abort_decode, false);
   for (r=0; r<harq_process->C; r++) {
     //printf("start rx segment %d\n",r);
     uint32_t E = nr_get_E(G, harq_process->C, dlsch->dlsch_config.qamModOrder, dlsch->Nl, r);
     decParams.R = nr_get_R_ldpc_decoder(dlsch->dlsch_config.rv, E, decParams.BG, decParams.Z, &harq_process->llrLen, harq_process->DLround);
-    union ldpcReqUnion id = {.s = {dlsch->rnti, frame, nr_slot_rx, 0, 0}};
-    notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(ldpcDecode_ue_t), id.p, &nf, &nr_processDLSegment);
-    ldpcDecode_ue_t * rdata=(ldpcDecode_ue_t *) NotifiedFifoData(req);
+    ldpcDecode_ue_t *rdata = &arr[r];
+    rdata->ans = &ans[r];
 
     rdata->phy_vars_ue = phy_vars_ue;
     rdata->harq_process = harq_process;
@@ -398,7 +398,8 @@ uint32_t nr_dlsch_decoding(PHY_VARS_NR_UE *phy_vars_ue,
     reset_meas(&rdata->ts_deinterleave);
     reset_meas(&rdata->ts_rate_unmatch);
     reset_meas(&rdata->ts_ldpc_decode);
-    pushTpool(&get_nrUE_params()->Tpool,req);
+    task_t t = {.args = rdata, .func = nr_processDLSegment};
+    pushTpool(&get_nrUE_params()->Tpool, t);
     LOG_D(PHY, "Added a block to decode, in pipe: %d\n", r);
     r_offset += E;
     offset += (Kr_bytes - (harq_process->F>>3) - ((harq_process->C>1)?3:0));
@@ -406,13 +407,11 @@ uint32_t nr_dlsch_decoding(PHY_VARS_NR_UE *phy_vars_ue,
   }
   int num_seg_ok = 0;
   int nbDecode = harq_process->C;
-  while (nbDecode) {
-    notifiedFIFO_elt_t *req=pullTpool(&nf,  &get_nrUE_params()->Tpool);
-    if (req == NULL)
-      break; // Tpool has been stopped
-    nr_ue_postDecode(phy_vars_ue, req, &nf, nbDecode == 1, b_size, b, &num_seg_ok, proc);
-    delNotifiedFIFO_elt(req);
-    nbDecode--;
+  if (nbDecode > 0) {
+    join_task_ans(ans, nbDecode);
+    for (size_t i = 0; i < nbDecode; ++i) {
+      nr_ue_postDecode(phy_vars_ue, &arr[i], i == nbDecode - 1, b_size, b, &num_seg_ok, proc);
+    }
   }
   LOG_D(PHY,
         "%d.%d DLSCH Decoded, harq_pid %d, round %d, result: %d TBS %d (%d) G %d nb_re_dmrs %d length dmrs %d mcs %d Nl %d "

openair1/PHY/NR_UE_TRANSPORT/nr_initial_sync.c

@@ -288,6 +288,8 @@ void nr_scan_ssb(void *arg)
       }
     }
   }
+
+  completed_task_ans(ssbInfo->ans);
 }
 
 nr_initial_sync_t nr_initial_sync(UE_nr_rxtx_proc_t *proc,
@@ -299,18 +301,18 @@ nr_initial_sync_t nr_initial_sync(UE_nr_rxtx_proc_t *proc,
 {
   NR_DL_FRAME_PARMS *fp = &ue->frame_parms;
 
-  notifiedFIFO_t nf;
-  initNotifiedFIFO(&nf);
-
   // Perform SSB scanning in parallel. One GSCN per thread.
   LOG_I(NR_PHY,
         "Starting cell search with center freq: %ld, bandwidth: %d. Scanning for %d number of GSCN.\n",
         fp->dl_CarrierFreq,
         fp->N_RB_DL,
         numGscn);
+
+  task_ans_t ans[numGscn];
+  memset(ans, 0, sizeof(ans));
+  nr_ue_ssb_scan_t ssb_info[numGscn];
   for (int s = 0; s < numGscn; s++) {
-    notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(nr_ue_ssb_scan_t), gscnInfo[s].gscn, &nf, &nr_scan_ssb);
-    nr_ue_ssb_scan_t *ssbInfo = (nr_ue_ssb_scan_t *)NotifiedFifoData(req);
+    nr_ue_ssb_scan_t *ssbInfo = &ssb_info[s];
     *ssbInfo = (nr_ue_ssb_scan_t){.gscnInfo = gscnInfo[s],
                                   .fp = &ue->frame_parms,
                                   .proc = proc,
@@ -329,32 +331,34 @@ nr_initial_sync_t nr_initial_sync(UE_nr_rxtx_proc_t *proc,
           ssbInfo->gscnInfo.gscn,
           ssbInfo->gscnInfo.ssbFirstSC,
           ssbInfo->gscnInfo.ssRef);
-    pushTpool(&get_nrUE_params()->Tpool, req);
+    ssbInfo->ans = &ans[s];
+    task_t t = {.func = nr_scan_ssb, .args = ssbInfo};
+    pushTpool(&get_nrUE_params()->Tpool, t);
   }
 
   // Collect the scan results
   nr_ue_ssb_scan_t res = {0};
-  while (numGscn) {
-    notifiedFIFO_elt_t *req = pullTpool(&nf, &get_nrUE_params()->Tpool);
-    nr_ue_ssb_scan_t *ssbInfo = (nr_ue_ssb_scan_t *)NotifiedFifoData(req);
-    if (ssbInfo->syncRes.cell_detected) {
-      LOG_A(NR_PHY,
-            "Cell Detected with GSCN: %d, SSB SC offset: %d, SSB Ref: %lf, PSS Corr peak: %d dB, PSS Corr Average: %d\n",
-            ssbInfo->gscnInfo.gscn,
-            ssbInfo->gscnInfo.ssbFirstSC,
-            ssbInfo->gscnInfo.ssRef,
-            ssbInfo->pssCorrPeakPower,
-            ssbInfo->pssCorrAvgPower);
-      if (!res.syncRes.cell_detected) { // take the first cell detected
-        res = *ssbInfo;
+  if (numGscn > 0) {
+    join_task_ans(ans, numGscn);
+    for (int i = 0; i < numGscn; i++) {
+      nr_ue_ssb_scan_t *ssbInfo = &ssb_info[i];
+      if (ssbInfo->syncRes.cell_detected) {
+        LOG_I(NR_PHY,
+              "Cell Detected with GSCN: %d, SSB SC offset: %d, SSB Ref: %lf, PSS Corr peak: %d dB, PSS Corr Average: %d\n",
+              ssbInfo->gscnInfo.gscn,
+              ssbInfo->gscnInfo.ssbFirstSC,
+              ssbInfo->gscnInfo.ssRef,
+              ssbInfo->pssCorrPeakPower,
+              ssbInfo->pssCorrAvgPower);
+        if (!res.syncRes.cell_detected) { // take the first cell detected
+          res = *ssbInfo;
+        }
       }
+      for (int ant = 0; ant < fp->nb_antennas_rx; ant++) {
+        free(ssbInfo->rxdata[ant]);
+      }
+      free(ssbInfo->rxdata);
     }
-    for (int ant = 0; ant < fp->nb_antennas_rx; ant++) {
-      free(ssbInfo->rxdata[ant]);
-    }
-    free(ssbInfo->rxdata);
-    delNotifiedFIFO_elt(req);
-    numGscn--;
   }
 
   // Set globals based on detected cell

openair1/PHY/defs_RU.h

@@ -39,8 +39,9 @@
 #include "time_meas.h"
 #include "defs_common.h"
 #include "nfapi_nr_interface_scf.h"
-#include <common/utils/threadPool/thread-pool.h>
 #include <executables/rt_profiling.h>
+#include "common/utils/threadPool/task_ans.h"
+#include "common/utils/threadPool/thread-pool.h"
 
 #define MAX_BANDS_PER_RRU 4
 #define MAX_RRU_CONFIG_SIZE 1024
@@ -175,7 +176,8 @@ typedef struct {
  struct RU_t_s *ru;
  int startSymbol;
  int endSymbol;
- int slot; 
+ int slot;
+ task_ans_t *ans;
 } feprx_cmd_t;
 
 typedef struct {
@@ -184,6 +186,7 @@ typedef struct {
  int slot; 
  int startSymbol;
  int numSymbols;
+ task_ans_t *ans;
 } feptx_cmd_t;
 
 typedef struct {
@@ -426,7 +429,7 @@ typedef enum {
 
 
 typedef struct RU_t_s {
-  /// ThreadPool for RU	
+  /// ThreadPool for RU        
   tpool_t *threadPool;
   /// index of this ru
   uint32_t idx;

openair1/PHY/defs_eNB.h

@@ -66,7 +66,7 @@
 #include "PHY/LTE_TRANSPORT/transport_eNB.h"
 #include "openair2/PHY_INTERFACE/IF_Module.h"
 #include "common/openairinterface5g_limits.h"
-
+#include "common/utils/threadPool/task_ans.h"
 
 #define PBCH_A 24
 #define MAX_NUM_RU_PER_eNB 64
@@ -786,6 +786,7 @@ typedef struct TurboDecode_s {
     int offset;
     int maxIterations;
     int decodeIterations;
+    task_ans_t *ans;
 } turboDecode_t;
 
 #define TURBO_SIMD_SOFTBITS   96+12+3+3*6144
@@ -802,6 +803,7 @@ typedef struct turboEncode_s {
   time_stats_t *rm_stats;
   time_stats_t *te_stats;
   time_stats_t *i_stats;
+  task_ans_t *ans;
 } turboEncode_t;
 
 

openair1/PHY/defs_gNB.h

@@ -575,7 +575,7 @@ typedef struct PHY_VARS_gNB_s {
   struct processingData_L1tx *msgDataTx;
   void *scopeData;
   /// structure for analyzing high-level RT measurements
-  rt_L1_profiling_t rt_L1_profiling; 
+  rt_L1_profiling_t rt_L1_profiling;
 } PHY_VARS_gNB;
 
 struct puschSymbolReqId {
@@ -609,6 +609,7 @@ typedef struct puschAntennaProc_s {
   NR_gNB_PUSCH *pusch_vars;
   NR_DL_FRAME_PARMS *frame_parms;
   c16_t ***rxdataF;
+  task_ans_t* ans;
 } puschAntennaProc_t;
 
 struct puschAntennaReqId {
@@ -641,6 +642,7 @@ typedef struct LDPCDecode_s {
   int offset;
   int decodeIterations;
   uint32_t tbslbrm;
+  task_ans_t *ans;
 } ldpcDecode_t;
 
 struct ldpcReqId {
@@ -661,6 +663,7 @@ typedef struct processingData_L1 {
   int slot_rx;
   openair0_timestamp timestamp_tx;
   PHY_VARS_gNB *gNB;
+  notifiedFIFO_elt_t *elt;
 } processingData_L1_t;
 
 typedef struct processingData_L1tx {

openair1/PHY/defs_nr_UE.h

@@ -576,6 +576,7 @@ typedef struct {
   int pssCorrPeakPower;
   int pssCorrAvgPower;
   int adjust_rxgain;
+  task_ans_t *ans;
 } nr_ue_ssb_scan_t;
 
 typedef struct nr_phy_data_tx_s {
@@ -632,6 +633,7 @@ typedef struct LDPCDecode_ue_s {
   time_stats_t ts_deinterleave;
   time_stats_t ts_rate_unmatch;
   time_stats_t ts_ldpc_decode;
+  task_ans_t *ans;
 } ldpcDecode_ue_t;
 
 static inline void start_meas_nr_ue_phy(PHY_VARS_NR_UE *ue, int meas_index) {

openair1/SCHED/phy_procedures_lte_eNb.c

@@ -1232,10 +1232,8 @@ uci_procedures(PHY_VARS_eNB *eNB,
   } // end loop for (int i = 0; i < NUMBER_OF_UCI_MAX; i++) {
 }
 
-void postDecode(L1_rxtx_proc_t *proc, notifiedFIFO_elt_t *req)
+void postDecode(L1_rxtx_proc_t *proc, turboDecode_t *rdata)
 {
-  turboDecode_t * rdata=(turboDecode_t *) NotifiedFifoData(req);
-
   LTE_eNB_ULSCH_t *ulsch = rdata->eNB->ulsch[rdata->UEid];
   LTE_UL_eNB_HARQ_t *ulsch_harq = rdata->ulsch_harq;
   PHY_VARS_eNB *eNB=rdata->eNB;
@@ -1330,6 +1328,10 @@ void pusch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc) {
   const int frame    = proc->frame_rx;
   uint32_t harq_pid0 = subframe2harq_pid(&eNB->frame_parms,frame,subframe);
 
+  turboDecode_t arr[64] = {0};
+  task_ans_t ans[64] = {0};
+  thread_info_tm_t t_info = {.ans = ans, .cap = 64, .len = 0, .buf = (uint8_t *)arr};
+
   for (i = 0; i < NUMBER_OF_ULSCH_MAX; i++) {
     ulsch = eNB->ulsch[i];
     if (!ulsch) continue; 
@@ -1398,7 +1400,8 @@ void pusch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc) {
                      i,
                      0, // control_only_flag
                      ulsch_harq->V_UL_DAI,
-                     ulsch_harq->nb_rb > 20 ? 1 : 0);
+                     ulsch_harq->nb_rb > 20 ? 1 : 0,
+                     &t_info);
     }
     else if ((ulsch) &&
              (ulsch->rnti>0) &&
@@ -1415,14 +1418,12 @@ void pusch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc) {
   }   //   for (i=0; i<NUMBER_OF_ULSCH_MAX; i++)
 
   const bool decode = proc->nbDecode;
-  while (proc->nbDecode > 0) {
-    notifiedFIFO_elt_t *req=pullTpool(proc->respDecode, proc->threadPool);
-    if (req == NULL)
-      break; // Tpool has been stopped
-    postDecode(proc, req);
-    const time_stats_t ts = exec_time_stats_NotifiedFIFO(req);
-    merge_meas(&eNB->ulsch_turbo_decoding_stats, &ts);
-    delNotifiedFIFO_elt(req);
+  DevAssert(t_info.len == proc->nbDecode);
+  if (proc->nbDecode > 0) {
+    join_task_ans(t_info.ans, t_info.len);
+    for (size_t i = 0; i < t_info.len; ++i) {
+      postDecode(proc, &arr[i]);
+    }
   }
   if (decode)
     stop_meas(&eNB->ulsch_decoding_stats);

openair1/SCHED_NR/nr_ru_procedures.c

@@ -289,47 +289,59 @@ void nr_feptx(void *arg)
 
       ////////////FEPTX////////////
   nr_feptx0(ru, slot, startSymbol, numSymbols, aa);
+
+  // Task completed in //
+  completed_task_ans(feptx->ans);
 }
 
 // RU FEP TX using thread-pool
 void nr_feptx_tp(RU_t *ru, int frame_tx, int slot)
 {
   nfapi_nr_config_request_scf_t *cfg = &ru->gNB_list[0]->gNB_config;
-  int nbfeptx = 0;
   if (nr_slot_select(cfg, frame_tx, slot) == NR_UPLINK_SLOT)
     return;
   if (ru->idx == 0)
     VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPTX_OFDM, 1);
   start_meas(&ru->ofdm_total_stats);
+
+  size_t const sz = ru->nb_tx + (ru->half_slot_parallelization > 0) * ru->nb_tx;
+  AssertFatal(sz < 64, "Please, increase the buffer size");
+  feptx_cmd_t arr[64] = {0};
+  task_ans_t ans[64] = {0};
+
+  int nbfeptx = 0;
   for (int aid = 0; aid < ru->nb_tx; aid++) {
-    notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(feptx_cmd_t), 2000 + aid, ru->respfeptx, nr_feptx);
-    feptx_cmd_t *feptx_cmd = (feptx_cmd_t*)NotifiedFifoData(req);
+    feptx_cmd_t *feptx_cmd = &arr[nbfeptx];
+    feptx_cmd->ans = &ans[nbfeptx];
+
     feptx_cmd->aid = aid;
     feptx_cmd->ru = ru;
     feptx_cmd->slot = slot;
     feptx_cmd->startSymbol = 0;
-    feptx_cmd->numSymbols = (ru->half_slot_parallelization > 0) ?
-                            ru->nr_frame_parms->symbols_per_slot >> 1 :
-                            ru->nr_frame_parms->symbols_per_slot;
-    pushTpool(ru->threadPool, req);
+    feptx_cmd->numSymbols =
+        (ru->half_slot_parallelization > 0) ? ru->nr_frame_parms->symbols_per_slot >> 1 : ru->nr_frame_parms->symbols_per_slot;
+
+    task_t t = {.func = nr_feptx, .args = feptx_cmd};
+    pushTpool(ru->threadPool, t);
     nbfeptx++;
     if (ru->half_slot_parallelization > 0) {
-      notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(feptx_cmd_t), 2000 + aid + ru->nb_tx, ru->respfeptx, nr_feptx);
-      feptx_cmd_t *feptx_cmd = (feptx_cmd_t*)NotifiedFifoData(req);
+      feptx_cmd_t *feptx_cmd = &arr[nbfeptx];
+      feptx_cmd->ans = &ans[nbfeptx];
+
       feptx_cmd->aid = aid;
       feptx_cmd->ru = ru;
       feptx_cmd->slot = slot;
       feptx_cmd->startSymbol = ru->nr_frame_parms->symbols_per_slot >> 1;
       feptx_cmd->numSymbols = ru->nr_frame_parms->symbols_per_slot >> 1;
-      pushTpool(ru->threadPool, req);
+
+      task_t t = {.func = nr_feptx, .args = feptx_cmd};
+      pushTpool(ru->threadPool, t);
       nbfeptx++;
     }
   }
-  while (nbfeptx > 0) {
-    notifiedFIFO_elt_t *req=pullTpool(ru->respfeptx, ru->threadPool);
-    delNotifiedFIFO_elt(req);
-    nbfeptx--;
-  }
+
+  join_task_ans(ans, nbfeptx);
+
   stop_meas(&ru->ofdm_total_stats);
   if (ru->idx == 0)
     VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPTX_OFDM, 0);
@@ -359,6 +371,9 @@ void nr_fep(void* arg)
                      tti_rx,
                      ru->N_TA_offset);
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX+aid, 0);
+
+  // Task completed in //
+  completed_task_ans(feprx_cmd->ans);
 }
 
 // RU RX FEP using thread-pool
@@ -367,33 +382,45 @@ void nr_fep_tp(RU_t *ru, int slot) {
   int nbfeprx=0;
   if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX, 1 );
   start_meas(&ru->ofdm_demod_stats);
+
+  size_t const sz = ru->nb_rx + (ru->half_slot_parallelization > 0) * ru->nb_rx;
+  AssertFatal(sz < 64, "Please, increase buffer size");
+  feprx_cmd_t arr[64] = {0};
+  task_ans_t ans[64] = {0};
+
   for (int aid=0;aid<ru->nb_rx;aid++) {
-       notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(feprx_cmd_t), 1000 + aid,ru->respfeprx,nr_fep);
-       feprx_cmd_t *feprx_cmd=(feprx_cmd_t*)NotifiedFifoData(req);       
-       feprx_cmd->aid          = aid;
-       feprx_cmd->ru           = ru;
-       feprx_cmd->slot         = ru->proc.tti_rx;
-       feprx_cmd->startSymbol  = 0;
-       feprx_cmd->endSymbol    = (ru->half_slot_parallelization > 0)?(ru->nr_frame_parms->symbols_per_slot>>1)-1:(ru->nr_frame_parms->symbols_per_slot-1);
-       pushTpool(ru->threadPool,req);
-       nbfeprx++;
-       if (ru->half_slot_parallelization>0) {
-         notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(feprx_cmd_t), 1000 + aid + ru->nb_rx,ru->respfeprx,nr_fep);
-         feprx_cmd_t *feprx_cmd=(feprx_cmd_t*)NotifiedFifoData(req);       
-         feprx_cmd->aid          = aid;
-         feprx_cmd->ru           = ru;
-         feprx_cmd->slot         = ru->proc.tti_rx;
-	 feprx_cmd->startSymbol  = ru->nr_frame_parms->symbols_per_slot>>1;
-         feprx_cmd->endSymbol    = ru->nr_frame_parms->symbols_per_slot-1;
-         pushTpool(ru->threadPool,req);
-         nbfeprx++;
-       }
-  }
-  while (nbfeprx>0) {
-    notifiedFIFO_elt_t *req=pullTpool(ru->respfeprx, ru->threadPool);
-    delNotifiedFIFO_elt(req);
-    nbfeprx--;
+    feprx_cmd_t *feprx_cmd = &arr[nbfeprx];
+    feprx_cmd->ans = &ans[nbfeprx];
+
+    feprx_cmd->aid = aid;
+    feprx_cmd->ru = ru;
+    feprx_cmd->slot = ru->proc.tti_rx;
+    feprx_cmd->startSymbol = 0;
+    feprx_cmd->endSymbol = (ru->half_slot_parallelization > 0) ? (ru->nr_frame_parms->symbols_per_slot >> 1) - 1
+                                                               : (ru->nr_frame_parms->symbols_per_slot - 1);
+
+    task_t t = {.func = nr_fep, .args = feprx_cmd};
+    pushTpool(ru->threadPool, t);
+    nbfeprx++;
+    if (ru->half_slot_parallelization > 0) {
+      feprx_cmd_t *feprx_cmd = &arr[nbfeprx];
+      feprx_cmd->ans = &ans[nbfeprx];
+
+      feprx_cmd->aid = aid;
+      feprx_cmd->ru = ru;
+      feprx_cmd->slot = ru->proc.tti_rx;
+      feprx_cmd->startSymbol = ru->nr_frame_parms->symbols_per_slot >> 1;
+      feprx_cmd->endSymbol = ru->nr_frame_parms->symbols_per_slot - 1;
+
+      task_t t = {.func = nr_fep, .args = feprx_cmd};
+      pushTpool(ru->threadPool, t);
+
+      nbfeprx++;
+    }
   }
+
+  join_task_ans(ans, nbfeprx);
+
   stop_meas(&ru->ofdm_demod_stats);
   if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX, 0 );
 }

openair1/SCHED_NR/phy_procedures_nr_gNB.c

@@ -39,6 +39,7 @@
 #include "nfapi/oai_integration/vendor_ext.h"
 #include "assertions.h"
 #include <time.h>
+#include <stdint.h>
 
 //#define DEBUG_RXDATA
 //#define SRS_IND_DEBUG
@@ -304,9 +305,8 @@ void phy_procedures_gNB_TX(processingData_L1tx_t *msgTx,
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_gNB_TX + gNB->CC_id, 0);
 }
 
-static void nr_postDecode(PHY_VARS_gNB *gNB, notifiedFIFO_elt_t *req)
+static void nr_postDecode(PHY_VARS_gNB *gNB, ldpcDecode_t *rdata)
 {
-  ldpcDecode_t *rdata = (ldpcDecode_t*) NotifiedFifoData(req);
   NR_UL_gNB_HARQ_t *ulsch_harq = rdata->ulsch_harq;
   NR_gNB_ULSCH_t *ulsch = rdata->ulsch;
   int r = rdata->segment_r;
@@ -415,7 +415,12 @@ static void nr_postDecode(PHY_VARS_gNB *gNB, notifiedFIFO_elt_t *req)
   }
 }
 
-static int nr_ulsch_procedures(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, int ULSCH_id, uint8_t harq_pid)
+static int nr_ulsch_procedures(PHY_VARS_gNB *gNB,
+                               int frame_rx,
+                               int slot_rx,
+                               int ULSCH_id,
+                               uint8_t harq_pid,
+                               thread_info_tm_t *t_info)
 {
   NR_DL_FRAME_PARMS *frame_parms = &gNB->frame_parms;
   nfapi_nr_pusch_pdu_t *pusch_pdu = &gNB->ulsch[ULSCH_id].harq_process->ulsch_pdu;
@@ -466,9 +471,17 @@ static int nr_ulsch_procedures(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, int
    * measurement per processed TB.*/
   if (gNB->max_nb_pusch == 1)
     start_meas(&gNB->ulsch_decoding_stats);
-  int nbDecode =
-      nr_ulsch_decoding(gNB, ULSCH_id, gNB->pusch_vars[ULSCH_id].llr, frame_parms, pusch_pdu, frame_rx, slot_rx, harq_pid, G);
 
+  int const nbDecode = nr_ulsch_decoding(gNB,
+                                         ULSCH_id,
+                                         gNB->pusch_vars[ULSCH_id].llr,
+                                         frame_parms,
+                                         pusch_pdu,
+                                         frame_rx,
+                                         slot_rx,
+                                         harq_pid,
+                                         G,
+                                         t_info);
   return nbDecode;
 }
 
@@ -827,6 +840,11 @@ int phy_procedures_gNB_uespec_RX(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx)
     }
   }
 
+  ldpcDecode_t arr[64];
+  task_ans_t ans[64] = {0};
+  thread_info_tm_t t_info = {.buf = (uint8_t *)arr, .len = 0, .cap = 64, .ans = ans};
+
+  // int64_t const t0 = time_now_ns();
   int totalDecode = 0;
   for (int ULSCH_id = 0; ULSCH_id < gNB->max_nb_pusch; ULSCH_id++) {
     NR_gNB_ULSCH_t *ulsch = &gNB->ulsch[ULSCH_id];
@@ -924,21 +942,20 @@ int phy_procedures_gNB_uespec_RX(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx)
       // LOG_M("rxdataF_comp.m","rxF_comp",gNB->pusch_vars[0]->rxdataF_comp[0],6900,1,1);
       // LOG_M("rxdataF_ext.m","rxF_ext",gNB->pusch_vars[0]->rxdataF_ext[0],6900,1,1);
       VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_NR_ULSCH_PROCEDURES_RX, 1);
-      int const tasks_added = nr_ulsch_procedures(gNB, frame_rx, slot_rx, ULSCH_id, ulsch->harq_pid);
+      int const tasks_added = nr_ulsch_procedures(gNB, frame_rx, slot_rx, ULSCH_id, ulsch->harq_pid, &t_info);
       if (tasks_added > 0)
         totalDecode += tasks_added; 
 
       VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_NR_ULSCH_PROCEDURES_RX, 0);
     }
   }
-    while (totalDecode > 0) {
-      notifiedFIFO_elt_t *req = pullTpool(&gNB->respDecode, &gNB->threadPool);
-      if (req == NULL)
-        break; // Tpool has been stopped
-      nr_postDecode(gNB, req);
-      delNotifiedFIFO_elt(req);
-      totalDecode--;
-    }
+
+  DevAssert(totalDecode == t_info.len);
+
+  join_task_ans(t_info.ans, t_info.len);
+  for (int i = 0; i < t_info.len; ++i) {
+    nr_postDecode(gNB, &arr[i]);
+  }
   /* Do ULSCH decoding time measurement only when number of PUSCH is limited to 1
    * (valid for unitary physical simulators). ULSCH processing loop is then executed
    * only once, which ensures exactly one start and stop of the ULSCH decoding time

openair1/SIMULATION/NR_PHY/dlschsim.c

@@ -661,6 +661,7 @@ int main(int argc, char **argv)
     free(gNB->gNB_config.tdd_table.max_tdd_periodicity_list[i].max_num_of_symbol_per_slot_list);
   free(gNB->gNB_config.tdd_table.max_tdd_periodicity_list);
 
+  abortTpool(&gNB->threadPool);
   phy_free_nr_gNB(gNB);
   free(RC.gNB[0]);
   free(RC.gNB);

openair1/SIMULATION/NR_PHY/ulschsim.c

@@ -45,7 +45,6 @@
 #include "openair1/SIMULATION/TOOLS/sim.h"
 #include "openair1/SIMULATION/RF/rf.h"
 #include "openair1/SIMULATION/NR_PHY/nr_unitary_defs.h"
-#include "common/utils/threadPool/thread-pool.h"
 #include "openair2/LAYER2/NR_MAC_COMMON/nr_mac_common.h"
 #include "executables/nr-uesoftmodem.h"
 #include "nfapi/oai_integration/vendor_ext.h"
@@ -91,9 +90,8 @@ void deref_sched_response(int _)
   exit(1);
 }
 
-int nr_postDecode_sim(PHY_VARS_gNB *gNB, notifiedFIFO_elt_t *req, int *nb_ok)
+int nr_postDecode_sim(PHY_VARS_gNB *gNB, ldpcDecode_t *rdata, int *nb_ok)
 {
-  ldpcDecode_t *rdata = (ldpcDecode_t*) NotifiedFifoData(req);
   NR_UL_gNB_HARQ_t *ulsch_harq = rdata->ulsch_harq;
   int r = rdata->segment_r;
 
@@ -107,8 +105,10 @@ int nr_postDecode_sim(PHY_VARS_gNB *gNB, notifiedFIFO_elt_t *req, int *nb_ok)
   }
 
   // if all segments are done
-  if (rdata->nbSegments == ulsch_harq->processedSegments)
+  if (rdata->nbSegments == ulsch_harq->processedSegments) {
     return *nb_ok == rdata->nbSegments;
+  }
+
   return 0;
 }
 
@@ -594,20 +594,23 @@ int main(int argc, char **argv)
       exit(-1);
 #endif
 
-     int nbDecode = nr_ulsch_decoding(gNB, UE_id, channel_output_fixed, frame_parms, rel15_ul, frame, subframe, harq_pid, G);
-     int nb_ok = 0;
-     if (nbDecode > 0)
-       while (nbDecode > 0) {
-         notifiedFIFO_elt_t *req = pullTpool(&gNB->respDecode, &gNB->threadPool);
-         ret = nr_postDecode_sim(gNB, req, &nb_ok);
-         delNotifiedFIFO_elt(req);
-         nbDecode--;
-       }
+      ldpcDecode_t arr[16] = {0};
+      task_ans_t ans[16] = {0};
+      thread_info_tm_t t_info = {.buf = (uint8_t *)arr, .cap = 16, .len = 0, .ans = ans};
+      int nbDecode =
+          nr_ulsch_decoding(gNB, UE_id, channel_output_fixed, frame_parms, rel15_ul, frame, subframe, harq_pid, G, &t_info);
+      DevAssert(nbDecode > 0);
 
+      int nb_ok = 0;
+      join_task_ans(t_info.ans, t_info.len);
+      for (size_t i = 0; i < nbDecode; ++i) {
+        ret = nr_postDecode_sim(gNB, &arr[i], &nb_ok);
+      }
+      nbDecode = 0;
       if (ret)
         n_errors++;
     }
-    
+
     printf("*****************************************\n");
     printf("SNR %f, BLER %f (false positive %f)\n", SNR,
            (float) n_errors / (float) n_trials,
@@ -633,6 +636,7 @@ int main(int argc, char **argv)
   term_nr_ue_signal(UE, 1);
   free(UE);
 
+  abortTpool(&gNB->threadPool);
   phy_free_nr_gNB(gNB);
   free(RC.gNB[0]);
   free(RC.gNB);

openair2/LAYER2/MAC/main.c

@@ -41,6 +41,7 @@
 #include "common/ran_context.h"
 #include "intertask_interface.h"
 #include <pthread.h>
+#include "common/utils/system.h"
 
 extern RAN_CONTEXT_t RC;
 extern int oai_exit;

openair2/PHY_INTERFACE/IF_Module.h

@@ -37,7 +37,6 @@
 //#include "openair1/PHY/LTE_TRANSPORT/transport_eNB.h"
 #include "nfapi_interface.h"
 #include "common/platform_types.h"
-#include <common/utils/threadPool/thread-pool.h>
 #include <radio/COMMON/common_lib.h>
 
 #define MAX_NUM_DL_PDU 100

radio/COMMON/common_lib.h

@@ -37,7 +37,7 @@
 #include <sys/types.h>
 #include <openair1/PHY/TOOLS/tools_defs.h>
 #include "record_player.h"
-#include <common/utils/threadPool/thread-pool.h>
+#include "common/utils/threadPool/notified_fifo.h"
 
 /* default name of shared library implementing the radio front end */
 #define OAI_RF_LIBNAME        "oai_device"

radio/ETHERNET/eth_udp.c

@@ -50,7 +50,6 @@
 #include "common_lib.h"
 #include "ethernet_lib.h"
 #include "openair1/PHY/sse_intrin.h"
-#include "common/utils/threadPool/thread-pool.h"
 
 //#define DEBUG 1
 

radio/ETHERNET/ethernet_lib.h

@@ -41,7 +41,6 @@
 #include <sys/socket.h>
 #include <net/if.h>
 #include <netinet/ether.h>
-#include <common/utils/threadPool/thread-pool.h>
 
 #define MAX_INST 4
 #define DEFAULT_IF "lo"

radio/USRP/usrp_lib.cpp

@@ -49,6 +49,7 @@
 #include "common/utils/LOG/log.h"
 #include "common_lib.h"
 #include "assertions.h"
+#include "system.h"
 
 #include "common/utils/LOG/vcd_signal_dumper.h"
 

radio/fhi_72/oran_isolate.c

@@ -31,7 +31,6 @@
 #include "common/utils/LOG/log.h"
 #include "common/utils/LOG/vcd_signal_dumper.h"
 #include "openair1/PHY/defs_gNB.h"
-#include "common/utils/threadPool/thread-pool.h"
 #include "oaioran.h"
 #include "oran-config.h"