Merge remote-tracking branch 'origin/tpool-nospinlock' into integration_2025_w03 (!3193)

Remove spinlock from threadpool task return

The idea to use atomic aligned to cache line for counting finished jobs
is reused, but instead of polling a semaphore was added.

common/utils/threadPool/task_ans.c

@@ -26,37 +26,50 @@
 #include <stdint.h>
 #include <stdlib.h>
 #include <time.h>
+#include "pthread_utils.h"
+#include "errno.h"
+#include <string.h>
 
-void completed_task_ans(task_ans_t* task)
+#define seminit(sem)                                                                           \
+  {                                                                                            \
+    int ret = sem_init(&sem, 0, 0);                                                            \
+    AssertFatal(ret == 0, "sem_init(): ret=%d, errno=%d (%s)\n", ret, errno, strerror(errno)); \
+  }
+#define sempost(sem)                                                                           \
+  {                                                                                            \
+    int ret = sem_post(&sem);                                                                  \
+    AssertFatal(ret == 0, "sem_post(): ret=%d, errno=%d (%s)\n", ret, errno, strerror(errno)); \
+  }
+#define semwait(sem)                                                                           \
+  {                                                                                            \
+    int ret = sem_wait(&sem);                                                                  \
+    AssertFatal(ret == 0, "sem_wait(): ret=%d, errno=%d (%s)\n", ret, errno, strerror(errno)); \
+  }
+#define semdestroy(sem)                                                                           \
+  {                                                                                               \
+    int ret = sem_destroy(&sem);                                                                  \
+    AssertFatal(ret == 0, "sem_destroy(): ret=%d, errno=%d (%s)\n", ret, errno, strerror(errno)); \
+  }
+
+void init_task_ans(task_ans_t* ans, uint num_jobs)
 {
-  DevAssert(task != NULL);
-
-  int status = atomic_load_explicit(&task->status, memory_order_acquire);
-  AssertFatal(status == 0, "Task not expected to be finished here. Status = %d\n", status);
-
-  atomic_store_explicit(&task->status, 1, memory_order_release);
+  ans->counter = num_jobs;
+  seminit(ans->sem);
 }
 
-void join_task_ans(task_ans_t* arr, size_t len)
+void completed_many_task_ans(task_ans_t* ans, uint num_completed_jobs)
 {
-  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);
-    }
+  DevAssert(ans != NULL);
+  // Using atomic counter in contention scenario to avoid locking in producers
+  int num_jobs = atomic_fetch_sub_explicit(&ans->counter, num_completed_jobs, memory_order_relaxed);
+  if (num_jobs == num_completed_jobs) {
+    // Using semaphore to enable blocking call in join_task_ans
+    sempost(ans->sem);
   }
 }
+
+void join_task_ans(task_ans_t* ans)
+{
+  semwait(ans->sem);
+  semdestroy(ans->sem);
+}

common/utils/threadPool/task_ans.h

@@ -21,7 +21,7 @@
 
 #ifndef TASK_ANSWER_THREAD_POOL_H
 #define TASK_ANSWER_THREAD_POOL_H
-
+#include "pthread_utils.h"
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -39,6 +39,7 @@ extern "C" {
 
 #include <stddef.h>
 #include <stdint.h>
+#include <semaphore.h>
 
 #if defined(__i386__) || defined(__x86_64__)
 #define LEVEL1_DCACHE_LINESIZE 64
@@ -52,9 +53,17 @@ extern "C" {
 #error Unknown CPU architecture
 #endif
 
+/** @brief
+ * A multi-producer - single-consumer synchronization mechanism built for efficiency under
+ * contention.
+ *
+ * @param sem semaphore to wait on
+ * @param counter atomic counter to keep track of the number of tasks completed. Atomic counter
+ * is used for efficiency under contention.
+ */
 typedef struct {
-  // Avoid false sharing
-  _Alignas(LEVEL1_DCACHE_LINESIZE) _Atomic(int) status;
+  sem_t sem;
+  _Alignas(LEVEL1_DCACHE_LINESIZE) _Atomic(int) counter;
 } task_ans_t;
 
 typedef struct {
@@ -64,10 +73,29 @@ typedef struct {
   task_ans_t* ans;
 } thread_info_tm_t;
 
+/// @brief Initialize a task_ans_t struct
+///
+/// @param ans task_ans_t struct
+/// @param num_jobs number of tasks to wait for
+void init_task_ans(task_ans_t* ans, unsigned int num_jobs);
 
-void join_task_ans(task_ans_t* arr, size_t len);
+/// @brief Wait for all tasks to complete
+/// @param ans task_ans_t struct
+void join_task_ans(task_ans_t* arr);
 
-void completed_task_ans(task_ans_t* task);
+/// @brief Mark a number of tasks as completed.
+///
+/// @param ans task_ans_t struct
+/// @param num_completed_jobs number of tasks to mark as completed
+void completed_many_task_ans(task_ans_t* ans, uint num_completed_jobs);
+
+/// @brief Mark 1 tasks as completed.
+///
+/// @param ans task_ans_t struct
+static inline void completed_task_ans(task_ans_t* ans)
+{
+  completed_many_task_ans(ans, 1);
+}
 
 #ifdef __cplusplus
 }

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

@@ -114,18 +114,18 @@ 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));
+    task_ans_t task_ans;
+    init_task_ans(&task_ans, parall);
     struct testData test_data[parall];
     memset(test_data, 0, sizeof(test_data));
     for (int j = 0; j < parall; j++) {
       task_t task = {.args = &test_data[j], .func = processing};
       struct testData *x = (struct testData *)task.args;
       x->id = i;
-      x->task_ans = &task_ans[j];
+      x->task_ans = &task_ans;
       pushTpool(&pool, task);
     }
-    join_task_ans(task_ans, parall);
+    join_task_ans(&task_ans);
     int sleepmax = 0;
     for (int j = 0; j < parall; j++) {
       if (test_data[j].sleepTime > sleepmax) {