Merge remote-tracking branch 'origin/nr-ue-threading-improvements' into integration_2024_w48 (!3091)

Update NR UE threading model

This updates the NR UE threading model by preventing calling threadpool
from threadpool and fixes deadlock issues related to processSlotTX. By
making processSlotTX run to completion instead of starting and waiting
for other threads to finish it prevents locking thread pool cores,
therefore prevening deadlocks.

There is a slight difference here compared to the previous version where
I do not use the UL actor as I believe it is not necessary,
processSlotTX satisfies the run-to-completion requirement now. If in the
future any UL procedure would be parallelized using the thread pool it
should either be done in a run-to-completion model or the processSlotTX
function should be taken off the thread pool.

The documentation has been updated.

common/utils/CMakeLists.txt

@@ -19,3 +19,5 @@ target_link_libraries(utils PRIVATE ${T_LIB})
 if (ENABLE_TESTS)
   add_subdirectory(hashtable/tests)
 endif()
+add_subdirectory(barrier)
+add_subdirectory(actor)

common/utils/actor/CMakeLists.txt

@@ -0,0 +1,6 @@
+add_library(actor actor.c)
+target_include_directories(actor PUBLIC ./)
+target_link_libraries(actor PUBLIC thread-pool)
+if (ENABLE_TESTS)
+  add_subdirectory(tests)
+endif()

common/utils/actor/README.md

@@ -0,0 +1,13 @@
+# Overview
+
+This is a simple actor implementation (see https://en.wikipedia.org/wiki/Actor_model).
+
+Actor is implemented as a single thread with single producer/consumer queue as input. You can think of it as a single thread threadpool
+
+If you need concurrency, consider allocating more than one actor.
+
+# Thread safety
+
+There is two ways to ensure thread safety between two actors
+ - Use core affinity to set both actor to run on the same core.
+ - Use data separation, like in testcase thread_safety_with_actor_specific_data

common/utils/actor/actor.c

@@ -0,0 +1,87 @@
+/*
+ * 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 "thread-pool.h"
+#include "actor.h"
+#include "system.h"
+#include "assertions.h"
+
+void *actor_thread(void *arg);
+
+void init_actor(Actor_t *actor, const char *name, int core_affinity)
+{
+  actor->terminate = false;
+  initNotifiedFIFO(&actor->fifo);
+  char actor_name[16];
+  snprintf(actor_name, sizeof(actor_name), "%s%s", name, "_actor");
+  threadCreate(&actor->thread, actor_thread, (void *)actor, actor_name, core_affinity, OAI_PRIORITY_RT_MAX);
+}
+
+/// @brief Main actor thread
+/// @param arg actor pointer
+/// @return NULL
+void *actor_thread(void *arg)
+{
+  Actor_t *actor = arg;
+
+  // Infinite loop to process requests
+  do {
+    notifiedFIFO_elt_t *elt = pullNotifiedFIFO(&actor->fifo);
+    if (elt == NULL) {
+      AssertFatal(actor->terminate, "pullNotifiedFIFO() returned NULL\n");
+      break;
+    }
+
+    elt->processingFunc(NotifiedFifoData(elt));
+    if (elt->reponseFifo) {
+      pushNotifiedFIFO(elt->reponseFifo, elt);
+    } else
+      delNotifiedFIFO_elt(elt);
+  } while (!actor->terminate);
+
+  return NULL;
+}
+
+void destroy_actor(Actor_t *actor)
+{
+  actor->terminate = true;
+  abortNotifiedFIFO(&actor->fifo);
+  pthread_join(actor->thread, NULL);
+}
+
+static void self_destruct_fun(void *arg) {
+  Actor_t *actor = arg;
+  actor->terminate = true;
+  abortNotifiedFIFO(&actor->fifo);
+}
+
+void shutdown_actor(Actor_t *actor)
+{
+  notifiedFIFO_t response_fifo;
+  initNotifiedFIFO(&response_fifo);
+  notifiedFIFO_elt_t *elt = newNotifiedFIFO_elt(0, 0, &response_fifo, self_destruct_fun);
+  elt->msgData = actor;
+  pushNotifiedFIFO(&actor->fifo, elt);
+  elt = pullNotifiedFIFO(&response_fifo);
+  delNotifiedFIFO_elt(elt);
+  abortNotifiedFIFO(&response_fifo);
+  pthread_join(actor->thread, NULL);
+}

common/utils/actor/actor.h

@@ -0,0 +1,52 @@
+/*
+ * 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 ACTOR_H
+#define ACTOR_H
+#include "thread-pool.h"
+
+#define INIT_ACTOR(ptr, name, core_affinity) init_actor((Actor_t *)ptr, name, core_affinity);
+
+#define DESTROY_ACTOR(ptr) destroy_actor((Actor_t *)ptr);
+
+#define SHUTDOWN_ACTOR(ptr) shutdown_actor((Actor_t *)ptr);
+
+typedef struct Actor_t {
+  notifiedFIFO_t fifo;
+  bool terminate;
+  pthread_t thread;
+} Actor_t;
+
+/// @brief Initialize the actor. Starts actor thread
+/// @param actor
+/// @param name Actor name. Thread name will be derived from it
+/// @param core_affinity Core affinity. Specify -1 for no affinity
+void init_actor(Actor_t *actor, const char *name, int core_affinity);
+
+/// @brief Destroy the actor. Free the memory, stop the thread.
+/// @param actor
+void destroy_actor(Actor_t *actor);
+
+/// @brief Gracefully shutdown the actor, letting all tasks previously started finish
+/// @param actor
+void shutdown_actor(Actor_t *actor);
+
+#endif

common/utils/actor/tests/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_executable(test_actor test_actor.cpp)
+target_link_libraries(test_actor PRIVATE actor GTest::gtest thread-pool LOG)
+add_dependencies(tests test_actor)
+add_test(NAME test_actor
+  COMMAND ./test_actor)

common/utils/actor/tests/test_actor.cpp

@@ -0,0 +1,134 @@
+/*
+ * 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 <stdio.h>
+#include <assert.h>
+#include <threads.h>
+#include <stdlib.h>
+#include <gtest/gtest.h>
+#include <unistd.h>
+#include "common/config/config_userapi.h"
+#include "log.h"
+
+extern "C" {
+#include "actor.h"
+configmodule_interface_t *uniqCfg;
+void exit_function(const char *file, const char *function, const int line, const char *s, const int assert)
+{
+  if (assert) {
+    abort();
+  } else {
+    exit(EXIT_SUCCESS);
+  }
+}
+}
+
+int num_processed = 0;
+
+void process(void *arg)
+{
+  num_processed++;
+}
+
+TEST(actor, schedule_one)
+{
+  Actor_t actor;
+  init_actor(&actor, "TEST", -1);
+  notifiedFIFO_elt_t *elt = newNotifiedFIFO_elt(0, 0, NULL, process);
+  pushNotifiedFIFO(&actor.fifo, elt);
+  shutdown_actor(&actor);
+  EXPECT_EQ(num_processed, 1);
+}
+
+TEST(actor, schedule_many)
+{
+  num_processed = 0;
+  Actor_t actor;
+  init_actor(&actor, "TEST", -1);
+  for (int i = 0; i < 100; i++) {
+    notifiedFIFO_elt_t *elt = newNotifiedFIFO_elt(0, 0, NULL, process);
+    pushNotifiedFIFO(&actor.fifo, elt);
+  }
+  shutdown_actor(&actor);
+  EXPECT_EQ(num_processed, 100);
+}
+
+// Thread safety can be ensured through core affinity - if two actors
+// are running on the same core they are thread safe
+TEST(DISABLED_actor, thread_safety_with_core_affinity)
+{
+  num_processed = 0;
+  int core = 0;
+  Actor_t actor;
+  init_actor(&actor, "TEST", core);
+  Actor_t actor2;
+  init_actor(&actor2, "TEST2", core);
+  for (int i = 0; i < 10000; i++) {
+    Actor_t *actor_ptr = i % 2 == 0 ? &actor : &actor2;
+    notifiedFIFO_elt_t *elt = newNotifiedFIFO_elt(0, 0, NULL, process);
+    pushNotifiedFIFO(&actor_ptr->fifo, elt);
+  }
+  shutdown_actor(&actor);
+  shutdown_actor(&actor2);
+  EXPECT_EQ(num_processed, 10000);
+}
+
+// Thread safety can be ensured through data separation, here using C inheritance-like
+// model
+typedef struct TestActor_t {
+  Actor_t actor;
+  int count;
+} TestActor_t;
+
+void process_thread_safe(void *arg)
+{
+  TestActor_t *actor = static_cast<TestActor_t *>(arg);
+  actor->count += 1;
+}
+
+TEST(actor, thread_safety_with_actor_specific_data)
+{
+  num_processed = 0;
+  int core = 0;
+  TestActor_t actor;
+  INIT_ACTOR(&actor, "TEST", core);
+  actor.count = 0;
+  TestActor_t actor2;
+  INIT_ACTOR(&actor2, "TEST2", core);
+  actor2.count = 0;
+  for (int i = 0; i < 10000; i++) {
+    TestActor_t *actor_ptr = i % 2 == 0 ? &actor : &actor2;
+    notifiedFIFO_elt_t *elt = newNotifiedFIFO_elt(0, 0, NULL, process_thread_safe);
+    elt->msgData = actor_ptr;
+    pushNotifiedFIFO(&actor_ptr->actor.fifo, elt);
+  }
+  SHUTDOWN_ACTOR(&actor);
+  SHUTDOWN_ACTOR(&actor2);
+  EXPECT_EQ(actor.count + actor2.count, 10000);
+}
+
+int main(int argc, char **argv)
+{
+  logInit();
+  g_log->log_component[UTIL].level = OAILOG_DEBUG;
+  testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

common/utils/barrier/CMakeLists.txt

@@ -0,0 +1,7 @@
+add_library(barrier barrier.c)
+target_include_directories(barrier PUBLIC ./)
+target_link_libraries(barrier PUBLIC pthread)
+
+if (ENABLE_TESTS)
+  add_subdirectory(tests)
+endif()

common/utils/barrier/barrier.c

@@ -0,0 +1,75 @@
+/*
+ * 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 "barrier.h"
+#include "assertions.h"
+
+void dynamic_barrier_init(dynamic_barrier_t* barrier)
+{
+  int ret = pthread_mutex_init(&barrier->barrier_lock, NULL);
+  AssertFatal(ret == 0, "Mutex error %d", ret);
+  barrier->callback = NULL;
+  barrier->callback_arg = NULL;
+  barrier->completed_jobs = 0;
+}
+
+void dynamic_barrier_reset(dynamic_barrier_t* barrier)
+{
+  barrier->callback = NULL;
+  if (barrier->callback_arg != NULL) {
+    free(barrier->callback_arg);
+  }
+  barrier->callback_arg = NULL;
+  barrier->completed_jobs = 0;
+}
+
+void dynamic_barrier_join(dynamic_barrier_t* barrier)
+{
+  int ret = pthread_mutex_lock(&barrier->barrier_lock);
+  AssertFatal(ret == 0, "Mutex error %d", ret);
+  barrier->completed_jobs++;
+  if (barrier->callback) {
+    if (barrier->completed_jobs == barrier->max_joins) {
+      barrier->completed_jobs = 0;
+      barrier->callback(barrier->callback_arg);
+      barrier->callback = NULL;
+      barrier->callback_arg = NULL;
+    }
+  }
+  ret = pthread_mutex_unlock(&barrier->barrier_lock);
+  AssertFatal(ret == 0, "Mutex error %d", ret);
+}
+
+void dynamic_barrier_update(dynamic_barrier_t* barrier, int max_joins, callback_t callback, void* arg)
+{
+  int ret = pthread_mutex_lock(&barrier->barrier_lock);
+  AssertFatal(ret == 0, "Mutex error %d", ret);
+  if (barrier->completed_jobs == max_joins) {
+    barrier->completed_jobs = 0;
+    callback(arg);
+  } else {
+    barrier->max_joins = max_joins;
+    barrier->callback = callback;
+    barrier->callback_arg = arg;
+  }
+  ret = pthread_mutex_unlock(&barrier->barrier_lock);
+  AssertFatal(ret == 0, "Mutex error %d", ret);
+}

common/utils/barrier/barrier.h

@@ -0,0 +1,57 @@
+/*
+ * 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 _BARRIER_H_
+#define _BARRIER_H_
+#include <stdint.h>
+#include <pthread.h>
+
+typedef void (*callback_t)(void*);
+
+/// @brief This thread barrier allows modifying the maximum joins after it is initialized
+typedef struct dynamic_barrier_t {
+  pthread_mutex_t barrier_lock;
+  int completed_jobs;
+  int max_joins;
+  callback_t callback;
+  void* callback_arg;
+} dynamic_barrier_t;
+
+/// @brief Initialize the barrier
+/// @param barrier
+void dynamic_barrier_init(dynamic_barrier_t* barrier);
+
+/// @brief Reset the barrier
+/// @param barrier
+void dynamic_barrier_reset(dynamic_barrier_t* barrier);
+
+/// @brief Perform join on the barrier. May run callback if it is already set
+/// @param barrier
+void dynamic_barrier_join(dynamic_barrier_t* barrier);
+
+/// @brief Set callback and max_joins. May run callback if the number of joins is already equal to max_joins
+/// @param barrier
+/// @param max_joins maximum number of joins before barrier triggers
+/// @param callback callback function
+/// @param arg callback function argument
+void dynamic_barrier_update(dynamic_barrier_t* barrier, int max_joins, callback_t callback, void* arg);
+
+#endif

common/utils/barrier/tests/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_executable(test_barrier test_barrier.cpp)
+target_link_libraries(test_barrier PRIVATE barrier GTest::gtest)
+add_dependencies(tests test_barrier)
+add_test(NAME test_barrier
+  COMMAND ./test_barrier)

common/utils/barrier/tests/test_barrier.cpp

@@ -0,0 +1,118 @@
+/*
+ * 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 <stdio.h>
+#include <assert.h>
+#include <threads.h>
+#include <stdlib.h>
+#include <gtest/gtest.h>
+
+extern "C" {
+#include "barrier.h"
+void exit_function(const char *file, const char *function, const int line, const char *s, const int assert)
+{
+  if (assert) {
+    abort();
+  } else {
+    exit(EXIT_SUCCESS);
+  }
+}
+}
+
+static int trigger_count = 0;
+static int triggered = 0;
+void f(void *arg)
+{
+  triggered = 1;
+  trigger_count++;
+}
+
+TEST(dynamic_barrier, trigger_once)
+{
+  dynamic_barrier_t barrier;
+  triggered = 0;
+  dynamic_barrier_init(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_update(&barrier, 4, f, NULL);
+  EXPECT_EQ(triggered, 1);
+}
+
+TEST(dynamic_barrier, trigger_twice)
+{
+  dynamic_barrier_t barrier;
+  triggered = 0;
+  dynamic_barrier_init(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_update(&barrier, 4, f, NULL);
+  EXPECT_EQ(triggered, 1);
+  triggered = 0;
+  dynamic_barrier_init(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_join(&barrier);
+  dynamic_barrier_update(&barrier, 4, f, NULL);
+  EXPECT_EQ(triggered, 1);
+}
+
+int dynamic_barrier_thread(void *thr_data)
+{
+  dynamic_barrier_t *barrier = (dynamic_barrier_t *)thr_data;
+  dynamic_barrier_join(barrier);
+  return 0;
+}
+
+int dynamic_barrier_update_thread(void *thr_data)
+{
+  dynamic_barrier_t *barrier = (dynamic_barrier_t *)thr_data;
+  dynamic_barrier_update(barrier, 3, f, NULL);
+  return 0;
+}
+
+TEST(dynamic_barrier, multithreaded)
+{
+  thrd_t thr[3];
+  dynamic_barrier_t barrier;
+  trigger_count = 0;
+  for (int i = 0; i < 300; i++) {
+    dynamic_barrier_init(&barrier);
+    thrd_t update_thread;
+    thrd_create(&update_thread, dynamic_barrier_update_thread, &barrier);
+    for (int n = 0; n < 3; ++n)
+      thrd_create(&thr[n], dynamic_barrier_thread, &barrier);
+    for (int n = 0; n < 3; ++n)
+      thrd_join(thr[n], NULL);
+    thrd_join(update_thread, NULL);
+  }
+  assert(trigger_count == 300);
+}
+
+int main(int argc, char **argv)
+{
+  testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

common/utils/nr/CMakeLists.txt

@@ -1,6 +1,6 @@
 add_library(nr_common nr_common.c)
 target_include_directories(nr_common PUBLIC .)
-target_link_libraries(nr_common PRIVATE UTIL)
+target_link_libraries(nr_common PUBLIC UTIL)
 if (ENABLE_TESTS)
   add_subdirectory(tests)
 endif()

common/utils/telnetsrv/CMakeLists.txt

@@ -53,7 +53,7 @@ message(STATUS "No specific telnet functions for 4Gue")
 message(STATUS "Add 5Gue specific telnet functions in libtelnetsrv_5Gue.so")
 add_library(telnetsrv_5Gue MODULE telnetsrv_5Gue_measurements.c)
 add_dependencies(telnetsrv telnetsrv_5Gue)
-target_link_libraries(telnetsrv_5Gue PRIVATE asn1_nr_rrc_hdrs asn1_lte_rrc_hdrs)
+target_link_libraries(telnetsrv_5Gue PRIVATE asn1_nr_rrc_hdrs asn1_lte_rrc_hdrs UTIL)
 
 message(STATUS "Add CI specific telnet functions in libtelnetsrv_ci.so")
 add_library(telnetsrv_ci MODULE telnetsrv_ci.c)
@@ -79,4 +79,3 @@ add_dependencies(telnetsrv telnetsrv_rrc)
 set_target_properties(telnetsrv telnetsrv_enb telnetsrv_5Gue telnetsrv_ci telnetsrv_ciUE telnetsrv_bearer telnetsrv_rrc
                       PROPERTIES LIBRARY_OUTPUT_DIRECTORY ../../..
 )
-

common/utils/threadPool/thread-pool.h

@@ -30,6 +30,7 @@
 #include <stdalign.h>
 #include <pthread.h>
 #include <unistd.h>
+#include <string.h>
 #include <sys/syscall.h>
 #include "assertions.h"
 #include "common/utils/time_meas.h"