<commit_before>#ifndef JSPROC_JOB_HPP_
#define JSPROC_JOB_HPP_

#include <stdarg.h>             // va_list

#include "arch/runtime/runtime_utils.hpp" // fd_t
#include "containers/archive/archive.hpp"
#include "containers/archive/fd_stream.hpp"

namespace jsproc {

// Abstract base class for jobs.
class job_t {
  public:
    virtual ~job_t() {}

    // Passed in to a job on the worker process side.
    class control_t : public unix_socket_stream_t {
      private:
        friend class spawner_t;
        control_t(pid_t pid, fd_t fd);

      public:
        void vlog(const char *fmt, va_list ap);
        void log(const char *fmt, ...)
            __attribute__((format (printf, 2, 3)));

      private:
        pid_t pid_;
    };

    // Sends us over a stream.
    // Returns 0 on success, -1 on error.
    int send(write_stream_t *stream);

    // Receives and runs a job. Called on worker process side. Returns 0 on
    // success, -1 on failure.
    static int accept_job(control_t *control);

    /* ----- Pure virtual methods ----- */

    // Serialization methods. Suggest implementing by invoking
    // RDB_MAKE_ME_SERIALIZABLE_#(..) in subclass definition.
    friend class write_message_t;
    friend class archive_deserializer_t;
    virtual void rdb_serialize(write_message_t &msg) const = 0;
    virtual archive_result_t rdb_deserialize(read_stream_t *s) = 0;

    // Called on worker process side.
    virtual void run_job(control_t *control) = 0;

    // Returns a function that deserializes & runs an instance of the
    // appropriate job type. Called on worker process side.
    typedef void (*func_t)(control_t*);
    virtual func_t job_runner() = 0;
};

template <class instance_t>
class auto_job_t : public job_t {
    static void job_runner_func(control_t *control) {
        // Get the job instance.
        instance_t job;
        archive_result_t res = deserialize(control, &job);

        if (res != ARCHIVE_SUCCESS) {
            control->log("Could not deserialize job: %s",
                         res == ARCHIVE_SOCK_ERROR ? "socket error" :
                         res == ARCHIVE_SOCK_EOF ? "end of file" :
                         res == ARCHIVE_RANGE_ERROR ? "range error" :
                         "unknown error");
            // TODO (rntz): shouldn't we just die here?
            return;
        }

        // Run it.
        job.run_job(control);
    }

    func_t job_runner() { return job_runner_func; }
};

} // namespace jsproc

#endif // JSPROC_JOB_HPP_<commit_msg>crash_or_trap on job failure<commit_after>#ifndef JSPROC_JOB_HPP_
#define JSPROC_JOB_HPP_

#include <stdarg.h>             // va_list
#include <stdlib.h>             // exit

#include "arch/runtime/runtime_utils.hpp" // fd_t
#include "containers/archive/archive.hpp"
#include "containers/archive/fd_stream.hpp"

namespace jsproc {

// Abstract base class for jobs.
class job_t {
  public:
    virtual ~job_t() {}

    // Passed in to a job on the worker process side.
    class control_t : public unix_socket_stream_t {
      private:
        friend class spawner_t;
        control_t(pid_t pid, fd_t fd);

      public:
        void vlog(const char *fmt, va_list ap);
        void log(const char *fmt, ...)
            __attribute__((format (printf, 2, 3)));

      private:
        pid_t pid_;
    };

    // Sends us over a stream.
    // Returns 0 on success, -1 on error.
    int send(write_stream_t *stream);

    // Receives and runs a job. Called on worker process side. Returns 0 on
    // success, -1 on failure.
    static int accept_job(control_t *control);

    /* ----- Pure virtual methods ----- */

    // Serialization methods. Suggest implementing by invoking
    // RDB_MAKE_ME_SERIALIZABLE_#(..) in subclass definition.
    friend class write_message_t;
    friend class archive_deserializer_t;
    virtual void rdb_serialize(write_message_t &msg) const = 0;
    virtual archive_result_t rdb_deserialize(read_stream_t *s) = 0;

    // Called on worker process side.
    virtual void run_job(control_t *control) = 0;

    // Returns a function that deserializes & runs an instance of the
    // appropriate job type. Called on worker process side.
    typedef void (*func_t)(control_t*);
    virtual func_t job_runner() = 0;
};

template <class instance_t>
class auto_job_t : public job_t {
    static void job_runner_func(control_t *control) {
        // Get the job instance.
        instance_t job;
        archive_result_t res = deserialize(control, &job);

        if (res != ARCHIVE_SUCCESS) {
            control->log("Could not deserialize job: %s",
                         res == ARCHIVE_SOCK_ERROR ? "socket error" :
                         res == ARCHIVE_SOCK_EOF ? "end of file" :
                         res == ARCHIVE_RANGE_ERROR ? "range error" :
                         "unknown error");
            crash_or_trap("worker: could not deserialize job");
        }

        // Run it.
        job.run_job(control);
    }

    func_t job_runner() { return job_runner_func; }
};

} // namespace jsproc

#endif // JSPROC_JOB_HPP_
{'fraction_non_alphanumeric': 0.09998031883487503, 'fraction_numerical': 0.00393623302499508, 'mean_word_length': 2.3107491856677522, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 10, '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': '2803888', 'n_tokens_mistral': 1624, 'n_tokens_neox': 1540, 'n_words': 462}