pip/tests/concurrent/ConditionVariableIntegrationTest.cpp

#include "gtest/gtest.h"
#include "piconditionvar.h"
#include "pithread.h"
#include "testutil.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();
	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) {
	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();
	msleep(WAIT_THREAD_TIME_MS);
	ASSERT_FALSE(thread->isRunning());
}