diff --git a/CMakeLists.txt b/CMakeLists.txt index f4047bebd7..9f2c76c390 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/common/utils/CMakeLists.txt b/common/utils/CMakeLists.txt index bce79d082d..ccb15bfd47 100644 --- a/common/utils/CMakeLists.txt +++ b/common/utils/CMakeLists.txt @@ -20,3 +20,4 @@ if (ENABLE_TESTS) add_subdirectory(hashtable/tests) endif() add_subdirectory(barrier) +add_subdirectory(actor) diff --git a/common/utils/actor/CMakeLists.txt b/common/utils/actor/CMakeLists.txt new file mode 100644 index 0000000000..40057d347e --- /dev/null +++ b/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() diff --git a/common/utils/actor/README.md b/common/utils/actor/README.md new file mode 100644 index 0000000000..13a577a7a1 --- /dev/null +++ b/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 diff --git a/common/utils/actor/actor.c b/common/utils/actor/actor.c new file mode 100644 index 0000000000..68661ddb6d --- /dev/null +++ b/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); +} diff --git a/common/utils/actor/actor.h b/common/utils/actor/actor.h new file mode 100644 index 0000000000..753ba8d00a --- /dev/null +++ b/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 diff --git a/common/utils/actor/tests/CMakeLists.txt b/common/utils/actor/tests/CMakeLists.txt new file mode 100644 index 0000000000..dd4a22a848 --- /dev/null +++ b/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) diff --git a/common/utils/actor/tests/test_actor.cpp b/common/utils/actor/tests/test_actor.cpp new file mode 100644 index 0000000000..5f613846a2 --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#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(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(); +} diff --git a/common/utils/threadPool/thread-pool.h b/common/utils/threadPool/thread-pool.h index b0e1b9e4e8..31395d7a9b 100644 --- a/common/utils/threadPool/thread-pool.h +++ b/common/utils/threadPool/thread-pool.h @@ -30,6 +30,7 @@ #include #include #include +#include #include #include "assertions.h" #include "common/utils/time_meas.h" diff --git a/executables/nr-ue.c b/executables/nr-ue.c index f18505dc64..2460081981 100644 --- a/executables/nr-ue.c +++ b/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; diff --git a/executables/nr-uesoftmodem.c b/executables/nr-uesoftmodem.c index 8a9f0c626f..1389d610ac 100644 --- a/executables/nr-uesoftmodem.c +++ b/executables/nr-uesoftmodem.c @@ -88,6 +88,7 @@ unsigned short config_frames[4] = {2,9,11,13}; #include #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) { diff --git a/openair1/PHY/defs_nr_UE.h b/openair1/PHY/defs_nr_UE.h index 5e46e2899a..99b24f6fc0 100644 --- a/openair1/PHY/defs_nr_UE.h +++ b/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 {