<commit_before>#include "extproc/pool.hpp"

#include "containers/scoped.hpp"
#include "logger.hpp"
#include "rpc/serialize_macros.hpp"

namespace extproc {

// ---------- pool_group_t ----------
const pool_group_t::config_t pool_group_t::DEFAULTS;

pool_group_t::pool_group_t(spawner_t::info_t *info, const config_t &config)
    : spawner_(info), config_(config),
      pool_maker_(this)
{
    // Check config for sanity
    guarantee(config_.max_workers >= config_.min_workers &&
              config_.max_workers > 0);
}


// ---------- pool_t ----------
pool_t::pool_t(pool_group_t *group)
    : group_(group),
      num_spawning_workers_(0),
      worker_semaphore_(group_->config_.max_workers)
{
    // Spawn initial worker pool.
    repair_invariants();
}

pool_t::~pool_t() {
    // All workers should be idle before we shut down (clients can interrupt
    // them if necessary).
    guarantee(busy_workers_.empty(), "Busy workers at pool shutdown!");

    // Kill worker processes.
    worker_t *w;
    while ((w = idle_workers_.head()))
        end_worker(&idle_workers_, w);
}

void pool_t::repair_invariants() {
    int need = config()->min_workers - num_workers();
    if (need > 0)
        spawn_workers(need);

    // Double-check.
    rassert(num_workers() >= config()->min_workers);
}

pool_t::worker_t *pool_t::acquire_worker() {
    assert_thread();

    // We're going to be using up a worker process, so we lock the semaphore.
    // This blocks if `config()->max_workers` workers are currently in use. We
    // unlock the semaphore in disconnect(), once we're done using the worker
    // process.
    worker_semaphore_.co_lock();

    // An `if` would suffice here in place of a `while`. However, this is only
    // because spawn_workers() performs no blocking/yielding operations between
    // when it adds the created workers to the idle_workers_ list and when it
    // returns. Otherwise, someone could call connect() and 'steal' our spawned
    // worker in the mean time. In case this changes later, we use a `while`.
    while (idle_workers_.empty()) {
        // Our usage of worker_semaphore_ should ensure this.
        rassert(num_workers() < config()->max_workers);
        spawn_workers(1);
    }
    guarantee(!idle_workers_.empty()); // sanity

    // Grab an idle worker, move it to the busy list, assign it to `handle`.
    worker_t *worker = idle_workers_.head();
    idle_workers_.remove(worker);
    busy_workers_.push_back(worker);
    return worker;
}

void pool_t::release_worker(worker_t *worker) THROWS_NOTHING {
    assert_thread();
    rassert(worker && worker->pool_ == this);

    // If the worker's stream isn't open, something bad has happened.
    if (!worker->is_read_open() || !worker->is_write_open()) {
        worker->on_error();

        // TODO(rntz): Currently worker_t::on_error() never returns. If we ever
        // change this, some code needs to get written here.
        unreachable();
    }

    // Move it from the busy list to the idle list.
    busy_workers_.remove(worker);
    idle_workers_.push_back(worker);

    // Free up worker slot for someone else to use.
    worker_semaphore_.unlock();
}

void pool_t::interrupt_worker(worker_t *worker) THROWS_NOTHING {
    assert_thread();
    rassert(worker && worker->pool_ == this);

    end_worker(&busy_workers_, worker);
    worker_semaphore_.unlock();

    repair_invariants();
}

void pool_t::detach_worker(worker_t *worker) {
    rassert(worker && worker->pool_ == this && worker->attached_);
    ASSERT_NO_CORO_WAITING;

    worker->attached_ = false;
    busy_workers_.remove(worker);
    worker_semaphore_.unlock();

    // Alas, we can't call repair_invariants now, since we're not allowed to
    // block.
}

void pool_t::cleanup_detached_worker(worker_t *worker) {
    rassert(worker && worker->pool_ == this && !worker->attached_);
    delete worker;
    repair_invariants();
}

// The root job that runs on spawned worker processes.
class job_acceptor_t :
        public auto_job_t<job_acceptor_t>
{
  public:
    void run_job(control_t *control, UNUSED void *extra) {
        while (-1 != accept_job(control, NULL)) {}<commit_msg>fix #854. that was easier to debug than expected, probably shouldn't have made an issue.<commit_after>#include "extproc/pool.hpp"

#include "containers/scoped.hpp"
#include "logger.hpp"
#include "rpc/serialize_macros.hpp"

namespace extproc {

// ---------- pool_group_t ----------
const pool_group_t::config_t pool_group_t::DEFAULTS;

pool_group_t::pool_group_t(spawner_t::info_t *info, const config_t &config)
    : spawner_(info), config_(config),
      pool_maker_(this)
{
    // Check config for sanity
    guarantee(config_.max_workers >= config_.min_workers &&
              config_.max_workers > 0);
}


// ---------- pool_t ----------
pool_t::pool_t(pool_group_t *group)
    : group_(group),
      num_spawning_workers_(0),
      worker_semaphore_(group_->config_.max_workers)
{
    // Spawn initial worker pool.
    repair_invariants();
}

pool_t::~pool_t() {
    // All workers should be idle before we shut down (clients can interrupt
    // them if necessary).
    guarantee(busy_workers_.empty(), "Busy workers at pool shutdown!");

    // Kill worker processes.
    worker_t *w;
    while ((w = idle_workers_.head()))
        end_worker(&idle_workers_, w);
}

void pool_t::repair_invariants() {
    int need = config()->min_workers - num_workers();
    if (need > 0)
        spawn_workers(need);

    // Double-check.
    rassert(num_workers() >= config()->min_workers);
}

pool_t::worker_t *pool_t::acquire_worker() {
    assert_thread();

    // We're going to be using up a worker process, so we lock the semaphore.
    // This blocks if `config()->max_workers` workers are currently in use. We
    // unlock the semaphore in disconnect(), once we're done using the worker
    // process.
    worker_semaphore_.co_lock();

    // An `if` would suffice here in place of a `while`. However, this is only
    // because spawn_workers() performs no blocking/yielding operations between
    // when it adds the created workers to the idle_workers_ list and when it
    // returns. Otherwise, someone could call connect() and 'steal' our spawned
    // worker in the mean time. In case this changes later, we use a `while`.
    while (idle_workers_.empty()) {
        // Our usage of worker_semaphore_ should ensure this.
        rassert(num_workers() < config()->max_workers);
        spawn_workers(1);
    }
    guarantee(!idle_workers_.empty()); // sanity

    // Grab an idle worker, move it to the busy list, assign it to `handle`.
    worker_t *worker = idle_workers_.head();
    idle_workers_.remove(worker);
    busy_workers_.push_back(worker);
    return worker;
}

void pool_t::release_worker(worker_t *worker) THROWS_NOTHING {
    assert_thread();
    rassert(worker && worker->pool_ == this);

    // If the worker's stream isn't open, something bad has happened.
    if (!worker->is_read_open() || !worker->is_write_open()) {
        worker->on_error();

        // TODO(rntz): Currently worker_t::on_error() never returns. If we ever
        // change this, some code needs to get written here.
        unreachable();
    }

    // Move it from the busy list to the idle list.
    busy_workers_.remove(worker);
    idle_workers_.push_back(worker);

    // Free up worker slot for someone else to use.
    worker_semaphore_.unlock();
}

void pool_t::interrupt_worker(worker_t *worker) THROWS_NOTHING {
    assert_thread();
    rassert(worker && worker->pool_ == this);

    end_worker(&busy_workers_, worker);
    worker_semaphore_.unlock();

    repair_invariants();
}

void pool_t::detach_worker(worker_t *worker) {
    rassert(worker && worker->pool_ == this && worker->attached_);
    ASSERT_NO_CORO_WAITING;

    worker->attached_ = false;
    busy_workers_.remove(worker);
    worker_semaphore_.unlock();

    guarantee_err(0 ==  kill(worker->pid_, SIGKILL), "could not kill worker");

    // Alas, we can't call repair_invariants now, since we're not allowed to
    // block.
}

void pool_t::cleanup_detached_worker(worker_t *worker) {
    rassert(worker && worker->pool_ == this && !worker->attached_);
    delete worker;
    repair_invariants();
}

// The root job that runs on spawned worker processes.
class job_acceptor_t :
        public auto_job_t<job_acceptor_t>
{
  public:
    void run_job(control_t *control, UNUSED void *extra) {
        while (-1 != accept_job(control, NULL)) {}
{'fraction_non_alphanumeric': 0.1278700105969622, 'fraction_numerical': 0.0016484163428705993, 'mean_word_length': 3.234297108673978, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 44, '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': '191645', 'n_tokens_mistral': 2682, 'n_tokens_neox': 2659, 'n_words': 723}