diff --git a/common/utils/CMakeLists.txt b/common/utils/CMakeLists.txt index 7cf067dd14..bce79d082d 100644 --- a/common/utils/CMakeLists.txt +++ b/common/utils/CMakeLists.txt @@ -19,3 +19,4 @@ target_link_libraries(utils PRIVATE ${T_LIB}) if (ENABLE_TESTS) add_subdirectory(hashtable/tests) endif() +add_subdirectory(barrier) 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..88c7d0f318 --- /dev/null +++ b/common/utils/barrier/barrier.c @@ -0,0 +1,65 @@ +/* + * 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_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..33abfcb46c --- /dev/null +++ b/common/utils/barrier/barrier.h @@ -0,0 +1,53 @@ +/* + * 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 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(); +}