Merge remote-tracking branch 'origin/tpool-fix' into integration_2026_w03 (!3799)

Fix for threadpool abort function

common/utils/CMakeLists.txt

@@ -23,3 +23,6 @@ endif()
 add_subdirectory(barrier)
 add_subdirectory(actor)
 add_subdirectory(shm_iq_channel)
+if (ENABLE_TESTS)
+  add_subdirectory(tests)
+endif()

common/utils/tests/CMakeLists.txt

@@ -0,0 +1,2 @@
+add_executable(test_tpool_vs_actors test_tpool_vs_actors.c)
+target_link_libraries(test_tpool_vs_actors PRIVATE thread-pool pthread LOG minimal_lib actor)

common/utils/tests/test_tpool_vs_actors.c

@@ -0,0 +1,126 @@
+/*
+ * 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 "actor.h"
+#include "thread-pool.h"
+#include "task.h"
+#include "log.h"
+#include <time.h>
+
+#define NUM_THREADS 10
+#define NUM_JOBS 200000
+
+typedef struct {
+  struct timespec ts;
+  int actor_index;
+} actor_task_t;
+
+long long delay_table[NUM_THREADS] = {0};
+
+void calculate_delay(struct timespec* send_ts, int thread_index)
+{
+  struct timespec ts;
+  clock_gettime(CLOCK_MONOTONIC, &ts);
+  long long recv_time = ts.tv_sec * 1000000000LL + ts.tv_nsec;
+  long long send_time = send_ts->tv_sec * 1000000000LL + send_ts->tv_nsec;
+  long long delay = recv_time - send_time;
+  delay_table[thread_index] += delay;
+}
+
+void tpool_function(void* args)
+{
+  int worker_id = get_tpool_worker_index();
+  calculate_delay((struct timespec*)args, worker_id);
+  free(args);
+}
+
+void actor_function(void* args)
+{
+  actor_task_t* actor_args = (actor_task_t*)args;
+  calculate_delay(&actor_args->ts, actor_args->actor_index);
+}
+
+int main()
+{
+  logInit();
+  tpool_t pool;
+  char params[NUM_THREADS * 4];
+  memset(params, 0, sizeof(params));
+  for (int i = 0; i < NUM_THREADS; i++) {
+    char buf[4];
+    snprintf(buf, sizeof(buf), "%d,", -1);
+    strcat(params, buf);
+  }
+
+  initTpool(params, &pool, true);
+
+  // Example task
+  task_t task;
+  task.func = tpool_function;
+  task.args = NULL;
+
+  // Push tasks to the thread pool
+  for (int i = 0; i < NUM_JOBS; i++) {
+    struct timespec* ts = malloc(sizeof(struct timespec));
+    clock_gettime(CLOCK_MONOTONIC, ts);
+    task.args = ts;
+    pushTpool(&pool, task);
+  }
+
+  // Abort the thread pool
+  abortTpool(&pool);
+
+  long long sum_delay = 0;
+  for (int i = 0; i < NUM_THREADS; i++) {
+    sum_delay += delay_table[i];
+  }
+  float average_delay = sum_delay / (NUM_JOBS * 1.0f);
+  printf("Average task delay on tpool: %.2f ns\n", average_delay);
+
+
+  memset(delay_table, 0, sizeof(delay_table));
+
+  Actor_t actors[NUM_THREADS];
+  for (int i = 0; i < NUM_THREADS; i++) {
+    init_actor(&actors[i], "example_actor", -1);
+  }
+
+  // Push tasks to the actors
+  for (int i = 0; i < NUM_JOBS; i++) {
+    notifiedFIFO_elt_t* task = newNotifiedFIFO_elt(sizeof(actor_task_t), 0, NULL, actor_function);
+    actor_task_t* arg_ts = (actor_task_t*)NotifiedFifoData(task);
+    arg_ts->actor_index = i % NUM_THREADS;
+    clock_gettime(CLOCK_MONOTONIC, &arg_ts->ts);
+    pushNotifiedFIFO(&actors[i % NUM_THREADS].fifo, task);
+  }
+
+  for (int i = 0; i < NUM_THREADS; i++) {
+    shutdown_actor(&actors[i]);
+  }
+
+  sum_delay = 0;
+  for (int i = 0; i < NUM_THREADS; i++) {
+    sum_delay += delay_table[i];
+  }
+  average_delay = sum_delay / (NUM_JOBS * 1.0f);
+  printf("Average task delay on actors: %.2f ns\n", average_delay);
+  return 0;
+}

common/utils/threadPool/thread-pool.c

@@ -55,12 +55,47 @@ void pushTpool(tpool_t* tpool, task_t task)
   push_not_q(&q_arr[index % len_thr], task);
 }
 
+// Same as above, but avoid certain tpool workers based on the mask
+void pushTpool_mask(tpool_t* tpool, task_t task, uint64_t mask)
+{
+  DevAssert(tpool != NULL);
+  if (tpool->len_thr == 0) {
+    task.func(task.args);
+    return;
+  }
+
+  size_t index = tpool->index++;
+  size_t const len_thr = tpool->len_thr;
+
+  not_q_t* q_arr = (not_q_t*)tpool->q_arr;
+
+  for (size_t i = 0; i < len_thr; ++i) {
+    int index = (i + tpool->index) % len_thr;
+    if ((1UL << index) & mask)
+      continue;
+    if (try_push_not_q(&q_arr[index], task)) {
+      return;
+    }
+  }
+
+  while ((1ULL << (index % len_thr)) & mask)
+    index++;
+  push_not_q(&q_arr[index % len_thr], task);
+}
+
+__thread int tpool_worker_index = -1;
+int get_tpool_worker_index(void)
+{
+  return tpool_worker_index;
+}
+
 static void* worker_thread(void* arg)
 {
   DevAssert(arg != NULL);
 
   task_thread_args_t* args = (task_thread_args_t*)arg;
   int const idx = args->idx;
+  tpool_worker_index = idx;
   tpool_t* tpool = args->tpool;
 
   uint32_t const len = tpool->len_thr;
@@ -87,7 +122,10 @@ static void* worker_thread(void* arg)
     }
 
     if (ret.t.func == NULL && ret.t.args == NULL) {
-      pushTpool(tpool, (task_t){.args = NULL, .func = NULL});
+      tpool->dead_mask |= 1ULL << idx;
+      if (__builtin_popcountll(tpool->dead_mask) == tpool->len_thr)
+        break;
+      pushTpool_mask(tpool, (task_t){.args = NULL, .func = NULL}, tpool->dead_mask);
       break;
     }
     ret.t.func(ret.t.args);

common/utils/threadPool/thread-pool.h

@@ -50,6 +50,7 @@ typedef struct {
   void* q_arr;
 
   pthread_barrier_t barrier;
+  _Atomic(uint64_t) dead_mask;
 } tpool_t;
 
 /// @brief Push job to threadpool. May run task inline in case there are no worker threads
@@ -82,4 +83,8 @@ void initFloatingCoresTpool(int nbThreads,tpool_t *pool, bool performanceMeas, c
 
 /// Convenience macro
 #define  initTpool(PARAMPTR,TPOOLPTR, MEASURFLAG) initNamedTpool(PARAMPTR,TPOOLPTR, MEASURFLAG, NULL)
+
+/// Returns the index of the worker thread in the thread pool
+/// @return index of the worker thread in a thread pool or -1 if not called from a thread pool worker thread
+int get_tpool_worker_index(void);
 #endif