<commit_before>#ifndef __CONCURRENCY_CORO_POOL_HPP__
#define	__CONCURRENCY_CORO_POOL_HPP__

#include "errors.hpp"
#include "concurrency/cond_var.hpp"
#include "concurrency/gate.hpp"


/* coro_pool_t allows tasks to be queued. These tasks are dispatched
to a number of coroutine workers. The workers themselves get allocated and destroyed
dynamically, obeying some limit on the maximal number of running worker coroutines. */

class coro_pool_t : public home_thread_mixin_t {
public:
    coro_pool_t(size_t worker_count_) :
            max_worker_count(worker_count_),
            active_worker_count(0),
            shutting_down(false) {

        rassert(max_worker_count > 0);

        worker_active_gate_opener.reset(new gate_t::open_t(&worker_active_gate));
    }
    ~coro_pool_t() {
        shutting_down = true;

        // Close the active gate. This automatically waits on all workers to finish whatever they are doing
        worker_active_gate_opener.reset();

        rassert(task_queue.empty());
        rassert(active_worker_count == 0);
    }

    // Must be called on home_thread. Use queue_task_on_home_thread if calling from a different thread
    void queue_task(const boost::function<void()> &task) {
        assert_thread();
        rassert(!shutting_down);
        rassert(active_worker_count <= max_worker_count);

        // Place the task on the queue
        task_queue.push_back(task);

        // Spawn a new worker if permitted
        if (active_worker_count < max_worker_count) {
            ++active_worker_count;
            coro_t::spawn(boost::bind(&coro_pool_t::worker_run, this));
        }<commit_msg>Add an additional safety net to coro_pool's shutdown procedure, just to be sure.<commit_after>#ifndef __CONCURRENCY_CORO_POOL_HPP__
#define	__CONCURRENCY_CORO_POOL_HPP__

#include "errors.hpp"
#include "concurrency/cond_var.hpp"
#include "concurrency/gate.hpp"


/* coro_pool_t allows tasks to be queued. These tasks are dispatched
to a number of coroutine workers. The workers themselves get allocated and destroyed
dynamically, obeying some limit on the maximal number of running worker coroutines. */

class coro_pool_t : public home_thread_mixin_t {
public:
    coro_pool_t(size_t worker_count_) :
            max_worker_count(worker_count_),
            active_worker_count(0),
            shutting_down(false) {

        rassert(max_worker_count > 0);

        worker_active_gate_opener.reset(new gate_t::open_t(&worker_active_gate));
    }
    ~coro_pool_t() {
        assert_thread();
        shutting_down = true;

        // Wait till queued tasks get handled...
        // (in case queue_task got called and spawned a new coroutine but that coroutine has not materialized yet)
        if (!task_queue.empty()) {
            coro_t::yield();
        }

        // Close the active gate. This automatically waits on all workers to finish whatever they are doing
        worker_active_gate_opener.reset();

        rassert(task_queue.empty());
        rassert(active_worker_count == 0);
    }

    // Must be called on home_thread. Use queue_task_on_home_thread if calling from a different thread
    void queue_task(const boost::function<void()> &task) {
        assert_thread();
        rassert(!shutting_down);
        rassert(active_worker_count <= max_worker_count);

        // Place the task on the queue
        task_queue.push_back(task);

        // Spawn a new worker if permitted
        if (active_worker_count < max_worker_count) {
            ++active_worker_count;
            coro_t::spawn(boost::bind(&coro_pool_t::worker_run, this));
        }
{'fraction_non_alphanumeric': 0.08257134940426711, 'fraction_numerical': 0.0016625103906899418, 'mean_word_length': 3.1589861751152073, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1337191', 'n_tokens_mistral': 1120, 'n_tokens_neox': 1082, 'n_words': 283}