Thread Pool APIs

ThreadPool

The espp::ThreadPool component provides a reusable pool of worker tasks for executing queued jobs asynchronously. Workers are implemented as espp::Task instances and pull work from an internal job queue (backed by std::deque) whose maximum size is optionally enforced by 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.

API Reference

Header File

• components/thread_pool/include/thread_pool.hpp

Classes

class ThreadPool : public espp::BaseComponent

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.

Lifecycle: start / stop / is_running / worker_count

    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()");

Submit Jobs

    logger.info("--- {} ---", name);
    bool passed = true;

    std::mutex done_mutex;
    std::condition_variable done_cv;
    constexpr int total_jobs = 8;
    std::atomic<int> 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();

try_submit - Non-Blocking Rejection When Full

    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<int> jobs_started{0};

    auto blocking_job = [&]() {
      // Signal that this job is now executing (off the queue)
      {
        std::lock_guard<std::mutex> lock(started_mutex);
        ++jobs_started;
      }
      started_cv.notify_one();
      // Block until the test releases the barrier
      std::unique_lock<std::mutex> 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<std::mutex> 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<std::mutex> lock(barrier_mutex);
      release_workers = true;
    }
    barrier_cv.notify_all();
    pool.stop();

Blocking Submit When Full

    logger.info("--- {} ---", name);
    bool passed = true;

    std::mutex done_mutex;
    std::condition_variable done_cv;
    std::atomic<int> 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();

Submit Rejected After stop()

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

Concurrent start / stop

    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<std::thread> 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 submit and try_submit

    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<int> completed_jobs{0};
    std::atomic<int> 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<std::thread> 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();

Chained Pools

    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<int> 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();

Self-Submit

    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<int> 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();

Public Types

using Job = std::function<void(void)>

A callable job that can be submitted to the pool.

Public Functions

explicit ThreadPool(const Config &config)

Construct the pool with the given configuration.

Parameters:

config – Pool configuration.

~ThreadPool()

Destroy the pool, stopping all workers gracefully.

bool start()

Start all worker threads.

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.

Returns:

True if all workers were successfully started, false otherwise.

void stop()

Stop all worker threads and reject further submissions.

Note

Blocks until every worker has exited; queued jobs may not be executed.

bool is_running() const

Query whether the pool is currently running.

Returns:

true if workers are active, false otherwise.

bool submit(Job &&job)

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().

Parameters:

job – Callable to enqueue; moved into the queue on acceptance.

Returns:

true if the job was accepted, false if it was rejected.

bool try_submit(Job &&job)

Attempt to submit a job without blocking.

Returns immediately with false when the queue is full.

Parameters:

job – Callable to enqueue; moved into the queue on acceptance.

Returns:

true if the job was accepted, false if it was rejected.

std::size_t queue_size() const

Return the number of jobs currently waiting in the queue.

Returns:

Pending job count.

std::size_t worker_count() const

Return the number of worker threads in the pool.

Returns:

Worker thread count.

Stats stats() const

Return a snapshot of the pool’s activity counters.

Returns:

Stats struct with submitted, executed, and rejected counts.

inline const std::string &get_name() const

Get the name of the component

Note

This is the tag of the logger

Returns:

A const reference to the name of the component

inline void set_log_tag(const std::string_view &tag)

Set the tag for the logger

Parameters:

tag – The tag to use for the logger

inline espp::Logger::Verbosity get_log_level() const

Get the log level for the logger

See also

Logger::Verbosity

See also

Logger::set_verbosity

Returns:

The verbosity level of the logger

inline void set_log_level(espp::Logger::Verbosity level)

Set the log level for the logger

See also

Logger::Verbosity

See also

Logger::set_verbosity

Parameters:

level – The verbosity level to use for the logger

inline void set_log_verbosity(espp::Logger::Verbosity level)

Set the log verbosity for the logger

See also

set_log_level

See also

Logger::Verbosity

See also

Logger::set_verbosity

Note

This is a convenience method that calls set_log_level

Parameters:

level – The verbosity level to use for the logger

inline espp::Logger::Verbosity get_log_verbosity() const

Get the log verbosity for the logger

See also

get_log_level

See also

Logger::Verbosity

See also

Logger::get_verbosity

Note

This is a convenience method that calls get_log_level

Returns:

The verbosity level of the logger

inline void set_log_rate_limit(std::chrono::duration<float> rate_limit)

Set the rate limit for the logger

See also

Logger::set_rate_limit

Note

Only calls to the logger that have _rate_limit suffix will be rate limited

Parameters:

rate_limit – The rate limit to use for the logger

struct Config

Configuration parameters for constructing a ThreadPool.

Public Members

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::Logger::Verbosity log_level = espp::Logger::Verbosity::WARN

Logger verbosity level.

struct Stats

Snapshot of pool activity counters.

Public Members

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

full) or dropped (due to stop, the enqueued jobs were dropped).

Total jobs rejected (invalid job, stopped/stopping, or queue