diff --git a/common/utils/CMakeLists.txt b/common/utils/CMakeLists.txt index c3c7720628..dc997db6c7 100644 --- a/common/utils/CMakeLists.txt +++ b/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() diff --git a/common/utils/tests/CMakeLists.txt b/common/utils/tests/CMakeLists.txt new file mode 100644 index 0000000000..de855b5944 --- /dev/null +++ b/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) diff --git a/common/utils/tests/test_tpool_vs_actors.c b/common/utils/tests/test_tpool_vs_actors.c new file mode 100644 index 0000000000..97f2867f42 --- /dev/null +++ b/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 + +#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; +} \ No newline at end of file diff --git a/common/utils/threadPool/thread-pool.c b/common/utils/threadPool/thread-pool.c index 2a31602e6e..2983c779a7 100644 --- a/common/utils/threadPool/thread-pool.c +++ b/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); diff --git a/common/utils/threadPool/thread-pool.h b/common/utils/threadPool/thread-pool.h index 8e5d329499..75fc1487bb 100644 --- a/common/utils/threadPool/thread-pool.h +++ b/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