SHS/pip
4
3
Fork 0
Code Issues 67 Pull Requests 1 Actions Releases Wiki Activity

fix concurrent

Browse Source 
git-svn-id: svn://db.shs.com.ru/pip@884 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5
This commit is contained in:
Бычков Андрей
2020-02-25 15:58:02 +00:00
parent f8f627360a
commit 92ac2b12cf
7 changed files with 408 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
src_concurrent/executor.cpp
View File

@@ -1,54 +1,55 @@
//
// Created by fomenko on 23.09.2019.
//
#include "executor.h" #include "executor.h"
PIThreadPoolExecutor::PIThreadPoolExecutor(size_t corePoolSize, PIBlockingDequeue<std::function<void()>> *taskQueue_,
PIThreadFactory *threadFactory) : isShutdown_(false), taskQueue(taskQueue_), threadFactory(threadFactory) { PIThreadPoolExecutor::PIThreadPoolExecutor(size_t corePoolSize, PIBlockingDequeue<std::function<void()>> *taskQueue_) : isShutdown_(false), taskQueue(taskQueue_) {
for (size_t i = 0; i < corePoolSize; ++i) { for (size_t i = 0; i < corePoolSize; ++i) {
AbstractThread* thread = threadFactory->newThread([&, i](){ PIThread * thread = new PIThread([&, i](){
auto runnable = taskQueue->poll(100, std::function<void()>()); auto runnable = taskQueue->poll(100, std::function<void()>());
if (runnable) { if (runnable) {
runnable(); runnable();
} }
if (isShutdown_ && taskQueue->size() == 0) threadPool[i]->stop(); if (isShutdown_ && taskQueue->size() == 0) threadPool[i]->stop();
}); });
threadPool.push_back(thread); threadPool.push_back(thread);
thread->start(); thread->start();
} }
} }
bool PIThreadPoolExecutor::awaitTermination(int timeoutMs) { bool PIThreadPoolExecutor::awaitTermination(int timeoutMs) {
PITimeMeasurer measurer; PITimeMeasurer measurer;
for (size_t i = 0; i < threadPool.size(); ++i) { for (size_t i = 0; i < threadPool.size(); ++i) {
int dif = timeoutMs - (int)measurer.elapsed_m(); int dif = timeoutMs - (int)measurer.elapsed_m();
if (dif < 0) return false; if (dif < 0) return false;
if (!threadPool[i]->waitForFinish(dif)) return false; if (!threadPool[i]->waitForFinish(dif)) return false;
} }
return true; return true;
} }
void PIThreadPoolExecutor::shutdownNow() { void PIThreadPoolExecutor::shutdownNow() {
isShutdown_ = true; isShutdown_ = true;
for (size_t i = 0; i < threadPool.size(); ++i) threadPool[i]->stop(); for (size_t i = 0; i < threadPool.size(); ++i) threadPool[i]->stop();
} }
PIThreadPoolExecutor::~PIThreadPoolExecutor() { PIThreadPoolExecutor::~PIThreadPoolExecutor() {
shutdownNow(); shutdownNow();
while (threadPool.size() > 0) delete threadPool.take_back(); while (threadPool.size() > 0) delete threadPool.take_back();
delete threadFactory; delete taskQueue;
delete taskQueue;
} }
void PIThreadPoolExecutor::execute(const std::function<void()> &runnable) { void PIThreadPoolExecutor::execute(const std::function<void()> &runnable) {
if (!isShutdown_) taskQueue->offer(runnable); if (!isShutdown_) taskQueue->offer(runnable);
} }
volatile bool PIThreadPoolExecutor::isShutdown() const { volatile bool PIThreadPoolExecutor::isShutdown() const {
return isShutdown_; return isShutdown_;
} }
void PIThreadPoolExecutor::shutdown() { void PIThreadPoolExecutor::shutdown() {
isShutdown_ = true; isShutdown_ = true;
} }

60
src_concurrent/piconditionlock.cpp
View File

@@ -1,7 +1,3 @@
//
// Created by fomenko on 25.09.2019.
//
#include "piconditionlock.h" #include "piconditionlock.h"
#ifdef WINDOWS #ifdef WINDOWS
#include "synchapi.h" #include "synchapi.h"
@@ -9,70 +5,82 @@
#include "pthread.h" #include "pthread.h"
#endif #endif
PRIVATE_DEFINITION_START(PIConditionLock) PRIVATE_DEFINITION_START(PIConditionLock)
#ifdef WINDOWS #ifdef WINDOWS
CRITICAL_SECTION CRITICAL_SECTION
#else #else
pthread_mutex_t pthread_mutex_t
#endif #endif
nativeHandle; nativeHandle;
PRIVATE_DEFINITION_END(PIConditionLock) PRIVATE_DEFINITION_END(PIConditionLock)
#ifdef WINDOWS #ifdef WINDOWS
PIConditionLock::PIConditionLock() { PIConditionLock::PIConditionLock() {
InitializeCriticalSection(&PRIVATE->nativeHandle); InitializeCriticalSection(&PRIVATE->nativeHandle);
} }
PIConditionLock::~PIConditionLock() { PIConditionLock::~PIConditionLock() {
DeleteCriticalSection(&PRIVATE->nativeHandle); DeleteCriticalSection(&PRIVATE->nativeHandle);
} }
void PIConditionLock::lock() { void PIConditionLock::lock() {
EnterCriticalSection(&PRIVATE->nativeHandle); EnterCriticalSection(&PRIVATE->nativeHandle);
} }
void PIConditionLock::unlock() { void PIConditionLock::unlock() {
LeaveCriticalSection(&PRIVATE->nativeHandle); LeaveCriticalSection(&PRIVATE->nativeHandle);
} }
void *PIConditionLock::handle() { void *PIConditionLock::handle() {
return &PRIVATE->nativeHandle; return &PRIVATE->nativeHandle;
} }
bool PIConditionLock::tryLock() { bool PIConditionLock::tryLock() {
return TryEnterCriticalSection(&PRIVATE->nativeHandle) != 0; return TryEnterCriticalSection(&PRIVATE->nativeHandle) != 0;
} }
#else #else
PIConditionLock::PIConditionLock() { PIConditionLock::PIConditionLock() {
pthread_mutexattr_t attr; pthread_mutexattr_t attr;
memset(&attr, 0, sizeof(attr)); memset(&attr, 0, sizeof(attr));
pthread_mutexattr_init(&attr); pthread_mutexattr_init(&attr);
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE); pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
memset(&(PRIVATE->nativeHandle), 0, sizeof(PRIVATE->nativeHandle)); memset(&(PRIVATE->nativeHandle), 0, sizeof(PRIVATE->nativeHandle));
pthread_mutex_init(&(PRIVATE->nativeHandle), &attr); pthread_mutex_init(&(PRIVATE->nativeHandle), &attr);
pthread_mutexattr_destroy(&attr); pthread_mutexattr_destroy(&attr);
} }
PIConditionLock::~PIConditionLock() { PIConditionLock::~PIConditionLock() {
pthread_mutex_destroy(&(PRIVATE->nativeHandle)); pthread_mutex_destroy(&(PRIVATE->nativeHandle));
} }
void PIConditionLock::lock() { void PIConditionLock::lock() {
pthread_mutex_lock(&(PRIVATE->nativeHandle)); pthread_mutex_lock(&(PRIVATE->nativeHandle));
} }
void PIConditionLock::unlock() { void PIConditionLock::unlock() {
pthread_mutex_unlock(&(PRIVATE->nativeHandle)); pthread_mutex_unlock(&(PRIVATE->nativeHandle));
} }
void *PIConditionLock::handle() { void *PIConditionLock::handle() {
return &PRIVATE->nativeHandle; return &PRIVATE->nativeHandle;
} }
bool PIConditionLock::tryLock() { bool PIConditionLock::tryLock() {
return (pthread_mutex_trylock(&(PRIVATE->nativeHandle)) == 0);; return (pthread_mutex_trylock(&(PRIVATE->nativeHandle)) == 0);
} }
#endif #endif

122
src_concurrent/piconditionvar.cpp
View File

@@ -1,8 +1,7 @@
//
// Created by fomenko on 20.09.2019.
//
#include "piplatform.h" #include "piplatform.h"
#include "piconditionvar.h"
#include "pithread.h"
#include "pitime.h"
#ifdef WINDOWS #ifdef WINDOWS
#define _WIN32_WINNT 0x0600 #define _WIN32_WINNT 0x0600
@@ -11,118 +10,113 @@
#include <winbase.h> #include <winbase.h>
#endif #endif
#include "piconditionvar.h"
#include "pithread.h"
#include "pitime.h"
PRIVATE_DEFINITION_START(PIConditionVariable) PRIVATE_DEFINITION_START(PIConditionVariable)
#ifdef WINDOWS #ifdef WINDOWS
CONDITION_VARIABLE nativeHandle; CONDITION_VARIABLE nativeHandle;
#else #else
pthread_cond_t nativeHandle; pthread_cond_t nativeHandle;
PIConditionLock* currentLock; PIConditionLock* currentLock;
#endif #endif
bool isDestroying; bool isDestroying;
PRIVATE_DEFINITION_END(PIConditionVariable) PRIVATE_DEFINITION_END(PIConditionVariable)
PIConditionVariable::PIConditionVariable() { PIConditionVariable::PIConditionVariable() {
#ifdef WINDOWS #ifdef WINDOWS
InitializeConditionVariable(&PRIVATE->nativeHandle); InitializeConditionVariable(&PRIVATE->nativeHandle);
#else #else
PRIVATE->isDestroying = false; PRIVATE->isDestroying = false;
PRIVATE->currentLock = nullptr; PRIVATE->currentLock = nullptr;
memset(&(PRIVATE->nativeHandle), 0, sizeof(PRIVATE->nativeHandle)); memset(&(PRIVATE->nativeHandle), 0, sizeof(PRIVATE->nativeHandle));
pthread_cond_init(&PRIVATE->nativeHandle, NULL); pthread_cond_init(&PRIVATE->nativeHandle, NULL);
#endif #endif
} }
PIConditionVariable::~PIConditionVariable() { PIConditionVariable::~PIConditionVariable() {
#ifdef WINDOWS #ifdef WINDOWS
#else #else
pthread_cond_destroy(&PRIVATE->nativeHandle); pthread_cond_destroy(&PRIVATE->nativeHandle);
#endif #endif
} }
void PIConditionVariable::wait(PIConditionLock& lk) { void PIConditionVariable::wait(PIConditionLock& lk) {
#ifdef WINDOWS #ifdef WINDOWS
SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), INFINITE); SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), INFINITE);
#else #else
pthread_cond_wait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle()); pthread_cond_wait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle());
#endif #endif
} }
void PIConditionVariable::wait(PIConditionLock& lk, const std::function<bool()>& condition) { void PIConditionVariable::wait(PIConditionLock& lk, const std::function<bool()>& condition) {
bool isCondition; bool isCondition;
while (true) { while (true) {
isCondition = condition(); isCondition = condition();
if (isCondition) break; if (isCondition) break;
#ifdef WINDOWS #ifdef WINDOWS
SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), INFINITE); SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), INFINITE);
#else #else
pthread_cond_wait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle()); pthread_cond_wait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle());
#endif #endif
if (PRIVATE->isDestroying) return; if (PRIVATE->isDestroying) return;
} }
} }
bool PIConditionVariable::waitFor(PIConditionLock &lk, int timeoutMs) { bool PIConditionVariable::waitFor(PIConditionLock &lk, int timeoutMs) {
bool isNotTimeout; bool isNotTimeout;
#ifdef WINDOWS #ifdef WINDOWS
isNotTimeout = SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), timeoutMs) != 0; isNotTimeout = SleepConditionVariableCS(&PRIVATE->nativeHandle, (PCRITICAL_SECTION)lk.handle(), timeoutMs) != 0;
#else #else
timespec abstime = {.tv_sec = timeoutMs / 1000, .tv_nsec = timeoutMs * 1000 * 1000}; timespec abstime = {.tv_sec = timeoutMs / 1000, .tv_nsec = timeoutMs * 1000 * 1000};
isNotTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0; isNotTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0;
#endif #endif
if (PRIVATE->isDestroying) return false; if (PRIVATE->isDestroying) return false;
return isNotTimeout; return isNotTimeout;
} }
bool PIConditionVariable::waitFor(PIConditionLock& lk, int timeoutMs, const std::function<bool()> &condition) { bool PIConditionVariable::waitFor(PIConditionLock& lk, int timeoutMs, const std::function<bool()> &condition) {
bool isCondition; bool isCondition;
PITimeMeasurer measurer; PITimeMeasurer measurer;
while (true) { while (true) {
isCondition = condition(); isCondition = condition();
if (isCondition) break; if (isCondition) break;
#ifdef WINDOWS #ifdef WINDOWS
WINBOOL isTimeout = SleepConditionVariableCS( WINBOOL isTimeout = SleepConditionVariableCS(
&PRIVATE->nativeHandle, &PRIVATE->nativeHandle,
(PCRITICAL_SECTION)lk.handle(), (PCRITICAL_SECTION)lk.handle(),
timeoutMs - (int)measurer.elapsed_m()); timeoutMs - (int)measurer.elapsed_m());
if (isTimeout == 0) return false; if (isTimeout == 0) return false;
#else #else
int timeoutCurr = timeoutMs - (int)measurer.elapsed_m(); int timeoutCurr = timeoutMs - (int)measurer.elapsed_m();
timespec abstime = {.tv_sec = timeoutCurr / 1000, .tv_nsec = timeoutCurr * 1000 * 1000}; 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; bool isTimeout = pthread_cond_timedwait(&PRIVATE->nativeHandle, (pthread_mutex_t*)lk.handle(), &abstime) == 0;
if (isTimeout) return false; if (isTimeout) return false;
#endif #endif
if (PRIVATE->isDestroying) return false; if (PRIVATE->isDestroying) return false;
} }
return true; return true;
} }
void PIConditionVariable::notifyOne() { void PIConditionVariable::notifyOne() {
#ifdef WINDOWS #ifdef WINDOWS
WakeConditionVariable(&PRIVATE->nativeHandle); WakeConditionVariable(&PRIVATE->nativeHandle);
#else #else
pthread_cond_signal(&PRIVATE->nativeHandle); pthread_cond_signal(&PRIVATE->nativeHandle);
#endif #endif
} }
void PIConditionVariable::notifyAll() { void PIConditionVariable::notifyAll() {
#ifdef WINDOWS #ifdef WINDOWS
WakeAllConditionVariable(&PRIVATE->nativeHandle); WakeAllConditionVariable(&PRIVATE->nativeHandle);
#else #else
pthread_cond_broadcast(&PRIVATE->nativeHandle); pthread_cond_broadcast(&PRIVATE->nativeHandle);
#endif #endif
} }
void StdFunctionThreadFuncAdapter::threadFuncStdFunctionAdapter(void *it) {
auto consumer = (StdFunctionThreadFuncAdapter*)it;
consumer->fun();
}
void StdFunctionThreadFuncAdapter::registerToInvoke(PIThread *thread) {
thread->setData(data());
thread->setSlot((ThreadFunc) threadFunc());
}

83
src_main/concurrent/executor.h
View File

@@ -5,46 +5,8 @@
#ifndef PIP_TESTS_EXECUTOR_H #ifndef PIP_TESTS_EXECUTOR_H
#define PIP_TESTS_EXECUTOR_H #define PIP_TESTS_EXECUTOR_H
#include <pithread.h>
#include <functional>
#include <utility>
#include "piblockingdequeue.h" #include "piblockingdequeue.h"
class AbstractThread {
public:
virtual bool start() = 0;
virtual bool waitForStart(int timeout_msecs) = 0;
virtual bool waitForFinish(int timeout_msecs) = 0;
virtual void stop() = 0;
virtual ~AbstractThread() = default;
};
class Thread : public AbstractThread {
public:
explicit Thread(const std::function<void()>& fun = [](){}) : adapter(fun) {
adapter.registerToInvoke(&thread);
}
virtual ~Thread() = default;
inline bool start() override { return thread.start(); }
inline bool waitForStart(int timeout_msecs) override { return thread.waitForStart(timeout_msecs); }
inline bool waitForFinish(int timeout_msecs) override { return thread.waitForFinish(timeout_msecs); }
inline void stop() override { thread.stop(); }
private:
PIThread thread;
StdFunctionThreadFuncAdapter adapter;
};
class PIThreadFactory {
public:
inline virtual AbstractThread* newThread(const std::function<void()>& fun) {
return new Thread(fun);
}
virtual ~PIThreadFactory() = default;
};
/** /**
* @brief Thread pools address two different problems: they usually provide improved performance when executing large * @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 * numbers of asynchronous tasks, due to reduced per-task invocation overhead, and they provide a means of bounding and
@@ -52,36 +14,35 @@ public:
*/ */
class PIThreadPoolExecutor { class PIThreadPoolExecutor {
public: public:
explicit PIThreadPoolExecutor(size_t corePoolSize = 1, PIBlockingDequeue<std::function<void()> >* taskQueue_ = new PIBlockingDequeue<std::function<void()> >(), PIThreadFactory* threadFactory = new PIThreadFactory()); explicit PIThreadPoolExecutor(size_t corePoolSize = 1, PIBlockingDequeue<std::function<void()> >* taskQueue_ = new PIBlockingDequeue<std::function<void()> >());
virtual ~PIThreadPoolExecutor(); virtual ~PIThreadPoolExecutor();
/** /**
* @brief Executes the given task sometime in the future. The task execute in an existing pooled thread. If the task * @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 * cannot be submitted for execution, either because this executor has been shutdown or because its capacity has been
* reached. * reached.
* *
* @param runnable not empty function for thread pool execution * @param runnable not empty function for thread pool execution
*/ */
void execute(const std::function<void()>& runnable); void execute(const std::function<void()>& runnable);
void shutdownNow(); void shutdownNow();
/** /**
* @brief Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be * @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 * 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. * submitted tasks to complete execution. Use awaitTermination to do that.
*/ */
void shutdown(); void shutdown();
volatile bool isShutdown() const; volatile bool isShutdown() const;
bool awaitTermination(int timeoutMs); bool awaitTermination(int timeoutMs);
private: private:
volatile bool isShutdown_; volatile bool isShutdown_;
PIBlockingDequeue<std::function<void()> >* taskQueue; PIBlockingDequeue<std::function<void()> >* taskQueue;
PIThreadFactory* threadFactory; PIVector<PIThread*> threadPool;
PIVector<AbstractThread*> threadPool;
}; };
#endif //PIP_TESTS_EXECUTOR_H #endif //PIP_TESTS_EXECUTOR_H

340
src_main/concurrent/piblockingdequeue.h
View File

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

35
src_main/concurrent/piconditionlock.h
View File

@@ -1,36 +1,31 @@
//
// Created by fomenko on 25.09.2019.
//
#ifndef AWRCANFLASHER_PICONDITIONLOCK_H #ifndef AWRCANFLASHER_PICONDITIONLOCK_H
#define AWRCANFLASHER_PICONDITIONLOCK_H #define AWRCANFLASHER_PICONDITIONLOCK_H
#include <pimutex.h> #include "pimutex.h"
#include <piinit.h>
/** /**
* @brief Continued * @brief Continued
*/ */
class PIP_EXPORT PIConditionLock { class PIP_EXPORT PIConditionLock {
public: public:
explicit PIConditionLock(); explicit PIConditionLock();
virtual ~PIConditionLock(); virtual ~PIConditionLock();
/** //! \brief lock
* @brief lock void lock();
*/
void lock();
/** //! \brief unlock
* @brief unlock void unlock();
*/
void unlock(); //! \brief tryLock
bool tryLock();
void * handle();
bool tryLock();
void* handle();
private: private:
NO_COPY_CLASS(PIConditionLock) NO_COPY_CLASS(PIConditionLock)
PRIVATE_DECLARATION PRIVATE_DECLARATION
}; };

168
src_main/concurrent/piconditionvar.h
View File

@@ -1,7 +1,3 @@
//
// Created by fomenko on 20.09.2019.
//
#ifndef PIP_TESTS_PICONDITIONVAR_H #ifndef PIP_TESTS_PICONDITIONVAR_H
#define PIP_TESTS_PICONDITIONVAR_H #define PIP_TESTS_PICONDITIONVAR_H
@@ -9,6 +5,7 @@
#include "pithread.h" #include "pithread.h"
#include "piinit.h" #include "piinit.h"
/** /**
* @brief A condition variable is an object able to block the calling thread until notified to resume. * @brief A condition variable is an object able to block the calling thread until notified to resume.
* *
@@ -17,103 +14,88 @@
*/ */
class PIP_EXPORT PIConditionVariable { class PIP_EXPORT PIConditionVariable {
public: public:
explicit PIConditionVariable(); explicit PIConditionVariable();
virtual ~PIConditionVariable(); virtual ~PIConditionVariable();
/** /**
* @brief Unblocks one of the threads currently waiting for this condition. If no threads are waiting, the function * @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. * does nothing. If more than one, it is unspecified which of the threads is selected.
*/ */
virtual void notifyOne(); virtual void notifyOne();
/** /**
* @brief Unblocks all threads currently waiting for this condition. If no threads are waiting, the function does * @brief Unblocks all threads currently waiting for this condition. If no threads are waiting, the function does
* nothing. * nothing.
*/ */
virtual void notifyAll(); virtual void notifyAll();
/** /**
* @brief see wait(PIConditionLock&, const std::function<bool()>&) * @brief see wait(PIConditionLock&, const std::function<bool()>&)
*/ */
virtual void wait(PIConditionLock& lk); virtual void wait(PIConditionLock& lk);
/** /**
* @brief Wait until notified * @brief Wait until notified
* *
* The execution of the current thread (which shall have locked with lk method PIConditionLock::lock()) is blocked * The execution of the current thread (which shall have locked with lk method PIConditionLock::lock()) is blocked
* until notified. * until notified.
* *
* At the moment of blocking the thread, the function automatically calls lk.unlock() (PIConditionLock::unlock()), * At the moment of blocking the thread, the function automatically calls lk.unlock() (PIConditionLock::unlock()),
* allowing other locked threads to continue. * allowing other locked threads to continue.
* *
* Once notified (explicitly, by some other thread), the function unblocks and calls lk.lock() (PIConditionLock::lock()), * Once notified (explicitly, by some other thread), the function unblocks and calls lk.lock() (PIConditionLock::lock()),
* leaving lk in the same state as when the function was called. Then the function returns (notice that this last mutex * 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). * 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 * 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 * 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. * 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 * 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). * 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 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 * @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. * as a bool. This is called repeatedly until it evaluates to true.
*/ */
virtual void wait(PIConditionLock& lk, const std::function<bool()>& condition); virtual void wait(PIConditionLock& lk, const std::function<bool()>& condition);
/** /**
* @brief see waitFor(PIConditionLock&, int, const std::function<bool()>&) * @brief see waitFor(PIConditionLock&, int, const std::function<bool()>&)
*/ */
virtual bool waitFor(PIConditionLock& lk, int timeoutMs); virtual bool waitFor(PIConditionLock& lk, int timeoutMs);
/**
* @brief Wait for timeout or until notified
*
* The execution of the current thread (which shall have locked with lk method PIConditionLock::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() (PIConditionLock::lock()), allowing
* other locked threads to continue.
*
* Once notified or once timeoutMs has passed, the function unblocks and calls lk.unlock() (PIConditionLock::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(PIConditionLock& lk, int timeoutMs, const std::function<bool()>& condition);
/**
* @brief Wait for timeout or until notified
*
* The execution of the current thread (which shall have locked with lk method PIConditionLock::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() (PIConditionLock::lock()), allowing
* other locked threads to continue.
*
* Once notified or once timeoutMs has passed, the function unblocks and calls lk.unlock() (PIConditionLock::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(PIConditionLock& lk, int timeoutMs, const std::function<bool()>& condition);
private: private:
NO_COPY_CLASS(PIConditionVariable) NO_COPY_CLASS(PIConditionVariable)
PRIVATE_DECLARATION
PRIVATE_DECLARATION
}; };
// FIXME: remove that!
class StdFunctionThreadFuncAdapter {
public:
static void threadFuncStdFunctionAdapter(void* it);
explicit StdFunctionThreadFuncAdapter(const std::function<void()>& fun_): fun(fun_) {}
void registerToInvoke(PIThread* thread);
void* data() const { return (void*)this; }
ThreadFunc threadFunc() const { return threadFuncStdFunctionAdapter; }
private:
std::function<void()> fun;
};
#endif //PIP_TESTS_PICONDITIONVAR_H #endif //PIP_TESTS_PICONDITIONVAR_H