Merge remote-tracking branch 'origin/master' into concurrent

# Conflicts:
#	lib/main/thread/pithreadpoolexecutor.cpp
#	lib/main/thread/pithreadpoolexecutor.h
#	tests/concurrent/ExecutorIntegrationTest.cpp
#	tests/concurrent/ExecutorUnitTest.cpp
#	tests/concurrent/testutil.h

lib/main/thread/piexecutor.h

@@ -0,0 +1,111 @@
+/*
+	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/>.
+*/
+
+#ifndef PIEXECUTOR_H
+#define PIEXECUTOR_H
+
+#include "piblockingdequeue.h"
+#include <atomic>
+
+/**
+ * @brief Thread pools address two different problems: they usually provide improved performance when executing large
+ * 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.
+ *
+ * TODO adapt documentation to template
+ */
+template <typename Thread_, typename Dequeue_>
+class PIThreadPoolExecutorTemplate {
+public:
+	explicit PIThreadPoolExecutorTemplate(size_t corePoolSize = 1) : isShutdown_(false) { makePool(corePoolSize); }
+
+	virtual ~PIThreadPoolExecutorTemplate() {
+		shutdownNow();
+		while (threadPool.size() > 0) delete threadPool.take_back();
+	}
+
+	/**
+	 * @brief Executes the given task sometime in the future. The task execute in an existing pooled thread. If the task
+	 * cannot be submitted for execution, either because this executor has been shutdown or because its capacity has been
+	 * reached.
+	 *
+	 * @param runnable not empty function for thread pool execution
+	 */
+	void execute(const std::function<void()> & runnable) {
+		if (!isShutdown_) taskQueue.offer(runnable);
+	}
+
+	/**
+	 * @brief Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be
+	 * accepted. Invocation has no additional effect if already shut down. This method does not wait for previously
+	 * submitted tasks to complete execution. Use awaitTermination to do that.
+	 */
+	void shutdown() {
+		isShutdown_ = true;
+	}
+
+	void shutdownNow() {
+		isShutdown_ = true;
+		for (size_t i = 0; i < threadPool.size(); ++i) threadPool[i]->stop();
+	}
+
+	bool isShutdown() const {
+		return isShutdown_;
+	}
+
+	bool 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;
+	}
+
+protected:
+	std::atomic_bool isShutdown_;
+	Dequeue_ taskQueue;
+	PIVector<Thread_*> threadPool;
+
+	template<typename Function>
+	PIThreadPoolExecutorTemplate(size_t corePoolSize, Function onBeforeStart) : isShutdown_(false) {
+		makePool(corePoolSize, onBeforeStart);
+	}
+
+	void makePool(size_t corePoolSize, std::function<void(Thread_*)> onBeforeStart = [](Thread_*){}) {
+		for (size_t i = 0; i < corePoolSize; ++i) {
+			auto* thread = new Thread_([&, i](){
+				auto runnable = taskQueue.poll(100);
+				if (runnable) {
+					runnable();
+				}
+				if (isShutdown_ && taskQueue.size() == 0) threadPool[i]->stop();
+			});
+			threadPool.push_back(thread);
+			onBeforeStart(thread);
+			thread->start();
+		}
+	}
+};
+
+typedef PIThreadPoolExecutorTemplate<PIThread, PIBlockingDequeue<std::function<void()> > > PIThreadPoolExecutor;
+
+
+#endif //PIEXECUTOR_H

lib/main/thread/pithreadmodule.h

@@ -25,7 +25,7 @@
 #include "pitimer.h"
 #include "pipipelinethread.h"
 #include "pigrabberbase.h"
-#include "pithreadpoolexecutor.h"
+#include "piexecutor.h"
 #include "piconditionvar.h"
 
 #endif // PITHREADMODULE_H

lib/main/thread/pithreadpoolexecutor.cpp

@@ -1,89 +0,0 @@
-/*
-	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 "pithreadpoolexecutor.h"
-#include "pisysteminfo.h"
-
-/*! \class PIThreadPoolExecutor
- * @brief Thread pools address two different problems: they usually provide improved performance when executing large
- * 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.
- */
-
-
-PIThreadPoolExecutor::PIThreadPoolExecutor(size_t corePoolSize, PIBlockingDequeue<std::function<void()> > * taskQueue_) : isShutdown_(false) {
-	queue_own = false;
-	if (corePoolSize <= 0)
-		corePoolSize = PISystemInfo::instance()->processorsCount;
-	if (!taskQueue_) {
-		taskQueue = new PIBlockingDequeue<std::function<void()> >();
-		queue_own = true;
-	}
-	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();
-	if (queue_own)
-		delete taskQueue;
-}
-
-
-void PIThreadPoolExecutor::execute(const std::function<void()> & runnable) {
-	if (!isShutdown_) taskQueue->offer(runnable);
-}
-
-
-bool PIThreadPoolExecutor::isShutdown() const {
-	return isShutdown_;
-}
-
-
-void PIThreadPoolExecutor::shutdown() {
-	isShutdown_ = true;
-}

lib/main/thread/pithreadpoolexecutor.h

@@ -1,63 +0,0 @@
-/*
-	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/>.
-*/
-
-#ifndef PITHREADPOOLEXECUTOR_H
-#define PITHREADPOOLEXECUTOR_H
-
-#include "piblockingdequeue.h"
-#include <atomic>
-
-
-class PIP_EXPORT PIThreadPoolExecutor {
-public:
-	explicit PIThreadPoolExecutor(size_t corePoolSize = -1, PIBlockingDequeue<std::function<void()> > * taskQueue_ = 0);
-
-	virtual ~PIThreadPoolExecutor();
-
-	/**
-	 * @brief Executes the given task sometime in the future. The task execute in an existing pooled thread. If the task
-	 * cannot be submitted for execution, either because this executor has been shutdown or because its capacity has been
-	 * reached.
-	 *
-	 * @param runnable not empty function for thread pool execution
-	 */
-	void execute(const std::function<void()> & runnable);
-
-	void shutdownNow();
-
-	/**
-	 * @brief Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be
-	 * accepted. Invocation has no additional effect if already shut down. This method does not wait for previously
-	 * submitted tasks to complete execution. Use awaitTermination to do that.
-	 */
-	void shutdown();
-
-	bool isShutdown() const;
-
-	bool awaitTermination(int timeoutMs);
-
-private:
-	std::atomic_bool isShutdown_;
-	PIBlockingDequeue<std::function<void()> > * taskQueue;
-	PIVector<PIThread*> threadPool;
-	bool queue_own;
-
-};
-
-#endif // PITHREADPOOLEXECUTOR_H

tests/concurrent/ExecutorIntegrationTest.cpp

@@ -1,8 +1,7 @@
 #include "gtest/gtest.h"
-#include "pithreadpoolexecutor.h"
+#include "piexecutor.h"
 #include "pimutex.h"
-
-const int WAIT_THREAD_TIME_MS = 30;
+#include "testutil.h"
 
 TEST(ExcutorIntegrationTest, execute_is_runnable_invoke) {
     PIMutex m;
@@ -14,11 +13,13 @@ TEST(ExcutorIntegrationTest, execute_is_runnable_invoke) {
         m.unlock();
     });
     piMSleep(WAIT_THREAD_TIME_MS);
+	m.lock();
     ASSERT_EQ(invokedRunnables, 1);
+    m.unlock();
 }
 
 TEST(ExcutorIntegrationTest, execute_is_not_execute_after_shutdown) {
-    bool isRunnableInvoke = false;
+    volatile bool isRunnableInvoke = false;
     PIThreadPoolExecutor executorService(1);
     executorService.shutdown();
     executorService.execute([&]() {
@@ -29,7 +30,7 @@ TEST(ExcutorIntegrationTest, execute_is_not_execute_after_shutdown) {
 }
 
 TEST(ExcutorIntegrationTest, execute_is_execute_before_shutdown) {
-    bool isRunnableInvoke = false;
+	volatile bool isRunnableInvoke = false;
     PIThreadPoolExecutor executorService(1);
     executorService.execute([&]() {
         piMSleep(WAIT_THREAD_TIME_MS);

tests/concurrent/ExecutorUnitTest.cpp

@@ -0,0 +1,102 @@
+#include "gtest/gtest.h"
+#include "gmock/gmock.h"
+#include "piexecutor.h"
+#include "testutil.h"
+
+using ::testing::_;
+using ::testing::SetArgReferee;
+using ::testing::DoAll;
+using ::testing::DeleteArg;
+using ::testing::Return;
+using ::testing::AtLeast;
+using ::testing::ByRef;
+using ::testing::Eq;
+using ::testing::Ge;
+using ::testing::Pointee;
+using ::testing::IsNull;
+using ::testing::NiceMock;
+
+typedef std::function<void()> VoidFunc;
+
+namespace std {
+    inline bool operator ==(const VoidFunc& s, const VoidFunc& v) {
+        // TODO VoidFunc operator ==
+        return true;
+    }
+}
+
+class MockThread {
+public:
+	std::function<void()> runnnable;
+
+	MockThread(std::function<void()> runnnable) : runnnable(runnnable) { }
+
+    MOCK_METHOD0(start, bool());
+    MOCK_METHOD0(stop, void());
+    MOCK_METHOD1(waitForStart, bool(int timeout_msecs));
+    MOCK_METHOD1(waitForFinish, bool(int timeout_msecs));
+};
+
+class MockDeque : public PIBlockingDequeue<VoidFunc> {
+public:
+    MOCK_METHOD1(offer, bool(const VoidFunc&));
+    MOCK_METHOD0(take, VoidFunc());
+    MOCK_METHOD1(poll, VoidFunc(int));
+    MOCK_METHOD0(capacity, size_t());
+    MOCK_METHOD0(remainingCapacity, size_t());
+};
+
+typedef PIThreadPoolExecutorTemplate<NiceMock<MockThread>, MockDeque> PIThreadPoolExecutorMoc_t;
+
+class PIThreadPoolExecutorMoc : public PIThreadPoolExecutorMoc_t {
+public:
+	explicit PIThreadPoolExecutorMoc(size_t corePoolSize) : PIThreadPoolExecutorMoc_t(corePoolSize) { }
+
+	template<typename Function>
+	explicit PIThreadPoolExecutorMoc(size_t corePoolSize, Function onBeforeStart) : PIThreadPoolExecutorMoc_t(corePoolSize, onBeforeStart) { }
+
+	PIVector<testing::NiceMock<MockThread>*>* getThreadPool() { return &threadPool; }
+	bool isShutdown() { return isShutdown_; }
+	MockDeque* getTaskQueue() { return &taskQueue; }
+};
+
+TEST(ExecutorUnitTest, is_corePool_created) {
+	PIThreadPoolExecutorMoc executor(THREAD_COUNT);
+    ASSERT_EQ(THREAD_COUNT, executor.getThreadPool()->size());
+}
+
+TEST(ExecutorUnitTest, is_corePool_started) {
+	PIThreadPoolExecutorMoc executor(THREAD_COUNT, [](MockThread* thread){
+		EXPECT_CALL(*thread, start())
+				.WillOnce(Return(true));
+	});
+	EXPECT_EQ(THREAD_COUNT, executor.getThreadPool()->size());
+	executor.getThreadPool()->forEach([](MockThread* thread){
+		EXPECT_CALL(*thread, stop())
+				.WillOnce(Return());
+	});
+}
+
+TEST(ExecutorUnitTest, execute_is_added_to_taskQueue) {
+    VoidFunc voidFunc = [](){};
+	PIThreadPoolExecutorMoc executor(THREAD_COUNT);
+	EXPECT_CALL(*executor.getTaskQueue(), offer(Eq(voidFunc)))
+			.WillOnce(Return(true));
+    executor.execute(voidFunc);
+}
+
+TEST(ExecutorUnitTest, is_corePool_execute_queue_elements) {
+	bool is_executed = false;
+	PIThreadPoolExecutorMoc executor(1);
+	EXPECT_EQ(executor.getThreadPool()->size(), 1);
+	EXPECT_CALL(*executor.getTaskQueue(), poll(Ge(0)))
+			.WillOnce(Return([&](){ is_executed = true; }));
+	executor.getThreadPool()->at(0)->runnnable();
+	ASSERT_TRUE(is_executed);
+}
+/* FIXME
+TEST(ExecutorUnitTest, shutdown_is_stop_threads) {
+	PIThreadPoolExecutorMoc executor(THREAD_COUNT);
+    executor.shutdown();
+}
+*/

tests/concurrent/testutil.h

@@ -1,60 +1,60 @@
-#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 = 40;
-
-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
+#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 = 10;
+
+const int THREAD_COUNT = 2;
+
+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