Have async procs and timers remove from active.

MoarVM · Jan 18, 2017 · 4fad811 · 4fad811
1 parent 4241f5e
commit 4fad811
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 1 deletion.
diff --git a/src/io/procops.c b/src/io/procops.c
@@ -366,7 +366,8 @@ typedef struct {
     MVMuint32          seq_stdout;
     MVMuint32          seq_stderr;
     uv_stream_t       *stdin_handle;
-    ProcessState      state;
+    ProcessState       state;
+    int                using;
 } SpawnInfo;
 
 /* Info we convey about a write task. */
@@ -413,6 +414,7 @@ static void on_write(uv_write_t *req, int status) {
     MVM_repr_push_o(tc, t->body.queue, arr);
     if (wi->str_data)
         MVM_free(wi->buf.base);
+    MVM_io_eventloop_remove_active_work(tc, &(wi->work_idx));
     MVM_free(wi->req);
 }
 
@@ -722,6 +724,8 @@ static void async_spawn_on_exit(uv_process_t *req, MVMint64 exit_status, int ter
     /* Close handle. */
     uv_close((uv_handle_t *)req, spawn_async_close);
     ((MVMIOAsyncProcessData *)((MVMOSHandle *)si->handle)->body.data)->handle = NULL;
+    if (--si->using == 0)
+        MVM_io_eventloop_remove_active_work(tc, &(si->work_idx));
 }
 
 /* Allocates a buffer of the suggested size. */
@@ -788,6 +792,8 @@ static void async_read(uv_stream_t *handle, ssize_t nread, const uv_buf_t *buf,
         if (buf->base)
             MVM_free(buf->base);
         uv_close((uv_handle_t *) handle, NULL);
+        if (--si->using == 0)
+            MVM_io_eventloop_remove_active_work(tc, &(si->work_idx));
     }
     else {
         MVM_repr_push_o(tc, arr, tc->instance->boot_types.BOOTInt);
@@ -804,6 +810,8 @@ static void async_read(uv_stream_t *handle, ssize_t nread, const uv_buf_t *buf,
         if (buf->base)
             MVM_free(buf->base);
         uv_close((uv_handle_t *) handle, NULL);
+        if (--si->using == 0)
+            MVM_io_eventloop_remove_active_work(tc, &(si->work_idx));
     }
     MVM_repr_push_o(tc, t->body.queue, arr);
 }
@@ -848,6 +856,7 @@ static void spawn_setup(MVMThreadContext *tc, uv_loop_t *loop, MVMObject *async_
     SpawnInfo *si = (SpawnInfo *)data;
     si->tc        = tc;
     si->work_idx  = MVM_io_eventloop_add_active_work(tc, async_task);
+    si->using     = 1;
 
     /* Create input/output handles as needed. */
     if (MVM_repr_exists_key(tc, si->callbacks, tc->instance->str_consts.write)) {
@@ -871,6 +880,7 @@ static void spawn_setup(MVMThreadContext *tc, uv_loop_t *loop, MVMObject *async_
         si->ds_stdout                = MVM_string_decodestream_create(tc, MVM_encoding_type_utf8, 0, 1);
         stdout_pipe                  = pipe;
         stdout_cb                    = async_spawn_stdout_chars_read;
+        si->using++;
     }
     else if (MVM_repr_exists_key(tc, si->callbacks, tc->instance->str_consts.stdout_bytes)) {
         uv_pipe_t *pipe = MVM_malloc(sizeof(uv_pipe_t));
@@ -880,6 +890,7 @@ static void spawn_setup(MVMThreadContext *tc, uv_loop_t *loop, MVMObject *async_
         process_stdio[1].data.stream = (uv_stream_t *)pipe;
         stdout_pipe                  = pipe;
         stdout_cb                    = async_spawn_stdout_bytes_read;
+        si->using++;
     }
     else {
         process_stdio[1].flags   = UV_INHERIT_FD;
@@ -894,6 +905,7 @@ static void spawn_setup(MVMThreadContext *tc, uv_loop_t *loop, MVMObject *async_
         si->ds_stderr                = MVM_string_decodestream_create(tc, MVM_encoding_type_utf8, 0, 1);
         stderr_pipe                  = pipe;
         stderr_cb                    = async_spawn_stderr_chars_read;
+        si->using++;
     }
     else if (MVM_repr_exists_key(tc, si->callbacks, tc->instance->str_consts.stderr_bytes)) {
         uv_pipe_t *pipe = MVM_malloc(sizeof(uv_pipe_t));
@@ -903,6 +915,7 @@ static void spawn_setup(MVMThreadContext *tc, uv_loop_t *loop, MVMObject *async_
         process_stdio[2].data.stream = (uv_stream_t *)pipe;
         stderr_pipe                  = pipe;
         stderr_cb                    = async_spawn_stderr_bytes_read;
+        si->using++;
     }
     else {
         process_stdio[2].flags   = UV_INHERIT_FD;
@@ -941,6 +954,7 @@ static void spawn_setup(MVMThreadContext *tc, uv_loop_t *loop, MVMObject *async_
                 MVM_repr_push_o(tc, ((MVMAsyncTask *)async_task)->body.queue, arr);
             });
             });
+            MVM_io_eventloop_remove_active_work(tc, &(si->work_idx));
         }
     }
     else {

diff --git a/src/io/timers.c b/src/io/timers.c
@@ -33,6 +33,7 @@ static void cancel(MVMThreadContext *tc, uv_loop_t *loop, MVMObject *async_task,
     uv_timer_stop(&ti->handle);
     MVM_io_eventloop_send_cancellation_notification(ti->tc,
         MVM_io_eventloop_get_active_work(tc, ti->work_idx));
+    MVM_io_eventloop_remove_active_work(tc, &(ti->work_idx));
 }
 
 /* Frees data associated with a timer async task. */