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/piblockingdequeue.h

@@ -0,0 +1,245 @@
+/*
+	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 PIBLOCKINGDEQUEUE_H
+#define PIBLOCKINGDEQUEUE_H
+
+#include "pideque.h"
+#include "piconditionvar.h"
+
+/**
+ * @brief A Queue that supports operations that wait for the queue to become non-empty when retrieving an element, and
+ * wait for space to become available in the queue when storing an element.
+ */
+template <typename T>
+class PIBlockingDequeue: private PIDeque<T> {
+public:
+
+	/**
+	 * @brief Constructor
+	 */
+	explicit inline PIBlockingDequeue(size_t capacity = SIZE_MAX,
+									  PIConditionVariable* cond_var_add = new PIConditionVariable(),
+									  PIConditionVariable* cond_var_rem = new PIConditionVariable())
+		: cond_var_add(cond_var_add), cond_var_rem(cond_var_rem), max_size(capacity) { }
+
+	/**
+	 * @brief Copy constructor. Initialize queue with copy of other queue elements. Not thread-safe for other queue.
+	 */
+	explicit inline PIBlockingDequeue(const PIDeque<T>& other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
+		mutex.lock();
+		max_size = SIZE_MAX;
+		PIDeque<T>::append(other);
+		mutex.unlock();
+	}
+
+	/**
+	 * @brief Thread-safe copy constructor. Initialize queue with copy of other queue elements.
+	 */
+	inline PIBlockingDequeue(PIBlockingDequeue<T> & other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
+		other.mutex.lock();
+		mutex.lock();
+		max_size = other.max_size;
+		PIDeque<T>::append(static_cast<PIDeque<T>&>(other));
+		mutex.unlock();
+		other.mutex.unlock();
+	}
+
+	~PIBlockingDequeue() {
+		delete cond_var_add;
+		delete cond_var_rem;
+	}
+
+	/**
+	 * @brief Inserts the specified element into this queue, waiting if necessary for space to become available.
+	 *
+	 * @param v the element to add
+	 */
+	void put(const T & v) {
+		mutex.lock();
+		cond_var_rem->wait(mutex, [&]() { return PIDeque<T>::size() < max_size; });
+		PIDeque<T>::push_back(v);
+		mutex.unlock();
+		cond_var_add->notifyOne();
+	}
+
+	/**
+	 * @brief Inserts the specified element at the end of this queue if it is possible to do so immediately without
+	 * exceeding the queue's capacity, returning true upon success and false if this queue is full.
+	 *
+	 * @param v the element to add
+	 * @return true if the element was added to this queue, else false
+	 */
+	bool offer(const T & v) {
+		mutex.lock();
+		if (PIDeque<T>::size() >= max_size) {
+			mutex.unlock();
+			return false;
+		}
+		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();
+		return isOk;
+	}
+
+	/**
+	 * @brief Retrieves and removes the head of this queue, waiting if necessary until an element becomes available.
+	 *
+	 * @return the head of this queue
+	 */
+	T take() {
+		T t;
+		mutex.lock();
+		cond_var_add->wait(mutex, [&]() { return !PIDeque<T>::isEmpty(); });
+		t = T(PIDeque<T>::take_front());
+		mutex.unlock();
+		cond_var_rem->notifyOne();
+		return t;
+	}
+
+	/**
+	 * @brief Retrieves and removes the head of this queue, waiting up to the specified wait time if necessary for an
+	 * element to become available.
+	 *
+	 * @param timeoutMs how long to wait before giving up, in milliseconds
+	 * @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(int timeoutMs, const T & defaultVal = T(), bool * isOk = nullptr) {
+		T t;
+		mutex.lock();
+		bool isNotEmpty = cond_var_add->waitFor(mutex, timeoutMs, [&]() { return !PIDeque<T>::isEmpty(); });
+		t = isNotEmpty ? T(PIDeque<T>::take_front()) : defaultVal;
+		mutex.unlock();
+		if (isNotEmpty) cond_var_rem->notifyOne();
+		if (isOk) *isOk = isNotEmpty;
+		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;
+		return t;
+	}
+
+	/**
+	 * @brief Returns the number of elements that this queue can ideally (in the absence of memory or resource
+	 * constraints) contains. This is always equal to the initial capacity of this queue less the current size of this queue.
+	 *
+	 * @return the capacity
+	 */
+	size_t capacity() {
+		size_t c;
+		mutex.lock();
+		c = max_size;
+		mutex.unlock();
+		return c;
+	}
+
+	/**
+	 * @brief Returns the number of additional elements that this queue can ideally (in the absence of memory or resource
+	 * constraints) accept. This is always equal to the initial capacity of this queue less the current size of this queue.
+	 *
+	 * @return the remaining capacity
+	 */
+	size_t remainingCapacity() {
+		mutex.lock();
+		size_t c = max_size - PIDeque<T>::size();
+		mutex.unlock();
+		return c;
+	}
+
+	/**
+	 * @brief Returns the number of elements in this collection.
+	 */
+	size_t size() {
+		mutex.lock();
+		size_t s = PIDeque<T>::size();
+		mutex.unlock();
+		return s;
+	}
+
+	/**
+	 * @brief Removes all available elements from this queue and adds them to other given queue.
+	 */
+	size_t drainTo(PIDeque<T>& other, size_t maxCount = SIZE_MAX) {
+		mutex.lock();
+		size_t count = ((maxCount > PIDeque<T>::size()) ? PIDeque<T>::size() : maxCount);
+		for (size_t i = 0; i < count; ++i) other.push_back(PIDeque<T>::take_front());
+		mutex.unlock();
+		return count;
+	}
+
+	/**
+	 * @brief Removes all available elements from this queue and adds them to other given queue.
+	 */
+	size_t drainTo(PIBlockingDequeue<T>& other, size_t maxCount = SIZE_MAX) {
+		mutex.lock();
+		other.mutex.lock();
+		size_t count = maxCount > PIDeque<T>::size() ? PIDeque<T>::size() : maxCount;
+		size_t otherRemainingCapacity = other.max_size - static_cast<PIDeque<T> >(other).size();
+		if (count > otherRemainingCapacity) count = otherRemainingCapacity;
+		for (size_t i = 0; i < count; ++i) other.push_back(PIDeque<T>::take_front());
+		other.mutex.unlock();
+		mutex.unlock();
+		return count;
+	}
+
+private:
+	PIMutex mutex;
+	PIConditionVariable * cond_var_add, * cond_var_rem;
+	size_t max_size;
+
+};
+
+
+#endif // PIBLOCKINGDEQUEUE_H

lib/main/thread/piconditionvar.cpp

@@ -0,0 +1,141 @@
+/*
+	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 "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;
+	PIMutex * 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(PIMutex& 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(PIMutex& 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(PIMutex &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(PIMutex& 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/main/thread/piconditionvar.h

@@ -0,0 +1,118 @@
+/*
+	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 PICONDITIONVAR_H
+#define PICONDITIONVAR_H
+
+#include "pithread.h"
+
+
+/**
+ * @brief A condition variable is an object able to block the calling thread until notified to resume.
+ *
+ * It uses a PIMutex to lock the thread when one of its wait functions is called. The thread remains
+ * blocked until woken up by another thread that calls a notification function on the same PIConditionVariable object.
+ */
+class PIP_EXPORT PIConditionVariable {
+public:
+	NO_COPY_CLASS(PIConditionVariable)
+	explicit PIConditionVariable();
+	virtual ~PIConditionVariable();
+
+	/**
+	 * @brief Unblocks one of the threads currently waiting for this condition. If no threads are waiting, the function
+	 * does nothing. If more than one, it is unspecified which of the threads is selected.
+	 */
+	void notifyOne();
+
+	/**
+	 * @brief Unblocks all threads currently waiting for this condition. If no threads are waiting, the function does
+	 * nothing.
+	 */
+	void notifyAll();
+
+	/**
+	 * @brief see wait(PIMutex &, const std::function<bool()>&)
+	 */
+	virtual void wait(PIMutex & lk);
+
+	/**
+	 * @brief Wait until notified
+	 *
+	 * The execution of the current thread (which shall have locked with lk method PIMutex::lock()) is blocked
+	 * until notified.
+	 *
+	 * At the moment of blocking the thread, the function automatically calls lk.unlock() (PIMutex::unlock()),
+	 * allowing other locked threads to continue.
+	 *
+	 * Once notified (explicitly, by some other thread), the function unblocks and calls lk.lock() (PIMutex::lock()),
+	 * leaving lk in the same state as when the function was called. Then the function returns (notice that this last mutex
+	 * locking may block again the thread before returning).
+	 *
+	 * Generally, the function is notified to wake up by a call in another thread either to member notifyOne() or to
+	 * member notifyAll(). But certain implementations may produce spurious wake-up calls without any of these functions
+	 * being called. Therefore, users of this function shall ensure their condition for resumption is met.
+	 *
+	 * If condition is specified, the function only blocks if condition returns false, and notifications can only unblock
+	 * the thread when it becomes true (which is specially useful to check against spurious wake-up calls).
+	 *
+	 * @param lk lock object used by method wait for data protection
+	 * @param condition A callable object or function that takes no arguments and returns a value that can be evaluated
+	 * as a bool. This is called repeatedly until it evaluates to true.
+	 */
+	virtual void wait(PIMutex& lk, const std::function<bool()>& condition);
+
+	/**
+	 * @brief see waitFor(PIMutex &, int, const std::function<bool()>&)
+	 */
+	virtual bool waitFor(PIMutex & lk, int timeoutMs);
+
+	/**
+	 * @brief Wait for timeout or until notified
+	 *
+	 * The execution of the current thread (which shall have locked with lk method PIMutex::lock()) is blocked
+	 * during timeoutMs, or until notified (if the latter happens first).
+	 *
+	 * At the moment of blocking the thread, the function automatically calls lk.lock() (PIMutex::lock()), allowing
+	 * other locked threads to continue.
+	 *
+	 * Once notified or once timeoutMs has passed, the function unblocks and calls lk.unlock() (PIMutex::unlock()),
+	 * leaving lk in the same state as when the function was called. Then the function returns (notice that this last
+	 * mutex locking may block again the thread before returning).
+	 *
+	 * Generally, the function is notified to wake up by a call in another thread either to member notifyOne() or to
+	 * member notifyAll(). But certain implementations may produce spurious wake-up calls without any of these functions
+	 * being called. Therefore, users of this function shall ensure their condition for resumption is met.
+	 *
+	 * If condition is specified, the function only blocks if condition returns false, and notifications can only unblock
+	 * the thread when it becomes true (which is especially useful to check against spurious wake-up calls).
+	 *
+	 * @param lk lock object used by method wait for data protection
+	 * @param condition A callable object or function that takes no arguments and returns a value that can be evaluated
+	 * as a bool. This is called repeatedly until it evaluates to true.
+	 * @return false if timeout reached or true if wakeup condition is true
+	 */
+	virtual bool waitFor(PIMutex& lk, int timeoutMs, const std::function<bool()>& condition);
+
+private:
+	PRIVATE_DECLARATION(PIP_EXPORT)
+};
+
+
+#endif // PICONDITIONVAR_H

lib/main/thread/pigrabberbase.h

@@ -28,7 +28,7 @@
 
 
 template<typename T = PIByteArray>
-class PIP_EXPORT PIGrabberBase: public PIThread
+class PIGrabberBase: public PIThread
 {
 	PIOBJECT_SUBCLASS(PIGrabberBase, PIThread)
 public:

lib/main/thread/pimutex.cpp

@@ -1,7 +1,7 @@
 /*
 	PIP - Platform Independent Primitives
 	Mutex
-	Ivan Pelipenko peri4ko@yandex.ru, Andrey Bychkov work.a.b@yandex.ru
+	Ivan Pelipenko peri4ko@yandex.ru, Stephan Fomenko, Andrey Bychkov work.a.b@yandex.ru
 
 	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
@@ -19,102 +19,88 @@
 
 /** \class PIMutex
  * \brief Mutex
- * \details 
+ * \details
  * \section PIMutex_sec0 Synopsis
  * %PIMutex provides synchronization blocks between several threads.
  * Using mutex guarantees execution of some code only one of threads.
  * Mutex contains logic state and functions to change it: \a lock(),
  * \a unlock() and \a tryLock().
- * 
+ *
  * \section PIMutex_sec1 Usage
  * Block of code that should to be executed only one thread simultaniously
  * should to be started with \a lock() and ended with \a unlock().
  * \snippet pimutex.cpp main
  * "mutex" in this example is one for all threads.
- * 
+ *
  * */
 
 #include "pimutex.h"
 #include "piincludes_p.h"
-#ifdef BLACKBERRY
+#ifdef WINDOWS
+#  include <synchapi.h>
+#else
 #  include <pthread.h>
 #endif
 
 
+
 PRIVATE_DEFINITION_START(PIMutex)
 #ifdef WINDOWS
-	HANDLE
+	CRITICAL_SECTION
 #else
 	pthread_mutex_t
 #endif
-		mutex;
+					mutex;
 PRIVATE_DEFINITION_END(PIMutex)
 
 
-PIMutex::PIMutex(): inited_(false) {
-	//printf("new Mutex %p\n", this);
-#ifdef WINDOWS
-	PRIVATE->mutex = 0;
-#endif
+PIMutex::PIMutex() {
 	init();
 }
 
 
 PIMutex::~PIMutex() {
-	//printf("del Mutex %p\n", this);
 	destroy();
 }
 
 
 void PIMutex::lock() {
 #ifdef WINDOWS
-//	std::cout << (ullong)PRIVATE->mutex << "locking..." << std::endl;
-//	DWORD wr =
-			WaitForSingleObject(PRIVATE->mutex, INFINITE);
-//	std::cout << (ullong)PRIVATE->mutex << " lock wr=" << wr << std::endl;
+	EnterCriticalSection(&(PRIVATE->mutex));
 #else
 	pthread_mutex_lock(&(PRIVATE->mutex));
 #endif
-	locked = true;
 }
 
 
 void PIMutex::unlock() {
 #ifdef WINDOWS
-//	BOOL wr =
-//			ReleaseMutex(PRIVATE->mutex);
-			SetEvent(PRIVATE->mutex);
-//	std::cout << (ullong)PRIVATE->mutex << " unlock wr=" << wr << std::endl;
+	LeaveCriticalSection(&(PRIVATE->mutex));
 #else
 	pthread_mutex_unlock(&(PRIVATE->mutex));
 #endif
-	locked = false;
 }
 
 
 bool PIMutex::tryLock() {
 	bool ret =
 #ifdef WINDOWS
-			(WaitForSingleObject(PRIVATE->mutex, 0) == WAIT_OBJECT_0);
+			(TryEnterCriticalSection(&(PRIVATE->mutex)) != 0);
 #else
 			(pthread_mutex_trylock(&(PRIVATE->mutex)) == 0);
 #endif
-	locked = true;
 	return ret;
 }
 
 
-bool PIMutex::isLocked() const {
-//	std::cout << "test " << (ullong)PRIVATE->mutex << std::endl;
-	return locked;
+void *  PIMutex::handle() {
+	return (void*)&(PRIVATE->mutex);
 }
 
 
 void PIMutex::init() {
-	if (inited_) destroy();
 #ifdef WINDOWS
-	PRIVATE->mutex = CreateEvent(NULL, FALSE, TRUE, NULL);
-//	std::cout << "create " << (ullong)PRIVATE->mutex << std::endl;
+	InitializeCriticalSection(&(PRIVATE->mutex));
 #else
 	pthread_mutexattr_t attr;
 	memset(&attr, 0, sizeof(attr));
@@ -124,20 +110,13 @@ void PIMutex::init() {
 	pthread_mutex_init(&(PRIVATE->mutex), &attr);
 	pthread_mutexattr_destroy(&attr);
 #endif
-	locked = false;
-	inited_ = true;
 }
 
 
 void PIMutex::destroy() {
-	if (inited_) {
 #ifdef WINDOWS
-//		std::cout << "destroy " << (ullong)PRIVATE->mutex << std::endl;
-		if (PRIVATE->mutex) CloseHandle(PRIVATE->mutex);
-		PRIVATE->mutex = 0;
+	DeleteCriticalSection(&(PRIVATE->mutex));
 #else
-		pthread_mutex_destroy(&(PRIVATE->mutex));
+	pthread_mutex_destroy(&(PRIVATE->mutex));
 #endif
-	}
-	locked = inited_ = false;
 }

lib/main/thread/pimutex.h

@@ -1,10 +1,10 @@
 /*! \file pimutex.h
- * \brief Mutex
+ * \brief PIMutexLocker
 */
 /*
 	PIP - Platform Independent Primitives
-	Mutex
-	Ivan Pelipenko peri4ko@yandex.ru, Andrey Bychkov work.a.b@yandex.ru
+	PIMutexLocker
+	Ivan Pelipenko peri4ko@yandex.ru, Stephan Fomenko, Andrey Bychkov work.a.b@yandex.ru
 
 	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
@@ -24,13 +24,17 @@
 #define PIMUTEX_H
 
 #include "piinit.h"
+#include <mutex>
+
 
 class PIP_EXPORT PIMutex
 {
 public:
+	NO_COPY_CLASS(PIMutex)
+
 	//! Constructs unlocked mutex
 	explicit PIMutex();
-	
+
 	//! Destroy mutex
 	~PIMutex();
 
@@ -39,39 +43,43 @@ public:
 	//! \details If mutex is unlocked it set to locked state and returns immediate.
 	//! If mutex is already locked function blocks until mutex will be unlocked
 	void lock();
-	
+
 	//! \brief Unlock mutex
 	//! \details In any case this function returns immediate
-	void unlock();
-	
+	void unlock() ;
+
 	//! \brief Try to lock mutex
 	//! \details If mutex is unlocked it set to locked state and returns "true" immediate.
 	//! If mutex is already locked function returns immediate an returns "false"
 	bool tryLock();
-	
-	//! Returns if mutex is locked
-	bool isLocked() const;
-	
-private:
-	NO_COPY_CLASS(PIMutex)
 
+	void * handle();
+
+private:
 	void init();
 	void destroy();
-	bool inited_;
-	bool locked;
-	PRIVATE_DECLARATION
+
+	PRIVATE_DECLARATION(PIP_EXPORT)
+
 };
 
 
+//! \brief PIMutexLocker
+//! \details Same as std::lock_guard<std::mutex>.
+//! When a PIMutexLocker object is created, it attempts to lock the mutex it is given, if "condition" true.
+//! When control leaves the scope in which the PIMutexLocker object was created,
+//! the PIMutexLocker is destructed and the mutex is released, if "condition" true.
+//! If "condition" false this class do nothing.
+//! The PIMutexLocker class is non-copyable.
 class PIP_EXPORT PIMutexLocker
 {
 public:
-	PIMutexLocker(PIMutex * m, bool condition = true): mutex(m), cond(condition) {if (cond) mutex->lock();}
-	PIMutexLocker(PIMutex & m, bool condition = true): mutex(&m), cond(condition) {if (cond) mutex->lock();}
-	~PIMutexLocker() {if (cond) mutex->unlock();}
+	NO_COPY_CLASS(PIMutexLocker)
+	PIMutexLocker(PIMutex & m, bool condition = true): mutex(m), cond(condition) {if (cond) mutex.lock();}
+	~PIMutexLocker() {if (cond) mutex.unlock();}
 private:
-	PIMutex * mutex;
-	volatile bool cond;
+	PIMutex & mutex;
+	bool cond;
 };
 
 #endif // PIMUTEX_H

lib/main/thread/pithread.cpp

@@ -184,7 +184,6 @@ PIThread::PIThread(void * data, ThreadFunc func, bool startNow, int timer_delay)
 }
 
 
-#ifdef PIP_CXX11_SUPPORT
 PIThread::PIThread(std::function<void ()> func, bool startNow, int timer_delay) {
 	PIINTROSPECTION_THREAD_NEW(this);
 	tid_ = -1;
@@ -196,7 +195,6 @@ PIThread::PIThread(std::function<void ()> func, bool startNow, int timer_delay)
 	delay_ = timer_delay;
 	if (startNow) start(timer_delay);
 }
-#endif
 
 
 PIThread::PIThread(bool startNow, int timer_delay): PIObject() {

lib/main/thread/pithread.h

@@ -60,25 +60,20 @@ public:
 static __PIThreadCollection_Initializer__ __PIThreadCollection_initializer__;
 
 
-#ifdef PIP_CXX11_SUPPORT
 typedef std::function<void(void *)> ThreadFunc;
-#else
-typedef void (*ThreadFunc)(void * );
-#endif
 
 class PIP_EXPORT PIThread: public PIObject
 {
 	PIOBJECT_SUBCLASS(PIThread, PIObject)
 	friend class PIIntrospectionThreads;
 public:
-	
+	NO_COPY_CLASS(PIThread)
+
 	//! Contructs thread with custom data "data", external function "func" and main loop delay "loop_delay".
 	PIThread(void * data, ThreadFunc func, bool startNow = false, int loop_delay = -1);
 
-#ifdef PIP_CXX11_SUPPORT
 	//! Contructs thread with external function "func" and main loop delay "loop_delay".
 	PIThread(std::function<void()> func, bool startNow = false, int loop_delay = -1);
-#endif
 
 	//! Contructs thread with main loop delay "loop_delay".
 	PIThread(bool startNow = false, int loop_delay = -1);
@@ -98,10 +93,8 @@ public:
 	EVENT_HANDLER1(bool, start, int, timer_delay);
 	bool start(ThreadFunc func) {return start(func, -1);}
 	bool start(ThreadFunc func, int timer_delay) {ret_func = func; return start(timer_delay);}
-#ifdef PIP_CXX11_SUPPORT
 	bool start(std::function<void()> func) {return start(func, -1);}
 	bool start(std::function<void()> func, int timer_delay) {ret_func = [func](void*){func();}; return start(timer_delay);}
-#endif
 	EVENT_HANDLER0(bool, startOnce);
 	EVENT_HANDLER1(bool, startOnce, ThreadFunc, func) {ret_func = func; return startOnce();}
 	EVENT_HANDLER0(void, stop) {stop(false);}
@@ -114,10 +107,8 @@ public:
 	//! \brief Set external function that will be executed after every \a run()
 	void setSlot(ThreadFunc func) {ret_func = func;}
 
-#ifdef PIP_CXX11_SUPPORT
 	//! \brief Set external function that will be executed after every \a run()
 	void setSlot(std::function<void()> func) {ret_func = [func](void*){func();};}
-#endif
 
 	//! \brief Set priority of thread
 	void setPriority(PIThread::Priority prior);
@@ -160,11 +151,9 @@ public:
 	//! and automatically delete it on function finish
 	static void runOnce(PIObject * object, const char * handler, const PIString & name = PIString());
 
-#ifdef PIP_CXX11_SUPPORT
 	//! \brief Start function \"func\" in separate thread with name \"name\"
 	//! and automatically delete it on function finish
 	static void runOnce(std::function<void()> func, const PIString & name = PIString());
-#endif
 
 //! \handlers
 //! \{
@@ -246,7 +235,7 @@ protected:
 	//! Function executed once at the end of thread.
 	virtual void end() {;}
 
-	volatile bool terminating, running_, lockRun;
+	std::atomic_bool terminating, running_, lockRun;
 	int delay_, policy_;
 	llong tid_;
 	void * data_;
@@ -254,11 +243,9 @@ protected:
 	PITimeMeasurer tmf_, tms_, tmr_;
 	PIThread::Priority priority_;
 	ThreadFunc ret_func;
-	PRIVATE_DECLARATION
+	PRIVATE_DECLARATION(PIP_EXPORT)
 
 private:
-	NO_COPY_CLASS(PIThread)
-
 	bool _startThread(void * func);
 	void _beginThread();
 	void _runThread();

lib/main/thread/pithreadmodule.h

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

lib/main/thread/pithreadpoolexecutor.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 PIP_EXECUTOR_H
+#define PIP_EXECUTOR_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 PIP_EXPORT 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 //PIP_EXECUTOR_H

lib/main/thread/pitimer.cpp

@@ -479,7 +479,6 @@ PITimer::PITimer(TimerEvent slot, void * data, PITimer::TimerImplementation ti):
 }
 
 
-#ifdef PIP_CXX11_SUPPORT
 PITimer::PITimer(std::function<void ()> slot, PITimer::TimerImplementation ti) {
 	imp_mode = ti;
 	initFirst();
@@ -493,7 +492,7 @@ PITimer::PITimer(std::function<void (void *)> slot, void * data, PITimer::TimerI
 	data_t = data;
 	ret_func = [slot](void *d, int){slot(d);};
 }
-#endif
+
 
 PITimer::~PITimer() {
 	destroy();

lib/main/thread/pitimer.h

@@ -26,11 +26,7 @@
 #include "pithread.h"
 #include "pitime.h"
 
-#ifdef PIP_CXX11_SUPPORT
 typedef std::function<void(void *, int)> TimerEvent;
-#else
-typedef void (*TimerEvent)(void *, int);
-#endif
 
 class PITimer;
 
@@ -68,7 +64,7 @@ protected:
 	
 	double interval_, deferred_delay;
 	bool deferred_, deferred_mode; // mode: true - date, false - delay
-	volatile bool running_;
+	std::atomic_bool running_;
 	PIDateTime deferred_datetime;
 };
 
@@ -78,7 +74,8 @@ protected:
 class PIP_EXPORT PITimer: public PIObject {
 	PIOBJECT_SUBCLASS(PITimer, PIObject)
 public:
-	
+	NO_COPY_CLASS(PITimer)
+
 	//! \brief Constructs timer with PITimer::Thread implementation
 	explicit PITimer();
 
@@ -96,13 +93,11 @@ public:
 	//! \brief Constructs timer with "slot" slot void(void *,int), "data" data and "ti" implementation
 	explicit PITimer(TimerEvent slot, void * data = 0, TimerImplementation ti = Thread);
 
-#ifdef PIP_CXX11_SUPPORT
 	//! \brief Constructs timer with "slot" slot void(), and "ti" implementation
 	explicit PITimer(std::function<void ()> slot, TimerImplementation ti = Thread);
 
 	//! \brief Constructs timer with "slot" slot void(void *), "data" data and "ti" implementation
 	explicit PITimer(std::function<void (void *)> slot, void * data, TimerImplementation ti = Thread);
-#endif
 	
 	virtual ~PITimer();
 	
@@ -159,13 +154,11 @@ public:
 	//! \brief Set timer tick function
 	void setSlot(TimerEvent slot) {ret_func = slot;}
 
-#ifdef PIP_CXX11_SUPPORT
 	//! \brief Set timer tick function
 	void setSlot(std::function<void ()> slot) {ret_func = [slot](void *, int){slot();};}
 
 	//! \brief Set timer tick function
 	void setSlot(std::function<void (void *)> slot) {ret_func = [slot](void *d, int){slot(d);};}
-#endif
 
 	//! \brief Returns common data passed to tick functions
 	void * data() const {return data_t;}
@@ -185,13 +178,11 @@ public:
 	//! \brief Add frequency delimiter \b delim with optional delimiter slot \b slot.
 	void addDelimiter(int delim, TimerEvent slot = 0) {delims << Delimiter(slot, delim);}
 
-#ifdef PIP_CXX11_SUPPORT
 	//! \brief Add frequency delimiter \b delim with optional delimiter slot \b slot.
 	void addDelimiter(int delim, std::function<void ()> slot) {delims << Delimiter([slot](void *, int){slot();}, delim);}
 
 	//! \brief Add frequency delimiter \b delim with optional delimiter slot \b slot.
 	void addDelimiter(int delim, std::function<void (void *)> slot) {delims << Delimiter([slot](void *d, int){slot(d);}, delim);}
-#endif
 	
 	//! \brief Remove all frequency delimiters \b delim.
 	void removeDelimiter(int delim) {for (int i = 0; i < delims.size_s(); ++i) if (delims[i].delim == delim) {delims.remove(i); i--;}}
@@ -254,17 +245,13 @@ protected:
 	virtual void tick(void * data_, int delimiter) {}
 	
 	void * data_t;
-	volatile bool lockRun, callEvents;
+	std::atomic_bool lockRun, callEvents;
 	PIMutex mutex_;
 	TimerEvent ret_func;
 	TimerImplementation imp_mode;
 	PIVector<Delimiter> delims;
 	
 	mutable _PITimerBase * imp;
-
-private:
-	NO_COPY_CLASS(PITimer)
-
 };