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

PIThreadPoolExecutor & PIBlockingDequeue improvements

Browse Source 
- add support move & copy semantic
- introduce submit method for executor with future result
This commit is contained in:
Фоменко Степан Владимирович
2020-08-05 22:59:33 +03:00
parent 3ec1ecfb5b
commit 3cfdda7365
8 changed files with 193 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
lib/main/thread/piblockingdequeue.h
View File

@@ -20,6 +20,7 @@
#ifndef PIBLOCKINGDEQUEUE_H
#define PIBLOCKINGDEQUEUE_H
#include <queue>
#include "pideque.h"
#include "piconditionvar.h"
@@ -28,8 +29,9 @@
* wait for space to become available in the queue when storing an element.
*/
template <typename T>
class PIBlockingDequeue: private PIDeque<T> {
class PIBlockingDequeue {
public:
typedef typename std::deque<T> QueueType;
/**
* @brief Constructor
@@ -42,21 +44,21 @@ public:
/**
* @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 QueueType& other) : cond_var_add(new PIConditionVariable()), cond_var_rem(new PIConditionVariable()) {
mutex.lock();
max_size = SIZE_MAX;
PIDeque<T>::append(other);
data_queue = QueueType(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()) {
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));
data_queue = QueueType(other.data_queue);
mutex.unlock();
other.mutex.unlock();
}
@@ -71,10 +73,10 @@ public:
*
* @param v the element to add
*/
void put(const T & v) {
void put(T && v) {
mutex.lock();
cond_var_rem->wait(mutex, [&]() { return PIDeque<T>::size() < max_size; });
PIDeque<T>::push_back(v);
cond_var_rem->wait(mutex, [&]() { return data_queue.size() < max_size; });
data_queue.push_back(std::forward<T>(v));
mutex.unlock();
cond_var_add->notifyOne();
}
@@ -86,13 +88,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(T && v) {
mutex.lock();
if (PIDeque<T>::size() >= max_size) {
if (data_queue.size() >= max_size) {
mutex.unlock();
return false;
}
PIDeque<T>::push_back(v);
data_queue.push_back(std::forward<T>(v));
mutex.unlock();
cond_var_add->notifyOne();
return true;
@@ -106,10 +108,10 @@ public:
* @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) {
bool offer(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);
bool isOk = cond_var_rem->waitFor(mutex, timeoutMs, [&]() { return data_queue.size() < max_size; } );
if (isOk) data_queue.push_back(std::forward<T>(v));
mutex.unlock();
if (isOk) cond_var_add->notifyOne();
return isOk;
@@ -121,10 +123,10 @@ public:
* @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());
cond_var_add->wait(mutex, [&]() { return !data_queue.empty(); });
T t = std::move(data_queue.front());
data_queue.pop_front();
mutex.unlock();
cond_var_rem->notifyOne();
return t;
@@ -140,11 +142,16 @@ 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 t;
T poll(int timeoutMs, T && defaultVal = T(), bool * isOk = nullptr) {
mutex.lock();
bool isNotEmpty = cond_var_add->waitFor(mutex, timeoutMs, [&]() { return !PIDeque<T>::isEmpty(); });
t = isNotEmpty ? T(PIDeque<T>::take_front()) : defaultVal;
bool isNotEmpty = cond_var_add->waitFor(mutex, timeoutMs, [&]() { return !data_queue.empty(); });
T t;
if (isNotEmpty) {
t = std::move(data_queue.front());
data_queue.pop_front();
} else {
t = std::move(defaultVal);
}
mutex.unlock();
if (isNotEmpty) cond_var_rem->notifyOne();
if (isOk) *isOk = isNotEmpty;
@@ -160,11 +167,16 @@ 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(const T & defaultVal = T(), bool * isOk = nullptr) {
T poll(T && defaultVal = T(), bool * isOk = nullptr) {
T t;
mutex.lock();
bool isNotEmpty = !PIDeque<T>::isEmpty();
t = isNotEmpty ? PIDeque<T>::take_front() : defaultVal;
bool isNotEmpty = !data_queue.empty();
if (isNotEmpty) {
t = std::move(data_queue.front());
data_queue.pop_front();
} else {
t = std::move(defaultVal);
}
mutex.unlock();
if (isNotEmpty) cond_var_rem->notifyOne();
if (isOk) *isOk = isNotEmpty;
@@ -193,7 +205,7 @@ public:
*/
size_t remainingCapacity() {
mutex.lock();
size_t c = max_size - PIDeque<T>::size();
size_t c = max_size - data_queue.size();
mutex.unlock();
return c;
}
@@ -203,7 +215,7 @@ public:
*/
size_t size() {
mutex.lock();
size_t s = PIDeque<T>::size();
size_t s = data_queue.size();
mutex.unlock();
return s;
}
@@ -211,10 +223,13 @@ public:
/**
* @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) {
size_t drainTo(QueueType& 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());
size_t count = ((maxCount > data_queue.size()) ? data_queue.size() : maxCount);
for (size_t i = 0; i < count; ++i) {
other.push_back(std::move(data_queue.front()));
data_queue.pop_front();
}
mutex.unlock();
return count;
}
@@ -225,10 +240,13 @@ public:
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();
size_t count = maxCount > data_queue.size() ? data_queue.size() : maxCount;
size_t otherRemainingCapacity = other.max_size - data_queue.size();
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.data_queue.push_back(std::move(data_queue.front()));
data_queue.pop_front();
}
other.mutex.unlock();
mutex.unlock();
return count;
@@ -237,6 +255,7 @@ public:
private:
PIMutex mutex;
PIConditionVariable * cond_var_add, * cond_var_rem;
QueueType data_queue;
size_t max_size;
};

74
lib/main/thread/piexecutor.h
View File

@@ -22,8 +22,47 @@
#include "piblockingdequeue.h"
#include <atomic>
#include <future>
template <typename Thread_, typename Dequeue_>
/**
* @brief Wrapper for custom invoke operator available function types.
* @note Source from: "Энтони Уильямс, Параллельное программирование на С++ в действии. Практика разработки многопоточных
* программ. Пер. с англ. Слинкин А. А. - M.: ДМК Пресс, 2012 - 672c.: ил." (page 387)
*/
class FunctionWrapper {
struct ImplBase {
virtual void call() = 0;
virtual ~ImplBase() = default;
};
std::unique_ptr<ImplBase> impl;
template<typename F>
struct ImplType: ImplBase {
F f;
explicit ImplType(F&& f): f(std::forward<F>(f)) {}
void call() final { f(); }
};
public:
template<typename F, typename = std::enable_if<!std::is_same<F, FunctionWrapper>::value> >
explicit FunctionWrapper(F&& f): impl(new ImplType<F>(std::forward<F>(f))) {}
void operator()() { impl->call(); }
explicit operator bool() const noexcept { return static_cast<bool>(impl); }
FunctionWrapper() = default;
FunctionWrapper(FunctionWrapper&& other) noexcept : impl(std::move(other.impl)) {}
FunctionWrapper& operator=(FunctionWrapper&& other) noexcept {
impl = std::move(other.impl);
return *this;
}
FunctionWrapper(const FunctionWrapper& other) = delete;
FunctionWrapper& operator=(const FunctionWrapper&) = delete;
};
template <typename Thread_, template<typename> class Dequeue_>
class PIThreadPoolExecutorTemplate {
public:
NO_COPY_CLASS(PIThreadPoolExecutorTemplate)
@@ -34,8 +73,27 @@ public:
while (threadPool.size() > 0) delete threadPool.take_back();
}
void execute(const std::function<void()> & runnable) {
if (!isShutdown_) taskQueue.offer(runnable);
template<typename FunctionType>
std::future<typename std::result_of<FunctionType()>::type> submit(FunctionType&& callable) {
typedef typename std::result_of<FunctionType()>::type ResultType;
if (!isShutdown_) {
std::packaged_task<ResultType()> callable_task(std::forward<FunctionType>(callable));
auto future = callable_task.get_future();
FunctionWrapper functionWrapper(callable_task);
taskQueue.offer(std::move(functionWrapper));
return future;
} else {
return std::future<ResultType>();
}
}
template<typename FunctionType>
void execute(FunctionType&& runnable) {
if (!isShutdown_) {
FunctionWrapper function_wrapper(std::forward<FunctionType>(runnable));
taskQueue.offer(std::move(function_wrapper));
}
}
void shutdown() {
@@ -63,15 +121,15 @@ public:
protected:
std::atomic_bool isShutdown_;
Dequeue_ taskQueue;
Dequeue_<FunctionWrapper> taskQueue;
PIVector<Thread_*> threadPool;
template<typename Function>
PIThreadPoolExecutorTemplate(size_t corePoolSize, Function onBeforeStart) : isShutdown_(false) {
makePool(corePoolSize, onBeforeStart);
PIThreadPoolExecutorTemplate(size_t corePoolSize, Function&& onBeforeStart) : isShutdown_(false) {
makePool(corePoolSize, std::forward<Function>(onBeforeStart));
}
void makePool(size_t corePoolSize, std::function<void(Thread_*)> onBeforeStart = [](Thread_*){}) {
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);
@@ -87,7 +145,7 @@ protected:
}
};
typedef PIThreadPoolExecutorTemplate<PIThread, PIBlockingDequeue<std::function<void()> > > PIThreadPoolExecutor;
typedef PIThreadPoolExecutorTemplate<PIThread, PIBlockingDequeue> PIThreadPoolExecutor;
#ifdef DOXYGEN
/**