Commit ea0ea2ff authored by Bartosz Podrygajlo's avatar Bartosz Podrygajlo

Thread barrier implementation

This commit introduces dynamic_barrier_t. Its a thread barrier that counts joins
but allows callback to be specified later. See std::barrier for a basic barrier
description.
parent 03946cd4
......@@ -19,3 +19,4 @@ target_link_libraries(utils PRIVATE ${T_LIB})
if (ENABLE_TESTS)
add_subdirectory(hashtable/tests)
endif()
add_subdirectory(barrier)
add_library(barrier barrier.c)
target_include_directories(barrier PUBLIC ./)
target_link_libraries(barrier PUBLIC pthread)
if (ENABLE_TESTS)
add_subdirectory(tests)
endif()
/*
* 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);
}
/*
* 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 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
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)
/*
* 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();
}
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment