Correctly free libuv handles

This ensures memory chunks for libuv handles are only freed after the event loop
no longer has references to it.

2 files changed, 30 insertions, 13 deletions

diff --git a/src/os/job.c b/src/os/job.c
index f4cbbfb670..c44f4d1d69 100644
--- a/src/os/job.c
+++ b/src/os/job.c
@@ -28,6 +28,9 @@ struct job {
   // We use this reference count to ensure the JobExit event is only emitted
   // when stdout/stderr are drained
   int pending_refs;
+  // Same as above, but for freeing the job memory which contains
+  // libuv handles. Only after all are closed the job can be safely freed.
+  int pending_closes;
   // If the job was already stopped
   bool stopped;
   // Data associated with the job
@@ -57,6 +60,7 @@ static void job_prepare_cb(uv_prepare_t *handle);
 static void write_cb(uv_write_t *req, int status);
 static void read_cb(RStream *rstream, void *data, bool eof);
 static void exit_cb(uv_process_t *proc, int64_t status, int term_signal);
+static void close_cb(uv_handle_t *handle);
 static void emit_exit_event(Job *job);
 
 void job_init()
@@ -137,6 +141,7 @@ int job_start(char **argv,
   // Initialize
   job->id = i + 1;
   job->pending_refs = 3;
+  job->pending_closes = 4;
   job->data = data;
   job->stdout_cb = stdout_cb;
   job->stderr_cb = stderr_cb;
@@ -261,13 +266,13 @@ static Job * find_job(int id)
 
 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);
+  uv_close((uv_handle_t *)&job->proc_stdout, close_cb);
+  uv_close((uv_handle_t *)&job->proc_stdin, close_cb);
+  uv_close((uv_handle_t *)&job->proc_stderr, close_cb);
+  uv_close((uv_handle_t *)&job->proc, close_cb);
   rstream_free(job->out);
   rstream_free(job->err);
-  free(job);
+  wstream_free(job->in);
 }
 
 /// Iterates the table, sending SIGTERM to stopped jobs and SIGKILL to those
@@ -332,3 +337,13 @@ static void emit_exit_event(Job *job)
   event_push(event);
 }
 
+static void close_cb(uv_handle_t *handle)
+{
+  Job *job = handle->data;
+
+  if (--job->pending_closes == 0) {
+    // Only free the job memory after all the associated handles are properly
+    // closed by libuv
+    free(job);
+  }
+}
diff --git a/src/os/rstream.c b/src/os/rstream.c
index 63dfb2aa39..d1468cf764 100644
--- a/src/os/rstream.c
+++ b/src/os/rstream.c
@@ -28,6 +28,7 @@ struct rstream {
 static void alloc_cb(uv_handle_t *, size_t, uv_buf_t *);
 static void read_cb(uv_stream_t *, ssize_t, const uv_buf_t *);
 static void fread_idle_cb(uv_idle_t *);
+static void close_cb(uv_handle_t *handle);
 static void emit_read_event(RStream *rstream, bool eof);
 
 RStream * rstream_new(rstream_cb cb,
@@ -53,11 +54,9 @@ void rstream_free(RStream *rstream)
 {
   if (rstream->free_handle) {
     if (rstream->fread_idle != NULL) {
-      uv_close((uv_handle_t *)rstream->fread_idle, NULL);
-      free(rstream->fread_idle);
+      uv_close((uv_handle_t *)rstream->fread_idle, close_cb);
     } else {
-      uv_close((uv_handle_t *)rstream->stream, NULL);
-      free(rstream->stream);
+      uv_close((uv_handle_t *)rstream->stream, close_cb);
     }
   }
 
@@ -79,11 +78,9 @@ void rstream_set_file(RStream *rstream, uv_file file)
     // If this is the second time we're calling this function, free the
     // previously allocated memory
     if (rstream->fread_idle != NULL) {
-      uv_close((uv_handle_t *)rstream->fread_idle, NULL);
-      free(rstream->fread_idle);
+      uv_close((uv_handle_t *)rstream->fread_idle, close_cb);
     } else {
-      uv_close((uv_handle_t *)rstream->stream, NULL);
-      free(rstream->stream);
+      uv_close((uv_handle_t *)rstream->stream, close_cb);
     }
   }
 
@@ -261,6 +258,11 @@ static void fread_idle_cb(uv_idle_t *handle)
   emit_read_event(rstream, false);
 }
 
+static void close_cb(uv_handle_t *handle)
+{
+  free(handle);
+}
+
 static void emit_read_event(RStream *rstream, bool eof)
 {
   if (rstream->async) {