diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 496dd420d..62f751f92 100755 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -231,6 +231,8 @@ jobs: target: esp32 - path: 'components/thermistor/example' target: esp32 + - path: 'components/thread_pool/example' + target: esp32 - path: 'components/timer/example' target: esp32 - path: 'components/touch/example' diff --git a/.github/workflows/upload_components.yml b/.github/workflows/upload_components.yml index f2f749d35..c6f873681 100755 --- a/.github/workflows/upload_components.yml +++ b/.github/workflows/upload_components.yml @@ -134,6 +134,7 @@ jobs: components/tabulate components/task components/thermistor + components/thread_pool components/timer components/touch components/tla2528 diff --git a/components/thread_pool/CMakeLists.txt b/components/thread_pool/CMakeLists.txt new file mode 100644 index 000000000..cb1e7ccab --- /dev/null +++ b/components/thread_pool/CMakeLists.txt @@ -0,0 +1,4 @@ +idf_component_register( + INCLUDE_DIRS "include" + SRC_DIRS "src" + REQUIRES base_component task) diff --git a/components/thread_pool/README.md b/components/thread_pool/README.md new file mode 100644 index 000000000..95d45ce11 --- /dev/null +++ b/components/thread_pool/README.md @@ -0,0 +1,16 @@ +# Thread Pool Component + +[![Badge](https://components.espressif.com/components/espp/thread_pool/badge.svg)](https://components.espressif.com/components/espp/thread_pool) + +The `ThreadPool` component provides a reusable pool of worker tasks for executing queued jobs asynchronously. + +It is implemented with `espp::Task` workers and `std::condition_variable` synchronization. + +## Features + +- Configurable worker count +- Bounded or unbounded queue +- Optional blocking submit mode for backpressure +- Manual `start()` / `stop()` control; `start()` returns `true` if all workers launched successfully (or the pool was already running), `false` if any worker failed to start and the pool was rolled back to stopped state +- Graceful stop (stops workers; queued jobs may not be executed) +- Thread-safe stats for submitted / executed / rejected jobs diff --git a/components/thread_pool/example/CMakeLists.txt b/components/thread_pool/example/CMakeLists.txt new file mode 100644 index 000000000..6019f3f82 --- /dev/null +++ b/components/thread_pool/example/CMakeLists.txt @@ -0,0 +1,22 @@ +# The following lines of boilerplate have to be in your project's CMakeLists +# in this exact order for cmake to work correctly +cmake_minimum_required(VERSION 3.20) + +set(CMAKE_CXX_STANDARD 20) + +set(ENV{IDF_COMPONENT_MANAGER} "0") +include($ENV{IDF_PATH}/tools/cmake/project.cmake) + +# add the component directories that we want to use +set(EXTRA_COMPONENT_DIRS + "../../../components/" +) + +set( + COMPONENTS + "main esptool_py thread_pool" + CACHE STRING + "List of components to include" + ) + +project(thread_pool_example) diff --git a/components/thread_pool/example/README.md b/components/thread_pool/example/README.md new file mode 100644 index 000000000..a34dd22cb --- /dev/null +++ b/components/thread_pool/example/README.md @@ -0,0 +1,42 @@ +# Thread Pool Example + +This example is a self-contained test suite that exercises the full public API +of `espp::ThreadPool` across a range of usage patterns, from basic single-pool +operation to more advanced concurrent and multi-pool scenarios. + +| # | Test | APIs covered | +|---|---|---| +| 1 | Lifecycle | `start()`, `stop()`, `is_running()`, `worker_count()` | +| 2 | Normal dispatch | `submit()`, `queue_size()`, `stats()` | +| 3 | Bounded-queue rejection | `try_submit()` with deterministic barrier | +| 4 | Blocking submit | `submit()` with `block_on_submit_when_full` | +| 5 | Post-stop rejection | `submit()` after `stop()` | +| 6 | Concurrent lifecycle | multi-thread `start()` / `stop()` stress | +| 7 | Concurrent submission | multi-thread `submit()` + `try_submit()` | +| 8 | Chained pools | a job in `pool_a` submitting work into `pool_b` | +| 9 | Self-submit | a running job submitting back to its own pool | + +Each test logs individual `PASS` / `FAIL` results inline. At the end of the run +a summary is printed: + +``` +==================== Results ==================== + PASS lifecycle: start/stop/is_running/worker_count + PASS submit: normal dispatch + queue_size + stats + ... +================================================= +9/9 tests passed +All tests passed! +``` + +The final summary line (`N/N tests passed` / `N test(s) FAILED`) is the key +indicator of overall correctness and is suitable for CI log parsing. + +## Build + +From this directory, run: + +```bash +idf.py set-target esp32 +idf.py build flash monitor +``` diff --git a/components/thread_pool/example/main/CMakeLists.txt b/components/thread_pool/example/main/CMakeLists.txt new file mode 100644 index 000000000..a941e22ba --- /dev/null +++ b/components/thread_pool/example/main/CMakeLists.txt @@ -0,0 +1,2 @@ +idf_component_register(SRC_DIRS "." + INCLUDE_DIRS ".") diff --git a/components/thread_pool/example/main/thread_pool_example.cpp b/components/thread_pool/example/main/thread_pool_example.cpp new file mode 100644 index 000000000..3d33a0dfb --- /dev/null +++ b/components/thread_pool/example/main/thread_pool_example.cpp @@ -0,0 +1,602 @@ +#include +#include +#include +#include +#include +#include +#include + +#include "logger.hpp" +#include "thread_pool.hpp" + +using namespace std::chrono_literals; + +static void wait_for_jobs(std::condition_variable &cv, std::mutex &mtx, std::atomic &completed, + int expected) { + std::unique_lock lock(mtx); + cv.wait(lock, [&]() { return completed.load() >= expected; }); +} + +extern "C" void app_main(void) { + espp::Logger logger({.tag = "ThreadPool Example", .level = espp::Logger::Verbosity::INFO}); + + struct TestResult { + std::string name; + bool passed; + }; + std::vector results; + + // Returns false and logs on failure; used to accumulate per-test pass/fail. + auto check = [&](const std::string &test, bool condition, const std::string &desc) -> bool { + if (condition) { + logger.info(" PASS [{}]: {}", test, desc); + } else { + logger.error(" FAIL [{}]: {}", test, desc); + } + return condition; + }; + + // ------------------------------------------------------------------------- + // 1. Manual start/stop + is_running() + worker_count() + // ------------------------------------------------------------------------- + { + const std::string name = "lifecycle: start/stop/is_running/worker_count"; + //! [lifecycle example] + logger.info("--- {} ---", name); + bool passed = true; + + espp::ThreadPool pool({ + .worker_count = 3, + .max_queue_size = 0, + .auto_start = false, + .worker_task_config = { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + passed &= check(name, !pool.is_running(), "pool should not be running before start()"); + passed &= check(name, pool.worker_count() == 3, "worker_count() should be 3"); + + passed &= check(name, pool.start(), "start() should return true on first call"); + passed &= check(name, pool.is_running(), "pool should be running after start()"); + + passed &= check(name, pool.start(), "start() should return true when already running (no-op)"); + passed &= check(name, pool.is_running(), "pool should still be running after duplicate start()"); + + pool.stop(); + passed &= check(name, !pool.is_running(), "pool should not be running after stop()"); + //! [lifecycle example] + + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // 2. submit() + queue_size() + stats() + // ------------------------------------------------------------------------- + { + const std::string name = "submit: normal dispatch + queue_size + stats"; + //! [submit example] + logger.info("--- {} ---", name); + bool passed = true; + + std::mutex done_mutex; + std::condition_variable done_cv; + constexpr int total_jobs = 8; + std::atomic completed_jobs{0}; + + espp::ThreadPool pool({ + .worker_count = 2, + .max_queue_size = 0, + .auto_start = true, + .worker_task_config = { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + int accepted_count = 0; + for (int i = 0; i < total_jobs; ++i) { + if (pool.submit([&, i]() { + std::this_thread::sleep_for(50ms); + ++completed_jobs; + done_cv.notify_one(); + })) { + ++accepted_count; + } + } + passed &= check(name, accepted_count == total_jobs, "all jobs should be accepted (unbounded queue)"); + + wait_for_jobs(done_cv, done_mutex, completed_jobs, total_jobs); + + auto s = pool.stats(); + logger.info(" stats: {}", s); + passed &= check(name, s.submitted == total_jobs, "submitted count should equal total_jobs"); + passed &= check(name, s.executed == total_jobs, "executed count should equal total_jobs"); + passed &= check(name, s.rejected == 0, "rejected count should be 0"); + passed &= check(name, pool.queue_size() == 0, "queue should be empty after all jobs finish"); + + pool.stop(); + //! [submit example] + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // 3. try_submit() — non-blocking rejection when queue is full + // ------------------------------------------------------------------------- + { + const std::string name = "try_submit: rejection when queue full"; + //! [try_submit example] + logger.info("--- {} ---", name); + bool passed = true; + + // 1 worker, queue capacity 2: 1 executing + 2 queued = 3 total slots. + // + // Jobs are gated by an explicit barrier so workers cannot drain the queue + // before we assert rejection, making the test fully deterministic. + // + // To guarantee the queue is provably full we must also ensure the worker + // has dequeued (and is executing) the first job before we fill the two + // remaining queue slots. We use a separate "started" CV for this. + std::mutex barrier_mutex; + std::condition_variable barrier_cv; + bool release_workers = false; + + std::mutex started_mutex; + std::condition_variable started_cv; + std::atomic jobs_started{0}; + + auto blocking_job = [&]() { + // Signal that this job is now executing (off the queue) + { + std::lock_guard lock(started_mutex); + ++jobs_started; + } + started_cv.notify_one(); + // Block until the test releases the barrier + std::unique_lock lock(barrier_mutex); + barrier_cv.wait(lock, [&]() { return release_workers; }); + }; + + espp::ThreadPool pool({ + .worker_count = 1, + .max_queue_size = 2, + .auto_start = true, + .block_on_submit_when_full = false, + .worker_task_config = { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + // Step 1: submit the first job and wait until it is executing + // (it has been removed from the queue by the worker). + int fill_accepted = 0; + if (pool.try_submit(blocking_job)) { + ++fill_accepted; + } + { + std::unique_lock lock(started_mutex); + started_cv.wait(lock, [&]() { return jobs_started.load() >= 1; }); + } + + // Step 2: fill the 2 remaining queue slots. + for (int i = 0; i < 2; ++i) { + if (pool.try_submit(blocking_job)) { + ++fill_accepted; + } + } + passed &= check(name, fill_accepted == 3, "first 3 try_submit calls should be accepted"); + + // Step 3: queue is now provably full — every additional try_submit must be rejected. + int rejected_count = 0; + for (int i = 0; i < 3; ++i) { + if (!pool.try_submit([&]() {})) { + ++rejected_count; + } + } + passed &= check(name, rejected_count == 3, "try_submit when full should return false"); + + auto s = pool.stats(); + logger.info(" stats: {}", s); + passed &= check(name, s.rejected == 3, "stats.rejected should be 3"); + + // Release the barrier so workers can finish, then stop cleanly. + { + std::lock_guard lock(barrier_mutex); + release_workers = true; + } + barrier_cv.notify_all(); + pool.stop(); + //! [try_submit example] + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // 4. submit() blocking when full (block_on_submit_when_full = true) + // ------------------------------------------------------------------------- + { + const std::string name = "submit: blocking when queue full"; + //! [blocking submit example] + logger.info("--- {} ---", name); + bool passed = true; + + std::mutex done_mutex; + std::condition_variable done_cv; + std::atomic completed_jobs{0}; + constexpr int total_jobs = 6; + + espp::ThreadPool pool({ + .worker_count = 1, + .max_queue_size = 2, + .auto_start = true, + .block_on_submit_when_full = true, + .worker_task_config = { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + int accepted_count = 0; + for (int i = 0; i < total_jobs; ++i) { + if (pool.submit([&, i]() { + std::this_thread::sleep_for(30ms); + ++completed_jobs; + done_cv.notify_one(); + })) { + ++accepted_count; + } + } + passed &= check(name, accepted_count == total_jobs, "all jobs should be accepted (blocking submit)"); + + wait_for_jobs(done_cv, done_mutex, completed_jobs, total_jobs); + + auto s = pool.stats(); + logger.info(" stats: {}", s); + passed &= check(name, s.submitted == total_jobs, "submitted count should equal total_jobs"); + passed &= check(name, s.executed == total_jobs, "executed count should equal total_jobs"); + passed &= check(name, s.rejected == 0, "rejected count should be 0"); + + pool.stop(); + //! [blocking submit example] + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // 5. submit() after stop() — rejected via is_running() guard + // ------------------------------------------------------------------------- + { + const std::string name = "submit: rejected after stop()"; + //! [submit after stop example] + logger.info("--- {} ---", name); + bool passed = true; + + espp::ThreadPool pool({ + .worker_count = 1, + .max_queue_size = 0, + .auto_start = true, + .worker_task_config = { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + pool.stop(); + bool accepted = pool.submit([]() {}); + passed &= check(name, !accepted, "submit() after stop() should return false"); + passed &= check(name, pool.stats().submitted == 0, "submitted count should be 0"); + passed &= check(name, pool.stats().rejected == 1, "rejected count should be 1"); + + auto s = pool.stats(); + logger.info(" stats: {}", s); + //! [submit after stop example] + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // 6. Concurrent start/stop from multiple threads + // ------------------------------------------------------------------------- + { + const std::string name = "concurrent: start/stop from multiple threads"; + //! [concurrent lifecycle example] + logger.info("--- {} ---", name); + bool passed = true; + + espp::ThreadPool pool({ + .worker_count = 2, + .max_queue_size = 0, + .auto_start = false, + .worker_task_config = { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + constexpr int num_threads = 4; + constexpr int iterations = 10; + std::vector threads; + threads.reserve(num_threads); + + for (int t = 0; t < num_threads; ++t) { + threads.emplace_back([&pool, t]() { + for (int i = 0; i < iterations; ++i) { + if ((t + i) % 2 == 0) { + pool.start(); + } else { + pool.stop(); + } + } + }); + } + for (auto &t : threads) { + t.join(); + } + + // Bring pool to a known stopped state and verify consistency + pool.stop(); + passed &= check(name, !pool.is_running(), "pool should reach a clean stopped state"); + + auto s = pool.stats(); + logger.info(" stats: {}", s); + // No jobs were submitted — all counters must be zero + passed &= check(name, s.submitted == 0 && s.executed == 0 && s.rejected == 0, + "stats should all be zero (no jobs submitted)"); + //! [concurrent lifecycle example] + + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // 7. Concurrent submit/try_submit from multiple producer threads + // ------------------------------------------------------------------------- + { + const std::string name = "concurrent: multi-thread submit and try_submit"; + //! [concurrent submit example] + logger.info("--- {} ---", name); + bool passed = true; + + std::mutex done_mutex; + std::condition_variable done_cv; + constexpr int num_submit_threads = 3; + constexpr int num_try_submit_threads = 2; + constexpr int jobs_per_thread = 10; + constexpr int total_jobs = + (num_submit_threads + num_try_submit_threads) * jobs_per_thread; + std::atomic completed_jobs{0}; + std::atomic total_accepted{0}; + + espp::ThreadPool pool({ + .worker_count = 4, + .max_queue_size = 0, + .auto_start = true, + .worker_task_config = { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + std::vector producers; + producers.reserve(num_submit_threads + num_try_submit_threads); + + // submit() producers + for (int p = 0; p < num_submit_threads; ++p) { + producers.emplace_back([&]() { + for (int i = 0; i < jobs_per_thread; ++i) { + if (pool.submit([&]() { + std::this_thread::sleep_for(10ms); + ++completed_jobs; + done_cv.notify_one(); + })) { + ++total_accepted; + } + } + }); + } + + // try_submit() producers + for (int p = 0; p < num_try_submit_threads; ++p) { + producers.emplace_back([&]() { + for (int i = 0; i < jobs_per_thread; ++i) { + if (pool.try_submit([&]() { + std::this_thread::sleep_for(10ms); + ++completed_jobs; + done_cv.notify_one(); + })) { + ++total_accepted; + } + } + }); + } + + for (auto &p : producers) { + p.join(); + } + + wait_for_jobs(done_cv, done_mutex, completed_jobs, total_accepted.load()); + + auto s = pool.stats(); + logger.info(" stats: {}", s); + passed &= check(name, s.submitted + s.rejected == total_jobs, + "submitted + rejected should equal total attempted"); + passed &= check(name, s.executed == s.submitted, + "all accepted jobs should be executed (unbounded queue)"); + passed &= check(name, s.rejected == 0, "unbounded queue should not reject any jobs"); + + pool.stop(); + //! [concurrent submit example] + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // 8. Chained pools: a job in pool A submits work to pool B + // ------------------------------------------------------------------------- + { + const std::string name = "chained: job in pool_a submits to pool_b"; + //! [chained pools example] + logger.info("--- {} ---", name); + bool passed = true; + + std::mutex done_mutex; + std::condition_variable done_cv; + constexpr int num_a_jobs = 5; + constexpr int b_jobs_per_a = 2; + constexpr int total_b_jobs = num_a_jobs * b_jobs_per_a; + std::atomic completed_b{0}; + + espp::ThreadPool pool_b({ + .worker_count = 2, + .max_queue_size = 0, + .auto_start = true, + .worker_task_config = { + .name = "pool_b_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + espp::ThreadPool pool_a({ + .worker_count = 2, + .max_queue_size = 0, + .auto_start = true, + .worker_task_config = { + .name = "pool_a_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + for (int i = 0; i < num_a_jobs; ++i) { + pool_a.submit([&pool_b, &completed_b, &done_cv]() { + for (int j = 0; j < b_jobs_per_a; ++j) { + pool_b.submit([&completed_b, &done_cv]() { + std::this_thread::sleep_for(20ms); + ++completed_b; + done_cv.notify_one(); + }); + } + }); + } + + wait_for_jobs(done_cv, done_mutex, completed_b, total_b_jobs); + + auto sa = pool_a.stats(); + auto sb = pool_b.stats(); + logger.info(" pool_a stats: {}", sa); + logger.info(" pool_b stats: {}", sb); + passed &= check(name, sa.executed == num_a_jobs, "pool_a should execute all A jobs"); + passed &= check(name, sb.executed == total_b_jobs, "pool_b should execute all chained B jobs"); + passed &= check(name, sb.rejected == 0, "pool_b should not reject any jobs"); + + pool_a.stop(); + pool_b.stop(); + //! [chained pools example] + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // 9. Self-submit: a job submits another job back to the same pool + // ------------------------------------------------------------------------- + { + const std::string name = "self-submit: job submits to its own pool"; + //! [self-submit example] + logger.info("--- {} ---", name); + bool passed = true; + + std::mutex done_mutex; + std::condition_variable done_cv; + constexpr int num_initial_jobs = 4; + // Each initial job submits one follow-up job → 4 initial + 4 follow-up = 8 total + constexpr int total_executions = num_initial_jobs * 2; + std::atomic completed{0}; + + espp::ThreadPool pool({ + .worker_count = 2, + .max_queue_size = 0, + .auto_start = true, + .worker_task_config = { + .name = "tp_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }, + .log_level = espp::Logger::Verbosity::WARN, + }); + + for (int i = 0; i < num_initial_jobs; ++i) { + pool.submit([&pool, &completed, &done_cv]() { + ++completed; + done_cv.notify_one(); + // Submit a follow-up job back to the same pool without deadlock + pool.submit([&completed, &done_cv]() { + std::this_thread::sleep_for(10ms); + ++completed; + done_cv.notify_one(); + }); + }); + } + + wait_for_jobs(done_cv, done_mutex, completed, total_executions); + + auto s = pool.stats(); + logger.info(" stats: {}", s); + passed &= check(name, s.submitted == total_executions, + "all initial + follow-up jobs should be submitted"); + passed &= check(name, s.executed == total_executions, + "all initial + follow-up jobs should be executed"); + passed &= check(name, s.rejected == 0, "no jobs should be rejected"); + + pool.stop(); + //! [self-submit example] + results.push_back({name, passed}); + } + + // ------------------------------------------------------------------------- + // Summary + // ------------------------------------------------------------------------- + logger.info("==================== Results ===================="); + int total_passed = 0; + for (const auto &r : results) { + if (r.passed) { + logger.info(" PASS {}", r.name); + ++total_passed; + } else { + logger.error(" FAIL {}", r.name); + } + } + logger.info("================================================="); + logger.info("{}/{} tests passed", total_passed, results.size()); + if (total_passed == static_cast(results.size())) { + logger.info("All tests passed!"); + } else { + logger.error("{} test(s) FAILED", static_cast(results.size()) - total_passed); + } + + while (true) { + std::this_thread::sleep_for(1s); + } +} diff --git a/components/thread_pool/example/sdkconfig.defaults b/components/thread_pool/example/sdkconfig.defaults new file mode 100644 index 000000000..c3667f3e3 --- /dev/null +++ b/components/thread_pool/example/sdkconfig.defaults @@ -0,0 +1,4 @@ +# Common ESP-related +# +CONFIG_ESP_SYSTEM_EVENT_TASK_STACK_SIZE=4096 +CONFIG_ESP_MAIN_TASK_STACK_SIZE=8192 diff --git a/components/thread_pool/idf_component.yml b/components/thread_pool/idf_component.yml new file mode 100644 index 000000000..71a535f71 --- /dev/null +++ b/components/thread_pool/idf_component.yml @@ -0,0 +1,21 @@ +## IDF Component Manager Manifest File +license: "MIT" +description: "Generic thread pool component implemented with espp::Task and condition variables." +url: "https://github.com/esp-cpp/espp/tree/main/components/thread_pool" +repository: "git://github.com/esp-cpp/espp.git" +maintainers: + - William Emfinger +documentation: "https://esp-cpp.github.io/espp/core/thread_pool.html" +examples: + - path: example +tags: + - cpp + - Component + - ThreadPool + - Task + - Concurrent +dependencies: + idf: + version: '>=5.0' + espp/base_component: '>=1.0' + espp/task: '>=1.0' diff --git a/components/thread_pool/include/thread_pool.hpp b/components/thread_pool/include/thread_pool.hpp new file mode 100644 index 000000000..41fd9336d --- /dev/null +++ b/components/thread_pool/include/thread_pool.hpp @@ -0,0 +1,145 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "base_component.hpp" +#include "task.hpp" + +namespace espp { + +/** + * @brief A thread pool that dispatches submitted jobs to a fixed set of worker threads. + * + * Workers are implemented as espp::Task instances. Jobs are queued and + * consumed in FIFO order. The queue can be optionally bounded; when full, + * new submissions are either rejected immediately or blocked until space + * becomes available, depending on the configuration. + * + * \section thread_pool_ex1 Lifecycle: start / stop / is_running / worker_count + * \snippet thread_pool_example.cpp lifecycle example + * \section thread_pool_ex2 Submit Jobs + * \snippet thread_pool_example.cpp submit example + * \section thread_pool_ex3 try_submit — Non-Blocking Rejection When Full + * \snippet thread_pool_example.cpp try_submit example + * \section thread_pool_ex4 Blocking Submit When Full + * \snippet thread_pool_example.cpp blocking submit example + * \section thread_pool_ex5 Submit Rejected After stop() + * \snippet thread_pool_example.cpp submit after stop example + * \section thread_pool_ex6 Concurrent start / stop + * \snippet thread_pool_example.cpp concurrent lifecycle example + * \section thread_pool_ex7 Concurrent submit and try_submit + * \snippet thread_pool_example.cpp concurrent submit example + * \section thread_pool_ex8 Chained Pools + * \snippet thread_pool_example.cpp chained pools example + * \section thread_pool_ex9 Self-Submit + * \snippet thread_pool_example.cpp self-submit example + */ +class ThreadPool : public espp::BaseComponent { +public: + /// @brief A callable job that can be submitted to the pool. + using Job = std::function; + + /// @brief Snapshot of pool activity counters. + struct Stats { + std::uint64_t submitted = 0; ///< Total jobs accepted into the queue. + std::uint64_t executed = 0; ///< Total jobs successfully executed. + std::uint64_t rejected = 0; ///< Total jobs rejected (invalid job, stopped/stopping, or queue full) or dropped (due to stop, the enqueued jobs were dropped). + }; + + /// @brief Configuration parameters for constructing a ThreadPool. + struct Config { + std::size_t worker_count = 1; ///< Number of worker threads to spawn. + std::size_t max_queue_size = 0; ///< Maximum pending jobs (0 = unbounded). + bool auto_start = true; ///< Start workers immediately on construction. + bool block_on_submit_when_full = false; ///< If true, submit() blocks when the queue is full instead of rejecting. + espp::Task::BaseConfig worker_task_config = { ///< Base configuration applied to every worker task. + .name = "thread_pool_worker", + .stack_size_bytes = 4096, + .priority = 5, + .core_id = -1, + }; + espp::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN; ///< Logger verbosity level. + }; + + /// @brief Construct the pool with the given configuration. + /// @param config Pool configuration. + explicit ThreadPool(const Config &config); + + /// @brief Destroy the pool, stopping all workers gracefully. + ~ThreadPool(); + + /// @brief Start all worker threads. + /// @return True if all workers were successfully started, false otherwise. + /// @note No-op if the pool is already running and return true immediately. + /// @note If any workers could not be started, the pool will roll back to the stopped state. + bool start(); + + /// @brief Stop all worker threads and reject further submissions. + /// @note Blocks until every worker has exited; queued jobs may not be executed. + void stop(); + + /// @brief Query whether the pool is currently running. + /// @return true if workers are active, false otherwise. + bool is_running() const; + + /// @brief Submit a job, optionally blocking when the queue is full. + /// + /// Blocks if Config::block_on_submit_when_full is true and the queue has + /// reached its capacity limit. Otherwise behaves identically to try_submit(). + /// @param job Callable to enqueue; moved into the queue on acceptance. + /// @return true if the job was accepted, false if it was rejected. + bool submit(Job &&job); + + /// @brief Attempt to submit a job without blocking. + /// + /// Returns immediately with false when the queue is full. + /// @param job Callable to enqueue; moved into the queue on acceptance. + /// @return true if the job was accepted, false if it was rejected. + bool try_submit(Job &&job); + + /// @brief Return the number of jobs currently waiting in the queue. + /// @return Pending job count. + std::size_t queue_size() const; + + /// @brief Return the number of worker threads in the pool. + /// @return Worker thread count. + std::size_t worker_count() const; + + /// @brief Return a snapshot of the pool's activity counters. + /// @return Stats struct with submitted, executed, and rejected counts. + Stats stats() const; + +private: + bool worker_task_fn(); + + bool submit_impl(Job &&job, bool allow_blocking_when_full); + + Config config_; + + std::mutex lifecycle_mutex_; + mutable std::mutex queue_mutex_; + std::condition_variable queue_has_work_cv_; + std::condition_variable queue_has_space_cv_; + std::deque queue_; + + std::vector> workers_; + std::atomic running_{false}; + std::atomic stopping_{false}; + + std::atomic submitted_{0}; + std::atomic executed_{0}; + std::atomic rejected_{0}; +}; + +} // namespace espp + +#include "thread_pool_format_helpers.hpp" diff --git a/components/thread_pool/include/thread_pool_format_helpers.hpp b/components/thread_pool/include/thread_pool_format_helpers.hpp new file mode 100644 index 000000000..f301b10fb --- /dev/null +++ b/components/thread_pool/include/thread_pool_format_helpers.hpp @@ -0,0 +1,17 @@ +#pragma once + +#include "format.hpp" + +// fmt formatter for espp::ThreadPool::Stats +template <> struct fmt::formatter { + template constexpr auto parse(ParseContext &ctx) const { + return ctx.begin(); + } + + template + auto format(espp::ThreadPool::Stats const &s, FormatContext &ctx) const { + return fmt::format_to(ctx.out(), + "ThreadPool::Stats{{submitted: {}, executed: {}, rejected: {}}}", + s.submitted, s.executed, s.rejected); + } +}; diff --git a/components/thread_pool/src/thread_pool.cpp b/components/thread_pool/src/thread_pool.cpp new file mode 100644 index 000000000..d42a0ba38 --- /dev/null +++ b/components/thread_pool/src/thread_pool.cpp @@ -0,0 +1,164 @@ +#include "thread_pool.hpp" + +#include + +using namespace espp; + +ThreadPool::ThreadPool(const Config &config) + : BaseComponent("ThreadPool", config.log_level), config_(config) { + if (config_.worker_count == 0) { + config_.worker_count = 1; + } + + workers_.reserve(config_.worker_count); + for (std::size_t i = 0; i < config_.worker_count; ++i) { + auto worker_config = config_.worker_task_config; + worker_config.name = config_.worker_task_config.name + "_" + std::to_string(i); + workers_.emplace_back(espp::Task::make_unique({ + .callback = [this]() { return worker_task_fn(); }, + .task_config = worker_config, + .log_level = config_.log_level, + })); + } + + if (config_.auto_start) { + start(); + } +} + +ThreadPool::~ThreadPool() { stop(); } + +bool ThreadPool::start() { + std::lock_guard lifecycle_lock(lifecycle_mutex_); + + if (running_.load()) { + return true; + } + stopping_ = false; + + bool all_workers_started = true; + for (std::size_t i = 0; i < workers_.size(); ++i) { + if ((!workers_[i]->start()) && (!workers_[i]->is_running())) { + logger_.warn("Failed to start worker {} (already started or insufficient memory)", i); + all_workers_started = false; + break; + } + } + + if (!all_workers_started) { + logger_.error("Not all workers started; rolling back pool start"); + stopping_ = true; + queue_has_work_cv_.notify_all(); + queue_has_space_cv_.notify_all(); + for (auto &worker : workers_) { + worker->stop(); + } + return false; + } + + running_.store(true); + return true; +} + +void ThreadPool::stop() { + std::lock_guard lifecycle_lock(lifecycle_mutex_); + + if (!running_.exchange(false)) { + return; + } + stopping_ = true; + + { + std::lock_guard lock(queue_mutex_); + rejected_ += static_cast(queue_.size()); + queue_.clear(); + } + + queue_has_work_cv_.notify_all(); + queue_has_space_cv_.notify_all(); + + for (auto &worker : workers_) { + worker->stop(); + } +} + +bool ThreadPool::is_running() const { return running_.load(); } + +bool ThreadPool::submit(Job &&job) { + return submit_impl(std::move(job), config_.block_on_submit_when_full); +} + +bool ThreadPool::try_submit(Job &&job) { return submit_impl(std::move(job), false); } + +bool ThreadPool::submit_impl(Job &&job, bool allow_blocking_when_full) { + if (!job) { + rejected_++; + return false; + } + + std::unique_lock lock(queue_mutex_); + if (!running_.load() || stopping_) { + rejected_++; + return false; + } + + if (config_.max_queue_size > 0) { + if (allow_blocking_when_full) { + queue_has_space_cv_.wait(lock, [&]() { + return stopping_ || queue_.size() < config_.max_queue_size; + }); + if (stopping_) { + rejected_++; + return false; + } + } else if (queue_.size() >= config_.max_queue_size) { + rejected_++; + return false; + } + } + + queue_.push_back(std::move(job)); + submitted_++; + lock.unlock(); + queue_has_work_cv_.notify_one(); + return true; +} + +std::size_t ThreadPool::queue_size() const { + std::lock_guard lock(queue_mutex_); + return queue_.size(); +} + +std::size_t ThreadPool::worker_count() const { return workers_.size(); } + +ThreadPool::Stats ThreadPool::stats() const { + return { + .submitted = submitted_.load(), + .executed = executed_.load(), + .rejected = rejected_.load(), + }; +} + +bool ThreadPool::worker_task_fn() { + Job job; + { + std::unique_lock lock(queue_mutex_); + queue_has_work_cv_.wait(lock, [&]() { return stopping_ || !queue_.empty(); }); + + if (stopping_) { + return true; + } + + job = std::move(queue_.front()); + queue_.pop_front(); + } + + if (config_.max_queue_size > 0) { + queue_has_space_cv_.notify_one(); + } + + job(); + executed_++; + + return false; +} diff --git a/doc/Doxyfile b/doc/Doxyfile index 63fd3069a..e5b3f886e 100755 --- a/doc/Doxyfile +++ b/doc/Doxyfile @@ -170,6 +170,7 @@ EXAMPLE_PATH = \ $(PROJECT_PATH)/components/t_keyboard/example/main/t_keyboard_example.cpp \ $(PROJECT_PATH)/components/task/example/main/task_example.cpp \ $(PROJECT_PATH)/components/thermistor/example/main/thermistor_example.cpp \ + $(PROJECT_PATH)/components/thread_pool/example/main/thread_pool_example.cpp \ $(PROJECT_PATH)/components/timer/example/main/timer_example.cpp \ $(PROJECT_PATH)/components/touch/example/main/touch_example.cpp \ $(PROJECT_PATH)/components/tla2528/example/main/tla2528_example.cpp \ @@ -383,6 +384,7 @@ INPUT = \ $(PROJECT_PATH)/components/task/include/task.hpp \ $(PROJECT_PATH)/components/task/include/run_on_core.hpp \ $(PROJECT_PATH)/components/thermistor/include/thermistor.hpp \ + $(PROJECT_PATH)/components/thread_pool/include/thread_pool.hpp \ $(PROJECT_PATH)/components/timer/include/high_resolution_timer.hpp \ $(PROJECT_PATH)/components/timer/include/timer.hpp \ $(PROJECT_PATH)/components/touch/include/touch.hpp \ diff --git a/doc/en/core/index.rst b/doc/en/core/index.rst index 07ec25562..4e7f07a0d 100644 --- a/doc/en/core/index.rst +++ b/doc/en/core/index.rst @@ -11,6 +11,7 @@ the command-line interface. base_component base_peripheral task + thread_pool runqueue timer interrupt diff --git a/doc/en/core/thread_pool.rst b/doc/en/core/thread_pool.rst new file mode 100644 index 000000000..f5ce0d47e --- /dev/null +++ b/doc/en/core/thread_pool.rst @@ -0,0 +1,29 @@ +Thread Pool APIs +**************** + +ThreadPool +---------- + +The :cpp:class:`espp::ThreadPool` component provides a reusable pool of worker tasks for +executing queued jobs asynchronously. Workers are implemented as +:cpp:class:`espp::Task` instances and pull work from an internal job queue +(backed by ``std::deque``) whose maximum size is optionally enforced by +:cpp:member:`espp::ThreadPool::Config::max_queue_size`. Submissions can either +reject immediately when the queue is full or block until space is available, +depending on configuration. + +Code examples for the thread pool API are provided in the ``thread_pool`` example +folder. + +.. ------------------------------- Example ------------------------------------- + +.. toctree:: + + thread_pool_example + +.. ---------------------------- API Reference ---------------------------------- + +API Reference +------------- + +.. include-build-file:: inc/thread_pool.inc diff --git a/doc/en/core/thread_pool_example.md b/doc/en/core/thread_pool_example.md new file mode 100644 index 000000000..c97ae4e96 --- /dev/null +++ b/doc/en/core/thread_pool_example.md @@ -0,0 +1,2 @@ +```{include} ../../components/thread_pool/example/README.md +```