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

add new PIThreadPoolWorker - rework of PIThreadPoolExecutor

Browse Source
This commit is contained in:
Пелипенко Иван
2026-03-24 14:09:41 +03:00
parent 93547beb38
commit 3102b985d5
9 changed files with 353 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
libs/main/thread/pithread.h
View File

@@ -217,6 +217,7 @@ public:
//! \~english Waits until the thread starts. Returns \b false if the timeout expires first.
//! \~russian Ожидает запуска потока. Возвращает \b false, если таймаут истек раньше.
bool waitForStart(PISystemTime timeout = {});
//! \~english Deprecated overload of \a waitForStart() that accepts milliseconds.
//! \~russian Устаревшая перегрузка \a waitForStart(), принимающая миллисекунды.
bool waitForStart(int timeout_msecs) DEPRECATEDM("use waitForStart(PISystemTime)") {
@@ -226,6 +227,7 @@ public:
//! \~english Waits for thread completion. Returns \b false if the timeout expires first.
//! \~russian Ожидает завершения потока. Возвращает \b false, если таймаут истек раньше.
bool waitForFinish(PISystemTime timeout = {});
//! \~english Deprecated overload of \a waitForFinish() that accepts milliseconds.
//! \~russian Устаревшая перегрузка \a waitForFinish(), принимающая миллисекунды.
bool waitForFinish(int timeout_msecs) DEPRECATEDM("use waitForFinish(PISystemTime)") {

2
libs/main/thread/pithreadmodule.h
View File

@@ -71,8 +71,8 @@
#include "pispinlock.h"
#include "pithread.h"
#include "pithreadnotifier.h"
#include "pithreadpoolexecutor.h"
#include "pithreadpoolloop.h"
#include "pithreadpoolworker.h"
#include "pitimer.h"
#endif // PITHREADMODULE_H

83
libs/main/thread/pithreadpoolexecutor.cpp
View File

@@ -1,83 +0,0 @@
/*
PIP - Platform Independent Primitives
Stephan Fomenko
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "pithreadpoolexecutor.h"
#include "piliterals_time.h"
/*! \class PIThreadPoolExecutor
* \brief Thread pools address two different problems: they usually provide improved performance when executing large
* numbers of asynchronous tasks, due to reduced per-task invocation overhead, and they provide a means of bounding and
* managing the resources, including threads, consumed when executing a collection of tasks.
*/
PIThreadPoolExecutor::PIThreadPoolExecutor(int corePoolSize): isShutdown_(false) {
for (int i = 0; i < corePoolSize; ++i) {
PIThread * thread = new PIThread([&, i]() {
auto runnable = taskQueue.poll(100_ms, std::function<void()>());
if (runnable) {
runnable();
}
if (isShutdown_ && taskQueue.size() == 0) threadPool[i]->stop();
});
threadPool.push_back(thread);
thread->start();
}
}
bool PIThreadPoolExecutor::awaitTermination(PISystemTime timeout) {
PITimeMeasurer measurer;
for (size_t i = 0; i < threadPool.size(); ++i) {
auto dif = timeout - measurer.elapsed();
if (dif.isNegative()) return false;
if (!threadPool[i]->waitForFinish(dif)) return false;
}
return true;
}
void PIThreadPoolExecutor::shutdownNow() {
isShutdown_ = true;
for (size_t i = 0; i < threadPool.size(); ++i)
threadPool[i]->stop();
}
PIThreadPoolExecutor::~PIThreadPoolExecutor() {
shutdownNow();
while (threadPool.size() > 0)
delete threadPool.take_back();
}
void PIThreadPoolExecutor::execute(std::function<void()> runnable) {
if (!isShutdown_) taskQueue.offer(std::move(runnable));
}
bool PIThreadPoolExecutor::isShutdown() const {
return isShutdown_;
}
void PIThreadPoolExecutor::shutdown() {
isShutdown_ = true;
}

99
libs/main/thread/pithreadpoolexecutor.h
View File

@@ -1,95 +1,6 @@
//! \~\file pithreadpoolexecutor.h
//! \~\ingroup Thread
//! \brief
//! \~english Thread pool executor
//! \~russian Исполнитель пула потоков
//!
//! \details
//! \~english Executes tasks in a pool of worker threads.
//! \~russian Выполняет задачи в пуле рабочих потоков.
/*
PIP - Platform Independent Primitives
#include "pithreadpoolworker.h"
Stephan Fomenko
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef PITHREADPOOLEXECUTOR_H
#define PITHREADPOOLEXECUTOR_H
#include "piblockingqueue.h"
#include "pithread.h"
#include <atomic>
//! \~\ingroup Thread
//! \~\brief
//! \~english Fixed-size pool of worker threads for fire-and-forget tasks.
//! \~russian Фиксированный пул рабочих потоков для задач без ожидания результата.
class PIP_EXPORT PIThreadPoolExecutor {
public:
//! \~english Constructs executor with \a corePoolSize worker threads.
//! \~russian Создает исполнитель с \a corePoolSize рабочими потоками.
explicit PIThreadPoolExecutor(int corePoolSize);
//! \~english Stops worker threads and destroys executor resources.
//! \~russian Останавливает рабочие потоки и уничтожает ресурсы исполнителя.
virtual ~PIThreadPoolExecutor();
//! \~\brief
//! \~english Submits \a runnable for asynchronous execution by a worker thread.
//! \~russian Передает \a runnable на асинхронное выполнение рабочим потоком.
//! \details
//! \~english
//! This is a best-effort fire-and-forget call and does not report whether the task was accepted.
//! \After shutdown requests new tasks are ignored.
//! \~russian
//! Это вызов по принципу best-effort без ожидания результата и без сообщения о том, была ли задача принята.
//! \После запроса на завершение новые задачи игнорируются.
void execute(std::function<void()> runnable);
//! \~english Requests immediate shutdown and stops worker threads without waiting for queued tasks to finish.
//! \~russian Запрашивает немедленное завершение и останавливает рабочие потоки без ожидания завершения задач в очереди.
void shutdownNow();
//! \~\brief
//! \~english Requests orderly shutdown: new tasks are rejected and workers stop after the current queue is drained.
//! \~russian Запрашивает упорядоченное завершение: новые задачи отклоняются, а рабочие потоки останавливаются после опустошения текущей
//! очереди.
//! \details
//! \~english This method does not wait for worker termination.
//! \~russian Этот метод не ожидает завершения рабочих потоков.
void shutdown();
//! \~english Returns \c true after \a shutdown() or \a shutdownNow() has been requested.
//! \~russian Возвращает \c true после запроса \a shutdown() или \a shutdownNow().
bool isShutdown() const;
//! \~\brief
//! \~english Waits up to \a timeout for all worker threads to finish.
//! \~russian Ожидает до \a timeout завершения всех рабочих потоков.
//! \return
//! \~english \c false if the timeout expires first.
//! \~russian \c false, если таймаут истек раньше.
bool awaitTermination(PISystemTime timeout);
private:
std::atomic_bool isShutdown_;
PIBlockingQueue<std::function<void()>> taskQueue;
PIVector<PIThread *> threadPool;
};
#endif // PITHREADPOOLEXECUTOR_H
namespace {
// [[deprecated("DeprecatedHeader.h is deprecated. Use NewHeader.h instead.")]]
DEPRECATEDM("Use pithreadpoolworker.h") const bool deprecated_header_is_used = true;
} // namespace

157
libs/main/thread/pithreadpoolworker.cpp Normal file
View File

@@ -0,0 +1,157 @@
/*
PIP - Platform Independent Primitives
Stephan Fomenko, Ivan Pelipenko
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 "pithreadpoolworker.h"
#include "piliterals_time.h"
#include "pisysteminfo.h"
//! \class PIThreadPoolWorker
//! \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.
//!
PIThreadPoolWorker::PIThreadPoolWorker(int threads_count) {
if (threads_count < 0) threads_count = PISystemInfo::instance()->processorsCount;
assertm(threads_count > 0, "Invalid threads count!");
for (int i = 0; i < threads_count; ++i) {
Worker * w = new Worker();
w->thread.setSlot([this, w]() { threadFunc(w); });
workers << w;
}
}
PIThreadPoolWorker::~PIThreadPoolWorker() {
stop();
for (auto * w: workers) {
if (!w->thread.waitForFinish(1_s)) w->thread.terminate();
}
piDeleteAllAndClear(workers);
}
void PIThreadPoolWorker::start() {
for (auto w: workers)
w->thread.start();
m_running = true;
}
void PIThreadPoolWorker::stop() {
m_running = false;
for (auto w: workers) {
w->thread.stop();
w->notifier.notify();
}
}
bool PIThreadPoolWorker::stopAndWait(PISystemTime timeout) {
stop();
return waitForFinish(timeout);
}
bool PIThreadPoolWorker::waitForStart(PISystemTime timeout) {
PITimeMeasurer tm;
for (auto w: workers) {
if (timeout.isNull())
w->thread.waitForStart();
else {
auto remains = timeout - tm.elapsed();
if (remains.isNegative()) return false;
if (!w->thread.waitForStart(remains)) return false;
}
}
return true;
}
bool PIThreadPoolWorker::waitForFinish(PISystemTime timeout) {
PITimeMeasurer tm;
for (auto w: workers) {
if (timeout.isNull())
w->thread.waitForFinish();
else {
auto remains = timeout - tm.elapsed();
if (remains.isNegative()) return false;
if (!w->thread.waitForFinish(remains)) return false;
}
}
return true;
}
int64_t PIThreadPoolWorker::enqueueTask(std::function<void(int64_t)> func, PIObject * context) {
int64_t id = ++next_task_id;
Task task;
task.id = id;
task.func = std::move(func);
task.context = context;
tasks_queue.getRef()->append(std::move(task));
for (auto * w: workers)
w->notifier.notify();
return id;
}
bool PIThreadPoolWorker::removeTask(int64_t id) {
auto qref = tasks_queue.getRef();
auto prev_size = qref->size();
qref->removeWhere([id](const Task & t) { return t.id == id; });
return prev_size != qref->size();
}
void PIThreadPoolWorker::clearTasks() {
tasks_queue.getRef()->clear();
}
PIThreadPoolWorker::TaskStatus PIThreadPoolWorker::taskStatus(int64_t id) const {
if (id <= 0) return TaskStatus::Unknown;
auto qref = tasks_queue.getRef();
for (auto w: workers)
if (w->in_work == id) return TaskStatus::InProgress;
for (const auto & t: *qref)
if (t.id == id) return TaskStatus::Enqueued;
return id <= next_task_id ? TaskStatus::Done : TaskStatus::Unknown;
}
void PIThreadPoolWorker::threadFunc(Worker * w) {
w->notifier.wait();
if (w->thread.isStopping()) return;
if (!m_running) return;
Task task;
{
auto ref = tasks_queue.getRef();
if (ref->isEmpty()) return;
task = ref->dequeue();
}
if (!task.isValid()) return;
w->in_work = task.id;
task.func(task.id);
w->in_work = -1;
w->notifier.notify();
}

150
libs/main/thread/pithreadpoolworker.h Normal file
View File

@@ -0,0 +1,150 @@
//! \~\file pithreadpoolworker.h
//! \~\ingroup Thread
//! \brief
//! \~english Thread pool worker
//! \~russian Исполнитель пула потоков
//!
//! \details
//! \~english Executes tasks in a pool of worker threads.
//! \~russian Выполняет задачи в пуле рабочих потоков.
/*
PIP - Platform Independent Primitives
Stephan Fomenko, Ivan Pelipenko
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 PITHREADPOOLWORKER_H
#define PITHREADPOOLWORKER_H
#include "piblockingqueue.h"
#include "piprotectedvariable.h"
#include "pithread.h"
//! \~\ingroup Thread
//! \~\brief
//! \~english Fixed-size pool of worker threads for fire-and-forget tasks.
//! \~russian Фиксированный пул рабочих потоков для задач без ожидания результата.
class PIP_EXPORT PIThreadPoolWorker {
public:
//! \~english Constructs executor with \a threads_count worker threads.
//! \~russian Создает исполнитель с \a threads_count рабочими потоками.
explicit PIThreadPoolWorker(int threads_count = -1);
//! \~english Stops worker threads and destroys executor resources.
//! \~russian Останавливает рабочие потоки и уничтожает ресурсы исполнителя.
virtual ~PIThreadPoolWorker();
enum class TaskStatus {
Unknown,
Enqueued,
InProgress,
Done
};
//! \~english Starts the threads.
//! \~russian Запускает потоки.
void start();
//! \~english Requests graceful threads shutdown.
//! \~russian Запрашивает корректное завершение потоков.
void stop();
//! \~english Requests stop and waits for threads completion. Returns \b false if the timeout expires.
//! \~russian Запрашивает остановку и ожидает завершения потоков. Возвращает \b false, если таймаут истек.
bool stopAndWait(PISystemTime timeout = {});
//! \~english Waits until the threads starts. Returns \b false if the timeout expires first.
//! \~russian Ожидает запуска потоков. Возвращает \b false, если таймаут истек раньше.
bool waitForStart(PISystemTime timeout = {});
//! \~english Waits for threads completion. Returns \b false if the timeout expires first.
//! \~russian Ожидает завершения потоков. Возвращает \b false, если таймаут истек раньше.
bool waitForFinish(PISystemTime timeout = {});
//! \~english Returns whether the threads are currently running.
//! \~russian Возвращает, выполняются ли потоки в данный момент.
bool isRunning() const { return m_running; }
int64_t enqueueTask(std::function<void(int64_t)> func, PIObject * context = nullptr);
int64_t enqueueTask(std::function<void()> func, PIObject * context = nullptr) {
return enqueueTask([func](int64_t) { func(); }, context);
}
template<typename O>
int64_t enqueueTask(O * obj, void (O::*member_func)(int64_t)) {
return enqueueTask([obj, member_func](int64_t id) { (obj->*member_func)(id); });
}
template<typename O>
int64_t enqueueTask(O * obj, void (O::*member_func)()) {
return enqueueTask([obj, member_func](int64_t) { (obj->*member_func)(); });
}
bool removeTask(int64_t id);
TaskStatus taskStatus(int64_t id) const;
// DEPRECATED
//! \~\brief
//! \~english Submits \a runnable for asynchronous execution by a worker thread.
//! \~russian Передает \a runnable на асинхронное выполнение рабочим потоком.
//! \details
//! \~english
//! This is a best-effort fire-and-forget call and does not report whether the task was accepted.
//! \After shutdown requests new tasks are ignored.
//! \~russian
//! Это вызов по принципу best-effort без ожидания результата и без сообщения о том, была ли задача принята.
//! \После запроса на завершение новые задачи игнорируются.
void execute(std::function<void()> runnable);
void shutdownNow() DEPRECATEDM("Use stopAndWait()") { stopAndWait(); }
void shutdown() DEPRECATEDM("Use stop()") { stop(); }
bool isShutdown() const DEPRECATEDM("Use !isRunning()") { return !isRunning(); }
bool awaitTermination(PISystemTime timeout) DEPRECATEDM("Use waitForFinish()") { return waitForFinish(timeout); }
private:
struct Worker {
PIThread thread;
PIThreadNotifier notifier;
std::atomic_int64_t in_work = {-1};
};
struct Task {
bool isValid() const { return func != nullptr; }
PIObject * context = nullptr;
std::function<void(int64_t)> func = nullptr;
int64_t id = -1;
};
void threadFunc(Worker * w);
mutable PIVector<Worker *> workers;
mutable PIProtectedVariable<PIQueue<Task>> tasks_queue;
std::atomic_bool m_running = {false};
std::atomic_int64_t next_task_id = {0};
};
typedef PIThreadPoolWorker PIThreadPoolExecutor DEPRECATEDM("Use PIThreadPoolWorker");
#endif // PITHREADPOOLWORKER_H

12
libs/main/thread/pitimer.cpp
View File

@@ -274,6 +274,18 @@ void PITimer::stopAndWait(PISystemTime timeout) {
}
void PITimer::setSlot(std::function<void()> func) {
ret_func_delim = nullptr;
ret_func = std::move(func);
}
void PITimer::setSlot(std::function<void(int)> func) {
ret_func = nullptr;
ret_func_delim = std::move(func);
}
void PITimer::addDelimiter(int delim, std::function<void(int)> func) {
delims << Delimiter(std::move(func), delim);
}

28
libs/main/thread/pitimer.h
View File

@@ -110,11 +110,6 @@ public:
//! \~english Deprecated overload of \a start() that accepts milliseconds.
//! \~russian Устаревшая перегрузка \a start(), принимающая миллисекунды.
bool start(double interval_ms) DEPRECATEDM("use start(PISystemTime)") { return start(PISystemTime::fromMilliseconds(interval_ms)); }
EVENT_HANDLER0(bool, start);
EVENT_HANDLER0(bool, restart);
EVENT_HANDLER0(void, stop);
//! \~english Deprecated overload of \a stopAndWait() that accepts milliseconds.
//! \~russian Устаревшая перегрузка \a stopAndWait(), принимающая миллисекунды.
@@ -126,23 +121,15 @@ public:
//! \~english Sets a tick callback that ignores the delimiter value.
//! \~russian Устанавливает обратный вызов тика, игнорирующий значение делителя.
void setSlot(std::function<void()> func) {
ret_func_delim = nullptr;
ret_func = std::move(func);
}
void setSlot(std::function<void()> func);
//! \~english Sets a tick callback that receives the current delimiter value.
//! \~russian Устанавливает обратный вызов тика, принимающий текущее значение делителя.
void setSlot(std::function<void(int)> func) {
ret_func = nullptr;
ret_func_delim = std::move(func);
}
void setSlot(std::function<void(int)> func);
//! \~english Enables locking of the internal mutex around tick processing.
//! \~russian Включает блокировку внутреннего мьютекса вокруг обработки тиков.
void needLockRun(bool need) { lockRun = need; }
EVENT_HANDLER0(void, lock) { mutex_.lock(); }
EVENT_HANDLER0(void, unlock) { mutex_.unlock(); }
//! \~english Returns whether the timer drains queued delivery for itself as performer on each main tick. By default \b true.
//! \~russian Возвращает, должен ли таймер обрабатывать отложенную доставку для себя как исполнителя на каждом основном тике. По
@@ -169,9 +156,6 @@ public:
//! \~russian Удаляет все делители со значением \a delim.
void removeDelimiter(int delim);
EVENT_HANDLER0(void, clearDelimiters) { delims.clear(); }
EVENT1(tickEvent, int, delimiter);
//! \handlers
//! \{
@@ -180,31 +164,37 @@ public:
//! \brief
//! \~english Starts the timer with the current \a interval().
//! \~russian Запускает таймер с текущим значением \a interval().
EVENT_HANDLER0(bool, start);
//! \fn bool restart()
//! \brief
//! \~english Stops the timer, then starts it again with the current \a interval().
//! \~russian Останавливает таймер, затем снова запускает его с текущим значением \a interval().
EVENT_HANDLER0(bool, restart);
//! \fn bool stop()
//! \brief
//! \~english Requests stop and wakes the timer thread, but does not wait for completion.
//! \~russian Запрашивает остановку и пробуждает поток таймера, но не ожидает завершения.
EVENT_HANDLER0(void, stop);
//! \fn void clearDelimiters()
//! \brief
//! \~english Remove all frequency delimiters
//! \~russian Удаляет все делители частоты
EVENT_HANDLER0(void, clearDelimiters) { delims.clear(); }
//! \fn void lock()
//! \brief
//! \~english Locks the internal timer mutex.
//! \~russian Блокирует внутренний мьютекс таймера.
EVENT_HANDLER0(void, lock) { mutex_.lock(); }
//! \fn void unlock()
//! \brief
//! \~english Unlocks the internal timer mutex.
//! \~russian Разблокирует внутренний мьютекс таймера.
EVENT_HANDLER0(void, unlock) { mutex_.unlock(); }
//! \}
//! \events
@@ -219,7 +209,7 @@ public:
//! "delimiter" is the frequency delimiter, 1 for the main loop.
//! \~russian
//! "delimiter" - делитель частоты, 1 для основного цикла.
EVENT1(tickEvent, int, delimiter);
//! \}