<commit_before>#include "jsproc/spawner.hpp"

#include <signal.h>             // sigaction
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/un.h>
#include <sys/wait.h>           // waitpid
#include <unistd.h>             // setpgid

#include "jsproc/job.hpp"
#include "utils.hpp"

namespace jsproc {

static bool spawner_created = false;

static void sigchld_handler(int signo) {
    guarantee(signo == SIGCHLD);
    guarantee(spawner_created);
    crash_or_trap("spawner process died");
}

spawner_t::spawner_t(const info_t &info)
    : pid_(info.pid), socket_(info.socket)
{
    guarantee(!spawner_created);
    spawner_created = true;

    // Check that the spawner hasn't already exited.
    guarantee(0 == waitpid(pid_, NULL, WNOHANG),
              "spawner process already died!");

    // Establish SIGCHLD handler for spawner.
    struct sigaction act;
    memset(&act, 0, sizeof act);
    act.sa_handler = sigchld_handler;
    guarantee_err(0 == sigemptyset(&act.sa_mask), "could not empty signal mask");
    guarantee_err(0 == sigaction(SIGCHLD, &act, NULL), "could not set SIGCHLD handler");
}

spawner_t::~spawner_t() {
    // De-establish SIGCHLD handler.
    struct sigaction act;
    memset(&act, 0, sizeof act);
    act.sa_handler = SIG_DFL;
    guarantee_err(0 == sigemptyset(&act.sa_mask), "could not empty signal mask");
    guarantee_err(0 == sigaction(SIGCHLD, &act, NULL), "could not unset SIGCHLD handler");
}

pid_t spawner_t::create(info_t *info) {
    int fds[2];
    int res = socketpair(AF_UNIX, SOCK_STREAM, 0, fds);
    if (res) return -1;

    pid_t pid = fork();
    if (-1 == pid) {
        guarantee_err(0 == close(fds[0]), "could not close fd");
        guarantee_err(0 == close(fds[1]), "could not close fd");
        return -1;
    }

    if (0 == pid) {
        // We're the child; run the spawner.
        guarantee_err(0 == close(fds[0]), "could not close fd");
        exec_spawner(fds[1]);
        unreachable();
    }

    // We're the parent. Return.
    guarantee_err(0 == close(fds[1]), "could not close fd");
    info->pid = pid;
    info->socket = fds[0];
    return 0;
}

pid_t spawner_t::spawn_process(fd_t *socket) {
    assert_thread();

    // Create a socket pair.
    fd_t fds[2];
    int res = socketpair(AF_UNIX, SOCK_STREAM, 0, fds);
    if (res) return -1;

    // We need to send the fd & receive the pid atomically with respect to other
    // calls to spawn_process().
    mutex_t::acq_t lock(&mutex_);

    // Send one half to the spawner process.
    guarantee(0 == socket_.send_fd(fds[1]));
    guarantee_err(0 == close(fds[1]), "could not close fd");
    *socket = fds[0];

    // Receive the pid from the spawner process.
    pid_t pid;
    guarantee(sizeof pid == force_read(&socket_, &pid, sizeof pid));
    guarantee(pid > 0);

    return pid;
}

void spawner_t::exec_worker(fd_t sockfd) {
    // TODO (rntz): worker should use prctl() to watch the spawner process.

    // Receive one job and run it.
    job_t::control_t control(getpid(), sockfd);
    exit(job_t::accept_job(&control));
}

void spawner_t::exec_spawner(fd_t sockfd) {
    // We set our PGID to our own PID (rather than inheriting our parent's PGID)
    // so that a signal (eg. SIGINT) sent to the parent's PGID (by eg. hitting
    // Ctrl-C at a terminal) will not propagate to us or our children.
    //
    // This is desirable because the RethinkDB engine deliberately crashes with
    // an error message if the spawner or a worker process dies; but a
    // command-line SIGINT should trigger a clean shutdown, not a crash.
    //
    // TODO(rntz): This is an ugly, hackish way to handle the problem.
    guarantee_err(0 == setpgid(0, 0), "spawner could not set PGID");

    // We ignore SIGCHLD so that we don't accumulate zombie children.
    {
        // NB. According to `man 2 sigaction` on linux, POSIX.1-2001 says that
        // this will prevent zombies, but may not be "fully portable".
        struct sigaction act;
        memset(&act, 0, sizeof act);
        act.sa_handler = SIG_IGN;
<commit_msg>workers now call prctl()<commit_after>#include "jsproc/spawner.hpp"

#include <signal.h>             // sigaction
#include <sys/prctl.h>          // prctl
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/un.h>
#include <sys/wait.h>           // waitpid
#include <unistd.h>             // setpgid

#include "jsproc/job.hpp"
#include "utils.hpp"

namespace jsproc {

static bool spawner_created = false;

static void sigchld_handler(int signo) {
    guarantee(signo == SIGCHLD);
    guarantee(spawner_created);
    crash_or_trap("spawner process died");
}

spawner_t::spawner_t(const info_t &info)
    : pid_(info.pid), socket_(info.socket)
{
    guarantee(!spawner_created);
    spawner_created = true;

    // Check that the spawner hasn't already exited.
    guarantee(0 == waitpid(pid_, NULL, WNOHANG),
              "spawner process already died!");

    // Establish SIGCHLD handler for spawner.
    struct sigaction act;
    memset(&act, 0, sizeof act);
    act.sa_handler = sigchld_handler;
    guarantee_err(0 == sigemptyset(&act.sa_mask), "could not empty signal mask");
    guarantee_err(0 == sigaction(SIGCHLD, &act, NULL), "could not set SIGCHLD handler");
}

spawner_t::~spawner_t() {
    // De-establish SIGCHLD handler.
    struct sigaction act;
    memset(&act, 0, sizeof act);
    act.sa_handler = SIG_DFL;
    guarantee_err(0 == sigemptyset(&act.sa_mask), "could not empty signal mask");
    guarantee_err(0 == sigaction(SIGCHLD, &act, NULL), "could not unset SIGCHLD handler");
}

pid_t spawner_t::create(info_t *info) {
    int fds[2];
    int res = socketpair(AF_UNIX, SOCK_STREAM, 0, fds);
    if (res) return -1;

    pid_t pid = fork();
    if (-1 == pid) {
        guarantee_err(0 == close(fds[0]), "could not close fd");
        guarantee_err(0 == close(fds[1]), "could not close fd");
        return -1;
    }

    if (0 == pid) {
        // We're the child; run the spawner.
        guarantee_err(0 == close(fds[0]), "could not close fd");
        exec_spawner(fds[1]);
        unreachable();
    }

    // We're the parent. Return.
    guarantee_err(0 == close(fds[1]), "could not close fd");
    info->pid = pid;
    info->socket = fds[0];
    return 0;
}

pid_t spawner_t::spawn_process(fd_t *socket) {
    assert_thread();

    // Create a socket pair.
    fd_t fds[2];
    int res = socketpair(AF_UNIX, SOCK_STREAM, 0, fds);
    if (res) return -1;

    // We need to send the fd & receive the pid atomically with respect to other
    // calls to spawn_process().
    mutex_t::acq_t lock(&mutex_);

    // Send one half to the spawner process.
    guarantee(0 == socket_.send_fd(fds[1]));
    guarantee_err(0 == close(fds[1]), "could not close fd");
    *socket = fds[0];

    // Receive the pid from the spawner process.
    pid_t pid;
    guarantee(sizeof pid == force_read(&socket_, &pid, sizeof pid));
    guarantee(pid > 0);

    return pid;
}

void spawner_t::exec_worker(fd_t sockfd) {
    // Makes sure we get SIGTERMed when our parent (the spawner) dies.
    // TODO(rntz): prctl is linux-specific.
    guarantee_err(0 == prctl(PR_SET_PDEATHSIG, SIGTERM),
                  "worker: could not set parent-death signal");

    // Receive one job and run it.
    job_t::control_t control(getpid(), sockfd);
    exit(job_t::accept_job(&control));
}

void spawner_t::exec_spawner(fd_t sockfd) {
    // We set our PGID to our own PID (rather than inheriting our parent's PGID)
    // so that a signal (eg. SIGINT) sent to the parent's PGID (by eg. hitting
    // Ctrl-C at a terminal) will not propagate to us or our children.
    //
    // This is desirable because the RethinkDB engine deliberately crashes with
    // an error message if the spawner or a worker process dies; but a
    // command-line SIGINT should trigger a clean shutdown, not a crash.
    //
    // TODO(rntz): This is an ugly, hackish way to handle the problem.
    guarantee_err(0 == setpgid(0, 0), "spawner could not set PGID");

    // We ignore SIGCHLD so that we don't accumulate zombie children.
    {
        // NB. According to `man 2 sigaction` on linux, POSIX.1-2001 says that
        // this will prevent zombies, but may not be "fully portable".
        struct sigaction act;
        memset(&act, 0, sizeof act);
        act.sa_handler = SIG_IGN;

{'fraction_non_alphanumeric': 0.12721609966459033, 'fraction_numerical': 0.010421657882127455, 'mean_word_length': 2.903225806451613, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 22, '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': '2810757', 'n_tokens_mistral': 2874, 'n_tokens_neox': 2856, 'n_words': 861}