version 1.22.0

source tree changed
detached PIConsole and PIScreen* in "pip_console" library

lib/concurrent/executor.cpp

@@ -0,0 +1,74 @@
+/*
+	PIP - Platform Independent Primitives
+
+	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/>.
+*/
+
+#include "executor.h"
+
+
+PIThreadPoolExecutor::PIThreadPoolExecutor(size_t corePoolSize, PIBlockingDequeue<std::function<void()>> *taskQueue_) : isShutdown_(false), taskQueue(taskQueue_) {
+	for (size_t i = 0; i < corePoolSize; ++i) {
+		PIThread * thread = new PIThread([&, i](){
+			auto runnable = taskQueue->poll(100, std::function<void()>());
+			if (runnable) {
+				runnable();
+			}
+			if (isShutdown_ && taskQueue->size() == 0) threadPool[i]->stop();
+		});
+		threadPool.push_back(thread);
+		thread->start();
+	}
+}
+
+
+bool PIThreadPoolExecutor::awaitTermination(int timeoutMs) {
+	PITimeMeasurer measurer;
+	for (size_t i = 0; i < threadPool.size(); ++i) {
+		int dif = timeoutMs - (int)measurer.elapsed_m();
+		if (dif < 0) return false;
+		if (!threadPool[i]->waitForFinish(dif)) return false;
+	}
+	return true;
+}
+
+
+void PIThreadPoolExecutor::shutdownNow() {
+	isShutdown_ = true;
+	for (size_t i = 0; i < threadPool.size(); ++i) threadPool[i]->stop();
+}
+
+
+PIThreadPoolExecutor::~PIThreadPoolExecutor() {
+	shutdownNow();
+	while (threadPool.size() > 0) delete threadPool.take_back();
+	delete taskQueue;
+}
+
+
+void PIThreadPoolExecutor::execute(const std::function<void()> &runnable) {
+	if (!isShutdown_) taskQueue->offer(runnable);
+}
+
+
+volatile bool PIThreadPoolExecutor::isShutdown() const {
+	return isShutdown_;
+}
+
+
+void PIThreadPoolExecutor::shutdown() {
+	isShutdown_ = true;
+}

lib/concurrent/piconditionlock.cpp

@@ -0,0 +1,108 @@
+/*
+	PIP - Platform Independent Primitives
+
+	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/>.
+*/
+
+#include "piconditionlock.h"
+#ifdef WINDOWS
+#include "synchapi.h"
+#else
+#include "pthread.h"
+#endif
+
+
+PRIVATE_DEFINITION_START(PIConditionLock)
+#ifdef WINDOWS
+CRITICAL_SECTION
+#else
+pthread_mutex_t
+#endif
+nativeHandle;
+PRIVATE_DEFINITION_END(PIConditionLock)
+
+
+#ifdef WINDOWS
+PIConditionLock::PIConditionLock() {
+	InitializeCriticalSection(&PRIVATE->nativeHandle);
+}
+
+
+PIConditionLock::~PIConditionLock() {
+	DeleteCriticalSection(&PRIVATE->nativeHandle);
+}
+
+
+void PIConditionLock::lock() {
+	EnterCriticalSection(&PRIVATE->nativeHandle);
+}
+
+
+void PIConditionLock::unlock() {
+	LeaveCriticalSection(&PRIVATE->nativeHandle);
+}
+
+
+void *PIConditionLock::handle() {
+	return &PRIVATE->nativeHandle;
+}
+
+
+bool PIConditionLock::tryLock() {
+	return TryEnterCriticalSection(&PRIVATE->nativeHandle) != 0;
+}
+#else
+
+
+PIConditionLock::PIConditionLock() {
+	pthread_mutexattr_t attr;
+	memset(&attr, 0, sizeof(attr));
+	pthread_mutexattr_init(&attr);
+	pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
+	memset(&(PRIVATE->nativeHandle), 0, sizeof(PRIVATE->nativeHandle));
+	pthread_mutex_init(&(PRIVATE->nativeHandle), &attr);
+	pthread_mutexattr_destroy(&attr);
+}
+
+
+PIConditionLock::~PIConditionLock() {
+	pthread_mutex_destroy(&(PRIVATE->nativeHandle));
+}
+
+
+void PIConditionLock::lock() {
+	pthread_mutex_lock(&(PRIVATE->nativeHandle));
+}
+
+
+void PIConditionLock::unlock() {
+	pthread_mutex_unlock(&(PRIVATE->nativeHandle));
+}
+
+
+void *PIConditionLock::handle() {
+	return &PRIVATE->nativeHandle;
+}
+
+
+bool PIConditionLock::tryLock() {
+	return (pthread_mutex_trylock(&(PRIVATE->nativeHandle)) == 0);
+}
+#endif
+
+
+
+

lib/concurrent/piconditionvar.cpp

@@ -0,0 +1,142 @@
+/*
+	PIP - Platform Independent Primitives
+
+	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/>.
+*/
+
+#include "piplatform.h"
+#include "piconditionvar.h"
+#include "pithread.h"
+#include "pitime.h"
+
+#ifdef WINDOWS
+#undef _WIN32_WINNT
+#define _WIN32_WINNT 0x0600
+#include "synchapi.h"
+#include <windef.h>
+#include <winbase.h>
+#endif
+
+
+PRIVATE_DEFINITION_START(PIConditionVariable)
+#ifdef WINDOWS
+CONDITION_VARIABLE nativeHandle;
+#else
+pthread_cond_t nativeHandle;
+PIConditionLock* currentLock;
+#endif
+bool isDestroying;
+PRIVATE_DEFINITION_END(PIConditionVariable)
+
+
+PIConditionVariable::PIConditionVariable() {
+#ifdef WINDOWS
+	InitializeConditionVariable(&PRIVATE->nativeHandle);
+#else
+	PRIVATE->isDestroying = false;
+	PRIVATE->currentLock = nullptr;
+	memset(&(PRIVATE->nativeHandle), 0, sizeof(PRIVATE->nativeHandle));
+	pthread_cond_init(&PRIVATE->nativeHandle, NULL);
+#endif
+}
+
+
+PIConditionVariable::~PIConditionVariable() {
+#ifdef WINDOWS
+#else
+	pthread_cond_destroy(&PRIVATE->nativeHandle);
+#endif
+}
+
+
+void PIConditionVariable::wait(PIConditionLock& lk) {
+#ifdef WINDOWS
+	SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), INFINITE);
+#else
+	pthread_cond_wait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle());
+#endif
+}
+
+
+void PIConditionVariable::wait(PIConditionLock& lk, const std::function<bool()>& condition) {
+	bool isCondition;
+	while (true) {
+		isCondition = condition();
+		if (isCondition) break;
+#ifdef WINDOWS
+		SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), INFINITE);
+#else
+		pthread_cond_wait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle());
+#endif
+		if (PRIVATE->isDestroying) return;
+	}
+}
+
+
+bool PIConditionVariable::waitFor(PIConditionLock &lk, int timeoutMs) {
+	bool isNotTimeout;
+#ifdef WINDOWS
+	isNotTimeout = SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), timeoutMs) != 0;
+#else
+	timespec abstime = {.tv_sec = timeoutMs / 1000, .tv_nsec = timeoutMs * 1000 * 1000};
+	isNotTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0;
+#endif
+	if (PRIVATE->isDestroying) return false;
+	return isNotTimeout;
+}
+
+
+bool PIConditionVariable::waitFor(PIConditionLock& lk, int timeoutMs, const std::function<bool()> &condition) {
+	bool isCondition;
+	PITimeMeasurer measurer;
+	while (true) {
+		isCondition = condition();
+		if (isCondition) break;
+#ifdef WINDOWS
+		WINBOOL isTimeout = SleepConditionVariableCS(
+					&PRIVATE->nativeHandle,
+					(PCRITICAL_SECTION)lk.handle(),
+					timeoutMs - (int)measurer.elapsed_m());
+		if (isTimeout == 0) return false;
+#else
+		int timeoutCurr = timeoutMs - (int)measurer.elapsed_m();
+		timespec abstime = {.tv_sec = timeoutCurr / 1000, .tv_nsec = timeoutCurr * 1000 * 1000};
+		bool isTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0;
+		if (isTimeout) return false;
+#endif
+		if (PRIVATE->isDestroying) return false;
+	}
+	return true;
+}
+
+
+void PIConditionVariable::notifyOne() {
+#ifdef WINDOWS
+	WakeConditionVariable(&PRIVATE->nativeHandle);
+#else
+	pthread_cond_signal(&PRIVATE->nativeHandle);
+#endif
+}
+
+
+void PIConditionVariable::notifyAll() {
+#ifdef WINDOWS
+	WakeAllConditionVariable(&PRIVATE->nativeHandle);
+#else
+	pthread_cond_broadcast(&PRIVATE->nativeHandle);
+#endif
+}
+

lib/concurrent/test/BlockingDequeueUnitTest.cpp

@@ -0,0 +1,238 @@
+#include "gtest/gtest.h"
+#include "piblockingdequeue.h"
+
+class MockConditionVar: public PIConditionVariable {
+public:
+    bool isWaitCalled = false;
+    bool isWaitForCalled = false;
+    bool isTrueCondition = false;
+    int timeout = -1;
+
+    void wait(PIConditionLock& lk) override {
+        isWaitCalled = true;
+    }
+
+    void wait(PIConditionLock& lk, const std::function<bool()>& condition) override {
+        isWaitCalled = true;
+        lk.lock();
+        isTrueCondition = condition();
+        lk.unlock();
+    }
+
+    bool waitFor(PIConditionLock& lk, int timeoutMs) override {
+        isWaitForCalled = true;
+        timeout = timeoutMs;
+        return false;
+    }
+
+    bool waitFor(PIConditionLock& lk, int timeoutMs, const std::function<bool()>& condition) override {
+        isWaitForCalled = true;
+        lk.lock();
+        isTrueCondition = condition();
+        timeout = timeoutMs;
+        lk.unlock();
+        return isTrueCondition;
+    }
+};
+
+TEST(BlockingDequeueUnitTest, put_is_block_when_capacity_reach) {
+    size_t capacity = 0;
+    auto conditionVarAdd = new MockConditionVar();
+    auto conditionVarRem = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVarAdd, conditionVarRem);
+    dequeue.put(11);
+    ASSERT_TRUE(conditionVarRem->isWaitCalled);
+    ASSERT_FALSE(conditionVarRem->isTrueCondition);
+}
+
+TEST(BlockingDequeueUnitTest, offer_timedout_is_false_when_capacity_reach) {
+    size_t capacity = 0;
+    int timeout = 11;
+    auto conditionVarAdd = new MockConditionVar();
+    auto conditionVarRem = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVarAdd, conditionVarRem);
+    ASSERT_FALSE(dequeue.offer(11, timeout));
+}
+
+TEST(BlockingDequeueUnitTest, offer_timedout_is_block_when_capacity_reach) {
+    size_t capacity = 0;
+    int timeout = 11;
+    auto conditionVarAdd = new MockConditionVar();
+    auto conditionVarRem = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVarAdd, conditionVarRem);
+    dequeue.offer(11, timeout);
+    EXPECT_TRUE(conditionVarRem->isWaitForCalled);
+    EXPECT_EQ(timeout, conditionVarRem->timeout);
+    ASSERT_FALSE(conditionVarRem->isTrueCondition);
+}
+
+TEST(BlockingDequeueUnitTest, offer_is_true_before_capacity_reach) {
+    size_t capacity = 1;
+    PIBlockingDequeue<int> dequeue(capacity);
+    ASSERT_TRUE(dequeue.offer(10));
+}
+
+TEST(BlockingDequeueUnitTest, offer_is_false_when_capacity_reach) {
+    size_t capacity = 1;
+    PIBlockingDequeue<int> dequeue(capacity);
+    dequeue.offer(11);
+    ASSERT_FALSE(dequeue.offer(10));
+}
+
+// TODO change take_is_block_when_empty to prevent segfault
+TEST(DISABLED_BlockingDequeueUnitTest, take_is_block_when_empty) {
+    size_t capacity = 1;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+    // May cause segfault because take front of empty queue
+    dequeue.take();
+    EXPECT_TRUE(conditionVar->isWaitCalled);
+    ASSERT_FALSE(conditionVar->isTrueCondition);
+}
+
+TEST(BlockingDequeueUnitTest, take_is_not_block_when_not_empty) {
+    size_t capacity = 1;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+    dequeue.offer(111);
+    dequeue.take();
+
+    EXPECT_TRUE(conditionVar->isWaitCalled);
+    ASSERT_TRUE(conditionVar->isTrueCondition);
+}
+
+TEST(BlockingDequeueUnitTest, take_is_value_eq_to_offer_value) {
+    size_t capacity = 1;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+
+    dequeue.offer(111);
+    ASSERT_EQ(dequeue.take(), 111);
+}
+
+TEST(BlockingDequeueUnitTest, take_is_last) {
+    size_t capacity = 10;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+    EXPECT_TRUE(dequeue.offer(111));
+    EXPECT_TRUE(dequeue.offer(222));
+    ASSERT_EQ(dequeue.take(), 111);
+    ASSERT_EQ(dequeue.take(), 222);
+}
+
+TEST(BlockingDequeueUnitTest, poll_is_block_when_empty) {
+    size_t capacity = 1;
+    int timeout = 11;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+    dequeue.poll(timeout, 111);
+    EXPECT_TRUE(conditionVar->isWaitForCalled);
+    EXPECT_EQ(timeout, conditionVar->timeout);
+    ASSERT_FALSE(conditionVar->isTrueCondition);
+}
+
+TEST(BlockingDequeueUnitTest, poll_is_default_value_when_empty) {
+    size_t capacity = 1;
+    int timeout = 11;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+    ASSERT_EQ(dequeue.poll(timeout, 111), 111);
+}
+
+TEST(BlockingDequeueUnitTest, poll_is_not_block_when_not_empty) {
+    size_t capacity = 1;
+    int timeout = 11;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+    dequeue.offer(111);
+    dequeue.poll(timeout, -1);
+
+    EXPECT_TRUE(conditionVar->isWaitForCalled);
+    ASSERT_TRUE(conditionVar->isTrueCondition);
+}
+
+TEST(BlockingDequeueUnitTest, poll_is_offer_value_when_not_empty) {
+    size_t capacity = 1;
+    int timeout = 11;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+    dequeue.offer(111);
+    ASSERT_EQ(dequeue.poll(timeout, -1), 111);
+}
+
+TEST(BlockingDequeueUnitTest, poll_is_last) {
+    size_t capacity = 10;
+    auto conditionVar = new MockConditionVar();
+    PIBlockingDequeue<int> dequeue(capacity, conditionVar);
+    dequeue.offer(111);
+    dequeue.offer(222);
+    ASSERT_EQ(dequeue.poll(10, -1), 111);
+    ASSERT_EQ(dequeue.poll(10, -1), 222);
+}
+
+TEST(BlockingDequeueUnitTest, capacity_is_eq_constructor_capacity) {
+    size_t capacity = 10;
+    PIBlockingDequeue<int> dequeue(capacity);
+    ASSERT_EQ(dequeue.capacity(), capacity);
+}
+
+TEST(BlockingDequeueUnitTest, remainingCapacity_is_dif_of_capacity_and_size) {
+    size_t capacity = 2;
+    PIBlockingDequeue<int> dequeue(capacity);
+    ASSERT_EQ(dequeue.remainingCapacity(), capacity);
+    dequeue.offer(111);
+    ASSERT_EQ(dequeue.remainingCapacity(), capacity - 1);
+}
+
+TEST(BlockingDequeueUnitTest, remainingCapacity_is_zero_when_capacity_reach) {
+    size_t capacity = 1;
+    PIBlockingDequeue<int> dequeue(capacity);
+    dequeue.offer(111);
+    dequeue.offer(111);
+    ASSERT_EQ(dequeue.remainingCapacity(), 0);
+}
+
+TEST(BlockingDequeueUnitTest, size_is_eq_to_num_of_elements) {
+    size_t capacity = 1;
+    PIBlockingDequeue<int> dequeue(capacity);
+    ASSERT_EQ(dequeue.size(), 0);
+    dequeue.offer(111);
+    ASSERT_EQ(dequeue.size(), 1);
+}
+
+TEST(BlockingDequeueUnitTest, size_is_eq_to_capacity_when_capacity_reach) {
+    size_t capacity = 1;
+    PIBlockingDequeue<int> dequeue(capacity);
+    dequeue.offer(111);
+    dequeue.offer(111);
+    ASSERT_EQ(dequeue.size(), capacity);
+}
+
+TEST(BlockingDequeueUnitTest, drainTo_is_elements_moved) {
+    size_t capacity = 10;
+    PIDeque<int> refDeque;
+    for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10);
+    PIBlockingDequeue<int> blockingDequeue(refDeque);
+    PIDeque<int> deque;
+    blockingDequeue.drainTo(deque);
+    ASSERT_EQ(blockingDequeue.size(), 0);
+    ASSERT_TRUE(deque == refDeque);
+}
+
+TEST(BlockingDequeueUnitTest, drainTo_is_ret_eq_to_size_when_all_moved) {
+    size_t capacity = 10;
+    PIDeque<int> refDeque;
+    for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10);
+    PIBlockingDequeue<int> blockingDequeue(refDeque);
+    PIDeque<int> deque;
+    ASSERT_EQ(blockingDequeue.drainTo(deque), refDeque.size());
+}
+
+TEST(BlockingDequeueUnitTest, drainTo_is_ret_eq_to_maxCount) {
+    size_t capacity = 10;
+    PIDeque<int> refDeque;
+    for (size_t i = 0; i < capacity / 2; ++i) refDeque.push_back(i * 10);
+    PIBlockingDequeue<int> blockingDequeue(refDeque);
+    PIDeque<int> deque;
+    ASSERT_EQ(blockingDequeue.drainTo(deque, refDeque.size() - 1), refDeque.size() - 1);
+}

lib/concurrent/test/ConditionLockIntegrationTest.cpp

@@ -0,0 +1,53 @@
+#include "gtest/gtest.h"
+#include "gmock/gmock.h"
+
+#include "piconditionvar.h"
+#include <pithread.h>
+#include "testutil.h"
+
+class ConditionLock : public ::testing::Test, public TestUtil {
+public:
+	PIConditionLock* m = new PIConditionLock();
+};
+
+TEST_F(ConditionLock, lock_is_protect) {
+    m->lock();
+    bool* isProtect = new bool(true);
+
+    createThread([&](){
+        m->lock();
+        *isProtect = false;
+    });
+	EXPECT_FALSE(thread->waitForFinish(WAIT_THREAD_TIME_MS));
+    ASSERT_TRUE(*isProtect);
+}
+
+TEST_F(ConditionLock, unlock_is_release) {
+    m->lock();
+    bool* isReleased = new bool(false);
+    m->unlock();
+
+    createThread([&](){
+        m->lock();
+        *isReleased = true;
+        m->unlock();
+    });
+    ASSERT_TRUE(*isReleased);
+}
+
+TEST_F(ConditionLock, tryLock_is_false_when_locked) {
+    createThread([&](){
+        m->lock();
+        piMSleep(WAIT_THREAD_TIME_MS);
+    });
+    ASSERT_FALSE(m->tryLock());
+}
+
+TEST_F(ConditionLock, tryLock_is_true_when_unlocked) {
+    ASSERT_TRUE(m->tryLock());
+}
+
+TEST_F(ConditionLock, tryLock_is_recursive_lock_enable) {
+    m->lock();
+    ASSERT_TRUE(m->tryLock());
+}

lib/concurrent/test/ConditionVariableIntegrationTest.cpp

@@ -0,0 +1,200 @@
+#include "gtest/gtest.h"
+#include "piconditionvar.h"
+#include "pithread.h"
+#include "testutil.h"
+
+class ConditionVariable : public ::testing::Test, public TestUtil {
+public:
+    PIConditionLock m;
+    PIConditionVariable* variable;
+
+protected:
+    void SetUp() override {
+        variable = new PIConditionVariable();
+        adapterFunctionDefault = [&](){
+            m.lock();
+            variable->wait(m);
+            m.unlock();
+        };
+    }
+};
+
+TEST_F(ConditionVariable, wait_is_block) {
+    createThread();
+    ASSERT_FALSE(thread->waitForFinish(WAIT_THREAD_TIME_MS));
+}
+
+TEST_F(ConditionVariable, wait_is_block_when_notifyOne_before_wait) {
+    variable->notifyOne();
+    createThread();
+    ASSERT_FALSE(thread->waitForFinish(WAIT_THREAD_TIME_MS));
+}
+
+TEST_F(ConditionVariable, wait_is_block_when_notifyAll_before_wait) {
+    variable->notifyAll();
+    createThread();
+    ASSERT_FALSE(thread->waitForFinish(WAIT_THREAD_TIME_MS));
+}
+
+TEST_F(ConditionVariable, wait_is_unblock_when_notifyOne_after_wait) {
+    createThread();
+    variable->notifyOne();
+    ASSERT_TRUE(thread->waitForFinish(WAIT_THREAD_TIME_MS));
+}
+
+TEST_F(ConditionVariable, wait_is_unblock_when_notifyAll_after_wait) {
+    PIVector<PIThread*> threads;
+
+    for (int i = 0; i < THREAD_COUNT; ++i) {
+        threads.push_back(new PIThread([=](){ adapterFunctionDefault(); }));
+    }
+
+    piForeach(PIThread* thread, threads) thread->startOnce();
+    piMSleep(WAIT_THREAD_TIME_MS * THREAD_COUNT);
+    variable->notifyAll();
+    PITimeMeasurer measurer;
+    piForeach(PIThread* thread, threads) {
+        int timeout = WAIT_THREAD_TIME_MS * THREAD_COUNT - (int)measurer.elapsed_m();
+        thread->waitForFinish(timeout > 0 ? timeout : 0);
+    }
+    for (size_t i = 0; i < threads.size(); ++i) EXPECT_FALSE(threads[i]->isRunning()) << "Thread " << i << " still running";
+    piForeach(PIThread* thread, threads) delete thread;
+}
+
+TEST_F(ConditionVariable, wait_is_one_unblock_when_notifyOne) {
+    PIVector<PIThread*> threads;
+
+    for (int i = 0; i < THREAD_COUNT; ++i) {
+        threads.push_back(new PIThread(adapterFunctionDefault));
+    }
+
+    piForeach(PIThread* thread, threads) thread->startOnce();
+    piMSleep(WAIT_THREAD_TIME_MS * THREAD_COUNT);
+    variable->notifyOne();
+    piMSleep(WAIT_THREAD_TIME_MS * THREAD_COUNT);
+    int runningThreadCount = 0;
+    piForeach(PIThread* thread, threads) if (thread->isRunning()) runningThreadCount++;
+    ASSERT_EQ(runningThreadCount, THREAD_COUNT - 1);
+}
+
+TEST_F(ConditionVariable, wait_is_protected_unblock_when_notifyOne) {
+    createThread([&](){
+        m.lock();
+        variable->wait(m);
+        piMSleep(2 * WAIT_THREAD_TIME_MS);
+        // Missing unlock
+    });
+    variable->notifyOne();
+    msleep(WAIT_THREAD_TIME_MS);
+    ASSERT_FALSE(m.tryLock());
+}
+
+TEST_F(ConditionVariable, wait_condition_is_block) {
+    createThread([&](){
+        m.lock();
+        variable->wait(m, [](){ return false; });
+        m.unlock();
+    });
+    ASSERT_FALSE(thread->waitForFinish(WAIT_THREAD_TIME_MS));
+}
+
+TEST_F(ConditionVariable, wait_condition_is_check_condition_before_block) {
+    bool isConditionChecked = false;
+    createThread([&](){
+        m.lock();
+        variable->wait(m, [&](){
+            isConditionChecked = true;
+            return false;
+        });
+        m.unlock();
+    });
+    m.lock();
+    ASSERT_TRUE(isConditionChecked);
+    m.unlock();
+}
+
+TEST_F(ConditionVariable, wait_condition_is_check_condition_when_notifyOne) {
+    bool isConditionChecked;
+    createThread([&](){
+        m.lock();
+        variable->wait(m, [&](){
+            isConditionChecked = true;
+            return false;
+        });
+        m.unlock();
+    });
+    m.lock();
+    isConditionChecked = false;
+    m.unlock();
+    variable->notifyOne();
+    msleep(threadStartTime + 1);
+    m.lock();
+    ASSERT_TRUE(isConditionChecked);
+    m.unlock();
+}
+
+TEST_F(ConditionVariable, wait_condition_is_unblock_when_condition_and_notifyOne) {
+    bool condition = false;
+    createThread([&](){
+        m.lock();
+        variable->wait(m, [&](){ return condition; });
+        m.unlock();
+    });
+    m.lock();
+    condition = true;
+    m.unlock();
+    variable->notifyOne();
+    ASSERT_TRUE(thread->waitForFinish(WAIT_THREAD_TIME_MS));
+}
+
+TEST_F(ConditionVariable, DISABLED_waitFor_is_block_before_timeout) {
+    createThread([&](){
+        PITimeMeasurer measurer;
+        m.lock();
+        variable->waitFor(m, WAIT_THREAD_TIME_MS * 2);
+        m.unlock();
+        // Not reliable because spurious wakeup may happen
+        ASSERT_GE(measurer.elapsed_m(), WAIT_THREAD_TIME_MS);
+    });
+    EXPECT_TRUE(thread->waitForFinish(WAIT_THREAD_TIME_MS * 3));
+}
+
+TEST_F(ConditionVariable, waitFor_is_unblock_when_timeout) {
+    volatile bool isUnblock = false;
+    createThread([&](){
+        m.lock();
+        variable->waitFor(m, WAIT_THREAD_TIME_MS);
+        isUnblock = true;
+        m.unlock();
+    });
+    // Test failed if suspend forever
+    EXPECT_TRUE(thread->waitForFinish(2 * WAIT_THREAD_TIME_MS));
+    ASSERT_TRUE(isUnblock);
+}
+
+TEST_F(ConditionVariable, waitFor_is_false_when_timeout) {
+    bool waitRet = true;
+    createThread([&](){
+        m.lock();
+        waitRet = variable->waitFor(m, WAIT_THREAD_TIME_MS);
+        m.unlock();
+    });
+    EXPECT_TRUE(thread->waitForFinish(2 * WAIT_THREAD_TIME_MS));
+    ASSERT_FALSE(waitRet);
+}
+
+TEST_F(ConditionVariable, waitFor_is_unblock_when_condition_and_notifyOne) {
+    bool condition = false;
+    createThread([&](){
+        m.lock();
+        variable->waitFor(m, 3 * WAIT_THREAD_TIME_MS, [&](){ return condition; });
+        m.unlock();
+    });
+    EXPECT_TRUE(thread->isRunning());
+    m.lock();
+    condition = true;
+    m.unlock();
+    variable->notifyOne();
+    msleep(WAIT_THREAD_TIME_MS);
+    ASSERT_FALSE(thread->isRunning());
+}

lib/concurrent/test/ExecutorIntegrationTest.cpp

@@ -0,0 +1,54 @@
+#include "gtest/gtest.h"
+#include "executor.h"
+#include "pimutex.h"
+
+const int WAIT_THREAD_TIME_MS = 30;
+
+TEST(ExcutorIntegrationTest, execute_is_runnable_invoke) {
+    PIMutex m;
+    int invokedRunnables = 0;
+    PIThreadPoolExecutor executorService(1);
+    executorService.execute([&]() {
+        m.lock();
+        invokedRunnables++;
+        m.unlock();
+    });
+    piMSleep(WAIT_THREAD_TIME_MS);
+    ASSERT_EQ(invokedRunnables, 1);
+}
+
+TEST(ExcutorIntegrationTest, execute_is_not_execute_after_shutdown) {
+    bool isRunnableInvoke = false;
+    PIThreadPoolExecutor executorService(1);
+    executorService.shutdown();
+    executorService.execute([&]() {
+        isRunnableInvoke = true;
+    });
+    piMSleep(WAIT_THREAD_TIME_MS);
+    ASSERT_FALSE(isRunnableInvoke);
+}
+
+TEST(ExcutorIntegrationTest, execute_is_execute_before_shutdown) {
+    bool isRunnableInvoke = false;
+    PIThreadPoolExecutor executorService(1);
+    executorService.execute([&]() {
+        piMSleep(WAIT_THREAD_TIME_MS);
+        isRunnableInvoke = true;
+    });
+    executorService.shutdown();
+    piMSleep(2 * WAIT_THREAD_TIME_MS);
+    ASSERT_TRUE(isRunnableInvoke);
+}
+
+TEST(ExcutorIntegrationTest, execute_is_awaitTermination_wait) {
+    PIThreadPoolExecutor executorService(1);
+    executorService.execute([&]() {
+        piMSleep(2 * WAIT_THREAD_TIME_MS);
+    });
+    executorService.shutdown();
+    PITimeMeasurer measurer;
+    ASSERT_TRUE(executorService.awaitTermination(3 * WAIT_THREAD_TIME_MS));
+    double waitTime = measurer.elapsed_m();
+    ASSERT_GE(waitTime, WAIT_THREAD_TIME_MS);
+    ASSERT_LE(waitTime, 4 * WAIT_THREAD_TIME_MS);
+}

lib/concurrent/test/testutil.h

@@ -0,0 +1,57 @@
+#ifndef AWRCANFLASHER_TESTUTIL_H
+#define AWRCANFLASHER_TESTUTIL_H
+
+#include "pithread.h"
+
+/**
+ * Minimum wait thread start, switch context or another interthread communication action time. Increase it if tests
+ * write "Start thread timeout reach!" message. You can reduce it if you want increase test performance.
+ */
+const int WAIT_THREAD_TIME_MS = 40;
+
+const int THREAD_COUNT = 5;
+
+class TestUtil: public PIObject {
+PIOBJECT(TestUtil)
+public:
+    double threadStartTime;
+    PIThread* thread = new PIThread();
+    volatile bool isRunning = false;
+    std::function<void()> adapterFunctionDefault;
+
+    bool createThread(const std::function<void()>&  fun = nullptr, PIThread* thread_ = nullptr) {
+    	std::function<void()> actualFun = fun == nullptr ? adapterFunctionDefault : fun;
+        if (thread_ == nullptr) thread_ = thread;
+        thread_->startOnce([=](void*){
+			isRunning = true;
+        	actualFun();
+        });
+        return waitThread(thread_);
+    }
+
+    bool waitThread(PIThread* thread_, bool runningStatus = true) {
+        PITimeMeasurer measurer;
+        bool isTimeout = !thread_->waitForStart(WAIT_THREAD_TIME_MS);
+        while (!isRunning) {
+            isTimeout = WAIT_THREAD_TIME_MS <= measurer.elapsed_m();
+            if (isTimeout) break;
+            piUSleep(100);
+        }
+
+        threadStartTime = measurer.elapsed_m();
+
+        if (isTimeout) piCout << "Start thread timeout reach!";
+
+        if (threadStartTime > 1) {
+            piCout << "Start time" << threadStartTime << "ms";
+        } else if (threadStartTime > 0.001) {
+            piCout << "Start time" << threadStartTime * 1000 << "mcs";
+        } else {
+            piCout << "Start time" << threadStartTime * 1000 * 1000 << "ns";
+        }
+
+        return !isTimeout;
+    }
+};
+
+#endif //AWRCANFLASHER_TESTUTIL_H