zzuummaa/concurrent_lib
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Remove remainder of PIP and use library instruction

Browse Source
This commit is contained in:
Фоменко Степан Владимирович
2020-08-28 14:18:56 +03:00
parent 49dfc86d46
commit 756eccc7e8
6 changed files with 57 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
include/piblockingdequeue.h → include/blockingdequeue.h
View File

@@ -1,24 +1,5 @@
/* #ifndef BLOCKINGDEQUEUE_H
PIP - Platform Independent Primitives #define BLOCKINGDEQUEUE_H
Stephan Fomenko
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef PIBLOCKINGDEQUEUE_H
#define PIBLOCKINGDEQUEUE_H
#include <queue> #include <queue>
#include <condition_variable> #include <condition_variable>
@@ -28,14 +9,14 @@
* wait for space to become available in the queue when storing an element. * wait for space to become available in the queue when storing an element.
*/ */
template <typename T, template<typename = T, typename...> class Queue_ = std::deque, typename ConditionVariable_ = std::condition_variable> template <typename T, template<typename = T, typename...> class Queue_ = std::deque, typename ConditionVariable_ = std::condition_variable>
class PIBlockingDequeue { class BlockingDequeue {
public: public:
typedef Queue_<T> QueueType; typedef Queue_<T> QueueType;
/** /**
* @brief Constructor * @brief Constructor
*/ */
explicit PIBlockingDequeue(size_t capacity = SIZE_MAX) explicit BlockingDequeue(size_t capacity = SIZE_MAX)
: cond_var_add(new ConditionVariable_()), cond_var_rem(new ConditionVariable_()), max_size(capacity) { } : cond_var_add(new ConditionVariable_()), cond_var_rem(new ConditionVariable_()), max_size(capacity) { }
/** /**
@@ -43,7 +24,7 @@ public:
*/ */
template<typename Iterable, template<typename Iterable,
typename std::enable_if<!std::is_arithmetic<Iterable>::value, int>::type = 0> typename std::enable_if<!std::is_arithmetic<Iterable>::value, int>::type = 0>
explicit PIBlockingDequeue(const Iterable& other): PIBlockingDequeue() { explicit BlockingDequeue(const Iterable& other): BlockingDequeue() {
mutex.lock(); mutex.lock();
for (const T& t : other) data_queue.push_back(t); for (const T& t : other) data_queue.push_back(t);
mutex.unlock(); mutex.unlock();
@@ -52,7 +33,7 @@ public:
/** /**
* @brief Thread-safe copy constructor. Initialize queue with copy of other queue elements. * @brief Thread-safe copy constructor. Initialize queue with copy of other queue elements.
*/ */
explicit PIBlockingDequeue(PIBlockingDequeue<T>& other): PIBlockingDequeue() { explicit BlockingDequeue(BlockingDequeue<T>& other): BlockingDequeue() {
other.mutex.lock(); other.mutex.lock();
mutex.lock(); mutex.lock();
max_size = other.max_size; max_size = other.max_size;
@@ -61,7 +42,7 @@ public:
other.mutex.unlock(); other.mutex.unlock();
} }
~PIBlockingDequeue() { ~BlockingDequeue() {
delete cond_var_add; delete cond_var_add;
delete cond_var_rem; delete cond_var_rem;
} }
@@ -244,7 +225,7 @@ public:
/** /**
* @brief Removes all available elements from this queue and adds them to other given queue. * @brief Removes all available elements from this queue and adds them to other given queue.
*/ */
size_t drainTo(PIBlockingDequeue<T>& other, size_t maxCount = SIZE_MAX) { size_t drainTo(BlockingDequeue<T>& other, size_t maxCount = SIZE_MAX) {
mutex.lock(); mutex.lock();
other.mutex.lock(); other.mutex.lock();
size_t count = maxCount > data_queue.size() ? data_queue.size() : maxCount; size_t count = maxCount > data_queue.size() ? data_queue.size() : maxCount;
@@ -269,4 +250,4 @@ protected:
}; };
#endif // PIBLOCKINGDEQUEUE_H #endif // BLOCKINGDEQUEUE_H

45
include/piexecutor.h → include/executor.h
View File

@@ -1,26 +1,7 @@
/* #ifndef EXECUTOR_H
PIP - Platform Independent Primitives #define EXECUTOR_H
Stephan Fomenko
This program is free software: you can redistribute it and/or modify #include "blockingdequeue.h"
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef PIEXECUTOR_H
#define PIEXECUTOR_H
#include "piblockingdequeue.h"
#include <atomic> #include <atomic>
#include <future> #include <future>
@@ -62,8 +43,8 @@ public:
FunctionWrapper& operator=(const FunctionWrapper&) = delete; FunctionWrapper& operator=(const FunctionWrapper&) = delete;
}; };
template <typename Thread_ = std::thread, typename Dequeue_ = PIBlockingDequeue<FunctionWrapper>> template <typename Thread_ = std::thread, typename Dequeue_ = BlockingDequeue<FunctionWrapper>>
class PIThreadPoolExecutorTemplate { class ThreadPoolExecutorTemplate {
protected: protected:
enum thread_command { enum thread_command {
run, run,
@@ -72,9 +53,9 @@ protected:
}; };
public: public:
explicit PIThreadPoolExecutorTemplate(size_t corePoolSize = 1) : thread_command_(thread_command::run) { makePool(corePoolSize); } explicit ThreadPoolExecutorTemplate(size_t corePoolSize = 1) : thread_command_(thread_command::run) { makePool(corePoolSize); }
virtual ~PIThreadPoolExecutorTemplate() { virtual ~ThreadPoolExecutorTemplate() {
shutdownNow(); shutdownNow();
awaitTermination(1000); awaitTermination(1000);
while (threadPool.size() > 0) { while (threadPool.size() > 0) {
@@ -139,7 +120,7 @@ protected:
std::vector<Thread_*> threadPool; std::vector<Thread_*> threadPool;
template<typename Function> template<typename Function>
PIThreadPoolExecutorTemplate(size_t corePoolSize, Function&& onBeforeStart) : thread_command_(thread_command::run) { ThreadPoolExecutorTemplate(size_t corePoolSize, Function&& onBeforeStart) : thread_command_(thread_command::run) {
makePool(corePoolSize, std::forward<Function>(onBeforeStart)); makePool(corePoolSize, std::forward<Function>(onBeforeStart));
} }
@@ -159,7 +140,7 @@ protected:
} }
}; };
typedef PIThreadPoolExecutorTemplate<> PIThreadPoolExecutor; typedef ThreadPoolExecutorTemplate<> ThreadPoolExecutor;
#ifdef DOXYGEN #ifdef DOXYGEN
/** /**
@@ -167,11 +148,11 @@ typedef PIThreadPoolExecutorTemplate<> PIThreadPoolExecutor;
* numbers of asynchronous tasks, due to reduced per-task invocation overhead, and they provide a means of bounding and * numbers of asynchronous tasks, due to reduced per-task invocation overhead, and they provide a means of bounding and
* managing the resources, including threads, consumed when executing a collection of tasks. * managing the resources, including threads, consumed when executing a collection of tasks.
*/ */
class PIThreadPoolExecutor { class ThreadPoolExecutor {
public: public:
explicit PIThreadPoolExecutor(size_t corePoolSize); explicit ThreadPoolExecutor(size_t corePoolSize);
virtual ~PIThreadPoolExecutor(); virtual ~ThreadPoolExecutor();
/** /**
* @brief Submits a Runnable task for execution and returns a Future representing that task. The Future's get method * @brief Submits a Runnable task for execution and returns a Future representing that task. The Future's get method
@@ -211,4 +192,4 @@ public:
}; };
#endif //DOXYGEN #endif //DOXYGEN
#endif //PIEXECUTOR_H #endif //EXECUTOR_H

8
readme.md
View File

@@ -33,6 +33,14 @@ cmake -DCMAKE_BUILD_TYPE=Release ..
cmake --build . --target install cmake --build . --target install
``` ```
4 Use in your project cmake files:
```cmake
find_package(Concurrent REQUIRED)
add_executable(your_app ${YOUR_SOURCES})
target_link_libraries(your_app concurrent)
```
## Build tests ## Build tests
1 Clone GTest: 1 Clone GTest:

18
test/src/BlockingDequeueUnitTest.cpp
View File

@@ -1,7 +1,7 @@
#include "gtest/gtest.h" #include "gtest/gtest.h"
#include "gmock/gmock.h" #include "gmock/gmock.h"
#include "testutil.h" #include "testutil.h"
#include "piblockingdequeue.h" #include "blockingdequeue.h"
using ::testing::_; using ::testing::_;
using ::testing::Return; using ::testing::Return;
@@ -77,15 +77,15 @@ public:
template<typename T> template<typename T>
class MockDeque: public NiceMock<MockDequeBase<T>> {}; class MockDeque: public NiceMock<MockDequeBase<T>> {};
class PIBlockingDequeuePrepare: public PIBlockingDequeue<QueueElement, MockDeque, NiceMock<MockConditionVar>> { class BlockingDequeuePrepare: public BlockingDequeue<QueueElement, MockDeque, NiceMock<MockConditionVar>> {
public: public:
typedef PIBlockingDequeue<QueueElement, MockDeque, NiceMock<MockConditionVar>> SuperClass; typedef BlockingDequeue<QueueElement, MockDeque, NiceMock<MockConditionVar>> SuperClass;
explicit PIBlockingDequeuePrepare(size_t capacity = SIZE_MAX): SuperClass(capacity) { } explicit BlockingDequeuePrepare(size_t capacity = SIZE_MAX): SuperClass(capacity) { }
template<typename Iterable, template<typename Iterable,
typename std::enable_if<!std::is_arithmetic<Iterable>::value, int>::type = 0> typename std::enable_if<!std::is_arithmetic<Iterable>::value, int>::type = 0>
explicit PIBlockingDequeuePrepare(const Iterable& other): SuperClass(other) { } explicit BlockingDequeuePrepare(const Iterable& other): SuperClass(other) { }
MockConditionVar* getCondVarAdd() { return this->cond_var_add; } MockConditionVar* getCondVarAdd() { return this->cond_var_add; }
MockConditionVar* getCondVarRem() { return this->cond_var_rem; } MockConditionVar* getCondVarRem() { return this->cond_var_rem; }
@@ -97,7 +97,7 @@ class BlockingDequeueUnitTest: public ::testing::Test {
public: public:
int timeout = 100; int timeout = 100;
size_t capacity; size_t capacity;
PIBlockingDequeuePrepare dequeue; BlockingDequeuePrepare dequeue;
QueueElement element; QueueElement element;
BlockingDequeueUnitTest(): capacity(1), dequeue(capacity), element(11) {} BlockingDequeueUnitTest(): capacity(1), dequeue(capacity), element(11) {}
@@ -108,12 +108,12 @@ public:
}; };
TEST_F(BlockingDequeueUnitTest, construct_default_is_max_size_eq_size_max) { TEST_F(BlockingDequeueUnitTest, construct_default_is_max_size_eq_size_max) {
PIBlockingDequeuePrepare dequeue; BlockingDequeuePrepare dequeue;
ASSERT_EQ(dequeue.getMaxSize(), SIZE_MAX); ASSERT_EQ(dequeue.getMaxSize(), SIZE_MAX);
} }
TEST_F(BlockingDequeueUnitTest, construct_from_constant_is_max_size_eq_capacity) { TEST_F(BlockingDequeueUnitTest, construct_from_constant_is_max_size_eq_capacity) {
PIBlockingDequeuePrepare dequeue(2); BlockingDequeuePrepare dequeue(2);
ASSERT_EQ(dequeue.getMaxSize(), 2); ASSERT_EQ(dequeue.getMaxSize(), 2);
} }
@@ -125,7 +125,7 @@ TEST_F(BlockingDequeueUnitTest, construct_from_iterable) {
std::vector<QueueElement> iterable; std::vector<QueueElement> iterable;
iterable.emplace_back(11); iterable.emplace_back(11);
iterable.emplace_back(22); iterable.emplace_back(22);
PIBlockingDequeuePrepare dequeue(iterable); BlockingDequeuePrepare dequeue(iterable);
} }
void BlockingDequeueUnitTest::put_is_wait_predicate(bool isCapacityReach) { void BlockingDequeueUnitTest::put_is_wait_predicate(bool isCapacityReach) {

10
test/src/ExecutorIntegrationTest.cpp
View File

@@ -1,6 +1,6 @@
#include "gtest/gtest.h" #include "gtest/gtest.h"
#include "testutil.h" #include "testutil.h"
#include "piexecutor.h" #include "executor.h"
using namespace std; using namespace std;
using namespace chrono; using namespace chrono;
@@ -8,7 +8,7 @@ using namespace chrono;
TEST(ExcutorIntegrationTest, execute_is_runnable_invoke) { TEST(ExcutorIntegrationTest, execute_is_runnable_invoke) {
std::mutex m; std::mutex m;
int invokedRunnables = 0; int invokedRunnables = 0;
PIThreadPoolExecutor executorService(1); ThreadPoolExecutor executorService(1);
executorService.execute([&]() { executorService.execute([&]() {
m.lock(); m.lock();
invokedRunnables++; invokedRunnables++;
@@ -22,7 +22,7 @@ TEST(ExcutorIntegrationTest, execute_is_runnable_invoke) {
TEST(ExcutorIntegrationTest, execute_is_not_execute_after_shutdown) { TEST(ExcutorIntegrationTest, execute_is_not_execute_after_shutdown) {
volatile bool isRunnableInvoke = false; volatile bool isRunnableInvoke = false;
PIThreadPoolExecutor executorService(1); ThreadPoolExecutor executorService(1);
executorService.shutdown(); executorService.shutdown();
executorService.execute([&]() { executorService.execute([&]() {
isRunnableInvoke = true; isRunnableInvoke = true;
@@ -33,7 +33,7 @@ TEST(ExcutorIntegrationTest, execute_is_not_execute_after_shutdown) {
TEST(ExcutorIntegrationTest, execute_is_execute_before_shutdown) { TEST(ExcutorIntegrationTest, execute_is_execute_before_shutdown) {
volatile bool isRunnableInvoke = false; volatile bool isRunnableInvoke = false;
PIThreadPoolExecutor executorService(1); ThreadPoolExecutor executorService(1);
executorService.execute([&]() { executorService.execute([&]() {
this_thread::sleep_for(milliseconds(WAIT_THREAD_TIME_MS)); this_thread::sleep_for(milliseconds(WAIT_THREAD_TIME_MS));
isRunnableInvoke = true; isRunnableInvoke = true;
@@ -45,7 +45,7 @@ TEST(ExcutorIntegrationTest, execute_is_execute_before_shutdown) {
// FIXME // FIXME
TEST(DISABLED_ExcutorIntegrationTest, execute_is_awaitTermination_wait) { TEST(DISABLED_ExcutorIntegrationTest, execute_is_awaitTermination_wait) {
PIThreadPoolExecutor executorService(1); ThreadPoolExecutor executorService(1);
executorService.execute([&]() { executorService.execute([&]() {
this_thread::sleep_for(milliseconds(2 * WAIT_THREAD_TIME_MS)); this_thread::sleep_for(milliseconds(2 * WAIT_THREAD_TIME_MS));
}); });

26
test/src/ExecutorUnitTest.cpp
View File

@@ -3,7 +3,7 @@
#include "gtest/gtest.h" #include "gtest/gtest.h"
#include "gmock/gmock.h" #include "gmock/gmock.h"
#include "testutil.h" #include "testutil.h"
#include "piexecutor.h" #include "executor.h"
using ::testing::_; using ::testing::_;
using ::testing::SetArgReferee; using ::testing::SetArgReferee;
@@ -40,7 +40,7 @@ public:
MOCK_METHOD0(join, void()); MOCK_METHOD0(join, void());
}; };
class MockDeque : public PIBlockingDequeue<FunctionWrapper> { class MockDeque : public BlockingDequeue<FunctionWrapper> {
public: public:
MOCK_METHOD1(offer, bool(const FunctionWrapper&)); MOCK_METHOD1(offer, bool(const FunctionWrapper&));
MOCK_METHOD0(take, FunctionWrapper()); MOCK_METHOD0(take, FunctionWrapper());
@@ -49,14 +49,14 @@ public:
MOCK_METHOD0(remainingCapacity, size_t()); MOCK_METHOD0(remainingCapacity, size_t());
}; };
typedef PIThreadPoolExecutorTemplate<NiceMock<MockThread>, MockDeque> PIThreadPoolExecutorMoc_t; typedef ThreadPoolExecutorTemplate<NiceMock<MockThread>, MockDeque> ThreadPoolExecutorMoc_t;
class PIThreadPoolExecutorMoc : public PIThreadPoolExecutorMoc_t { class ThreadPoolExecutorMoc : public ThreadPoolExecutorMoc_t {
public: public:
explicit PIThreadPoolExecutorMoc(size_t corePoolSize) : PIThreadPoolExecutorMoc_t(corePoolSize) { } explicit ThreadPoolExecutorMoc(size_t corePoolSize) : ThreadPoolExecutorMoc_t(corePoolSize) { }
template<typename Function> template<typename Function>
explicit PIThreadPoolExecutorMoc(size_t corePoolSize, Function onBeforeStart) : PIThreadPoolExecutorMoc_t(corePoolSize, onBeforeStart) { } explicit ThreadPoolExecutorMoc(size_t corePoolSize, Function onBeforeStart) : ThreadPoolExecutorMoc_t(corePoolSize, onBeforeStart) { }
std::vector<testing::NiceMock<MockThread>*>* getThreadPool() { return &threadPool; } std::vector<testing::NiceMock<MockThread>*>* getThreadPool() { return &threadPool; }
bool isShutdown() { return thread_command_ != thread_command::run; } bool isShutdown() { return thread_command_ != thread_command::run; }
@@ -64,18 +64,18 @@ public:
}; };
TEST(ExecutorUnitTest, is_corePool_created) { TEST(ExecutorUnitTest, is_corePool_created) {
PIThreadPoolExecutorMoc executor(THREAD_COUNT); ThreadPoolExecutorMoc executor(THREAD_COUNT);
ASSERT_EQ(THREAD_COUNT, executor.getThreadPool()->size()); ASSERT_EQ(THREAD_COUNT, executor.getThreadPool()->size());
} }
TEST(ExecutorUnitTest, is_corePool_started) { TEST(ExecutorUnitTest, is_corePool_started) {
PIThreadPoolExecutorMoc executor(THREAD_COUNT); ThreadPoolExecutorMoc executor(THREAD_COUNT);
for (auto* thread : *executor.getThreadPool()) ASSERT_TRUE(thread->is_executed); for (auto* thread : *executor.getThreadPool()) ASSERT_TRUE(thread->is_executed);
} }
TEST(ExecutorUnitTest, submit_is_added_to_taskQueue) { TEST(ExecutorUnitTest, submit_is_added_to_taskQueue) {
VoidFunc voidFunc = [](){}; VoidFunc voidFunc = [](){};
PIThreadPoolExecutorMoc executor(THREAD_COUNT); ThreadPoolExecutorMoc executor(THREAD_COUNT);
// TODO add check of offered // TODO add check of offered
EXPECT_CALL(*executor.getTaskQueue(), offer) EXPECT_CALL(*executor.getTaskQueue(), offer)
.WillOnce(Return(true)); .WillOnce(Return(true));
@@ -84,7 +84,7 @@ TEST(ExecutorUnitTest, submit_is_added_to_taskQueue) {
TEST(ExecutorUnitTest, submit_is_return_valid_future) { TEST(ExecutorUnitTest, submit_is_return_valid_future) {
VoidFunc voidFunc = [](){}; VoidFunc voidFunc = [](){};
PIThreadPoolExecutorMoc executor(THREAD_COUNT); ThreadPoolExecutorMoc executor(THREAD_COUNT);
// TODO add check of offered // TODO add check of offered
EXPECT_CALL(*executor.getTaskQueue(), offer) EXPECT_CALL(*executor.getTaskQueue(), offer)
.WillOnce(Return(true)); .WillOnce(Return(true));
@@ -94,7 +94,7 @@ TEST(ExecutorUnitTest, submit_is_return_valid_future) {
TEST(ExecutorUnitTest, execute_is_added_to_taskQueue) { TEST(ExecutorUnitTest, execute_is_added_to_taskQueue) {
VoidFunc voidFunc = [](){}; VoidFunc voidFunc = [](){};
PIThreadPoolExecutorMoc executor(THREAD_COUNT); ThreadPoolExecutorMoc executor(THREAD_COUNT);
// TODO add check of offered // TODO add check of offered
EXPECT_CALL(*executor.getTaskQueue(), offer) EXPECT_CALL(*executor.getTaskQueue(), offer)
.WillOnce(Return(true)); .WillOnce(Return(true));
@@ -104,7 +104,7 @@ TEST(ExecutorUnitTest, execute_is_added_to_taskQueue) {
// TODO fix // TODO fix
TEST(DISABLED_ExecutorUnitTest, is_corePool_execute_queue_elements) { TEST(DISABLED_ExecutorUnitTest, is_corePool_execute_queue_elements) {
bool is_executed = false; bool is_executed = false;
PIThreadPoolExecutorMoc executor(1); ThreadPoolExecutorMoc executor(1);
EXPECT_EQ(executor.getThreadPool()->size(), 1); EXPECT_EQ(executor.getThreadPool()->size(), 1);
EXPECT_CALL(*executor.getTaskQueue(), poll(Ge(0))) EXPECT_CALL(*executor.getTaskQueue(), poll(Ge(0)))
.WillOnce([&is_executed](int){ .WillOnce([&is_executed](int){
@@ -117,7 +117,7 @@ TEST(DISABLED_ExecutorUnitTest, is_corePool_execute_queue_elements) {
// FIXME // FIXME
TEST(DISABLED_ExecutorUnitTest, shutdown_is_stop_threads) { TEST(DISABLED_ExecutorUnitTest, shutdown_is_stop_threads) {
// Exclude stop calls when executor deleting // Exclude stop calls when executor deleting
auto* executor = new PIThreadPoolExecutorMoc(THREAD_COUNT, [](MockThread* thread){ auto* executor = new ThreadPoolExecutorMoc(THREAD_COUNT, [](MockThread* thread){
testing::Mock::AllowLeak(thread); testing::Mock::AllowLeak(thread);
EXPECT_CALL(*thread, join()) EXPECT_CALL(*thread, join())
.WillOnce(Return()); .WillOnce(Return());