Implement job control

- Add a job control module for spawning and controlling co-processes
- Add three vimscript functions for interfacing with the module
- Use dedicated header files for typedefs/structs in event/job modules

src/eval.c

@@ -63,6 +63,7 @@
 #include "version.h"
 #include "window.h"
 #include "os/os.h"
+#include "os/job.h"
 #include "os/shell.h"
 
 #if defined(FEAT_FLOAT)
@@ -388,6 +389,7 @@ static struct vimvar {
   {VV_NAME("hlsearch",         VAR_NUMBER), 0},
   {VV_NAME("oldfiles",         VAR_LIST), 0},
   {VV_NAME("windowid",         VAR_NUMBER), VV_RO},
+  {VV_NAME("job_data",         VAR_LIST), 0}
 };
 
 /* shorthand */
@@ -401,6 +403,11 @@ static struct vimvar {
 static dictitem_T vimvars_var;                  /* variable used for v: */
 #define vimvarht  vimvardict.dv_hashtab
 
+static void apply_job_autocmds(int id,
+                               void *data,
+                               char *buffer,
+                               uint32_t len,
+                               bool from_stdout);
 static void prepare_vimvar(int idx, typval_T *save_tv);
 static void restore_vimvar(int idx, typval_T *save_tv);
 static int ex_let_vars(char_u *arg, typval_T *tv, int copy,
@@ -629,6 +636,9 @@ static void f_invert(typval_T *argvars, typval_T *rettv);
 static void f_isdirectory(typval_T *argvars, typval_T *rettv);
 static void f_islocked(typval_T *argvars, typval_T *rettv);
 static void f_items(typval_T *argvars, typval_T *rettv);
+static void f_job_start(typval_T *argvars, typval_T *rettv);
+static void f_job_stop(typval_T *argvars, typval_T *rettv);
+static void f_job_write(typval_T *argvars, typval_T *rettv);
 static void f_join(typval_T *argvars, typval_T *rettv);
 static void f_keys(typval_T *argvars, typval_T *rettv);
 static void f_last_buffer_nr(typval_T *argvars, typval_T *rettv);
@@ -1366,6 +1376,7 @@ int eval_to_number(char_u *expr)
   return retval;
 }
 
+
 /*
  * Prepare v: variable "idx" to be used.
  * Save the current typeval in "save_tv".
@@ -6946,6 +6957,9 @@ static struct fst {
   {"isdirectory",     1, 1, f_isdirectory},
   {"islocked",        1, 1, f_islocked},
   {"items",           1, 1, f_items},
+  {"jobstart",        2, 3, f_job_start},
+  {"jobstop",         1, 1, f_job_stop},
+  {"jobwrite",        2, 2, f_job_write},
   {"join",            1, 2, f_join},
   {"keys",            1, 1, f_keys},
   {"last_buffer_nr",  0, 0, f_last_buffer_nr},  /* obsolete */
@@ -11001,6 +11015,143 @@ static void f_items(typval_T *argvars, typval_T *rettv)
   dict_list(argvars, rettv, 2);
 }
 
+// "jobstart()" function
+static void f_job_start(typval_T *argvars, typval_T *rettv)
+{
+  list_T *args = NULL;
+  listitem_T *arg;
+  int i, argvl, argsl;
+  char **argv = NULL;
+
+  rettv->v_type = VAR_NUMBER;
+  rettv->vval.v_number = 0;
+
+  if (check_restricted() || check_secure()) {
+    goto cleanup;
+  }
+
+  if (argvars[0].v_type != VAR_STRING
+      || argvars[1].v_type != VAR_STRING
+      || (argvars[2].v_type != VAR_LIST
+      && argvars[2].v_type != VAR_UNKNOWN)) {
+    // Wrong argument types
+    EMSG(_(e_invarg));
+    goto cleanup;
+  }
+
+  argsl = 0;
+  if (argvars[2].v_type == VAR_LIST) {
+    args = argvars[2].vval.v_list;
+    argsl = args->lv_len;
+    // Assert that all list items are strings
+    for (arg = args->lv_first; arg != NULL; arg = arg->li_next) {
+      if (arg->li_tv.v_type != VAR_STRING) {
+        EMSG(_(e_invarg));
+        goto cleanup;
+      }
+    }
+  }
+
+  if (!os_can_exe(get_tv_string(&argvars[1]))) {
+    // String is not executable
+    EMSG2(e_jobexe, get_tv_string(&argvars[1]));
+    goto cleanup;
+  }
+
+  // Allocate extra memory for the argument vector and the NULL pointer
+  argvl = argsl + 2;
+  argv = xmalloc(sizeof(char_u *) * argvl);
+
+  // Copy program name
+  argv[0] = xstrdup((char *)argvars[1].vval.v_string);
+
+  i = 1;
+  // Copy arguments to the vector
+  if (argsl > 0) {
+    for (arg = args->lv_first; arg != NULL; arg = arg->li_next) {
+      argv[i++] = xstrdup((char *)arg->li_tv.vval.v_string);
+    }
+  }
+
+  // The last item of argv must be NULL
+  argv[i] = NULL;
+
+  rettv->vval.v_number = job_start(argv,
+                                   xstrdup((char *)argvars[0].vval.v_string),
+                                   apply_job_autocmds);
+
+  if (rettv->vval.v_number <= 0) {
+    if (rettv->vval.v_number == 0) {
+      EMSG(_(e_jobtblfull));
+    } else {
+      EMSG(_(e_jobexe));
+    }
+  }
+
+cleanup:
+  if (rettv->vval.v_number > 0) {
+    // Success
+    return;
+  }
+  // Cleanup argv memory in case the `job_start` call failed
+  shell_free_argv(argv);
+}
+
+// "jobstop()" function
+static void f_job_stop(typval_T *argvars, typval_T *rettv)
+{
+  rettv->v_type = VAR_NUMBER;
+  rettv->vval.v_number = 0;
+
+  if (check_restricted() || check_secure()) {
+    return;
+  }
+
+  if (argvars[0].v_type != VAR_NUMBER) {
+    // Only argument is the job id
+    EMSG(_(e_invarg));
+    return;
+  }
+
+  if (!job_stop(argvars[0].vval.v_number)) {
+    // Probably an invalid job id
+    EMSG(_(e_invjob));
+    return;
+  }
+
+  rettv->vval.v_number = 1;
+}
+
+// "jobwrite()" function
+static void f_job_write(typval_T *argvars, typval_T *rettv)
+{
+  bool res;
+  rettv->v_type = VAR_NUMBER;
+  rettv->vval.v_number = 0;
+
+  if (check_restricted() || check_secure()) {
+    return;
+  }
+
+  if (argvars[0].v_type != VAR_NUMBER || argvars[1].v_type != VAR_STRING) {
+    // First argument is the job id and second is the string to write to 
+    // the job's stdin
+    EMSG(_(e_invarg));
+    return;
+  }
+
+  res = job_write(argvars[0].vval.v_number,
+                  xstrdup((char *)argvars[1].vval.v_string),
+                  strlen((char *)argvars[1].vval.v_string));
+
+  if (!res) {
+    // Invalid job id
+    EMSG(_(e_invjob));
+  }
+
+  rettv->vval.v_number = 1;
+}
+
 /*
  * "join()" function
  */
@@ -11045,7 +11196,7 @@ static void f_keys(typval_T *argvars, typval_T *rettv)
 static void f_last_buffer_nr(typval_T *argvars, typval_T *rettv)
 {
   int n = 0;
-  buf_T       *buf;
+  buf_T *buf;
 
   for (buf = firstbuf; buf != NULL; buf = buf->b_next)
     if (n < buf->b_fnum)
@@ -19593,3 +19744,20 @@ char_u *do_string_sub(char_u *str, char_u *pat, char_u *sub, char_u *flags)
   return ret;
 }
 
+static void apply_job_autocmds(int id,
+                               void *data,
+                               char *buffer,
+                               uint32_t len,
+                               bool from_stdout)
+{
+  list_T *list;
+
+  // Call JobActivity autocommands
+  list = list_alloc();
+  list_append_number(list, id);
+  list_append_string(list, (char_u *)buffer, len);
+  list_append_string(list, (char_u *)(from_stdout ? "out" : "err"), 3);
+  set_vim_var_list(VV_JOB_DATA, list);
+  apply_autocmds(EVENT_JOBACTIVITY, (char_u *)data, NULL, TRUE, NULL);
+}
+

src/fileio.c

@@ -5950,6 +5950,7 @@ static struct event_name {
   {"InsertEnter",     EVENT_INSERTENTER},
   {"InsertLeave",     EVENT_INSERTLEAVE},
   {"InsertCharPre",   EVENT_INSERTCHARPRE},
+  {"JobActivity",     EVENT_JOBACTIVITY},
   {"MenuPopup",       EVENT_MENUPOPUP},
   {"QuickFixCmdPost", EVENT_QUICKFIXCMDPOST},
   {"QuickFixCmdPre",  EVENT_QUICKFIXCMDPRE},
@@ -7394,7 +7395,7 @@ apply_autocmds_group (
   } else {
     sfname = vim_strsave(fname);
     /* Don't try expanding FileType, Syntax, FuncUndefined, WindowID,
-     * ColorScheme or QuickFixCmd* */
+     * ColorScheme, QuickFixCmd or JobActivity */
     if (event == EVENT_FILETYPE
         || event == EVENT_SYNTAX
         || event == EVENT_FUNCUNDEFINED
@@ -7402,7 +7403,8 @@ apply_autocmds_group (
         || event == EVENT_SPELLFILEMISSING
         || event == EVENT_QUICKFIXCMDPRE
         || event == EVENT_COLORSCHEME
-        || event == EVENT_QUICKFIXCMDPOST)
+        || event == EVENT_QUICKFIXCMDPOST
+        || event == EVENT_JOBACTIVITY)
       fname = vim_strsave(fname);
     else
       fname = FullName_save(fname, FALSE);

src/globals.h

@@ -1009,6 +1009,9 @@ EXTERN char_u e_invexpr2[] INIT(= N_("E15: Invalid expression: %s"));
 EXTERN char_u e_invrange[] INIT(= N_("E16: Invalid range"));
 EXTERN char_u e_invcmd[] INIT(= N_("E476: Invalid command"));
 EXTERN char_u e_isadir2[] INIT(= N_("E17: \"%s\" is a directory"));
+EXTERN char_u e_invjob[] INIT(= N_("E900: Invalid job id"));
+EXTERN char_u e_jobtblfull[] INIT(= N_("E901: Job table is full"));
+EXTERN char_u e_jobexe[] INIT(= N_("E902: \"%s\" is not an executable"));
 #ifdef FEAT_LIBCALL
 EXTERN char_u e_libcall[] INIT(= N_("E364: Library call failed for \"%s()\""));
 #endif

src/os/event.c

@@ -8,6 +8,7 @@
 #include "os/event.h"
 #include "os/input.h"
 #include "os/signal.h"
+#include "os/job.h"
 #include "vim.h"
 #include "memory.h"
 #include "misc2.h"
@@ -34,6 +35,8 @@ void event_init()
   // `event_poll`
   // Signals
   signal_init();
+  // Jobs
+  job_init();
   uv_timer_init(uv_default_loop(), &timer);
   // This prepare handle that actually starts the timer
   uv_prepare_init(uv_default_loop(), &timer_prepare);
@@ -88,6 +91,9 @@ static void process_all_events()
       case kEventSignal:
         signal_handle(event);
         break;
+      case kEventJobActivity:
+        job_handle(event);
+        break;
       default:
         abort();
     }

src/os/event.h

@@ -4,16 +4,8 @@
 #include <stdint.h>
 #include <stdbool.h>
 
-typedef enum {
+#include "os/event_defs.h"
-  kEventSignal
+#include "os/job_defs.h"
-} EventType;
-
-typedef struct {
-  EventType type;
-  union {
-    int signum;
-  } data;
-} Event;
 
 void event_init(void);
 bool event_poll(int32_t ms);

src/os/event_defs.h

@@ -0,0 +1,19 @@
+#ifndef NEOVIM_OS_EVENT_DEFS_H
+#define NEOVIM_OS_EVENT_DEFS_H
+
+#include "os/job_defs.h"
+
+typedef enum {
+  kEventSignal,
+  kEventJobActivity
+} EventType;
+
+typedef struct {
+  EventType type;
+  union {
+    int signum;
+    Job *job;
+  } data;
+} Event;
+
+#endif  // NEOVIM_OS_EVENT_H

src/os/job.c

@@ -0,0 +1,345 @@
+#include <stdint.h>
+#include <stdbool.h>
+
+#include <uv.h>
+
+#include "os/job_defs.h"
+#include "os/job.h"
+#include "os/time.h"
+#include "os/shell.h"
+#include "vim.h"
+#include "memory.h"
+#include "term.h"
+
+#define EXIT_TIMEOUT 25
+#define MAX_RUNNING_JOBS 100
+#define JOB_BUFFER_SIZE 1024
+
+/// Possible lock states of the job buffer
+typedef enum {
+  kBufferLockNone = 0, ///< No data was read
+  kBufferLockStdout,   ///< Data read from stdout
+  kBufferLockStderr    ///< Data read from stderr
+} BufferLock;
+
+struct _Job {
+  // Job id the index in the job table plus one. 
+  int id;
+  // Number of polls after a SIGTERM that will trigger a SIGKILL
+  int exit_timeout;
+  // If the job was already stopped
+  bool stopped;
+  // Data associated with the job
+  void *data;
+  // Buffer for reading from stdout or stderr
+  char buffer[JOB_BUFFER_SIZE];
+  // Size of the data from the last read
+  uint32_t length;
+  // Buffer lock state
+  BufferLock lock;
+  // Callback for consuming data from the buffer
+  job_read_cb read_cb;
+  // 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};
+static uv_prepare_t job_prepare;
+
+// Some helpers shared in this module
+static bool is_alive(Job *job);
+static Job * find_job(int id);
+static void free_job(Job *job);
+
+// Callbacks for libuv
+static void job_prepare_cb(uv_prepare_t *handle, int status);
+static void alloc_cb(uv_handle_t *handle, size_t suggested, uv_buf_t *buf);
+static void read_cb(uv_stream_t *stream, ssize_t cnt, const uv_buf_t *buf);
+static void write_cb(uv_write_t *req, int status);
+static void exit_cb(uv_process_t *proc, int64_t status, int term_signal);
+
+void job_init()
+{
+  uv_disable_stdio_inheritance();
+  uv_prepare_init(uv_default_loop(), &job_prepare);
+  uv_prepare_start(&job_prepare, job_prepare_cb);
+}
+
+void job_teardown()
+{
+  // 20 tries will give processes about 1 sec to exit cleanly
+  uint32_t remaining_tries = 20;
+  bool all_dead = true;
+  int i;
+  Job *job;
+
+  // Politely ask each job to terminate
+  for (i = 0; i < MAX_RUNNING_JOBS; i++) {
+    if ((job = table[i]) != NULL) {
+      all_dead = false;
+      uv_process_kill(&job->proc, SIGTERM);
+    }
+  }
+  
+  if (all_dead) {
+    return;
+  }
+
+  os_delay(10, 0);
+  // Right now any exited process are zombies waiting for us to acknowledge
+  // their status with `wait` or handling SIGCHLD. libuv does that
+  // automatically (and then calls `exit_cb`) but we have to give it a chance
+  // by running the loop one more time
+  uv_run(uv_default_loop(), UV_RUN_NOWAIT);
+
+  // Prepare to start shooting
+  for (i = 0; i < MAX_RUNNING_JOBS; i++) {
+    if ((job = table[i]) == NULL) {
+      continue;
+    }
+
+    // Still alive
+    while (is_alive(job) && remaining_tries--) {
+      // Since this is the first time we're checking, wait 300ms so
+      // every job has a chance to exit normally
+      os_delay(50, 0);
+      // Acknowledge child exits
+      uv_run(uv_default_loop(), UV_RUN_NOWAIT);
+    }
+
+    if (is_alive(job)) {
+      uv_process_kill(&job->proc, SIGKILL);
+    }
+  }
+}
+
+int job_start(char **argv, void *data, job_read_cb cb)
+{
+  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
+    return 0;
+  }
+
+  job = xmalloc(sizeof(Job)); 
+  // Initialize
+  job->id = i + 1;
+  job->data = data;
+  job->read_cb = cb;
+  job->stopped = false;
+  job->exit_timeout = EXIT_TIMEOUT;
+  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;
+
+  // Initialize the job std{in,out,err}
+  uv_pipe_init(uv_default_loop(), &job->proc_stdin, 0);
+  job->proc_stdin.data = job;
+  job->stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE;
+  job->stdio[0].data.stream = (uv_stream_t *)&job->proc_stdin;
+
+  uv_pipe_init(uv_default_loop(), &job->proc_stdout, 0);
+  job->proc_stdout.data = job;
+  job->stdio[1].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE;
+  job->stdio[1].data.stream = (uv_stream_t *)&job->proc_stdout;
+
+  uv_pipe_init(uv_default_loop(), &job->proc_stderr, 0);
+  job->proc_stderr.data = job;
+  job->stdio[2].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE;
+  job->stdio[2].data.stream = (uv_stream_t *)&job->proc_stderr;
+
+  // Spawn the job
+  if (uv_spawn(uv_default_loop(), &job->proc, &job->proc_opts) != 0) {
+    free_job(job);
+    return -1;
+  }
+
+  // Start the readable streams
+  uv_read_start((uv_stream_t *)&job->proc_stdout, alloc_cb, read_cb);
+  uv_read_start((uv_stream_t *)&job->proc_stderr, alloc_cb, read_cb);
+  // Give the callback a reference to the job
+  job->proc.data = job;
+  // Save the job to the table
+  table[i] = job;
+
+  return job->id;
+}
+
+bool job_stop(int id)
+{
+  Job *job = find_job(id);
+
+  if (job == NULL || job->stopped) {
+    return false;
+  }
+
+  uv_read_stop((uv_stream_t *)&job->proc_stdout);
+  uv_read_stop((uv_stream_t *)&job->proc_stderr);
+  job->stopped = true;
+
+  return true;
+}
+
+bool job_write(int id, char *data, uint32_t len)
+{
+  uv_buf_t uvbuf;
+  uv_write_t *req;
+  Job *job = find_job(id);
+
+  if (job == NULL || job->stopped) {
+    free(data);
+    return false;
+  }
+
+  req = xmalloc(sizeof(uv_write_t));
+  req->data = data;
+  uvbuf.base = data;
+  uvbuf.len = len;
+  uv_write(req, (uv_stream_t *)&job->proc_stdin, &uvbuf, 1, write_cb);
+
+  return true;
+}
+
+void job_handle(Event event)
+{
+  Job *job = event.data.job;
+
+  // Invoke the job callback
+  job->read_cb(job->id,
+               job->data,
+               job->buffer,
+               job->length,
+               job->lock == kBufferLockStdout);
+
+  shell_resized();
+  // restart reading
+  job->lock = kBufferLockNone;
+  uv_read_start((uv_stream_t *)&job->proc_stdout, alloc_cb, read_cb);
+  uv_read_start((uv_stream_t *)&job->proc_stderr, alloc_cb, read_cb);
+}
+
+static bool is_alive(Job *job)
+{
+  return uv_process_kill(&job->proc, 0) == 0;
+}
+
+static Job * find_job(int id)
+{
+  if (id <= 0 || id > MAX_RUNNING_JOBS) {
+    return NULL;
+  }
+   
+  return table[id - 1];
+}
+
+static void free_job(Job *job)
+{
+  uv_close((uv_handle_t *)&job->proc_stdout, NULL);
+  uv_close((uv_handle_t *)&job->proc_stdin, NULL);
+  uv_close((uv_handle_t *)&job->proc_stderr, NULL);
+  uv_close((uv_handle_t *)&job->proc, NULL);
+  free(job);
+}
+
+/// 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_prepare_cb(uv_prepare_t *handle, int status)
+{
+  Job *job;
+  int i;
+
+  for (i = 0; i < MAX_RUNNING_JOBS; i++) {
+    if ((job = table[i]) == NULL || !job->stopped) {
+      continue;
+    }
+  
+    if ((job->exit_timeout--) == EXIT_TIMEOUT) {
+      // Job was just stopped, close all stdio handles and send SIGTERM
+      uv_process_kill(&job->proc, SIGTERM);
+    } else if (job->exit_timeout == 0) {
+      // We've waited long enough, send SIGKILL
+      uv_process_kill(&job->proc, SIGKILL);
+    }
+  }
+}
+
+/// Puts the job into a 'reading state' which 'locks' the job buffer
+/// until the data is consumed
+static void alloc_cb(uv_handle_t *handle, size_t suggested, uv_buf_t *buf)
+{
+  Job *job = (Job *)handle->data;
+
+  if (job->lock != kBufferLockNone) {
+    // Already reserved the buffer for reading from stdout or stderr.
+    buf->len = 0;
+    return;
+  }
+
+  buf->base = job->buffer;
+  buf->len = JOB_BUFFER_SIZE;
+  // Avoid `alloc_cb`, `alloc_cb` sequences on windows and also mark which
+  // stream we are reading from
+  job->lock =
+    (handle == (uv_handle_t *)&job->proc_stdout) ?
+    kBufferLockStdout :
+    kBufferLockStderr;
+}
+
+/// Pushes a event object to the event queue, which will be handled later by
+/// `job_handle`
+static void read_cb(uv_stream_t *stream, ssize_t cnt, const uv_buf_t *buf)
+{
+  Event event;
+  Job *job = (Job *)stream->data;
+  // pause reading on both streams
+  uv_read_stop((uv_stream_t *)&job->proc_stdout);
+  uv_read_stop((uv_stream_t *)&job->proc_stderr);
+
+  if (cnt <= 0) {
+    if (cnt != UV_ENOBUFS) {
+      // Assume it's EOF and exit the job. Doesn't harm sending a SIGTERM
+      // at this point
+      uv_process_kill(&job->proc, SIGTERM);
+    }
+    return;
+  }
+
+  job->length = cnt;
+  event.type = kEventJobActivity;
+  event.data.job = job;
+  event_push(event);
+}
+
+static void write_cb(uv_write_t *req, int status)
+{
+  free(req->data);
+  free(req);
+}
+
+/// Cleanup all the resources associated with the job
+static void exit_cb(uv_process_t *proc, int64_t status, int term_signal)
+{
+  Job *job = proc->data;
+
+  table[job->id - 1] = NULL;
+  shell_free_argv(job->proc_opts.args);
+  free_job(job);
+}
+

src/os/job.h

@@ -0,0 +1,72 @@
+// Job is a short name we use to refer to child processes that run in parallel
+// with the editor, probably executing long-running tasks and sending updates
+// asynchronously. Communication happens through anonymous pipes connected to
+// the job's std{in,out,err}. They are more like bash/zsh co-processes than the
+// usual shell background job. The name 'Job' was chosen because it applies to
+// the concept while being significantly shorter.
+#ifndef NEOVIM_OS_JOB_H
+#define NEOVIM_OS_JOB_H
+
+#include <stdint.h>
+#include <stdbool.h>
+
+#include "os/event.h"
+
+/// Function called when the job reads data
+///
+/// @param id The job is
+/// @param data Some data associated with the job by the caller
+/// @param buffer Buffer containing the data read. It must be copied
+///        immediately.
+/// @param len Amount of bytes that must be read from `buffer`
+/// @param from_stdout This is true if data was read from the job's stdout,
+///        false if it came from stderr.
+typedef void (*job_read_cb)(int id,
+                            void *data,
+                            char *buffer,
+                            uint32_t len,
+                            bool from_stdout);
+
+/// Initializes job control resources
+void job_init(void);
+
+/// Releases job control resources and terminates running jobs
+void job_teardown(void);
+
+/// Tries to start a new job.
+///
+/// @param argv Argument vector for the process. The first item is the
+///        executable to run.
+/// @param data Caller data that will be associated with the job
+/// @param cb Callback that will be invoked everytime data is available in
+///        the job's stdout/stderr
+/// @return The job id if the job started successfully. If the the first item /
+///         of `argv`(the program) could not be executed, -1 will be returned.
+//          0 will be returned if the job table is full.
+int job_start(char **argv, void *data, job_read_cb cb);
+
+/// 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 id The job id
+/// @return true if the stop request was successfully sent, false if the job
+///              id is invalid(probably because it has already stopped)
+bool job_stop(int id);
+
+/// Writes data to the job's stdin. This is a non-blocking operation, it
+/// returns when the write request was sent.
+///
+/// @param id The job id
+/// @param data Buffer containing the data to be written
+/// @param len Size of the data
+/// @return true if the write request was successfully sent, false if the job
+///              id is invalid(probably because it has already stopped)
+bool job_write(int id, char *data, uint32_t len);
+
+/// Runs the read callback associated with the job/event
+///
+/// @param event Object containing data necessary to invoke the callback
+void job_handle(Event event);
+
+#endif  // NEOVIM_OS_JOB_H
+

src/os/job_defs.h

@@ -0,0 +1,6 @@
+#ifndef NEOVIM_OS_JOB_DEFS_H
+#define NEOVIM_OS_JOB_DEFS_H
+
+typedef struct _Job Job;
+
+#endif  // NEOVIM_OS_JOB_DEFS_H

src/os/signal.c

@@ -11,6 +11,7 @@
 #include "memory.h"
 #include "misc1.h"
 #include "misc2.h"
+#include "os/event_defs.h"
 #include "os/event.h"
 #include "os/signal.h"
 

src/os/signal.h

@@ -1,7 +1,7 @@
 #ifndef NEOVIM_OS_SIGNAL_H
 #define NEOVIM_OS_SIGNAL_H
 
-#include "os/event.h"
+#include "os/event_defs.h"
 
 void signal_init(void);
 void signal_stop(void);

src/os_unix.c

@@ -56,6 +56,7 @@
 #include "os/input.h"
 #include "os/shell.h"
 #include "os/signal.h"
+#include "os/job.h"
 
 #include "os_unixx.h"       /* unix includes for os_unix.c only */
 
@@ -589,6 +590,7 @@ void mch_exit(int r)
 {
   exiting = TRUE;
 
+  job_teardown();
 
   {
     settmode(TMODE_COOK);

src/vim.h

@@ -792,6 +792,7 @@ enum auto_event {
   EVENT_INSERTCHANGE,           /* when changing Insert/Replace mode */
   EVENT_INSERTENTER,            /* when entering Insert mode */
   EVENT_INSERTLEAVE,            /* when leaving Insert mode */
+  EVENT_JOBACTIVITY,            /* when job sent some data */
   EVENT_MENUPOPUP,              /* just before popup menu is displayed */
   EVENT_QUICKFIXCMDPOST,        /* after :make, :grep etc. */
   EVENT_QUICKFIXCMDPRE,         /* before :make, :grep etc. */
@@ -1304,6 +1305,7 @@ enum {
     VV_HLSEARCH,
     VV_OLDFILES,
     VV_WINDOWID,
+    VV_JOB_DATA,
     VV_LEN, /* number of v: vars */
 };