<commit_before>/*
 * Fork a process and connect its stdin and stdout to istreams.
 *
 * author: <NAME> <<EMAIL>>
 */

#include "fork.h"
#include "socket-util.h"
#include "istream-buffer.h"
#include "buffered-io.h"
#include "event2.h"
#include "fd-util.h"

#ifdef __linux
#include <fcntl.h>
#endif

#include <daemon/log.h>

#include <assert.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <limits.h>

struct fork {
    struct istream output;
    int output_fd;
    struct event2 output_event;
    fifo_buffer_t buffer;

    istream_t input;
    int input_fd;
    struct event2 input_event;

    pid_t pid;

    child_callback_t callback;
    void *callback_ctx;
};


static void
fork_close(struct fork *f)
{
    assert(f->output_fd >= 0);

    if (f->input != NULL) {
        assert(f->input_fd >= 0);

        event2_set(&f->input_event, 0);
        close(f->input_fd);
        istream_close_handler(f->input);
    }

    event2_set(&f->output_event, 0);

    close(f->output_fd);
    f->output_fd = -1;
    f->buffer = NULL;

    if (f->pid >= 0) {
        kill(f->pid, SIGTERM);
        child_clear(f->pid);
        /* XXX SIGKILL? */
    }
}

/*
 * input handler
 *
 */

static size_t
fork_input_data(const void *data, size_t length, void *ctx)
{
    struct fork *f = ctx;
    ssize_t nbytes;

    assert(f->input_fd >= 0);

    nbytes = write(f->input_fd, data, length);
    if (nbytes < 0) {
        if (errno == EAGAIN) {
            event2_or(&f->input_event, EV_WRITE);
            return 0;
        }

        daemon_log(1, "write() to subprocess failed: %s\n",
                   strerror(errno));
        event2_set(&f->input_event, 0);
        close(f->input_fd);
        f->input_fd = -1;
        istream_free_handler(&f->input);
        return 0;
    }

    return (size_t)nbytes;
}

#ifdef __linux
static ssize_t
fork_input_direct(__attr_unused istream_direct_t type,
                  int fd, size_t max_length, void *ctx)
{
    struct fork *f = ctx;
    ssize_t nbytes;

    assert(f->input_fd >= 0);

    nbytes = splice(fd, NULL, f->input_fd, NULL, max_length,
                    SPLICE_F_NONBLOCK | SPLICE_F_MORE | SPLICE_F_MOVE);
    if (nbytes < 0) {
        if (errno == EAGAIN) {
            if (!fd_ready_for_writing(f->input_fd)) {
                event2_or(&f->input_event, EV_WRITE);
                return -2;
            }

            /* try again, just in case connection->fd has become ready
               between the first splice() call and
               fd_ready_for_writing() */
            nbytes = splice(fd, NULL, f->input_fd, NULL, max_length,
                            SPLICE_F_NONBLOCK | SPLICE_F_MORE | SPLICE_F_MOVE);
        }
    }

    return nbytes;
}
#endif

static void
fork_input_eof(void *ctx)
{
    struct fork *f = ctx;

    assert(f->input_fd >= 0);

    event2_set(&f->input_event, 0);
    close(f->input_fd);
    f->input_fd = -1;

    f->input = NULL;
}

static void
fork_input_abort(void *ctx)
{
    struct fork *f = ctx;

    assert(f->input_fd >= 0);

    f->input = NULL;

    fork_close(f);
    istream_deinit_abort(&f->output);
}

static const struct istream_handler fork_input_handler = {
    .data = fork_input_data,
#ifdef __linux
    .direct = fork_input_direct,
#endif
    .eof = fork_input_eof,
    .abort = fork_input_abort,
};


/*
 * event for fork.output_fd<commit_msg>fork: unregister event in input_abort() handler<commit_after>/*
 * Fork a process and connect its stdin and stdout to istreams.
 *
 * author: <NAME> <<EMAIL>>
 */

#include "fork.h"
#include "socket-util.h"
#include "istream-buffer.h"
#include "buffered-io.h"
#include "event2.h"
#include "fd-util.h"

#ifdef __linux
#include <fcntl.h>
#endif

#include <daemon/log.h>

#include <assert.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <limits.h>

struct fork {
    struct istream output;
    int output_fd;
    struct event2 output_event;
    fifo_buffer_t buffer;

    istream_t input;
    int input_fd;
    struct event2 input_event;

    pid_t pid;

    child_callback_t callback;
    void *callback_ctx;
};


static void
fork_close(struct fork *f)
{
    assert(f->output_fd >= 0);

    if (f->input != NULL) {
        assert(f->input_fd >= 0);

        event2_set(&f->input_event, 0);
        close(f->input_fd);
        istream_close_handler(f->input);
    }

    event2_set(&f->output_event, 0);

    close(f->output_fd);
    f->output_fd = -1;
    f->buffer = NULL;

    if (f->pid >= 0) {
        kill(f->pid, SIGTERM);
        child_clear(f->pid);
        /* XXX SIGKILL? */
    }
}

/*
 * input handler
 *
 */

static size_t
fork_input_data(const void *data, size_t length, void *ctx)
{
    struct fork *f = ctx;
    ssize_t nbytes;

    assert(f->input_fd >= 0);

    nbytes = write(f->input_fd, data, length);
    if (nbytes < 0) {
        if (errno == EAGAIN) {
            event2_or(&f->input_event, EV_WRITE);
            return 0;
        }

        daemon_log(1, "write() to subprocess failed: %s\n",
                   strerror(errno));
        event2_set(&f->input_event, 0);
        close(f->input_fd);
        f->input_fd = -1;
        istream_free_handler(&f->input);
        return 0;
    }

    return (size_t)nbytes;
}

#ifdef __linux
static ssize_t
fork_input_direct(__attr_unused istream_direct_t type,
                  int fd, size_t max_length, void *ctx)
{
    struct fork *f = ctx;
    ssize_t nbytes;

    assert(f->input_fd >= 0);

    nbytes = splice(fd, NULL, f->input_fd, NULL, max_length,
                    SPLICE_F_NONBLOCK | SPLICE_F_MORE | SPLICE_F_MOVE);
    if (nbytes < 0) {
        if (errno == EAGAIN) {
            if (!fd_ready_for_writing(f->input_fd)) {
                event2_or(&f->input_event, EV_WRITE);
                return -2;
            }

            /* try again, just in case connection->fd has become ready
               between the first splice() call and
               fd_ready_for_writing() */
            nbytes = splice(fd, NULL, f->input_fd, NULL, max_length,
                            SPLICE_F_NONBLOCK | SPLICE_F_MORE | SPLICE_F_MOVE);
        }
    }

    return nbytes;
}
#endif

static void
fork_input_eof(void *ctx)
{
    struct fork *f = ctx;

    assert(f->input_fd >= 0);

    event2_set(&f->input_event, 0);
    close(f->input_fd);
    f->input_fd = -1;

    f->input = NULL;
}

static void
fork_input_abort(void *ctx)
{
    struct fork *f = ctx;

    assert(f->input_fd >= 0);

    event2_set(&f->input_event, 0);
    f->input = NULL;

    fork_close(f);
    istream_deinit_abort(&f->output);
}

static const struct istream_handler fork_input_handler = {
    .data = fork_input_data,
#ifdef __linux
    .direct = fork_input_direct,
#endif
    .eof = fork_input_eof,
    .abort = fork_input_abort,
};


/*
 * event for fork.output_fd
{'fraction_non_alphanumeric': 0.1330998248686515, 'fraction_numerical': 0.008756567425569177, 'mean_word_length': 2.3642611683848798, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 108, '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': '1614422', 'n_tokens_mistral': 2586, 'n_tokens_neox': 2486, 'n_words': 550}