fix PIThreadPoolExecutor and PIBlockingDequeue

lib/main/thread/piblockingdequeue.h

@@ -86,29 +86,13 @@ public:
 	 * @param v the element to add
 	 * @return true if the element was added to this queue, else false
 	 */
-	bool offer(const T & v) {
+	bool offer(const T & v, int timeoutMs = 0) {
+		bool isOk;
 		mutex.lock();
-		if (PIDeque<T>::size() >= max_size) {
+		if (timeoutMs == 0)
-			mutex.unlock();
+			isOk = PIDeque<T>::size() < max_size;
-			return false;
+		else
-		}
+			isOk = cond_var_rem->waitFor(mutex, timeoutMs, [&]() { return PIDeque<T>::size() < max_size; } );
-		PIDeque<T>::push_back(v);
-		mutex.unlock();
-		cond_var_add->notifyOne();
-		return true;
-	}
-
-	/**
-	 * @brief Inserts the specified element into this queue, waiting up to the specified wait time if necessary for
-	 * space to become available.
-	 *
-	 * @param v the element to add
-	 * @param timeoutMs how long to wait before giving up, in milliseconds
-	 * @return true if successful, or false if the specified waiting time elapses before space is available
-	 */
-	bool offer(const T & v, int timeoutMs) {
-		mutex.lock();
-		bool isOk = cond_var_rem->waitFor(mutex, timeoutMs, [&]() { return PIDeque<T>::size() < max_size; } );
 		if (isOk) PIDeque<T>::push_back(v);
 		mutex.unlock();
 		if (isOk) cond_var_add->notifyOne();
@@ -140,31 +124,15 @@ public:
 	 * return value is retrieved value
 	 * @return the head of this queue, or defaultVal if the specified waiting time elapses before an element is available
 	 */
-	T poll(int timeoutMs, const T & defaultVal = T(), bool * isOk = nullptr) {
+	T poll(int timeoutMs = 0, const T & defaultVal = T(), bool * isOk = nullptr) {
-		T t;
+		T t = defaultVal;
+		bool isNotEmpty;
 		mutex.lock();
-		bool isNotEmpty = cond_var_add->waitFor(mutex, timeoutMs, [&]() { return !PIDeque<T>::isEmpty(); });
+		if (timeoutMs == 0)
-		t = isNotEmpty ? T(PIDeque<T>::take_front()) : defaultVal;
+			isNotEmpty = !PIDeque<T>::isEmpty();
-		mutex.unlock();
+		else
-		if (isNotEmpty) cond_var_rem->notifyOne();
+			isNotEmpty = cond_var_add->waitFor(mutex, timeoutMs, [&]() { return !PIDeque<T>::isEmpty(); });
-		if (isOk) *isOk = isNotEmpty;
+		if (isNotEmpty) t = PIDeque<T>::take_front();
-		return t;
-	}
-
-	/**
-	 * @brief Retrieves and removes the head of this queue and return it if queue not empty, otherwise return defaultVal.
-	 * Do it immediately without waiting.
-	 *
-	 * @param defaultVal value, which returns if the specified waiting time elapses before an element is available
-	 * @param isOk flag, which indicates result of method execution. It will be set to false if timeout, or true if
-	 * return value is retrieved value
-	 * @return the head of this queue, or defaultVal if the specified waiting time elapses before an element is available
-	 */
-	T poll(const T & defaultVal = T(), bool * isOk = nullptr) {
-		T t;
-		mutex.lock();
-		bool isNotEmpty = !PIDeque<T>::isEmpty();
-		t = isNotEmpty ? PIDeque<T>::take_front() : defaultVal;
 		mutex.unlock();
 		if (isNotEmpty) cond_var_rem->notifyOne();
 		if (isOk) *isOk = isNotEmpty;

lib/main/thread/pithreadpoolexecutor.cpp

@@ -18,7 +18,6 @@
 */
 
 #include "pithreadpoolexecutor.h"
-#include "pisysteminfo.h"
 
 /*! \class PIThreadPoolExecutor
  * @brief Thread pools address two different problems: they usually provide improved performance when executing large
@@ -27,21 +26,14 @@
  */
 
 
-PIThreadPoolExecutor::PIThreadPoolExecutor(size_t corePoolSize, PIBlockingDequeue<std::function<void()> > * taskQueue_) : isShutdown_(false) {
+PIThreadPoolExecutor::PIThreadPoolExecutor(int corePoolSize) : isShutdown_(false) {
-	queue_own = false;
+	for (int i = 0; i < corePoolSize; ++i) {
-	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()>());
+			auto runnable = taskQueue.poll(100, std::function<void()>());
 			if (runnable) {
 				runnable();
 			}
-			if (isShutdown_ && taskQueue->size() == 0) threadPool[i]->stop();
+			if (isShutdown_ && taskQueue.size() == 0) threadPool[i]->stop();
 		});
 		threadPool.push_back(thread);
 		thread->start();
@@ -69,13 +61,11 @@ void PIThreadPoolExecutor::shutdownNow() {
 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);
+	if (!isShutdown_) taskQueue.offer(runnable);
 }
 
 

lib/main/thread/pithreadpoolexecutor.h

@@ -26,7 +26,7 @@
 
 class PIP_EXPORT PIThreadPoolExecutor {
 public:
-	explicit PIThreadPoolExecutor(size_t corePoolSize = -1, PIBlockingDequeue<std::function<void()> > * taskQueue_ = 0);
+	explicit PIThreadPoolExecutor(int corePoolSize);
 
 	virtual ~PIThreadPoolExecutor();
 
@@ -54,7 +54,7 @@ public:
 
 private:
 	std::atomic_bool isShutdown_;
-	PIBlockingDequeue<std::function<void()> > * taskQueue;
+	PIBlockingDequeue<std::function<void()> > taskQueue;
 	PIVector<PIThread*> threadPool;
 	bool queue_own;
 

tests/concurrent/BlockingDequeueUnitTest.cpp

@@ -69,7 +69,7 @@ TEST(BlockingDequeueUnitTest, offer_timedout_is_block_when_capacity_reach) {
 TEST(BlockingDequeueUnitTest, offer_is_true_before_capacity_reach) {
     size_t capacity = 1;
     PIBlockingDequeue<int> dequeue(capacity);
-    ASSERT_TRUE(dequeue.offer(10));
+	ASSERT_TRUE(dequeue.offer(10));
 }
 
 TEST(BlockingDequeueUnitTest, offer_is_false_when_capacity_reach) {
@@ -125,7 +125,7 @@ TEST(BlockingDequeueUnitTest, poll_is_not_block_when_empty) {
 	bool isOk;
 	auto conditionVar = new MockConditionVar();
 	PIBlockingDequeue<int> dequeue(capacity, conditionVar);
-	dequeue.poll(111, &isOk);
+	dequeue.poll(0, 111, &isOk);
 	EXPECT_FALSE(conditionVar->isWaitForCalled);
 }
 
@@ -134,7 +134,7 @@ TEST(BlockingDequeueUnitTest, poll_is_default_value_when_empty) {
 	bool isOk;
 	auto conditionVar = new MockConditionVar();
 	PIBlockingDequeue<int> dequeue(capacity, conditionVar);
-	ASSERT_EQ(dequeue.poll(111, &isOk), 111);
+	ASSERT_EQ(dequeue.poll(0, 111, &isOk), 111);
 }
 
 TEST(BlockingDequeueUnitTest, poll_is_offer_value_when_not_empty) {
@@ -143,7 +143,7 @@ TEST(BlockingDequeueUnitTest, poll_is_offer_value_when_not_empty) {
 	auto conditionVar = new MockConditionVar();
 	PIBlockingDequeue<int> dequeue(capacity, conditionVar);
 	dequeue.offer(111);
-	ASSERT_EQ(dequeue.poll(-1, &isOk), 111);
+	ASSERT_EQ(dequeue.poll(0, -1, &isOk), 111);
 }
 
 TEST(BlockingDequeueUnitTest, poll_timeouted_is_block_when_empty) {
@@ -152,8 +152,8 @@ TEST(BlockingDequeueUnitTest, poll_timeouted_is_block_when_empty) {
     auto conditionVar = new MockConditionVar();
     PIBlockingDequeue<int> dequeue(capacity, conditionVar);
     dequeue.poll(timeout, 111);
-    EXPECT_TRUE(conditionVar->isWaitForCalled);
+	EXPECT_TRUE(conditionVar->isWaitForCalled);
-    EXPECT_EQ(timeout, conditionVar->timeout);
+	EXPECT_EQ(timeout, conditionVar->timeout);
     ASSERT_FALSE(conditionVar->isTrueCondition);
 }