Merge PR #2037 'Job control fixes and pseudo terminal support'

12 files changed, 689 insertions, 227 deletions

diff --git a/src/nvim/CMakeLists.txt b/src/nvim/CMakeLists.txt
index 1c2dad6094..c59de2b5de 100644
--- a/src/nvim/CMakeLists.txt
+++ b/src/nvim/CMakeLists.txt
@@ -175,6 +175,7 @@ list(APPEND NVIM_LINK_LIBRARIES
     ${LIBTERMKEY_LIBRARIES}
     ${LIBUNIBILIUM_LIBRARIES}
     m
+    util
     ${CMAKE_THREAD_LIBS_INIT}
     )
 
diff --git a/src/nvim/eval.c b/src/nvim/eval.c
index a47473930a..b826ddcc50 100644
--- a/src/nvim/eval.c
+++ b/src/nvim/eval.c
@@ -454,8 +454,8 @@ static dictitem_T vimvars_var;                  /* variable used for v: */
 // Memory pool for reusing JobEvent structures
 typedef struct {
   int id;
-  char *name, *type, *received;
-  size_t received_len;
+  char *name, *type;
+  list_T *received;
 } JobEvent;
 #define JobEventFreer(x)
 KMEMPOOL_INIT(JobEventPool, JobEvent, JobEventFreer)
@@ -6488,8 +6488,9 @@ static struct fst {
   {"isdirectory",     1, 1, f_isdirectory},
   {"islocked",        1, 1, f_islocked},
   {"items",           1, 1, f_items},
+  {"jobresize",       3, 3, f_jobresize},
   {"jobsend",         2, 2, f_jobsend},
-  {"jobstart",        2, 3, f_jobstart},
+  {"jobstart",        2, 4, f_jobstart},
   {"jobstop",         1, 1, f_jobstop},
   {"join",            1, 2, f_join},
   {"keys",            1, 1, f_keys},
@@ -10665,6 +10666,39 @@ static void f_jobsend(typval_T *argvars, typval_T *rettv)
   rettv->vval.v_number = job_write(job, buf);
 }
 
+// "jobresize()" function
+static void f_jobresize(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 || argvars[1].v_type != VAR_NUMBER
+      || argvars[2].v_type != VAR_NUMBER) {
+    // job id, width, height
+    EMSG(_(e_invarg));
+    return;
+  }
+
+  Job *job = job_find(argvars[0].vval.v_number);
+
+  if (!job) {
+    // Probably an invalid job id
+    EMSG(_(e_invjob));
+    return;
+  }
+
+  if (!job_resize(job, argvars[1].vval.v_number, argvars[2].vval.v_number)) {
+    EMSG(_(e_jobnotpty));
+    return;
+  }
+
+  rettv->vval.v_number = 1;
+}
+
 // "jobstart()" function
 static void f_jobstart(typval_T *argvars, typval_T *rettv)
 {
@@ -10682,8 +10716,7 @@ static void f_jobstart(typval_T *argvars, typval_T *rettv)
 
   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)) {
+      || (argvars[2].v_type != VAR_LIST && argvars[2].v_type != VAR_UNKNOWN)) {
     // Wrong argument types
     EMSG(_(e_invarg));
     return;
@@ -10725,15 +10758,31 @@ static void f_jobstart(typval_T *argvars, typval_T *rettv)
 
   // The last item of argv must be NULL
   argv[i] = NULL;
+  JobOptions opts = JOB_OPTIONS_INIT;
+  opts.argv = argv;
+  opts.data = xstrdup((char *)argvars[0].vval.v_string);
+  opts.stdout_cb = on_job_stdout;
+  opts.stderr_cb = on_job_stderr;
+  opts.exit_cb = on_job_exit;
 
-  job_start(argv,
-            xstrdup((char *)argvars[0].vval.v_string),
-            true,
-            on_job_stdout,
-            on_job_stderr,
-            on_job_exit,
-            0,
-            &rettv->vval.v_number);
+  if (args && argvars[3].v_type == VAR_DICT) {
+    dict_T *job_opts = argvars[3].vval.v_dict;
+    opts.pty = true;
+    uint16_t width = get_dict_number(job_opts, (uint8_t *)"width");
+    if (width > 0) {
+      opts.width = width;
+    }
+    uint16_t height = get_dict_number(job_opts, (uint8_t *)"height");
+    if (height > 0) {
+      opts.height = height;
+    }
+    char *term = (char *)get_dict_string(job_opts, (uint8_t *)"TERM", true);
+    if (term) {
+      opts.term_name = term;
+    }
+  }
+
+  job_start(opts, &rettv->vval.v_number);
 
   if (rettv->vval.v_number <= 0) {
     if (rettv->vval.v_number == 0) {
@@ -19705,72 +19754,73 @@ char_u *do_string_sub(char_u *str, char_u *pat, char_u *sub, char_u *flags)
 
 // JobActivity autocommands will execute vimscript code, so it must be executed
 // on Nvim main loop
-#define push_job_event(j, r, t, eof)                                 \
-  do {                                                               \
-    JobEvent *event_data = kmp_alloc(JobEventPool, job_event_pool);  \
-    event_data->received = NULL;                                     \
-    size_t read_count = 0;                                           \
-    if (r) {                                                         \
-      if (eof) {                                                     \
-        read_count = rstream_pending(r);                             \
-      } else {                                                       \
-        char *read = rstream_read_ptr(r);                            \
-        char *lastnl = xmemrchr(read, NL, rstream_pending(r));       \
-        if (lastnl) {                                                \
-          read_count = (size_t) (lastnl - read) + 1;                 \
-        } else if (rstream_available(r) == 0) {                      \
-          /* No newline or room to grow; flush everything. */        \
-          read_count = rstream_pending(r);                           \
-        }                                                            \
-      }                                                              \
-      if (read_count == 0) {                                         \
-        /* Either we're at EOF or we need to wait until next time */ \
-        /* to receive a '\n. */                                      \
-        kmp_free(JobEventPool, job_event_pool, event_data);          \
-        return;                                                      \
-      }                                                              \
-      event_data->received_len = read_count;                         \
-      event_data->received = xmallocz(read_count);                   \
-      rstream_read(r, event_data->received, read_count);             \
-    }                                                                \
-    event_data->id = job_id(j);                                      \
-    event_data->name = job_data(j);                                  \
-    event_data->type = t;                                            \
-    event_push((Event) {                                             \
-      .handler = on_job_event,                                       \
-      .data = event_data                                             \
-    }, true);                                                        \
-  } while(0)
+static inline void push_job_event(Job *job, RStream *rstream, char *type)
+{
+  JobEvent *event_data = kmp_alloc(JobEventPool, job_event_pool);
+  event_data->received = NULL;
+  if (rstream) {
+    event_data->received = list_alloc();
+    char *ptr = rstream_read_ptr(rstream);
+    size_t count = rstream_pending(rstream);
+    size_t remaining = count;
+    size_t off = 0;
+
+    while (off < remaining) {
+      // append the line
+      if (ptr[off] == NL) {
+        list_append_string(event_data->received, (uint8_t *)ptr, off);
+        size_t skip = off + 1;
+        ptr += skip;
+        remaining -= skip;
+        off = 0;
+        continue;
+      }
+      if (ptr[off] == NUL) {
+        // Translate NUL to NL
+        ptr[off] = NL;
+      }
+      off++;
+    }
+    list_append_string(event_data->received, (uint8_t *)ptr, off);
+    rbuffer_consumed(rstream_buffer(rstream), count);
+  }
+  event_data->id = job_id(job);
+  event_data->name = job_data(job);
+  event_data->type = type;
+  event_push((Event) {
+    .handler = on_job_event,
+    .data = event_data
+  }, true);
+}
 
 static void on_job_stdout(RStream *rstream, void *data, bool eof)
 {
-  if (rstream_pending(rstream)) {
-    push_job_event(data, rstream, "stdout", eof);
+  if (!eof) {
+    push_job_event(data, rstream, "stdout");
   }
 }
 
 static void on_job_stderr(RStream *rstream, void *data, bool eof)
 {
-  if (rstream_pending(rstream)) {
-    push_job_event(data, rstream, "stderr", eof);
+  if (!eof) {
+    push_job_event(data, rstream, "stderr");
   }
 }
 
 static void on_job_exit(Job *job, void *data)
 {
-  push_job_event(job, NULL, "exit", true);
+  push_job_event(job, NULL, "exit");
 }
 
 static void on_job_event(Event event)
 {
   JobEvent *data = event.data;
-  apply_job_autocmds(data->id, data->name, data->type,
-                     data->received, data->received_len);
+  apply_job_autocmds(data->id, data->name, data->type, data->received);
   kmp_free(JobEventPool, job_event_pool, data);
 }
 
 static void apply_job_autocmds(int id, char *name, char *type,
-                               char *received, size_t received_len)
+                               list_T *received)
 {
   // Create the list which will be set to v:job_data
   list_T *list = list_alloc();
@@ -19781,12 +19831,9 @@ static void apply_job_autocmds(int id, char *name, char *type,
     listitem_T *str_slot = listitem_alloc();
     str_slot->li_tv.v_type = VAR_LIST;
     str_slot->li_tv.v_lock = 0;
-    str_slot->li_tv.vval.v_list =
-      string_to_list((char_u *) received, received_len, false);
+    str_slot->li_tv.vval.v_list = received;
     str_slot->li_tv.vval.v_list->lv_refcount++;
     list_append(list, str_slot);
-
-    free(received);
   }
 
   // Update v:job_data for the autocommands
diff --git a/src/nvim/globals.h b/src/nvim/globals.h
index e6a8ee33c5..c0d5217fc2 100644
--- a/src/nvim/globals.h
+++ b/src/nvim/globals.h
@@ -1127,6 +1127,7 @@ 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"));
+EXTERN char_u e_jobnotpty[] INIT(= N_("E904: Job is not connected to a pty"));
 EXTERN char_u e_libcall[] INIT(= N_("E364: Library call failed for \"%s()\""));
 EXTERN char_u e_markinval[] INIT(= N_("E19: Mark has invalid line number"));
 EXTERN char_u e_marknotset[] INIT(= N_("E20: Mark not set"));
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 3df3200d3d..00b8cd072f 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -132,14 +132,13 @@ uint64_t channel_from_job(char **argv)
   incref(channel);  // job channels are only closed by the exit_cb
 
   int status;
-  channel->data.job = job_start(argv,
-                                channel,
-                                true,
-                                job_out,
-                                job_err,
-                                job_exit,
-                                0,
-                                &status);
+  JobOptions opts = JOB_OPTIONS_INIT;
+  opts.argv = argv;
+  opts.data = channel;
+  opts.stdout_cb = job_out;
+  opts.stderr_cb = job_err;
+  opts.exit_cb = job_exit;
+  channel->data.job = job_start(opts, &status);
 
   if (status <= 0) {
     if (status == 0) {  // Two decrefs needed if status == 0.
diff --git a/src/nvim/os/job.c b/src/nvim/os/job.c
index 8db5e3cc39..9855a19269 100644
--- a/src/nvim/os/job.c
+++ b/src/nvim/os/job.c
@@ -6,13 +6,14 @@
 #include "nvim/os/uv_helpers.h"
 #include "nvim/os/job.h"
 #include "nvim/os/job_defs.h"
+#include "nvim/os/job_private.h"
+#include "nvim/os/pty_process.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"
@@ -29,8 +30,8 @@
     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);         \
+      if (!uv_is_closing((uv_handle_t *)job->proc_std##stream)) {          \
+        uv_close((uv_handle_t *)job->proc_std##stream, close_cb);          \
       }                                                                    \
     }                                                                      \
   } while (0)
@@ -39,37 +40,9 @@
 #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};
+Job *table[MAX_RUNNING_JOBS] = {NULL};
 size_t stop_requests = 0;
-static uv_timer_t job_stop_timer;
+uv_timer_t job_stop_timer;
 
 // Some helpers shared in this module
 
@@ -92,6 +65,8 @@ void job_teardown(void)
   for (int i = 0; i < MAX_RUNNING_JOBS; i++) {
     Job *job;
     if ((job = table[i]) != NULL) {
+      uv_kill(job->pid, SIGTERM);
+      job->term_sent = true;
       job_stop(job);
     }
   }
@@ -104,29 +79,10 @@ 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.
-///             [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)
+Job *job_start(JobOptions opts, int *status)
 {
   int i;
   Job *job;
@@ -140,7 +96,7 @@ Job *job_start(char **argv,
 
   if (i == MAX_RUNNING_JOBS) {
     // No free slots
-    shell_free_argv(argv);
+    shell_free_argv(opts.argv);
     *status = 0;
     return NULL;
   }
@@ -151,92 +107,64 @@ Job *job_start(char **argv,
   *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;
+  job->opts = opts;
+  job->closed = false;
 
-  // 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;
+  process_init(job);
 
-  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);
+  if (opts.writable) {
+    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);
+  if (opts.stdout_cb) {
+    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);
+  if (opts.stderr_cb) {
+    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) {
+  if (!process_spawn(job)) {
+    if (opts.writable) {
       uv_close((uv_handle_t *)&job->proc_stdin, close_cb);
     }
-    if (stdout_cb) {
+    if (opts.stdout_cb) {
       uv_close((uv_handle_t *)&job->proc_stdout, close_cb);
     }
-    if (stderr_cb) {
+    if (opts.stderr_cb) {
       uv_close((uv_handle_t *)&job->proc_stderr, close_cb);
     }
-    uv_close((uv_handle_t *)&job->proc, close_cb);
+    process_close(job);
     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);
+  if (opts.writable) {
+    job->in = wstream_new(opts.maxmem);
+    wstream_set_stream(job->in, job->proc_stdin);
   }
 
   // Start the readable streams
-  if (stdout_cb) {
+  if (opts.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_set_stream(job->out, job->proc_stdout);
     rstream_start(job->out);
   }
 
-  if (stderr_cb) {
+  if (opts.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_set_stream(job->err, job->proc_stderr);
     rstream_start(job->err);
   }
   // Save the job to the table
@@ -325,7 +253,8 @@ int job_wait(Job *job, int ms) FUNC_ATTR_NONNULL_ALL
     // Job exited, collect status and manually invoke close_cb to free the job
     // resources
     status = job->status;
-    close_cb((uv_handle_t *)&job->proc);
+    job_close_streams(job);
+    job_decref(job);
   } else {
     job->refcount--;
   }
@@ -389,25 +318,17 @@ int job_id(Job *job)
 /// @return The job data
 void *job_data(Job *job)
 {
-  return job->data;
+  return job->opts.data;
 }
 
-static void job_exit_callback(Job *job)
+/// Resize the window for a pty job
+bool job_resize(Job *job, uint16_t width, uint16_t height)
 {
-  // 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);
+  if (!job->opts.pty) {
+    return false;
   }
+  pty_process_resize(job, width, height);
+  return true;
 }
 
 /// Iterates the table, sending SIGTERM to stopped jobs and SIGKILL to those
@@ -426,11 +347,12 @@ static void job_stop_timer_cb(uv_timer_t *handle)
 
     if (!job->term_sent && elapsed >= TERM_TIMEOUT) {
       ILOG("Sending SIGTERM to job(id: %d)", job->id);
-      uv_process_kill(&job->proc, SIGTERM);
+      uv_kill(job->pid, 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);
+      uv_kill(job->pid, SIGKILL);
+      process_close(job);
     }
   }
 }
@@ -441,48 +363,26 @@ static void read_cb(RStream *rstream, void *data, bool eof)
   Job *job = data;
 
   if (rstream == job->out) {
-    job->stdout_cb(rstream, data, eof);
+    job->opts.stdout_cb(rstream, data, eof);
     if (eof) {
       close_job_out(job);
     }
   } else {
-    job->stderr_cb(rstream, data, eof);
+    job->opts.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)
+void job_close_streams(Job *job)
 {
-  Job *job = handle_get_job((uv_handle_t *)proc);
-
-  job->status = (int)status;
-  uv_close((uv_handle_t *)&job->proc, close_cb);
+  close_job_in(job);
+  close_job_out(job);
+  close_job_err(job);
 }
 
 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);
-  }
+  job_decref(handle_get_job(handle));
 }
diff --git a/src/nvim/os/job_defs.h b/src/nvim/os/job_defs.h
index a9caa169a8..ac9a37b366 100644
--- a/src/nvim/os/job_defs.h
+++ b/src/nvim/os/job_defs.h
@@ -1,7 +1,9 @@
 #ifndef NVIM_OS_JOB_DEFS_H
 #define NVIM_OS_JOB_DEFS_H
 
+#include <uv.h>
 #include "nvim/os/rstream_defs.h"
+#include "nvim/os/wstream_defs.h"
 
 typedef struct job Job;
 
@@ -11,4 +13,50 @@ typedef struct job Job;
 /// @param data Some data associated with the job by the caller
 typedef void (*job_exit_cb)(Job *job, void *data);
 
+// Job startup options
+// job_exit_cb Callback that will be invoked when the job exits
+// maxmem Maximum amount of memory used by the job WStream
+typedef struct {
+  // Argument vector for the process. The first item is the
+  // executable to run.
+  // [consumed]
+  char **argv;
+  // Caller data that will be associated with the job
+  void *data;
+  // If true the job stdin will be available for writing with job_write,
+  // otherwise it will be redirected to /dev/null
+  bool writable;
+  // Callback that will be invoked when data is available on stdout. If NULL
+  // stdout will be redirected to /dev/null.
+  rstream_cb stdout_cb;
+  // Callback that will be invoked when data is available on stderr. If NULL
+  // stderr will be redirected to /dev/null.
+  rstream_cb  stderr_cb;
+  // Callback that will be invoked when the job has exited and will not send
+  // data
+  job_exit_cb exit_cb;
+  // Maximum memory used by the job's WStream
+  size_t maxmem;
+  // Connect the job to a pseudo terminal
+  bool pty;
+  // Initial window dimensions if the job is connected to a pseudo terminal
+  uint16_t width, height;
+  // Value for the $TERM environment variable. A default value of "ansi" is
+  // assumed if NULL
+  char *term_name;
+} JobOptions;
+
+#define JOB_OPTIONS_INIT ((JobOptions) {                     \
+    .argv = NULL,                                            \
+    .data = NULL,                                            \
+    .writable = true,                                        \
+    .stdout_cb = NULL,                                       \
+    .stderr_cb = NULL,                                       \
+    .exit_cb = NULL,                                         \
+    .maxmem = 0,                                             \
+    .pty = false,                                            \
+    .width = 80,                                             \
+    .height = 24,                                            \
+    .term_name = NULL                                        \
+    })
 #endif  // NVIM_OS_JOB_DEFS_H
diff --git a/src/nvim/os/job_private.h b/src/nvim/os/job_private.h
new file mode 100644
index 0000000000..b1d5e13feb
--- /dev/null
+++ b/src/nvim/os/job_private.h
@@ -0,0 +1,117 @@
+#ifndef NVIM_OS_JOB_PRIVATE_H
+#define NVIM_OS_JOB_PRIVATE_H
+
+#include <stdlib.h>
+
+#include <uv.h>
+
+#include "nvim/os/rstream_defs.h"
+#include "nvim/os/wstream_defs.h"
+#include "nvim/os/pipe_process.h"
+#include "nvim/os/pty_process.h"
+#include "nvim/os/shell.h"
+#include "nvim/log.h"
+
+struct job {
+  // Job id the index in the job table plus one.
+  int id;
+  // Process id
+  int pid;
+  // 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;
+  // Readable streams(std{out,err})
+  RStream *out, *err;
+  // Writable stream(stdin)
+  WStream *in;
+  // Libuv streams representing stdin/stdout/stderr
+  uv_stream_t *proc_stdin, *proc_stdout, *proc_stderr;
+  // Extra data set by the process spawner
+  void *process;
+  // If process_close has been called on this job
+  bool closed;
+  // Startup options
+  JobOptions opts;
+};
+
+extern Job *table[];
+extern size_t stop_requests;
+extern uv_timer_t job_stop_timer;
+
+static inline bool process_spawn(Job *job)
+{
+  return job->opts.pty ? pty_process_spawn(job) : pipe_process_spawn(job);
+}
+
+static inline void process_init(Job *job)
+{
+  if (job->opts.pty) {
+    pty_process_init(job);
+  } else {
+    pipe_process_init(job);
+  }
+}
+
+static inline void process_close(Job *job)
+{
+  if (job->closed) {
+    return;
+  }
+  job->closed = true;
+  if (job->opts.pty) {
+    pty_process_close(job);
+  } else {
+    pipe_process_close(job);
+  }
+}
+
+static inline void process_destroy(Job *job)
+{
+  if (job->opts.pty) {
+    pty_process_destroy(job);
+  } else {
+    pipe_process_destroy(job);
+  }
+}
+
+static inline 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->opts.exit_cb) {
+    // Invoke the exit callback
+    job->opts.exit_cb(job, job->opts.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);
+  }
+}
+
+static inline void job_decref(Job *job)
+{
+  if (--job->refcount == 0) {
+    // Invoke the exit_cb
+    job_exit_callback(job);
+    // Free all memory allocated for the job
+    free(job->proc_stdin->data);
+    free(job->proc_stdout->data);
+    free(job->proc_stderr->data);
+    shell_free_argv(job->opts.argv);
+    process_destroy(job);
+    free(job);
+  }
+}
+
+
+#endif  // NVIM_OS_JOB_PRIVATE_H
diff --git a/src/nvim/os/pipe_process.c b/src/nvim/os/pipe_process.c
new file mode 100644
index 0000000000..5535c3fe93
--- /dev/null
+++ b/src/nvim/os/pipe_process.c
@@ -0,0 +1,110 @@
+#include <stdbool.h>
+#include <stdlib.h>
+
+#include <uv.h>
+
+#include "nvim/os/uv_helpers.h"
+#include "nvim/os/job.h"
+#include "nvim/os/job_defs.h"
+#include "nvim/os/job_private.h"
+#include "nvim/os/pipe_process.h"
+#include "nvim/memory.h"
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "os/pipe_process.c.generated.h"
+#endif
+
+typedef struct {
+  // 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;
+} UvProcess;
+
+void pipe_process_init(Job *job)
+{
+  UvProcess *pipeproc = xmalloc(sizeof(UvProcess));
+  pipeproc->proc_opts.file = job->opts.argv[0];
+  pipeproc->proc_opts.args = job->opts.argv;
+  pipeproc->proc_opts.stdio = pipeproc->stdio;
+  pipeproc->proc_opts.stdio_count = 3;
+  pipeproc->proc_opts.flags = UV_PROCESS_WINDOWS_HIDE;
+  pipeproc->proc_opts.exit_cb = exit_cb;
+  pipeproc->proc_opts.cwd = NULL;
+  pipeproc->proc_opts.env = NULL;
+  pipeproc->proc.data = NULL;
+  pipeproc->proc_stdin.data = NULL;
+  pipeproc->proc_stdout.data = NULL;
+  pipeproc->proc_stderr.data = NULL;
+
+  // Initialize the job std{in,out,err}
+  pipeproc->stdio[0].flags = UV_IGNORE;
+  pipeproc->stdio[1].flags = UV_IGNORE;
+  pipeproc->stdio[2].flags = UV_IGNORE;
+
+  handle_set_job((uv_handle_t *)&pipeproc->proc, job);
+
+  if (job->opts.writable) {
+    uv_pipe_init(uv_default_loop(), &pipeproc->proc_stdin, 0);
+    pipeproc->stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE;
+    pipeproc->stdio[0].data.stream = (uv_stream_t *)&pipeproc->proc_stdin;
+  }
+
+  if (job->opts.stdout_cb) {
+    uv_pipe_init(uv_default_loop(), &pipeproc->proc_stdout, 0);
+    pipeproc->stdio[1].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE;
+    pipeproc->stdio[1].data.stream = (uv_stream_t *)&pipeproc->proc_stdout;
+  }
+
+  if (job->opts.stderr_cb) {
+    uv_pipe_init(uv_default_loop(), &pipeproc->proc_stderr, 0);
+    pipeproc->stdio[2].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE;
+    pipeproc->stdio[2].data.stream = (uv_stream_t *)&pipeproc->proc_stderr;
+  }
+
+  job->proc_stdin = (uv_stream_t *)&pipeproc->proc_stdin;
+  job->proc_stdout = (uv_stream_t *)&pipeproc->proc_stdout;
+  job->proc_stderr = (uv_stream_t *)&pipeproc->proc_stderr;
+  job->process = pipeproc;
+}
+
+void pipe_process_destroy(Job *job)
+{
+  UvProcess *pipeproc = job->process;
+  free(pipeproc->proc.data);
+  free(pipeproc);
+  job->process = NULL;
+}
+
+bool pipe_process_spawn(Job *job)
+{
+  UvProcess *pipeproc = job->process;
+
+  if (uv_spawn(uv_default_loop(), &pipeproc->proc, &pipeproc->proc_opts) != 0) {
+    return false;
+  }
+
+  job->pid = pipeproc->proc.pid;
+  return true;
+}
+
+void pipe_process_close(Job *job)
+{
+  UvProcess *pipeproc = job->process;
+  uv_close((uv_handle_t *)&pipeproc->proc, close_cb);
+}
+
+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;
+  pipe_process_close(job);
+}
+
+static void close_cb(uv_handle_t *handle)
+{
+  Job *job = handle_get_job(handle);
+  job_close_streams(job);
+  job_decref(job);
+}
diff --git a/src/nvim/os/pipe_process.h b/src/nvim/os/pipe_process.h
new file mode 100644
index 0000000000..17a4255ddc
--- /dev/null
+++ b/src/nvim/os/pipe_process.h
@@ -0,0 +1,7 @@
+#ifndef NVIM_OS_PIPE_PROCESS_H
+#define NVIM_OS_PIPE_PROCESS_H
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "os/pipe_process.h.generated.h"
+#endif
+#endif  // NVIM_OS_PIPE_PROCESS_H
diff --git a/src/nvim/os/pty_process.c b/src/nvim/os/pty_process.c
new file mode 100644
index 0000000000..bd7247c741
--- /dev/null
+++ b/src/nvim/os/pty_process.c
@@ -0,0 +1,225 @@
+// Some of the code came from pangoterm and libuv
+#include <stdbool.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <unistd.h>
+#include <termios.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <sys/ioctl.h>
+
+// forkpty is not in POSIX, so headers are platform-specific
+#if defined(__FreeBSD__)
+# include <libutil.h>
+#elif defined(__OpenBSD__) || defined(__NetBSD__) || defined(__APPLE__)
+# include <util.h>
+#else
+# include <pty.h>
+#endif
+
+#include <uv.h>
+
+#include "nvim/os/job.h"
+#include "nvim/os/job_defs.h"
+#include "nvim/os/job_private.h"
+#include "nvim/os/pty_process.h"
+#include "nvim/memory.h"
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "os/pty_process.c.generated.h"
+#endif
+
+typedef struct {
+  struct winsize winsize;
+  uv_pipe_t proc_stdin, proc_stdout, proc_stderr;
+  uv_signal_t schld;
+  int tty_fd;
+} PtyProcess;
+
+void pty_process_init(Job *job)
+{
+  PtyProcess *ptyproc = xmalloc(sizeof(PtyProcess));
+
+  if (job->opts.writable) {
+    uv_pipe_init(uv_default_loop(), &ptyproc->proc_stdin, 0);
+    ptyproc->proc_stdin.data = NULL;
+  }
+
+  if (job->opts.stdout_cb) {
+    uv_pipe_init(uv_default_loop(), &ptyproc->proc_stdout, 0);
+    ptyproc->proc_stdout.data = NULL;
+  }
+
+  if (job->opts.stderr_cb) {
+    uv_pipe_init(uv_default_loop(), &ptyproc->proc_stderr, 0);
+    ptyproc->proc_stderr.data = NULL;
+  }
+
+  job->proc_stdin = (uv_stream_t *)&ptyproc->proc_stdin;
+  job->proc_stdout = (uv_stream_t *)&ptyproc->proc_stdout;
+  job->proc_stderr = (uv_stream_t *)&ptyproc->proc_stderr;
+  job->process = ptyproc;
+}
+
+void pty_process_destroy(Job *job)
+{
+  free(job->opts.term_name);
+  free(job->process);
+  job->process = NULL;
+}
+
+bool pty_process_spawn(Job *job)
+{
+  int master;
+  PtyProcess *ptyproc = job->process;
+  ptyproc->winsize = (struct winsize){job->opts.height, job->opts.width, 0, 0};
+  struct termios termios;
+  init_termios(&termios);
+  uv_disable_stdio_inheritance();
+
+  int pid = forkpty(&master, NULL, &termios, &ptyproc->winsize);
+
+  if (pid < 0) {
+    return false;
+  } else if (pid == 0) {
+    init_child(job);
+    abort();
+  }
+
+  // make sure the master file descriptor is non blocking
+  fcntl(master, F_SETFL, fcntl(master, F_GETFL) | O_NONBLOCK);
+
+  if (job->opts.writable) {
+    uv_pipe_open(&ptyproc->proc_stdin, dup(master));
+  }
+
+  if (job->opts.stdout_cb) {
+    uv_pipe_open(&ptyproc->proc_stdout, dup(master));
+  }
+
+  if (job->opts.stderr_cb) {
+    uv_pipe_open(&ptyproc->proc_stderr, dup(master));
+  }
+
+  uv_signal_init(uv_default_loop(), &ptyproc->schld);
+  uv_signal_start(&ptyproc->schld, chld_handler, SIGCHLD);
+  ptyproc->schld.data = job;
+  ptyproc->tty_fd = master;
+  job->pid = pid;
+  return true;
+}
+
+void pty_process_close(Job *job)
+{
+  PtyProcess *ptyproc = job->process;
+  uv_signal_stop(&ptyproc->schld);
+  uv_close((uv_handle_t *)&ptyproc->schld, NULL);
+  job_close_streams(job);
+  job_decref(job);
+}
+
+void pty_process_resize(Job *job, uint16_t width, uint16_t height)
+{
+  PtyProcess *ptyproc = job->process;
+  ptyproc->winsize = (struct winsize){height, width, 0, 0};
+  ioctl(ptyproc->tty_fd, TIOCSWINSZ, &ptyproc->winsize);
+}
+
+static void init_child(Job *job)
+{
+  unsetenv("COLUMNS");
+  unsetenv("LINES");
+  unsetenv("TERMCAP");
+  unsetenv("COLORTERM");
+  unsetenv("COLORFGBG");
+
+  signal(SIGCHLD, SIG_DFL);
+  signal(SIGHUP, SIG_DFL);
+  signal(SIGINT, SIG_DFL);
+  signal(SIGQUIT, SIG_DFL);
+  signal(SIGTERM, SIG_DFL);
+  signal(SIGALRM, SIG_DFL);
+
+  setenv("TERM", job->opts.term_name ? job->opts.term_name : "ansi", 1);
+  execvp(job->opts.argv[0], job->opts.argv);
+  fprintf(stderr, "execvp failed: %s\n", strerror(errno));
+}
+
+static void chld_handler(uv_signal_t *handle, int signum)
+{
+  Job *job = handle->data;
+  int stat = 0;
+
+  if (waitpid(job->pid, &stat, 0) < 0) {
+    fprintf(stderr, "Waiting for pid %d failed: %s\n", job->pid,
+        strerror(errno));
+    return;
+  }
+
+  if (WIFSTOPPED(stat) || WIFCONTINUED(stat)) {
+    // Did not exit
+    return;
+  }
+
+  if (WIFEXITED(stat)) {
+    job->status = WEXITSTATUS(stat);
+  } else if (WIFSIGNALED(stat)) {
+    job->status = WTERMSIG(stat);
+  }
+
+  pty_process_close(job);
+}
+
+static void init_termios(struct termios *termios)
+{
+  memset(termios, 0, sizeof(struct termios));
+  // Taken from pangoterm
+  termios->c_iflag = ICRNL|IXON;
+  termios->c_oflag = OPOST|ONLCR|TAB0;
+  termios->c_cflag = CS8|CREAD;
+  termios->c_lflag = ISIG|ICANON|IEXTEN|ECHO|ECHOE|ECHOK;
+
+  cfsetspeed(termios, 38400);
+
+#ifdef IUTF8
+  termios->c_iflag |= IUTF8;
+#endif
+#ifdef NL0
+  termios->c_oflag |= NL0;
+#endif
+#ifdef CR0
+  termios->c_oflag |= CR0;
+#endif
+#ifdef BS0
+  termios->c_oflag |= BS0;
+#endif
+#ifdef VT0
+  termios->c_oflag |= VT0;
+#endif
+#ifdef FF0
+  termios->c_oflag |= FF0;
+#endif
+#ifdef ECHOCTL
+  termios->c_lflag |= ECHOCTL;
+#endif
+#ifdef ECHOKE
+  termios->c_lflag |= ECHOKE;
+#endif
+
+  termios->c_cc[VINTR]    = 0x1f & 'C';
+  termios->c_cc[VQUIT]    = 0x1f & '\\';
+  termios->c_cc[VERASE]   = 0x7f;
+  termios->c_cc[VKILL]    = 0x1f & 'U';
+  termios->c_cc[VEOF]     = 0x1f & 'D';
+  termios->c_cc[VEOL]     = _POSIX_VDISABLE;
+  termios->c_cc[VEOL2]    = _POSIX_VDISABLE;
+  termios->c_cc[VSTART]   = 0x1f & 'Q';
+  termios->c_cc[VSTOP]    = 0x1f & 'S';
+  termios->c_cc[VSUSP]    = 0x1f & 'Z';
+  termios->c_cc[VREPRINT] = 0x1f & 'R';
+  termios->c_cc[VWERASE]  = 0x1f & 'W';
+  termios->c_cc[VLNEXT]   = 0x1f & 'V';
+  termios->c_cc[VMIN]     = 1;
+  termios->c_cc[VTIME]    = 0;
+}
diff --git a/src/nvim/os/pty_process.h b/src/nvim/os/pty_process.h
new file mode 100644
index 0000000000..62fcd1671f
--- /dev/null
+++ b/src/nvim/os/pty_process.h
@@ -0,0 +1,7 @@
+#ifndef NVIM_OS_PTY_PROCESS_H
+#define NVIM_OS_PTY_PROCESS_H
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "os/pty_process.h.generated.h"
+#endif
+#endif  // NVIM_OS_PTY_PROCESS_H
diff --git a/src/nvim/os/shell.c b/src/nvim/os/shell.c
index 32c7ea564d..8cf7e7161d 100644
--- a/src/nvim/os/shell.c
+++ b/src/nvim/os/shell.c
@@ -201,14 +201,14 @@ static int shell(const char *cmd,
   char **argv = shell_build_argv(cmd, extra_args);
 
   int status;
-  Job *job = job_start(argv,
-                       &buf,
-                       input != NULL,
-                       data_cb,
-                       data_cb,
-                       NULL,
-                       0,
-                       &status);
+  JobOptions opts = JOB_OPTIONS_INIT;
+  opts.argv = argv;
+  opts.data = &buf;
+  opts.writable = input != NULL;
+  opts.stdout_cb = data_cb;
+  opts.stderr_cb = data_cb;
+  opts.exit_cb = NULL;
+  Job *job = job_start(opts, &status);
 
   if (status <= 0) {
     // Failed, probably due to `sh` not being executable