Actor model imlementation

Add actor library which implements the Actor model (see  https://en.wikipedia.org/wiki/Actor_model).

CMakeLists.txt

@@ -636,7 +636,7 @@ if (cap_FOUND)
   target_link_libraries(UTIL PRIVATE cap)
   target_compile_definitions(UTIL PRIVATE HAVE_LIB_CAP)
 endif()
-target_link_libraries(UTIL PUBLIC ${T_LIB} pthread LOG thread-pool utils barrier)
+target_link_libraries(UTIL PUBLIC ${T_LIB} pthread LOG thread-pool utils barrier actor)
 
 set(SECURITY_SRC
   ${OPENAIR3_DIR}/SECU/secu_defs.c

common/utils/CMakeLists.txt

@@ -20,3 +20,4 @@ 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/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"

executables/nr-ue.c

@@ -872,7 +872,7 @@ void *UE_thread(void *arg)
       }
       syncMsg->UE = UE;
       memset(&syncMsg->proc, 0, sizeof(syncMsg->proc));
-      pushTpool(&(get_nrUE_params()->Tpool), Msg);
+      pushNotifiedFIFO(&UE->sync_actor.fifo, Msg);
       trashed_frames = 0;
       syncRunning = true;
       continue;

executables/nr-uesoftmodem.c

@@ -88,6 +88,7 @@ unsigned short config_frames[4] = {2,9,11,13};
 #include <openair1/PHY/MODULATION/nr_modulation.h>
 #include "openair2/GNB_APP/gnb_paramdef.h"
 #include "pdcp.h"
+#include "actor.h"
 
 extern const char *duplex_mode[];
 THREAD_STRUCT thread_struct;
@@ -495,6 +496,7 @@ int main(int argc, char **argv)
         }
 
         UE[CC_id]->sl_mode = get_softmodem_params()->sl_mode;
+        init_actor(&UE[CC_id]->sync_actor, "SYNC_", -1);
         init_nr_ue_vars(UE[CC_id], inst);
 
         if (UE[CC_id]->sl_mode) {

openair1/PHY/defs_nr_UE.h

@@ -51,6 +51,7 @@
 #include "fapi_nr_ue_interface.h"
 #include "assertions.h"
 #include "barrier.h"
+#include "actor.h"
 //#include "openair1/SCHED_NR_UE/defs.h"
 
 #ifdef MEX
@@ -529,6 +530,7 @@ typedef struct PHY_VARS_NR_UE_s {
   // Sidelink parameters
   sl_nr_sidelink_mode_t sl_mode;
   sl_nr_ue_phy_params_t SL_UE_PHY_PARAMS;
+  Actor_t sync_actor;
 } PHY_VARS_NR_UE;
 
 typedef struct {