diff --git a/CMakeLists.txt b/CMakeLists.txt index 2ca3de460f..351c40d89a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -43,7 +43,7 @@ include("cmake_targets/CPM.cmake") option(CCACHE_ACTIVE "CCache" ON) find_program(CCACHE_FOUND ccache) if(CCACHE_FOUND AND CCACHE_ACTIVE) - if(${CMAKE_VERSION} VERSION_LESS "3.4.0") + if(${CMAKE_VERSION} VERSION_LESS "3.4.0") set_property(GLOBAL PROPERTY RULE_LAUNCH_COMPILE "${CCACHE_PROGRAM}") message(STATUS "Found ccache in ${CCACHE_FOUND}. Using ccache. CMake < 3.4") else() @@ -105,13 +105,13 @@ add_list_string_option(CMAKE_BUILD_TYPE "RelWithDebInfo" "Choose the type of bui # in case /proc/cpuinfo exists we want to inspect available Intrinsics # -so not to go always through SIMDE emulation -# -so to avoid AVX512 instructions generation by gcc +# -so to avoid AVX512 instructions generation by gcc if(EXISTS "/proc/cpuinfo" AND NOT CROSS_COMPILE) file(STRINGS "/proc/cpuinfo" CPUFLAGS REGEX flags LIMIT_COUNT 1) else() message(WARNING "did not find /proc/cpuinfo -- not setting any x86-specific compilation variables") endif() - + eval_boolean(AUTODETECT_AVX512 DEFINED CPUFLAGS AND CPUFLAGS MATCHES "avx512") add_boolean_option(AVX512 ${AUTODETECT_AVX512} "Whether AVX512 intrinsics is available on the host processor" ON) @@ -641,7 +641,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) +target_link_libraries(UTIL PUBLIC ${T_LIB} pthread LOG thread-pool utils barrier actor) target_link_libraries(UTIL PUBLIC instrumentation) set(SECURITY_SRC @@ -683,7 +683,7 @@ set(SCHED_NR_SRC ${OPENAIR1_DIR}/SCHED_NR/phy_frame_config_nr.c ) add_library(SCHED_NR_LIB ${SCHED_NR_SRC}) -target_link_libraries(SCHED_NR_LIB PRIVATE asn1_nr_rrc_hdrs) +target_link_libraries(SCHED_NR_LIB PRIVATE asn1_nr_rrc_hdrs UTIL) set(SCHED_SRC_RU ${OPENAIR1_DIR}/SCHED/ru_procedures.c @@ -701,7 +701,7 @@ set(SCHED_SRC_UE ${OPENAIR1_DIR}/SCHED_UE/srs_pc.c ) add_library(SCHED_UE_LIB ${SCHED_SRC_UE}) -target_link_libraries(SCHED_UE_LIB PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs) +target_link_libraries(SCHED_UE_LIB PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs UTIL) set(SCHED_SRC_NR_UE ${OPENAIR1_DIR}/SCHED_NR_UE/phy_procedures_nr_ue.c @@ -764,7 +764,7 @@ set(NFAPI_USER_SRC ${NFAPI_USER_DIR}/gnb_ind_vars.c ) add_library(NFAPI_USER_LIB ${NFAPI_USER_SRC}) -target_link_libraries(NFAPI_USER_LIB PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs) +target_link_libraries(NFAPI_USER_LIB PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs UTIL) include_directories(${NFAPI_USER_DIR}) # Layer 1 @@ -1124,7 +1124,7 @@ set(PHY_SRC_UE ${PHY_SRC_UE} ${OPENAIR1_DIR}/PHY/LTE_UE_TRANSPORT/dlsch_llr_comp add_library(PHY_COMMON ${PHY_SRC_COMMON}) target_link_libraries(PHY_COMMON PRIVATE shlib_loader) add_dependencies(PHY_COMMON dfts) -target_link_libraries(PHY_COMMON PRIVATE asn1_lte_rrc_hdrs) +target_link_libraries(PHY_COMMON PRIVATE asn1_lte_rrc_hdrs PUBLIC UTIL) add_library(PHY ${PHY_SRC}) target_link_libraries(PHY PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs) @@ -1133,7 +1133,7 @@ pkg_check_modules(blas REQUIRED blas) pkg_check_modules(lapacke REQUIRED lapacke) add_library(PHY_UE ${PHY_SRC_UE}) -target_link_libraries(PHY_UE PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs) +target_link_libraries(PHY_UE PRIVATE asn1_lte_rrc_hdrs asn1_nr_rrc_hdrs UTIL) # RHEL needs also cblas, but Ubuntu does not have it. So `cblas_LIBRARIES` will # be empty for Ubuntu, a no-op pkg_check_modules(cblas cblas) @@ -1141,6 +1141,7 @@ target_link_libraries(PHY_UE PRIVATE ${blas_LIBRARIES} ${cblas_LIBRARIES} ${lapa target_include_directories(PHY_UE PRIVATE ${blas_INCLUDE_DIRS} ${lapacke_INCLUDE_DIRS}) add_library(PHY_NR_COMMON ${PHY_NR_SRC_COMMON}) +target_link_libraries(PHY_NR_COMMON PUBLIC UTIL) add_library(PHY_NR ${PHY_NR_SRC}) target_link_libraries(PHY_NR nr_phy_common nr_common) @@ -1153,7 +1154,7 @@ add_library(PHY_NR_UE ${PHY_NR_UE_SRC}) target_link_libraries(PHY_NR_UE PRIVATE asn1_nr_rrc_hdrs nr_phy_common nr_common UTIL) add_library(PHY_RU ${PHY_SRC_RU}) -target_link_libraries(PHY_RU PRIVATE asn1_lte_rrc_hdrs) +target_link_libraries(PHY_RU PRIVATE asn1_lte_rrc_hdrs UTIL) #Library for mex functions #########################3 @@ -1222,7 +1223,7 @@ set(NR_SDAP_SRC ${OPENAIR2_DIR}/SDAP/nr_sdap/nr_sdap.c ${OPENAIR2_DIR}/SDAP/nr_sdap/nr_sdap_entity.c ) - + set(L2_SRC ${PDCP_DIR}/pdcp.c ${PDCP_DIR}/pdcp_fifo.c @@ -2070,7 +2071,7 @@ if(E2_AGENT) target_link_libraries(nr-cuup PRIVATE e2_agent e2_agent_arg e2_ran_func_cuup) target_compile_definitions(nr-cuup PRIVATE ${E2AP_VERSION} ${KPM_VERSION} E2_AGENT) endif() - + # nr-uesoftmodem is UE implementation ####################################### diff --git a/common/utils/CMakeLists.txt b/common/utils/CMakeLists.txt index 7cf067dd14..ccb15bfd47 100644 --- a/common/utils/CMakeLists.txt +++ b/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) 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/barrier/CMakeLists.txt b/common/utils/barrier/CMakeLists.txt new file mode 100644 index 0000000000..7957f3e1cf --- /dev/null +++ b/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() diff --git a/common/utils/barrier/barrier.c b/common/utils/barrier/barrier.c new file mode 100644 index 0000000000..574d6be4a3 --- /dev/null +++ b/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); +} diff --git a/common/utils/barrier/barrier.h b/common/utils/barrier/barrier.h new file mode 100644 index 0000000000..e27b56e766 --- /dev/null +++ b/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 +#include + +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 diff --git a/common/utils/barrier/tests/CMakeLists.txt b/common/utils/barrier/tests/CMakeLists.txt new file mode 100644 index 0000000000..bbe774b596 --- /dev/null +++ b/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) diff --git a/common/utils/barrier/tests/test_barrier.cpp b/common/utils/barrier/tests/test_barrier.cpp new file mode 100644 index 0000000000..985b18c298 --- /dev/null +++ b/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 +#include +#include +#include +#include + +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(); +} diff --git a/common/utils/nr/CMakeLists.txt b/common/utils/nr/CMakeLists.txt index 979958d877..c99b9a06a7 100644 --- a/common/utils/nr/CMakeLists.txt +++ b/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() diff --git a/common/utils/telnetsrv/CMakeLists.txt b/common/utils/telnetsrv/CMakeLists.txt index 959812d73a..a7da081f09 100644 --- a/common/utils/telnetsrv/CMakeLists.txt +++ b/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 ../../.. ) - 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/doc/nr-ue-design.md b/doc/nr-ue-design.md index 7ee42335d6..f1cef858e1 100644 --- a/doc/nr-ue-design.md +++ b/doc/nr-ue-design.md @@ -13,13 +13,17 @@ The `UE_thread` function in `nr-ue.c` is the main top level thread that interact The UE exits when at any point in operation it gets out of synchronization. When the command line option `--non-stop` is used, the UE goes to 'Initial Synchronization' mode when it loses synchronization with gNB. However, this feature is not fully implemented and it is a work in progress at the time of writing this documentation. This will be the default behavior (not a command line option) when the feature is fully implemented. +UE uses actors which are threads dedicated to particular activity. Sync Actor handles initial sync. DL Actors handle DLSCH PHY procedures. UL procedures are processed directly on the threadpool + +![design](nr-ue-threads.svg) + ## Initial Synchronization Block ```mermaid graph TD start(Start) -->|UE_thread| rxRu["RU read
(Reads two frames)"] - rxRu --> |Tpool thread| sync["SSB detection
(PSS & SSS detection
PBCH decoding
SIB decoding)"] + rxRu --> |Sync Actor| sync["SSB detection
(PSS & SSS detection
PBCH decoding
SIB decoding)"] rxRu --> |UE_thread| rxRuDummy["RU read
(Dummy read till sync detection to avoid buffer overflow at radio)"] - sync --> |Tpool thread| frameSync["Frame synchronization
(Shift received samples to align with frame)"] + sync --> |UE_thread| frameSync["Frame synchronization
(Shift received samples to align with frame)"] rxRuDummy --> |UE_thread| frameSync ``` ## Regular Slot Processing @@ -27,9 +31,8 @@ graph TD graph TD sync["Frame synchronization
(Shift received samples to align with frame)"] -->|UE_thread| hw_read["RU read (slot n)"] hw_read --> |UE_thread| rxPreProc["PBCH & PDCCH decoding (slot n)"] - hw_read --> |Tpool thread| txProc["Tx processing (slot n+3)
PUSCH encode
PUCCH encode (wait for DLSCH in slot n+3-k1 to finish)
RU write"] - rxPreProc --> |Tpool thread| rxProc["PDSCH decoding (slot n)"] + hw_read --> |UL Actor| txProc["Tx processing (slot n+3)
PUSCH encode
PUCCH encode (wait for DLSCH in slot n+3-k1 to finish)
RU write"] + rxPreProc --> |DL Actor| rxProc["PDSCH decoding (slot n)"] rxPreProc --> |UE_thread| join(Merge) - txProc --> |Tpool thread| join join --> |Go to next slot
UE_thread| hw_read ``` diff --git a/doc/nr-ue-threads.svg b/doc/nr-ue-threads.svg new file mode 100644 index 0000000000..0ae01ea8aa --- /dev/null +++ b/doc/nr-ue-threads.svg @@ -0,0 +1 @@ + diff --git a/executables/nr-ue.c b/executables/nr-ue.c index c2b733030d..0aa966340b 100644 --- a/executables/nr-ue.c +++ b/executables/nr-ue.c @@ -100,6 +100,11 @@ static void *NRUE_phy_stub_standalone_pnf_task(void *arg); +static void start_process_slot_tx(void* arg) { + notifiedFIFO_elt_t *newTx = arg; + pushTpool(&(get_nrUE_params()->Tpool), newTx); +} + static size_t dump_L1_UE_meas_stats(PHY_VARS_NR_UE *ue, char *output, size_t max_len) { const char *begin = output; @@ -488,56 +493,10 @@ void processSlotTX(void *arg) nr_phy_data_tx_t phy_data = {0}; bool sl_tx_action = false; - // Force sequential execution, even if we launch in // for all slots - // at least ULstatus variable is a pure race condition that is quickly detected by assert() in the code because one thread sets it - // to active, so the other thread try to steal&run the ul work - if (rxtxD->stream_status == STREAM_STATUS_SYNCED) { - notifiedFIFO_elt_t *res = pullNotifiedFIFO(UE->tx_resume_ind_fifo + proc->nr_slot_tx); - delNotifiedFIFO_elt(res); - } - if (UE->if_inst) UE->if_inst->slot_indication(UE->Mod_id); - LOG_D(PHY, - "SlotTx %d.%d => slot type %d, wait: %d \n", - proc->frame_tx, - proc->nr_slot_tx, - proc->tx_slot_type, - rxtxD->tx_wait_for_dlsch); if (proc->tx_slot_type == NR_UPLINK_SLOT || proc->tx_slot_type == NR_MIXED_SLOT) { - if (rxtxD->tx_wait_for_dlsch) - LOG_D(PHY, "enter wait for tx, slot %d, nb events to wait %d; ", proc->nr_slot_tx, rxtxD->tx_wait_for_dlsch); - // wait for rx slots to send indication (if any) that DLSCH decoding is finished - for(int i=0; i < rxtxD->tx_wait_for_dlsch; i++) { - notifiedFIFO_elt_t *res = pullNotifiedFIFO(UE->tx_resume_ind_fifo + proc->nr_slot_tx); - delNotifiedFIFO_elt(res); - } - LOG_D(PHY, "completed wait for tx, slot %d\n", proc->nr_slot_tx); - - /* - This herafter code is costing some perfomance for a check that should be useless - But, we face today several bugs arround the matching between events in UE->tx_resume_ind_fifo[slot] - and the corresponding tx_wait_for_dlsch[slot] - The algorithm is we accumlate the actions that should end before processing a tx slot in tx_wait_for_dlsch[slot] - later, other threads push events in UE->tx_resume_ind_fifo[slot] - so, the tx encoding starts only when related actions are done (mainly DLSCH ACK/NACK to encode PUCCH) - if there is a bug that misses to send a event in UE->tx_resume_ind_fifo[slot], the process hangs, we detect the issue - if there is a bug that makes a extra event in UE->tx_resume_ind_fifo[slot], and if we drop the hereafter check - the system runs with random race conditions, very hard to debug - - Likely we should later remove completly UE->tx_resume_ind_fifo with notifications, - instead, - we may run in place the processSlotTX() when the conditions are met (when a decreasing tx_wait_for_dlsch[slot] will become 0) - It will remove the condition signals (for a thread safe semaphore or counter) and make the system simpler - */ - notifiedFIFO_elt_t *res = pollNotifiedFIFO(UE->tx_resume_ind_fifo + proc->nr_slot_tx); - if (res) - LOG_E(NR_PHY, - "Internal error: extra event on Tx waiting queue for slot %d, event comes from rx slot %d\n", - proc->nr_slot_tx, - *(int *)NotifiedFifoData(res)); - if (UE->sl_mode == 2 && proc->tx_slot_type == NR_SIDELINK_SLOT) { // trigger L2 to run ue_sidelink_scheduler thru IF module if (UE->if_inst != NULL && UE->if_inst->sl_indication != NULL) { @@ -586,13 +545,10 @@ void processSlotTX(void *arg) } } - notifiedFIFO_elt_t *newElt = newNotifiedFIFO_elt(sizeof(int), 0, NULL, NULL); - int *msgData = (int *)NotifiedFifoData(newElt); int slots_per_frame = (UE->sl_mode == 2) ? UE->SL_UE_PHY_PARAMS.sl_frame_params.slots_per_frame : UE->frame_parms.slots_per_frame; - int newslot = (proc->nr_slot_tx + 1) % slots_per_frame; - *msgData = newslot; - pushNotifiedFIFO(UE->tx_resume_ind_fifo + newslot, newElt); + int next_slot = (proc->nr_slot_tx + 1) % slots_per_frame; + dynamic_barrier_join(&UE->process_slot_tx_barriers[next_slot]); RU_write(rxtxD, sl_tx_action); TracyCZoneEnd(ctx); } @@ -849,7 +805,7 @@ void *UE_thread(void *arg) int num_ind_fifo = nb_slot_frame; for(int i = 0; i < num_ind_fifo; i++) { - initNotifiedFIFO(UE->tx_resume_ind_fifo + i); + dynamic_barrier_init(&UE->process_slot_tx_barriers[i]); } int shiftForNextFrame = 0; int intialSyncOffset = 0; @@ -924,7 +880,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; @@ -964,9 +920,7 @@ void *UE_thread(void *arg) // We have resynchronized, maybe after RF loss so we need to purge any existing context memset(tx_wait_for_dlsch, 0, sizeof(tx_wait_for_dlsch)); for (int i = 0; i < num_ind_fifo; i++) { - notifiedFIFO_elt_t *res; - while ((res = pollNotifiedFIFO(UE->tx_resume_ind_fifo + i))) - delNotifiedFIFO_elt(res); + dynamic_barrier_reset(&UE->process_slot_tx_barriers[i]); } continue; } @@ -1055,7 +1009,7 @@ void *UE_thread(void *arg) int ret = UE_dl_preprocessing(UE, &curMsgRx->proc, tx_wait_for_dlsch, &curMsgRx->phy_data, &stats_printed); if (ret != INT_MAX) shiftForNextFrame = ret; - pushTpool(&(get_nrUE_params()->Tpool), newRx); + pushNotifiedFIFO(&UE->dl_actors[curMsg.proc.nr_slot_rx % NUM_DL_ACTORS].fifo, newRx); // Start TX slot processing here. It runs in parallel with RX slot processing // in current code, DURATION_RX_TO_TX constant is the limit to get UL data to encode from a RX slot @@ -1065,11 +1019,16 @@ void *UE_thread(void *arg) curMsgTx->writeBlockSize = writeBlockSize; curMsgTx->proc.timestamp_tx = writeTimestamp; curMsgTx->UE = UE; - curMsgTx->tx_wait_for_dlsch = tx_wait_for_dlsch[curMsgTx->proc.nr_slot_tx]; curMsgTx->stream_status = stream_status; + + int sync_to_previous_thread = stream_status == STREAM_STATUS_SYNCED ? 1 : 0; + int slot = curMsgTx->proc.nr_slot_tx; + dynamic_barrier_update(&UE->process_slot_tx_barriers[slot], + tx_wait_for_dlsch[slot] + sync_to_previous_thread, + start_process_slot_tx, + newTx); stream_status = STREAM_STATUS_SYNCED; - tx_wait_for_dlsch[curMsgTx->proc.nr_slot_tx] = 0; - pushTpool(&(get_nrUE_params()->Tpool), newTx); + tx_wait_for_dlsch[slot] = 0; } return NULL; diff --git a/executables/nr-uesoftmodem.c b/executables/nr-uesoftmodem.c index 8a9f0c626f..07d3e532f5 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,10 @@ 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); + for (int i = 0; i < NUM_DL_ACTORS; i++) { + init_actor(&UE[CC_id]->dl_actors[i], "DL_", -1); + } init_nr_ue_vars(UE[CC_id], inst); if (UE[CC_id]->sl_mode) { diff --git a/openair1/PHY/INIT/phy_init.h b/openair1/PHY/INIT/phy_init.h index daf6eafa42..6b278f899f 100644 --- a/openair1/PHY/INIT/phy_init.h +++ b/openair1/PHY/INIT/phy_init.h @@ -25,7 +25,6 @@ #include "PHY/defs_eNB.h" #include "PHY/defs_UE.h" #include "PHY/defs_gNB.h" -#include "PHY/defs_nr_UE.h" #include "LTE_SystemInformationBlockType2.h" //#include "RadioResourceConfigCommonSIB.h" diff --git a/openair1/PHY/TOOLS/CMakeLists.txt b/openair1/PHY/TOOLS/CMakeLists.txt index ada3d8c863..574899f3e7 100644 --- a/openair1/PHY/TOOLS/CMakeLists.txt +++ b/openair1/PHY/TOOLS/CMakeLists.txt @@ -53,7 +53,7 @@ if(ENABLE_UESCOPE OR ENABLE_ENBSCOPE OR ENABLE_NRSCOPE) target_link_libraries(uescope PUBLIC forms PRIVATE asn1_nr_rrc_hdrs asn1_lte_rrc_hdrs) add_library(nrscope MODULE nr_phy_scope.c) - target_link_libraries(nrscope PUBLIC forms PRIVATE asn1_nr_rrc_hdrs asn1_lte_rrc_hdrs) + target_link_libraries(nrscope PUBLIC forms PRIVATE asn1_nr_rrc_hdrs asn1_lte_rrc_hdrs UTIL) # all libraries should be written to root build dir (default creates the same hierarchie under build as existing for sources) set_target_properties(enbscope uescope nrscope diff --git a/openair1/PHY/defs_nr_UE.h b/openair1/PHY/defs_nr_UE.h index 7178faefd1..182450f0e7 100644 --- a/openair1/PHY/defs_nr_UE.h +++ b/openair1/PHY/defs_nr_UE.h @@ -50,6 +50,8 @@ #include "common_lib.h" #include "fapi_nr_ue_interface.h" #include "assertions.h" +#include "barrier.h" +#include "actor.h" //#include "openair1/SCHED_NR_UE/defs.h" #ifdef MEX @@ -74,16 +76,6 @@ #define openair_free(y,x) free((y)) #define PAGE_SIZE 4096 -#ifdef NR_UNIT_TEST - #define FILE_NAME " " - #define LINE_FILE (0) - #define NR_TST_PHY_PRINTF(...) printf(__VA_ARGS__) -#else - #define FILE_NAME (__FILE__) - #define LINE_FILE (__LINE__) - #define NR_TST_PHY_PRINTF(...) -#endif - #define PAGE_MASK 0xfffff000 #define virt_to_phys(x) (x) #define openair_sched_exit() exit(-1) @@ -91,6 +83,7 @@ #define bzero(s,n) (memset((s),0,(n))) /// suppress compiler warning for unused arguments #define UNUSED(x) (void)x; +#define NUM_DL_ACTORS 4 #include "impl_defs_top.h" #include "impl_defs_nr.h" @@ -530,7 +523,7 @@ typedef struct PHY_VARS_NR_UE_s { void *phy_sim_pdsch_dl_ch_estimates_ext; uint8_t *phy_sim_dlsch_b; - notifiedFIFO_t tx_resume_ind_fifo[NR_MAX_SLOTS_PER_FRAME]; + dynamic_barrier_t process_slot_tx_barriers[NR_MAX_SLOTS_PER_FRAME]; // Gain change required for automation RX gain change int adjust_rxgain; @@ -538,6 +531,8 @@ 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; + Actor_t dl_actors[NUM_DL_ACTORS]; } PHY_VARS_NR_UE; typedef struct { diff --git a/openair1/SCHED_NR/phy_frame_config_nr.c b/openair1/SCHED_NR/phy_frame_config_nr.c index 250dcc2f95..da7cc26e1a 100644 --- a/openair1/SCHED_NR/phy_frame_config_nr.c +++ b/openair1/SCHED_NR/phy_frame_config_nr.c @@ -31,10 +31,19 @@ #include "PHY/defs_nr_common.h" #include "PHY/defs_gNB.h" -#include "PHY/defs_nr_UE.h" #include "SCHED_NR/phy_frame_config_nr.h" #include "common/utils/nr/nr_common.h" +#ifdef NR_UNIT_TEST + #define FILE_NAME " " + #define LINE_FILE (0) + #define NR_TST_PHY_PRINTF(...) printf(__VA_ARGS__) +#else + #define FILE_NAME (__FILE__) + #define LINE_FILE (__LINE__) + #define NR_TST_PHY_PRINTF(...) +#endif + /******************************************************************* * * NAME : set_tdd_configuration diff --git a/openair1/SCHED_NR_UE/phy_procedures_nr_ue.c b/openair1/SCHED_NR_UE/phy_procedures_nr_ue.c index a8cb33421b..aab41da209 100644 --- a/openair1/SCHED_NR_UE/phy_procedures_nr_ue.c +++ b/openair1/SCHED_NR_UE/phy_procedures_nr_ue.c @@ -635,18 +635,6 @@ static int nr_ue_pdsch_procedures(PHY_VARS_NR_UE *ue, return 0; } -// This function release the Tx working thread for one pending information, like dlsch ACK/NACK -static void send_dl_done_to_tx_thread(notifiedFIFO_t *nf, int rx_slot) -{ - if (nf) { - notifiedFIFO_elt_t *newElt = newNotifiedFIFO_elt(sizeof(int), 0, NULL, NULL); - // We put rx slot only for tracing purpose - int *msgData = (int *) NotifiedFifoData(newElt); - *msgData = rx_slot; - pushNotifiedFIFO(nf, newElt); - } -} - static uint32_t compute_csi_rm_unav_res(fapi_nr_dl_config_dlsch_pdu_rel15_t *dlsch_config) { uint32_t unav_res = 0; @@ -740,7 +728,7 @@ static bool nr_ue_dlsch_procedures(PHY_VARS_NR_UE *ue, // don't wait anymore LOG_E(NR_PHY, "Internal error nr_ue_dlsch_procedure() called but no active cw on slot %d, harq %d\n", nr_slot_rx, harq_pid); const int ack_nack_slot = (proc->nr_slot_rx + dlsch[0].dlsch_config.k1_feedback) % ue->frame_parms.slots_per_frame; - send_dl_done_to_tx_thread(ue->tx_resume_ind_fifo + ack_nack_slot, proc->nr_slot_rx); + dynamic_barrier_join(&ue->process_slot_tx_barriers[ack_nack_slot]); return false; } @@ -876,7 +864,7 @@ static bool nr_ue_dlsch_procedures(PHY_VARS_NR_UE *ue, // DLSCH decoding finished! don't wait anymore in Tx process, we know if we should answer ACK/NACK PUCCH if (dlsch[0].rnti_type == TYPE_C_RNTI_) { const int ack_nack_slot = (proc->nr_slot_rx + dlsch[0].dlsch_config.k1_feedback) % ue->frame_parms.slots_per_frame; - send_dl_done_to_tx_thread(ue->tx_resume_ind_fifo + ack_nack_slot, proc->nr_slot_rx); + dynamic_barrier_join(&ue->process_slot_tx_barriers[ack_nack_slot]); } if (ue->phy_sim_dlsch_b) @@ -1158,9 +1146,8 @@ void pdsch_processing(PHY_VARS_NR_UE *ue, const UE_nr_rxtx_proc_t *proc, nr_phy_ nr_ue_dlsch_procedures(ue, proc, dlsch, llr, G); else { LOG_E(NR_PHY, "Demodulation impossible, internal error\n"); - send_dl_done_to_tx_thread( - ue->tx_resume_ind_fifo + (proc->nr_slot_rx + dlsch_config->k1_feedback) % ue->frame_parms.slots_per_frame, - proc->nr_slot_rx); + int ack_nack_slot = (proc->nr_slot_rx + dlsch_config->k1_feedback) % ue->frame_parms.slots_per_frame; + dynamic_barrier_join(&ue->process_slot_tx_barriers[ack_nack_slot]); LOG_W(NR_PHY, "nr_ue_pdsch_procedures failed in slot %d\n", proc->nr_slot_rx); }