Merge branch 'master' into pitimer_slot

libs/main/thread/pitimer.cpp

@@ -122,13 +122,13 @@ PITimer::PITimer(): PIObject() {
 
 PITimer::PITimer(std::function<void(int)> func) {
 	initFirst();
-	ret_func = func;
+	ret_func_delim = std::move(func);
 }
 
 
 PITimer::PITimer(std::function<void()> func) {
 	initFirst();
-	ret_func = [func](int) { func(); };
+	ret_func = std::move(func);
 }
 
 
@@ -224,7 +224,8 @@ void PITimer::adjustTimes() {
 void PITimer::execTick() {
 	if (!isRunning()) return;
 	if (lockRun) lock();
-	if (ret_func) ret_func(1);
+	if (ret_func) ret_func();
+	if (ret_func_delim) ret_func_delim(1);
 	tick(1);
 	tickEvent(1);
 	if (callEvents) maybeCallQueuedEvents();
@@ -233,8 +234,8 @@ void PITimer::execTick() {
 		i.tick = 0;
 		if (i.func)
 			i.func(i.delim);
-		else if (ret_func)
+		else if (ret_func_delim)
-			ret_func(i.delim);
+			ret_func_delim(i.delim);
 		tick(i.delim);
 		tickEvent(i.delim);
 	}
@@ -262,7 +263,7 @@ bool PITimer::start(PISystemTime interval) {
 bool PITimer::start(PISystemTime interval, std::function<void()> func) {
 	if (isRunning()) stopAndWait();
 	setInterval(interval);
-	setSlot(func);
+	setSlot(std::move(func));
 	return start();
 }
 
@@ -274,7 +275,7 @@ void PITimer::stopAndWait(PISystemTime timeout) {
 
 
 void PITimer::addDelimiter(int delim, std::function<void(int)> func) {
-	delims << Delimiter(func, delim);
+	delims << Delimiter(std::move(func), delim);
 }
 
 

libs/main/thread/pitimer.h

@@ -127,12 +127,16 @@ public:
 	//! \~english Sets a tick callback that ignores the delimiter value.
 	//! \~russian Устанавливает обратный вызов тика, игнорирующий значение делителя.
 	void setSlot(std::function<void()> func) {
-		ret_func = [func](int) { func(); };
+		ret_func_delim = nullptr;
+		ret_func       = std::move(func);
 	}
 
 	//! \~english Sets a tick callback that receives the current delimiter value.
 	//! \~russian Устанавливает обратный вызов тика, принимающий текущее значение делителя.
-	void setSlot(std::function<void(int)> func) { ret_func = func; }
+	void setSlot(std::function<void(int)> func) {
+		ret_func       = nullptr;
+		ret_func_delim = std::move(func);
+	}
 
 	//! \~english Enables locking of the internal mutex around tick processing.
 	//! \~russian Включает блокировку внутреннего мьютекса вокруг обработки тиков.
@@ -141,7 +145,8 @@ public:
 	EVENT_HANDLER0(void, unlock) { mutex_.unlock(); }
 
 	//! \~english Returns whether the timer drains queued delivery for itself as performer on each main tick. By default \b true.
-	//! \~russian Возвращает, должен ли таймер обрабатывать отложенную доставку для себя как исполнителя на каждом основном тике. По умолчанию \b true.
+	//! \~russian Возвращает, должен ли таймер обрабатывать отложенную доставку для себя как исполнителя на каждом основном тике. По
+	//! умолчанию \b true.
 	bool isCallQueuedEvents() const { return callEvents; }
 
 	//! \~english Enables or disables queued-delivery draining through \a maybeCallQueuedEvents() on each main tick.
@@ -221,7 +226,7 @@ public:
 protected:
 	struct PIP_EXPORT Delimiter {
 		Delimiter(std::function<void(int)> func_ = nullptr, int delim_ = 1) {
-			func  = func_;
+			func  = std::move(func_);
 			delim = delim_;
 		}
 		std::function<void(int)> func;
@@ -245,7 +250,8 @@ protected:
 	PIMutex mutex_;
 	PISystemTime m_interval, m_interval_x5;
 	PISystemTime m_time_next;
-	std::function<void(int)> ret_func = nullptr;
+	std::function<void()> ret_func          = nullptr;
+	std::function<void(int)> ret_func_delim = nullptr;
 	PIVector<Delimiter> delims;
 	PIConditionVariable event;
 };

tests/CMakeLists.txt

@@ -13,7 +13,6 @@ FetchContent_Declare(
 )
 # For Windows: Prevent overriding the parent project's compiler/linker settings
 set(gtest_force_shared_crt ON CACHE BOOL "" FORCE)
-set(BUILD_GMOCK OFF CACHE BOOL "Build Google Mock" FORCE)
 FetchContent_MakeAvailable(googletest)
 
 enable_testing()
@@ -33,6 +32,7 @@ macro(pip_test NAME)
 	set(PIP_TESTS_LIST ${PIP_TESTS_LIST} PARENT_SCOPE)
 endmacro()
 
+#pip_test(concurrent)
 pip_test(math)
 pip_test(core)
 pip_test(piobject)

tests/concurrent/BlockingDequeueUnitTest.cpp

@@ -0,0 +1,262 @@
+#include "piblockingqueue.h"
+
+#include "gtest/gtest.h"
+
+class MockConditionVar: public PIConditionVariable {
+public:
+	bool isWaitCalled    = false;
+	bool isWaitForCalled = false;
+	bool isTrueCondition = false;
+	int timeout          = -1;
+
+	void wait(PIMutex & lk) override { isWaitCalled = true; }
+
+	void wait(PIMutex & lk, const std::function<bool()> & condition) override {
+		isWaitCalled    = true;
+		isTrueCondition = condition();
+	}
+
+	bool waitFor(PIMutex & lk, int timeoutMs) override {
+		isWaitForCalled = true;
+		timeout         = timeoutMs;
+		return false;
+	}
+
+	bool waitFor(PIMutex & lk, int timeoutMs, const std::function<bool()> & condition) override {
+		isWaitForCalled = true;
+		isTrueCondition = condition();
+		timeout         = timeoutMs;
+		return isTrueCondition;
+	}
+};
+
+TEST(BlockingDequeueUnitTest, put_is_block_when_capacity_reach) {
+	size_t capacity      = 0;
+	auto conditionVarAdd = new MockConditionVar();
+	auto conditionVarRem = new MockConditionVar();
+	PIBlockingQueue<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();
+	PIBlockingQueue<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();
+	PIBlockingQueue<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;
+	PIBlockingQueue<int> dequeue(capacity);
+	ASSERT_TRUE(dequeue.offer(10));
+}
+
+TEST(BlockingDequeueUnitTest, offer_is_false_when_capacity_reach) {
+	size_t capacity = 1;
+	PIBlockingQueue<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();
+	PIBlockingQueue<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();
+	PIBlockingQueue<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();
+	PIBlockingQueue<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();
+	PIBlockingQueue<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_not_block_when_empty) {
+	size_t capacity = 1;
+	bool isOk;
+	auto conditionVar = new MockConditionVar();
+	PIBlockingQueue<int> dequeue(capacity, conditionVar);
+	dequeue.poll(0, 111, &isOk);
+	EXPECT_FALSE(conditionVar->isWaitForCalled);
+}
+
+TEST(BlockingDequeueUnitTest, poll_is_default_value_when_empty) {
+	size_t capacity = 1;
+	bool isOk;
+	auto conditionVar = new MockConditionVar();
+	PIBlockingQueue<int> dequeue(capacity, conditionVar);
+	ASSERT_EQ(dequeue.poll(0, 111, &isOk), 111);
+}
+
+TEST(BlockingDequeueUnitTest, poll_is_offer_value_when_not_empty) {
+	size_t capacity = 1;
+	bool isOk;
+	auto conditionVar = new MockConditionVar();
+	PIBlockingQueue<int> dequeue(capacity, conditionVar);
+	dequeue.offer(111);
+	ASSERT_EQ(dequeue.poll(0, -1, &isOk), 111);
+}
+
+TEST(BlockingDequeueUnitTest, poll_timeouted_is_block_when_empty) {
+	size_t capacity   = 1;
+	int timeout       = 11;
+	auto conditionVar = new MockConditionVar();
+	PIBlockingQueue<int> dequeue(capacity, conditionVar);
+	dequeue.poll(timeout, 111);
+	EXPECT_TRUE(conditionVar->isWaitForCalled);
+	EXPECT_EQ(timeout, conditionVar->timeout);
+	ASSERT_FALSE(conditionVar->isTrueCondition);
+}
+
+TEST(BlockingDequeueUnitTest, poll_timeouted_is_default_value_when_empty) {
+	size_t capacity   = 1;
+	int timeout       = 11;
+	auto conditionVar = new MockConditionVar();
+	PIBlockingQueue<int> dequeue(capacity, conditionVar);
+	ASSERT_EQ(dequeue.poll(timeout, 111), 111);
+}
+
+TEST(BlockingDequeueUnitTest, poll_timeouted_is_not_block_when_not_empty) {
+	size_t capacity   = 1;
+	int timeout       = 11;
+	auto conditionVar = new MockConditionVar();
+	PIBlockingQueue<int> dequeue(capacity, conditionVar);
+	dequeue.offer(111);
+	dequeue.poll(timeout, -1);
+
+	EXPECT_TRUE(conditionVar->isWaitForCalled);
+	ASSERT_TRUE(conditionVar->isTrueCondition);
+}
+
+TEST(BlockingDequeueUnitTest, poll_timeouted_is_offer_value_when_not_empty) {
+	size_t capacity   = 1;
+	int timeout       = 11;
+	auto conditionVar = new MockConditionVar();
+	PIBlockingQueue<int> dequeue(capacity, conditionVar);
+	dequeue.offer(111);
+	ASSERT_EQ(dequeue.poll(timeout, -1), 111);
+}
+
+TEST(BlockingDequeueUnitTest, poll_timeouted_is_last) {
+	size_t capacity   = 10;
+	auto conditionVar = new MockConditionVar();
+	PIBlockingQueue<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;
+	PIBlockingQueue<int> dequeue(capacity);
+	ASSERT_EQ(dequeue.capacity(), capacity);
+}
+
+TEST(BlockingDequeueUnitTest, remainingCapacity_is_dif_of_capacity_and_size) {
+	size_t capacity = 2;
+	PIBlockingQueue<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;
+	PIBlockingQueue<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;
+	PIBlockingQueue<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;
+	PIBlockingQueue<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);
+	PIBlockingQueue<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);
+	PIBlockingQueue<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);
+	PIBlockingQueue<int> blockingDequeue(refDeque);
+	PIDeque<int> deque;
+	ASSERT_EQ(blockingDequeue.drainTo(deque, refDeque.size() - 1), refDeque.size() - 1);
+}

tests/concurrent/ConditionLockIntegrationTest.cpp

@@ -0,0 +1,57 @@
+#include "piconditionvar.h"
+#include "pithread.h"
+#include "testutil.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+class ConditionLock
+	: public ::testing::Test
+	, public TestUtil {
+public:
+	PIMutex * m = new PIMutex();
+	bool isProtect;
+	bool isReleased;
+};
+
+
+TEST_F(ConditionLock, DISABLED_lock_is_protect) {
+	m->lock();
+	isProtect = true;
+	createThread([&]() {
+		m->lock();
+		isProtect = false;
+	});
+	EXPECT_FALSE(thread->waitForFinish(WAIT_THREAD_TIME_MS));
+	ASSERT_TRUE(isProtect);
+}
+
+TEST_F(ConditionLock, DISABLED_unlock_is_release) {
+	m->lock();
+	isReleased = 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());
+}

tests/concurrent/ConditionVariableIntegrationTest.cpp

@@ -0,0 +1,209 @@
+#include "piconditionvar.h"
+#include "pithread.h"
+#include "testutil.h"
+
+#include "gtest/gtest.h"
+
+class ConditionVariable
+	: public ::testing::Test
+	, public TestUtil {
+public:
+	~ConditionVariable() { delete variable; }
+	PIMutex 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();
+	piMSleep(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();
+	piMSleep(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) {
+	std::atomic_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();
+	piMSleep(WAIT_THREAD_TIME_MS);
+	ASSERT_FALSE(thread->isRunning());
+}

tests/concurrent/ExecutorIntegrationTest.cpp

@@ -0,0 +1,51 @@
+#include "pimutex.h"
+#include "pithreadpoolexecutor.h"
+
+#include "gtest/gtest.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);
+}

tests/concurrent/pithreadnotifier_test.cpp

@@ -0,0 +1,57 @@
+#include "pithreadnotifier.h"
+
+#include "gtest/gtest.h"
+
+
+TEST(PIThreadNotifierTest, One) {
+	PIThreadNotifier n;
+	int cnt = 0;
+	PIThread t1(
+		[&n, &cnt]() {
+			n.wait();
+			cnt++;
+		},
+		true);
+	piMSleep(10);
+	n.notifyOnce();
+	piMSleep(10);
+	ASSERT_EQ(cnt, 1);
+	n.notifyOnce();
+	piMSleep(10);
+	ASSERT_EQ(cnt, 2);
+}
+
+
+TEST(PIThreadNotifierTest, Two) {
+	PIThreadNotifier n;
+	int cnt1 = 0;
+	int cnt2 = 0;
+	int cnt3 = 0;
+	PIThread t1(
+		[&n, &cnt1]() {
+			n.wait();
+			cnt1++;
+			piMSleep(2);
+		},
+		true);
+	PIThread t2(
+		[&n, &cnt2]() {
+			n.wait();
+			cnt2++;
+			piMSleep(2);
+		},
+		true);
+	PIThread t3(
+		[&n, &cnt3]() {
+			n.notifyOnce();
+			cnt3++;
+			piMSleep(1);
+		},
+		true);
+	piMSleep(20);
+	t3.stop(true);
+	piMSleep(100);
+	t1.stop();
+	t2.stop();
+	ASSERT_EQ(cnt1 + cnt2, cnt3);
+}

tests/concurrent/testutil.h

@@ -0,0 +1,62 @@
+#ifndef AWRCANFLASHER_TESTUTIL_H
+#define AWRCANFLASHER_TESTUTIL_H
+
+#include "pithread.h"
+
+#include <atomic>
+
+/**
+ * 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 = 400;
+
+const int THREAD_COUNT        = 5;
+
+class TestUtil: public PIObject {
+	PIOBJECT(TestUtil)
+
+public:
+	double threadStartTime;
+	PIThread * thread = new PIThread();
+	std::atomic_bool isRunning;
+	std::function<void()> adapterFunctionDefault;
+
+	TestUtil(): isRunning(false) {}
+
+	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