#include #include #include #include "nvim/os/uv_helpers.h" #include "nvim/os/job.h" #include "nvim/os/job_defs.h" #include "nvim/os/rstream.h" #include "nvim/os/rstream_defs.h" #include "nvim/os/wstream.h" #include "nvim/os/wstream_defs.h" #include "nvim/os/event.h" #include "nvim/os/event_defs.h" #include "nvim/os/shell.h" #include "nvim/os/time.h" #include "nvim/vim.h" #include "nvim/memory.h" // {SIGNAL}_TIMEOUT is the time (in nanoseconds) that a job has to cleanly exit // before we send SIGNAL to it #define TERM_TIMEOUT 1000000000 #define KILL_TIMEOUT (TERM_TIMEOUT * 2) #define MAX_RUNNING_JOBS 100 #define JOB_BUFFER_SIZE 0xFFFF #define close_job_stream(job, stream, type) \ do { \ if (job->stream) { \ type##stream_free(job->stream); \ job->stream = NULL; \ if (!uv_is_closing((uv_handle_t *)&job->proc_std##stream)) { \ uv_close((uv_handle_t *)&job->proc_std##stream, close_cb); \ } \ } \ } while (0) #define close_job_in(job) close_job_stream(job, in, w) #define close_job_out(job) close_job_stream(job, out, r) #define close_job_err(job) close_job_stream(job, err, r) struct job { // Job id the index in the job table plus one. int id; // Exit status code of the job process int status; // Number of references to the job. The job resources will only be freed by // close_cb when this is 0 int refcount; // Time when job_stop was called for the job. uint64_t stopped_time; // If SIGTERM was already sent to the job(only send one before SIGKILL) bool term_sent; // Data associated with the job void *data; // Callbacks job_exit_cb exit_cb; rstream_cb stdout_cb, stderr_cb; // Readable streams(std{out,err}) RStream *out, *err; // Writable stream(stdin) WStream *in; // Structures for process spawning/management used by libuv uv_process_t proc; uv_process_options_t proc_opts; uv_stdio_container_t stdio[3]; uv_pipe_t proc_stdin, proc_stdout, proc_stderr; }; static Job *table[MAX_RUNNING_JOBS] = {NULL}; size_t stop_requests = 0; static uv_timer_t job_stop_timer; // Some helpers shared in this module #ifdef INCLUDE_GENERATED_DECLARATIONS # include "os/job.c.generated.h" #endif // Callbacks for libuv /// Initializes job control resources void job_init(void) { uv_disable_stdio_inheritance(); uv_timer_init(uv_default_loop(), &job_stop_timer); } /// Releases job control resources and terminates running jobs void job_teardown(void) { // Stop all jobs for (int i = 0; i < MAX_RUNNING_JOBS; i++) { Job *job; if ((job = table[i]) != NULL) { job_stop(job); } } // Wait until all jobs are closed event_poll_until(-1, !stop_requests); // Close the timer uv_close((uv_handle_t *)&job_stop_timer, NULL); } /// Tries to start a new job. /// /// @param argv Argument vector for the process. The first item is the /// executable to run. /// [consumed] /// @param data Caller data that will be associated with the job /// @param writable If true the job stdin will be available for writing with /// job_write, otherwise it will be redirected to /dev/null /// @param stdout_cb Callback that will be invoked when data is available /// on stdout. If NULL stdout will be redirected to /dev/null. /// @param stderr_cb Callback that will be invoked when data is available /// on stderr. If NULL stderr will be redirected to /dev/null. /// @param job_exit_cb Callback that will be invoked when the job exits /// @param maxmem Maximum amount of memory used by the job WStream /// @param[out] status The job id if the job started successfully, 0 if the job /// table is full, -1 if the program could not be executed. /// @return The job pointer if the job started successfully, NULL otherwise Job *job_start(char **argv, void *data, bool writable, rstream_cb stdout_cb, rstream_cb stderr_cb, job_exit_cb job_exit_cb, size_t maxmem, int *status) { int i; Job *job; // Search for a free slot in the table for (i = 0; i < MAX_RUNNING_JOBS; i++) { if (table[i] == NULL) { break; } } if (i == MAX_RUNNING_JOBS) { // No free slots shell_free_argv(argv); *status = 0; return NULL; } job = xmalloc(sizeof(Job)); // Initialize job->id = i + 1; *status = job->id; job->status = -1; job->refcount = 1; job->data = data; job->stdout_cb = stdout_cb; job->stderr_cb = stderr_cb; job->exit_cb = job_exit_cb; job->stopped_time = 0; job->term_sent = false; job->proc_opts.file = argv[0]; job->proc_opts.args = argv; job->proc_opts.stdio = job->stdio; job->proc_opts.stdio_count = 3; job->proc_opts.flags = UV_PROCESS_WINDOWS_HIDE; job->proc_opts.exit_cb = exit_cb; job->proc_opts.cwd = NULL; job->proc_opts.env = NULL; job->proc.data = NULL; job->proc_stdin.data = NULL; job->proc_stdout.data = NULL; job->proc_stderr.data = NULL; job->in = NULL; job->out = NULL; job->err = NULL; // Initialize the job std{in,out,err} job->stdio[0].flags = UV_IGNORE; job->stdio[1].flags = UV_IGNORE; job->stdio[2].flags = UV_IGNORE; if (writable) { uv_pipe_init(uv_default_loop(), &job->proc_stdin, 0); job->stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE; job->stdio[0].data.stream = (uv_stream_t *)&job->proc_stdin; handle_set_job((uv_handle_t *)&job->proc_stdin, job); job->refcount++; } if (stdout_cb) { uv_pipe_init(uv_default_loop(), &job->proc_stdout, 0); job->stdio[1].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE; job->stdio[1].data.stream = (uv_stream_t *)&job->proc_stdout; handle_set_job((uv_handle_t *)&job->proc_stdout, job); job->refcount++; } if (stderr_cb) { uv_pipe_init(uv_default_loop(), &job->proc_stderr, 0); job->stdio[2].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE; job->stdio[2].data.stream = (uv_stream_t *)&job->proc_stderr; handle_set_job((uv_handle_t *)&job->proc_stderr, job); job->refcount++; } handle_set_job((uv_handle_t *)&job->proc, job); // Spawn the job if (uv_spawn(uv_default_loop(), &job->proc, &job->proc_opts) != 0) { if (writable) { uv_close((uv_handle_t *)&job->proc_stdin, close_cb); } if (stdout_cb) { uv_close((uv_handle_t *)&job->proc_stdout, close_cb); } if (stderr_cb) { uv_close((uv_handle_t *)&job->proc_stderr, close_cb); } uv_close((uv_handle_t *)&job->proc, close_cb); event_poll(0); // Manually invoke the close_cb to free the job resources *status = -1; return NULL; } if (writable) { job->in = wstream_new(maxmem); wstream_set_stream(job->in, (uv_stream_t *)&job->proc_stdin); } // Start the readable streams if (stdout_cb) { job->out = rstream_new(read_cb, rbuffer_new(JOB_BUFFER_SIZE), job); rstream_set_stream(job->out, (uv_stream_t *)&job->proc_stdout); rstream_start(job->out); } if (stderr_cb) { job->err = rstream_new(read_cb, rbuffer_new(JOB_BUFFER_SIZE), job); rstream_set_stream(job->err, (uv_stream_t *)&job->proc_stderr); rstream_start(job->err); } // Save the job to the table table[i] = job; return job; } /// Finds a job instance by id /// /// @param id The job id /// @return the Job instance Job *job_find(int id) { Job *job; if (id <= 0 || id > MAX_RUNNING_JOBS || !(job = table[id - 1]) || job->stopped_time) { return NULL; } return job; } /// Terminates a job. This is a non-blocking operation, but if the job exists /// it's guaranteed to succeed(SIGKILL will eventually be sent) /// /// @param job The Job instance void job_stop(Job *job) { if (job->stopped_time) { return; } job->stopped_time = os_hrtime(); // Close the job's stdin. If the job doesn't close it's own stdout/stderr, // they will be closed when the job exits(possibly due to being terminated // after a timeout) close_job_in(job); if (!stop_requests++) { // When there's at least one stop request pending, start a timer that // will periodically check if a signal should be send to a to the job DLOG("Starting job kill timer"); uv_timer_start(&job_stop_timer, job_stop_timer_cb, 100, 100); } } /// job_wait - synchronously wait for a job to finish /// /// @param job The job instance /// @param ms Number of milliseconds to wait, 0 for not waiting, -1 for /// waiting until the job quits. /// @return returns the status code of the exited job. -1 if the job is /// still running and the `timeout` has expired. Note that this is /// indistinguishable from the process returning -1 by itself. Which /// is possible on some OS. int job_wait(Job *job, int ms) FUNC_ATTR_NONNULL_ALL { // The default status is -1, which represents a timeout int status = -1; // Increase refcount to stop the job from being freed before we have a // chance to get the status. job->refcount++; event_poll_until(ms, // Until... got_int || // interrupted by the user job->refcount == 1); // job exited // we'll assume that a user frantically hitting interrupt doesn't like // the current job. Signal that it has to be killed. if (got_int) { got_int = false; job_stop(job); if (ms == -1) { // We can only return, if all streams/handles are closed and the job // exited. event_poll_until(-1, job->refcount == 1); } else { event_poll(0); } } if (job->refcount == 1) { // Job exited, collect status and manually invoke close_cb to free the job // resources status = job->status; close_cb((uv_handle_t *)&job->proc); } else { job->refcount--; } return status; } /// Close the pipe used to write to the job. /// /// This can be used for example to indicate to the job process that no more /// input is coming, and that it should shut down cleanly. /// /// It has no effect when the input pipe doesn't exist or was already /// closed. /// /// @param job The job instance void job_close_in(Job *job) FUNC_ATTR_NONNULL_ALL { close_job_in(job); } /// All writes that complete after calling this function will be reported /// to `cb`. /// /// Use this function to be notified about the status of an in-flight write. /// /// @see {wstream_set_write_cb} /// /// @param job The job instance /// @param cb The function that will be called on write completion or /// failure. It will be called with the job as the `data` argument. void job_write_cb(Job *job, wstream_cb cb) FUNC_ATTR_NONNULL_ALL { wstream_set_write_cb(job->in, cb, job); } /// Writes data to the job's stdin. This is a non-blocking operation, it /// returns when the write request was sent. /// /// @param job The Job instance /// @param buffer The buffer which contains the data to be written /// @return true if the write request was successfully sent, false if writing /// to the job stream failed (possibly because the OS buffer is full) bool job_write(Job *job, WBuffer *buffer) { return wstream_write(job->in, buffer); } /// Get the job id /// /// @param job A pointer to the job /// @return The job id int job_id(Job *job) { return job->id; } /// Get data associated with a job /// /// @param job A pointer to the job /// @return The job data void *job_data(Job *job) { return job->data; } static void job_exit_callback(Job *job) { // Free the slot now, 'exit_cb' may want to start another job to replace // this one table[job->id - 1] = NULL; if (job->exit_cb) { // Invoke the exit callback job->exit_cb(job, job->data); } if (stop_requests && !--stop_requests) { // Stop the timer if no more stop requests are pending DLOG("Stopping job kill timer"); uv_timer_stop(&job_stop_timer); } } /// Iterates the table, sending SIGTERM to stopped jobs and SIGKILL to those /// that didn't die from SIGTERM after a while(exit_timeout is 0). static void job_stop_timer_cb(uv_timer_t *handle) { Job *job; uint64_t now = os_hrtime(); for (size_t i = 0; i < MAX_RUNNING_JOBS; i++) { if ((job = table[i]) == NULL || !job->stopped_time) { continue; } uint64_t elapsed = now - job->stopped_time; if (!job->term_sent && elapsed >= TERM_TIMEOUT) { ILOG("Sending SIGTERM to job(id: %d)", job->id); uv_process_kill(&job->proc, SIGTERM); job->term_sent = true; } else if (elapsed >= KILL_TIMEOUT) { ILOG("Sending SIGKILL to job(id: %d)", job->id); uv_process_kill(&job->proc, SIGKILL); } } } // Wraps the call to std{out,err}_cb and emits a JobExit event if necessary. static void read_cb(RStream *rstream, void *data, bool eof) { Job *job = data; if (rstream == job->out) { job->stdout_cb(rstream, data, eof); if (eof) { close_job_out(job); } } else { job->stderr_cb(rstream, data, eof); if (eof) { close_job_err(job); } } } // Emits a JobExit event if both rstreams are closed static void exit_cb(uv_process_t *proc, int64_t status, int term_signal) { Job *job = handle_get_job((uv_handle_t *)proc); job->status = (int)status; uv_close((uv_handle_t *)&job->proc, close_cb); } static void close_cb(uv_handle_t *handle) { Job *job = handle_get_job(handle); if (handle == (uv_handle_t *)&job->proc) { // Make sure all streams are properly closed to trigger callback invocation // when job->proc is closed close_job_in(job); close_job_out(job); close_job_err(job); } if (--job->refcount == 0) { // Invoke the exit_cb job_exit_callback(job); // Free all memory allocated for the job free(job->proc.data); free(job->proc_stdin.data); free(job->proc_stdout.data); free(job->proc_stderr.data); shell_free_argv(job->proc_opts.args); free(job); } }