libs/main/thread/pithread.h

@@ -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)") {

libs/main/thread/pithreadmodule.h

@@ -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

libs/main/thread/pithreadpoolexecutor.cpp

@@ -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;
-}

libs/main/thread/pithreadpoolexecutor.h

@@ -1,95 +1,23 @@
-//! \~\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
-
-    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 "pithreadpoolworker.h"
-#include "pithread.h"
 
-#include <atomic>
+namespace {
+DEPRECATEDM("Use pithreadpoolworker.h") const bool deprecated_header_is_used = true;
+}
 
-
+class DEPRECATEDM("Use PIThreadPoolWorker") PIThreadPoolExecutor: public PIThreadPoolWorker {
-//! \~\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.
+	PIThreadPoolExecutor(int threads_count): PIThreadPoolWorker(threads_count) { start(); }
-	//! \~russian Создает исполнитель с \a corePoolSize рабочими потоками.
+	~PIThreadPoolExecutor() { stopAndWait(); }
-	explicit PIThreadPoolExecutor(int corePoolSize);
 
-	//! \~english Stops worker threads and destroys executor resources.
+	void execute(std::function<void()> runnable) DEPRECATEDM("Use enqueueTask()") { enqueueTask(runnable); }
-	//! \~russian Останавливает рабочие потоки и уничтожает ресурсы исполнителя.
+	void shutdownNow() DEPRECATEDM("Use stopAndWait()") { stopAndWait(); }
-	virtual ~PIThreadPoolExecutor();
+	void shutdown() DEPRECATEDM("Use stop()") { stop(); }
-
+	bool isShutdown() const DEPRECATEDM("Use !isRunning()") { return !isRunning(); }
-	//! \~\brief
+	bool awaitTermination(PISystemTime timeout) DEPRECATEDM("Use waitForFinish()") { return waitForFinish(timeout); }
-	//! \~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
+
+#endif

libs/main/thread/pithreadpoolworker.cpp

@@ -0,0 +1,238 @@
+/*
+    PIP - Platform Independent Primitives
+
+	Ivan Pelipenko, 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 "pithreadpoolworker.h"
+
+#include "pisysteminfo.h"
+
+//! \addtogroup Thread
+//! \{
+//! \class PIThreadPoolWorker pithreadpoolworker.h
+//! \~\details
+//! \~english
+//! \PIThreadPoolWorker is a class that implements a fixed-size pool of worker threads for general-purpose task execution. It
+//! allows starting threads, enqueuing tasks as functors or class member methods, and managing their execution. The class provides methods
+//! to wait for all tasks or specific tasks by ID to complete, as well as to gracefully stop the pool with timeouts. The lifecycle of each
+//! task can be monitored using the taskStarted and taskFinished events. It is a versatile tool for multithreaded asynchronous operations.
+//! \~russian
+//! PIThreadPoolWorker — это класс, реализующий фиксированный пул рабочих потоков для выполнения задач общего назначения. Он
+//! позволяет запускать потоки, добавлять в очередь задачи в виде функторов или методов класса, а также управлять их выполнением. Класс
+//! предоставляет методы для ожидания завершения всех задач или отдельных задач по их идентификатору, а также для корректной остановки пула
+//! с таймаутами. С помощью событий taskStarted и taskFinished можно отслеживать жизненный цикл каждой задачи. Это универсальный инструмент
+//! для многопоточного асинхронного выполнения операций.
+//!
+//! \}
+
+
+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() {
+	piDeleteAllAndClear(workers);
+}
+
+
+void PIThreadPoolWorker::start() {
+	for (auto w: workers) {
+		w->thread.start();
+		w->notifier.notify();
+	}
+}
+
+
+void PIThreadPoolWorker::stop() {
+	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;
+}
+
+
+bool PIThreadPoolWorker::isRunning() const {
+	return workers.every([](Worker * w) { return w->thread.isRunning(); });
+}
+
+
+bool PIThreadPoolWorker::waitForTasks(PISystemTime timeout) {
+	if (!isRunning()) return tasks_queue.getRef()->isEmpty();
+	auto checkWorking = [this] {
+		if (tasks_queue.getRef()->isNotEmpty()) return true;
+		for (auto * w: workers)
+			if (w->in_work > 0) return true;
+		return false;
+	};
+	if (timeout.isNull()) {
+		for (;;) {
+			if (!checkWorking()) break;
+			piMinSleep();
+		}
+		return true;
+	}
+	PITimeMeasurer tm;
+	while (tm.elapsed() < timeout) {
+		if (!checkWorking()) return true;
+		piMinSleep();
+	}
+	return tm.elapsed() < timeout;
+}
+
+
+bool PIThreadPoolWorker::waitForTask(int64_t id, PISystemTime timeout) {
+	if (!isRunning()) return tasks_queue.getRef()->every([id](const Task & t) { return t.id != id; });
+	auto checkWorking = [this, id] {
+		if (tasks_queue.getRef()->any([id](const Task & t) { return t.id == id; })) return true;
+		for (auto * w: workers)
+			if (w->in_work == id) return true;
+		return false;
+	};
+	if (timeout.isNull()) {
+		for (;;) {
+			if (!checkWorking()) break;
+			piMinSleep();
+		}
+		return true;
+	}
+	PITimeMeasurer tm;
+	while (tm.elapsed() < timeout) {
+		if (!checkWorking()) return true;
+		piMinSleep();
+	}
+	return tm.elapsed() < timeout;
+}
+
+
+void PIThreadPoolWorker::exec() {
+	start();
+	waitForStart();
+	waitForTasks();
+	stopAndWait();
+}
+
+
+int64_t PIThreadPoolWorker::enqueueTask(std::function<void(int64_t)> func, PIObject * context) {
+	if (context) {
+		if (!contexts[context]) {
+			contexts << context;
+			CONNECTL(context, deleted, ([this, context](PIObject *) {
+						 contexts.remove(context);
+						 auto qref = tasks_queue.getRef();
+						 // auto prev_size = qref->size();
+						 // piCout << "deleted" << (void *)context << qref->map<void *>([](const Task & t) { return t.context; });
+						 qref->removeWhere([context](const Task & t) { return t.context == context; });
+						 // piCout << prev_size << qref->size() << qref->map<void *>([](const Task & t) { return t.context; });
+					 }));
+		}
+	}
+	int64_t id = ++next_task_id;
+	Task task;
+	task.id      = id;
+	task.func    = std::move(func);
+	task.context = context;
+	tasks_queue.getRef()->enqueue(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::DoneOrCancelled : TaskStatus::Unknown;
+}
+
+
+void PIThreadPoolWorker::threadFunc(Worker * w) {
+	w->notifier.wait();
+	if (w->thread.isStopping()) return;
+	Task task;
+	{
+		auto ref = tasks_queue.getRef();
+		if (ref->isEmpty()) return;
+		task = ref->dequeue();
+	}
+	if (!task.isValid()) return;
+	w->in_work = task.id;
+	taskStarted(task.id);
+	task.func(task.id);
+	w->in_work = -1;
+	taskFinished(task.id);
+	w->notifier.notify();
+}

libs/main/thread/pithreadpoolworker.h

@@ -0,0 +1,177 @@
+//! \~\file pithreadpoolworker.h
+//! \~\ingroup Thread
+//! \brief
+//! \~english Thread pool worker
+//! \~russian Исполнитель пула потоков
+/*
+    PIP - Platform Independent Primitives
+
+	Ivan Pelipenko, 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 PITHREADPOOLWORKER_H
+#define PITHREADPOOLWORKER_H
+
+#include "piprotectedvariable.h"
+#include "pithread.h"
+
+
+//! \~\ingroup Thread
+//! \~\brief
+//! \~english Fixed-size pool of worker threads for generic-purpose tasks.
+//! \~russian Фиксированный пул рабочих потоков для задач общего назначения.
+class PIP_EXPORT PIThreadPoolWorker: public PIObject {
+	PIOBJECT(PIThreadPoolWorker)
+
+public:
+	//! \~english Constructs executor with \a threads_count worker threads. If \a threads_count < 0 processor threads count used.
+	//! \~russian Создает исполнитель с \a threads_count рабочими потоками. Если \a threads_count < 0 используется количество потоков
+	//! процессора.
+	explicit PIThreadPoolWorker(int threads_count = -1);
+
+	//! \~english Destroy worker threads. Call \a stopAndWait() before.
+	//! \~russian Уничтожает рабочие потоки. Вызывайте перед этим \a stopAndWait().
+	virtual ~PIThreadPoolWorker();
+
+
+	//! \~english Task status.
+	//! \~russian Статус задачи.
+	enum class TaskStatus {
+		Unknown /** \~english ID <= 0 or not queued yet \~russian ID <= 0 или не поставлена в очередь */,
+		Enqueued /** \~english Wait for execution \~russian Ожидает выполнения */,
+		InProgress /** \~english In execution now \~russian В процессе выполнения */,
+		DoneOrCancelled /** \~english Done or cancelled \~russian Выполнена или отменена */
+	};
+
+
+	//! \~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;
+
+	//! \~english Waits for all tasks completion. Returns \b false if the timeout expires first.
+	//! \~russian Ожидает завершения всех задач. Возвращает \b false, если таймаут истек раньше.
+	bool waitForTasks(PISystemTime timeout = {});
+
+	//! \~english Waits for task with id \a id completion. Returns \b false if the timeout expires first.
+	//! \~russian Ожидает завершения задачи с id \a id. Возвращает \b false, если таймаут истек раньше.
+	bool waitForTask(int64_t id, PISystemTime timeout = {});
+
+	//! \~english Starts threads, wait for all tasks complete and threads stop.
+	//! \~russian Запускает потоки, ожидает завершения всех задач и остановки потоков.
+	void exec();
+
+
+	//! \~english Queue functor to execution. Pass task ID in functor. Returns task ID.
+	//! \~russian Запланировать функтор на выполнение. В функтор передастся ID задачи. Возвращает ID задачи.
+	int64_t enqueueTask(std::function<void(int64_t)> func, PIObject * context = nullptr);
+
+	//! \~english Queue functor to execution. Returns task ID.
+	//! \~russian Запланировать функтор на выполнение. Возвращает ID задачи.
+	int64_t enqueueTask(std::function<void()> func, PIObject * context = nullptr) {
+		return enqueueTask([func](int64_t) { func(); }, context);
+	}
+
+	//! \~english Queue class member method to execution. Pass task ID in method. Returns task ID.
+	//! \~russian Запланировать член-метод класса на выполнение. В метод передастся ID задачи. Возвращает ID задачи.
+	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); },
+						   PIObject::isPIObject(obj) ? dynamic_cast<PIObject *>(obj) : nullptr);
+	}
+
+	//! \~english Queue class member method to execution. Returns task ID.
+	//! \~russian Запланировать член-метод класса на выполнение. Возвращает ID задачи.
+	template<typename O>
+	int64_t enqueueTask(O * obj, void (O::*member_func)()) {
+		return enqueueTask([obj, member_func](int64_t) { (obj->*member_func)(); },
+						   PIObject::isPIObject(obj) ? dynamic_cast<PIObject *>(obj) : nullptr);
+	}
+
+	//! \~english Remove task with id \a id from queue. Returns if task delete.
+	//! \~russian Удаляет задачу с id \a id из очереди. Возвращиает была ли задача удалена.
+	bool removeTask(int64_t id);
+
+	//! \~english Remove all queued tasks.
+	//! \~russian Удаляет все задачи из очереди.
+	void clearTasks();
+
+	//! \~english Returns task status with id \a id.
+	//! \~russian Возвращиает статус задачи с id \a id.
+	TaskStatus taskStatus(int64_t id) const;
+
+
+	//! \events
+	//! \{
+
+	//! \fn void taskStarted(int64_t id)
+	//! \brief
+	//! \~english Raised on start execution task with id \a id.
+	//! \~russian Вызывается при старте выполнения задачи с id \a id.
+	EVENT1(taskStarted, int64_t, id);
+
+	//! \fn void taskFinished(int64_t id)
+	//! \brief
+	//! \~english Raised on finish execution task with id \a id.
+	//! \~russian Вызывается при завершении выполнения задачи с id \a id.
+	EVENT1(taskFinished, int64_t, id);
+
+	//! \}
+
+
+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;
+
+	PISet<PIObject *> contexts;
+	std::atomic_int64_t next_task_id = {0};
+};
+
+
+#endif // PITHREADPOOLWORKER_H

libs/main/thread/pitimer.cpp

@@ -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);
 }

libs/main/thread/pitimer.h

@@ -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) {
+	void setSlot(std::function<void()> func);
-		ret_func_delim = nullptr;
-		ret_func       = std::move(func);
-	}
 
 	//! \~english Sets a tick callback that receives the current delimiter value.
 	//! \~russian Устанавливает обратный вызов тика, принимающий текущее значение делителя.
-	void setSlot(std::function<void(int)> func) {
+	void setSlot(std::function<void(int)> func);
-		ret_func       = nullptr;
-		ret_func_delim = std::move(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);
 
 	//! \}
 

main.cpp

@@ -1,143 +1,53 @@
-#include "libs/http_client/curl_thread_pool_p.h"
-#include "picodeparser.h"
-#include "pidigest.h"
-#include "pihttpclient.h"
-#include "piliterals.h"
 #include "pip.h"
-#include "piunits.h"
+#include "pithreadpoolexecutor.h"
-#include "pivaluetree_conversions.h"
 
-using namespace PICoutManipulators;
+class A: public PIObject {
-using namespace PIHTTP;
+	PIOBJECT(A)
-using namespace PIUnits::Class;
-
-int rcnt = 0, scnt = 0;
-
-inline PIByteArray SMBusTypeInfo_genHash(PIString n) {
-	PICrypt c;
-	return piSerialize(c.shorthash(n.removeAll(" "), PIString("SMBusDataHashKey").toByteArray()));
-}
 
+public:
+	A(PIString n = {}) { setName(n); }
+	void foo() {
+		100_ms .sleep();
+		piCoutObj << "foo!";
+		100_ms .sleep();
+	}
+};
 
 int main(int argc, char * argv[]) {
-	PICrypt _crypt;
+	PIVector<A *> objects;
-	// auto ba = PIFile::readAll("logo.png");
+	objects << new A("1") << new A("2") << new A("3");
-	PIString str    = "hello!"_a;
-	PIByteArray ba  = str.toAscii();
-	PIByteArray key = PIString("SMBusDataHashKey").toByteArray();
 
-	const int times = 1000000;
+	PITimer status_timer;
-	PITimeMeasurer tm;
+	PIThreadPoolWorker pool(2);
-	PISystemTime el;
+	pool.start();
 
-	tm.reset();
+	// int64_t id = -1;
-	piForTimes(times) {
+	// status_timer.start(10_Hz, [&id, &pool] { piCout << "[timer] status" << (int)pool.taskStatus(id); });
-		PIDigest::calculateWithKey(ba, key, PIDigest::Type::SipHash_2_4_128);
-	}
-	el = tm.elapsed();
-	piCout << "PIDigest" << el.toString();
 
-	tm.reset();
+	100_ms .sleep();
-	piForTimes(times) {
+	// pool.enqueueTask([](int64_t id) {
-		_crypt.shorthash(str, key);
+	// 	piCout << "[task ] start, id" << id;
-	}
+	// 	// 500_ms .sleep();
-	el = tm.elapsed();
+	// 	piCout << "[task ] done";
-	piCout << "  sodium" << el.toString();
+	// });
+	pool.enqueueTask(objects[0], &A::foo);
+	pool.enqueueTask(objects[1], &A::foo);
+	pool.enqueueTask(objects[1], &A::foo);
+	pool.enqueueTask(objects[2], &A::foo);
 
-	tm.reset();
+	10_ms .sleep();
-	piForTimes(times) {
+	delete objects[1];
-		PIDigest::calculateWithKey(ba, key, PIDigest::Type::BLAKE2b_128);
+	objects.remove(1);
-	}
+	// piCout << "[main ]" << "enqueued, id" << id;
-	el = tm.elapsed();
-	piCout << "   blake" << el.toString();
 
-	return 0;
+	// 200_ms .sleep();
+	piCout << "[main ]" << "wait ...";
+	piCout << "[main ]" << "wait done";
 
+	1000_ms .sleep();
 
-	PIEthernet *eth_r, *eth_s;
+	pool.stopAndWait();
-	eth_r = PIIODevice::createFromFullPath("eth://udp: 192.168.1.25 :10000")->cast<PIEthernet>();
+	status_timer.stopAndWait();
-	eth_s = PIIODevice::createFromFullPath("eth://udp: : : 192.168.1.25:10000")->cast<PIEthernet>();
+	piDeleteAll(objects);
-
-	eth_r->setReadBufferSize(1_MiB);
-	CONNECTL(eth_r, threadedReadEvent, [](const uchar * readed, ssize_t size) {
-		// piCout << "rec";
-		piMSleep(1);
-		++rcnt;
-	});
-	eth_r->startThreadedRead();
-
-	PIByteArray _ba(1400);
-	for (int i = 0; i < 100; ++i) {
-		eth_s->write(_ba);
-		++scnt;
-	}
-
-	0.2_s .sleep();
-
-	piCout << "snd" << scnt;
-	piCout << "rec" << rcnt;
-
-	piDeleteSafety(eth_r);
-	piDeleteSafety(eth_s);
-	return 0;
-
-	PITranslator::loadLang("ru");
-	/*auto ucl = PIUnits::allClasses();
-	for (auto c: ucl) {
-		piCout << (c->className() + ":");
-		for (auto t: c->allTypes()) {
-			piCout << " " << c->name(t) << "->" << c->unit(t);
-		}
-	}*/
-
-	// PIUnits::Value(1);
-	// piCout << PIUnits::name(PIUnits::Class::Information::Bit);
-	// piCout << PIUnits::name(PIUnits::Class::Information::Byte);
-	// piCout << PIUnits::name(PIUnits::Class::Information::_LastType);
-	// piCout << PIUnits::name((int)PIUnits::Class::Angle::Degree);
-
-	// piCout << PIUnits::unit(PIUnits::Class::Information::Bit);
-	// piCout << PIUnits::unit(PIUnits::Class::Information::Byte);
-	// piCout << PIUnits::unit(PIUnits::Class::Information::_LastType);
-	// piCout << PIUnits::unit((int)PIUnits::Class::Angle::Degree);
-
-	// for (int i = -10; i < 10; ++i)
-	// 	piCout << PIUnits::Value(pow10(i * 0.99), PIUnits::Class::Distance::Meter).toString();
-
-	auto v = PIUnits::Value(M_PI, Angle::Radian);
-	piCout << v << "=" << v.converted(Angle::Degree);
-
-	v = PIUnits::Value(45, Angle::Degree);
-	piCout << v << "=" << v.converted(Angle::Radian);
-
-	piCout << PIUnits::Value(5E-5, Time::Second);
-	piCout << PIUnits::Value(3E-3, Time::Second);
-	piCout << PIUnits::Value(0.8, Time::Second);
-	piCout << PIUnits::Value(1.2, Time::Second);
-	piCout << PIUnits::Value(1001, Time::Second);
-	piCout << PIUnits::Value(1000001, Time::Second);
-
-	piCout << PIUnits::Value(1_KB, Information::Byte);
-	piCout << PIUnits::Value(1_MB, Information::Byte);
-	piCout << PIUnits::Value(1_MiB, Information::Byte);
-	piCout << PIUnits::Value(1_MB, Information::Byte).converted(Information::Bit);
-	piCout << PIUnits::Value(1_MiB, Information::Byte).converted(Information::Bit);
-
-	piCout << PIUnits::Value(0., Temperature::Celsius).converted(Temperature::Kelvin);
-	piCout << PIUnits::Value(0., Temperature::Celsius).converted(Temperature::Fahrenheit);
-	piCout << PIUnits::Value(100., Temperature::Celsius).converted(Temperature::Fahrenheit);
-
-	piCout << PIUnits::Value(1., Pressure::Atmosphere).converted(Pressure::Pascal);
-	piCout << PIUnits::Value(1., Pressure::Atmosphere).converted(Pressure::MillimetreOfMercury);
-	piCout << PIUnits::Value(766., Pressure::MillimetreOfMercury).converted(Pressure::Atmosphere);
-
-	piCout << PIUnits::Value(5E-5, Time::Second).converted(Time::Hertz);
-	piCout << PIUnits::Value(3E-3, Time::Second).converted(Time::Hertz);
-	piCout << PIUnits::Value(0.8, Time::Second).converted(Time::Hertz);
-	piCout << PIUnits::Value(1.2, Time::Second).converted(Time::Hertz);
-	piCout << PIUnits::Value(1001, Time::Second).converted(Time::Hertz);
-	piCout << PIUnits::Value(1000001, Time::Second).converted(Time::Hertz);
-	// piCout << PIUnits::Value(0.2, Time::Second).converted(Time::Hertz);
-	// piCout << PIUnits::Value(5E-5, Time::Second).converted(Time::Hertz);
 	return 0;
 }