Add a libuv-managed event loop thread.

This will be used in timers, async I/O, file system notifications, and
so forth. It will deliver completion notifications into a queue - one
that is presumably being used as a thread pool input stream.

Author: jnthn

build/Makefile.in

@@ -82,6 +82,7 @@ OBJECTS = src/core/callsite@obj@ \
           src/gc/worklist@obj@ \
           src/gc/roots@obj@ \
           src/io/io@obj@ \
+          src/io/eventloop@obj@ \
           src/io/syncfile@obj@ \
           src/io/syncstream@obj@ \
           src/io/syncpipe@obj@ \
@@ -132,6 +133,7 @@ OBJECTS = src/core/callsite@obj@ \
           src/6model/reprs/ConditionVariable@obj@ \
           src/6model/reprs/Semaphore@obj@ \
           src/6model/reprs/ConcBlockingQueue@obj@ \
+          src/6model/reprs/MVMAsyncTask@obj@ \
           src/6model/6model@obj@ \
           src/6model/bootstrap@obj@ \
           src/6model/sc@obj@ \
@@ -186,6 +188,7 @@ HEADERS = src/moar.h \
           src/core/continuation.h \
           src/core/intcache.h \
           src/io/io.h \
+          src/io/eventloop.h \
           src/io/syncfile.h \
           src/io/syncstream.h \
           src/io/syncpipe.h \
@@ -241,6 +244,7 @@ HEADERS = src/moar.h \
           src/6model/reprs/ConditionVariable.h \
           src/6model/reprs/Semaphore.h \
           src/6model/reprs/ConcBlockingQueue.h \
+          src/6model/reprs/MVMAsyncTask.h \
           src/6model/sc.h \
           src/mast/compiler.h \
           src/mast/driver.h \

src/6model/bootstrap.c

@@ -576,7 +576,7 @@ void MVM_6model_bootstrap(MVMThreadContext *tc) {
 
     /* Create stub BOOTInt, BOOTNum, BOOTStr, BOOTArray, BOOTHash, BOOTCCode,
      * BOOTCode, BOOTThread, BOOTIter, BOOTContext, SCRef, Lexotic,
-     * CallCapture, BOOTIO and BOOTException types. */
+     * CallCapture, BOOTIO, BOOTException, and BOOTQueue types. */
 #define create_stub_boot_type(tc, reprid, slot, makeboolspec, boolspec) do { \
     const MVMREPROps *repr = MVM_repr_get_by_id(tc, reprid); \
     MVMObject *type = tc->instance->slot = repr->type_object_for(tc, NULL); \
@@ -608,6 +608,7 @@ void MVM_6model_bootstrap(MVMThreadContext *tc) {
     create_stub_boot_type(tc, MVM_REPR_ID_MVMMultiCache, boot_types.BOOTMultiCache, 0, MVM_BOOL_MODE_NOT_TYPE_OBJECT);
     create_stub_boot_type(tc, MVM_REPR_ID_MVMContinuation, boot_types.BOOTContinuation, 0, MVM_BOOL_MODE_NOT_TYPE_OBJECT);
     create_stub_boot_type(tc, MVM_REPR_ID_MVMThread, Thread, 0, MVM_BOOL_MODE_NOT_TYPE_OBJECT);
+    create_stub_boot_type(tc, MVM_REPR_ID_ConcBlockingQueue, boot_types.BOOTQueue, 0, MVM_BOOL_MODE_NOT_TYPE_OBJECT);
 
     /* Bootstrap the KnowHOW type, giving it a meta-object. */
     bootstrap_KnowHOW(tc);
@@ -638,6 +639,7 @@ void MVM_6model_bootstrap(MVMThreadContext *tc) {
     meta_objectifier(tc, boot_types.BOOTMultiCache, "BOOTMultiCache");
     meta_objectifier(tc, boot_types.BOOTContinuation, "BOOTContinuation");
     meta_objectifier(tc, Thread, "Thread");
+    meta_objectifier(tc, boot_types.BOOTQueue, "BOOTQueue");
 
     /* Create the KnowHOWAttribute type. */
     create_KnowHOWAttribute(tc);

src/6model/reprs.c

@@ -215,6 +215,7 @@ void MVM_repr_initialize_registry(MVMThreadContext *tc) {
     register_core_repr(ConditionVariable);
     register_core_repr(Semaphore);
     register_core_repr(ConcBlockingQueue);
+    register_core_repr(AsyncTask);
 
     tc->instance->num_reprs = MVM_REPR_CORE_COUNT;
 }

src/6model/reprs.h

@@ -36,6 +36,7 @@
 #include "6model/reprs/ConditionVariable.h"
 #include "6model/reprs/Semaphore.h"
 #include "6model/reprs/ConcBlockingQueue.h"
+#include "6model/reprs/MVMAsyncTask.h"
 
 /* REPR related functions. */
 void MVM_repr_initialize_registry(MVMThreadContext *tc);
@@ -82,8 +83,9 @@ const MVMREPROps * MVM_repr_get_by_name(MVMThreadContext *tc, MVMString *name);
 #define MVM_REPR_ID_ConditionVariable       34
 #define MVM_REPR_ID_Semaphore               35
 #define MVM_REPR_ID_ConcBlockingQueue       36
+#define MVM_REPR_ID_MVMAsyncTask            37
 
-#define MVM_REPR_CORE_COUNT                 37
+#define MVM_REPR_CORE_COUNT                 38
 #define MVM_REPR_MAX_COUNT                  64
 
 /* Default attribute functions for a REPR that lacks them. */

src/6model/reprs/MVMAsyncTask.c

@@ -0,0 +1,91 @@
+#include "moar.h"
+
+/* This representation's function pointer table. */
+static const MVMREPROps this_repr;
+
+/* Creates a new type object of this representation, and associates it with
+ * the given HOW. */
+static MVMObject * type_object_for(MVMThreadContext *tc, MVMObject *HOW) {
+    MVMSTable *st  = MVM_gc_allocate_stable(tc, &this_repr, HOW);
+
+    MVMROOT(tc, st, {
+        MVMObject *obj = MVM_gc_allocate_type_object(tc, st);
+        MVM_ASSIGN_REF(tc, &(st->header), st->WHAT, obj);
+        st->size = sizeof(MVMAsyncTask);
+    });
+
+    return st->WHAT;
+}
+
+/* Copies the body of one object to another. */
+static void copy_to(MVMThreadContext *tc, MVMSTable *st, void *src, MVMObject *dest_root, void *dest) {
+    MVM_exception_throw_adhoc(tc, "Cannot copy object with repr MVMAsyncTask");
+}
+
+static void deserialize_stable_size(MVMThreadContext *tc, MVMSTable *st, MVMSerializationReader *reader) {
+    st->size = sizeof(MVMAsyncTask);
+}
+
+/* Called by the VM to mark any GCable items. */
+static void gc_mark(MVMThreadContext *tc, MVMSTable *st, void *data, MVMGCWorklist *worklist) {
+    MVMAsyncTaskBody *task = (MVMAsyncTaskBody *)data;
+    MVM_gc_worklist_add(tc, worklist, &task->queue);
+    MVM_gc_worklist_add(tc, worklist, &task->schedulee);
+    if (task->ops && task->ops->gc_mark)
+        task->ops->gc_mark(tc, task->data, worklist);
+}
+
+/* Called by the VM in order to free memory associated with this object. */
+static void gc_free(MVMThreadContext *tc, MVMObject *obj) {
+    MVMAsyncTask *task = (MVMAsyncTask *)obj;
+    if (task->body.ops && task->body.ops->gc_free)
+        task->body.ops->gc_free(tc, obj, task->body.data);
+}
+
+/* Gets the storage specification for this representation. */
+static MVMStorageSpec get_storage_spec(MVMThreadContext *tc, MVMSTable *st) {
+    MVMStorageSpec spec;
+    spec.inlineable      = MVM_STORAGE_SPEC_REFERENCE;
+    spec.boxed_primitive = MVM_STORAGE_SPEC_BP_NONE;
+    spec.can_box         = 0;
+    return spec;
+}
+
+/* Compose the representation. */
+static void compose(MVMThreadContext *tc, MVMSTable *st, MVMObject *info) {
+    /* Nothing to do for this REPR. */
+}
+
+/* Initializes the representation. */
+const MVMREPROps * MVMAsyncTask_initialize(MVMThreadContext *tc) {
+    return &this_repr;
+}
+
+static const MVMREPROps this_repr = {
+    type_object_for,
+    MVM_gc_allocate_object,
+    NULL, /* initialize */
+    copy_to,
+    MVM_REPR_DEFAULT_ATTR_FUNCS,
+    MVM_REPR_DEFAULT_BOX_FUNCS,
+    MVM_REPR_DEFAULT_POS_FUNCS,
+    MVM_REPR_DEFAULT_ASS_FUNCS,
+    MVM_REPR_DEFAULT_ELEMS,
+    get_storage_spec,
+    NULL, /* change_type */
+    NULL, /* serialize */
+    NULL, /* deserialize */
+    NULL, /* serialize_repr_data */
+    NULL, /* deserialize_repr_data */
+    deserialize_stable_size,
+    gc_mark,
+    gc_free,
+    NULL, /* gc_cleanup */
+    NULL, /* gc_mark_repr_data */
+    NULL, /* gc_free_repr_data */
+    compose,
+    NULL, /* spesh */
+    "AsyncTask", /* name */
+    MVM_REPR_ID_MVMAsyncTask,
+    0, /* refs_frames */
+};

src/6model/reprs/MVMAsyncTask.h

@@ -0,0 +1,21 @@
+/* Representation serving as a handle to an asynchronous task. */
+struct MVMAsyncTaskBody {
+    /* The queue to schedule a result handler on. */
+    MVMObject *queue;
+
+    /* The result handler to schedule. */
+    MVMObject *schedulee;
+
+    /* Async task operation table. */
+    const MVMAsyncTaskOps *ops;
+
+    /* Data stored by operation type. */
+    void *data;
+};
+struct MVMAsyncTask {
+    MVMObject common;
+    MVMAsyncTaskBody body;
+};
+
+/* Function for REPR setup. */
+const MVMREPROps * MVMAsyncTask_initialize(MVMThreadContext *tc);

src/core/instance.h

@@ -20,6 +20,7 @@ struct MVMBootTypes {
     MVMObject *BOOTCompUnit;
     MVMObject *BOOTMultiCache;
     MVMObject *BOOTContinuation;
+    MVMObject *BOOTQueue;
 };
 
 /* Various raw types that don't need a HOW */
@@ -78,9 +79,6 @@ typedef struct _MVMCallsiteProfileData {
 
 /* Represents a MoarVM instance. */
 struct MVMInstance {
-    /* libuv loop */
-    uv_loop_t *default_loop;
-
     /* The main thread. */
     MVMThreadContext *main_thread;
 
@@ -90,6 +88,15 @@ struct MVMInstance {
     /* The number of active user threads. */
     MVMuint16 num_user_threads;
 
+    /* The event loop thread, a mutex to avoid start-races, a concurrent
+     * queue of tasks that need to be processed by the event loop thread
+     * and an array of active tasks, for the purpose of keeping them GC
+     * marked. */
+    MVMThreadContext *event_loop_thread;
+    uv_mutex_t        mutex_event_loop_start;
+    MVMObject        *event_loop_todo_queue;
+    MVMObject        *event_loop_active;
+
     /* The KnowHOW meta-object; all other meta-objects (which are
      * built in user-space) are built out of this. */
     MVMObject *KnowHOW;

src/core/threadcontext.c

@@ -34,7 +34,8 @@ MVMThreadContext * MVM_tc_create(MVMInstance *instance) {
     tc->frame_pool_table_size = MVMInitialFramePoolTableSize;
     tc->frame_pool_table = calloc(MVMInitialFramePoolTableSize, sizeof(MVMFrame *));
 
-    tc->loop = instance->default_loop ? uv_loop_new() : uv_default_loop();
+    /* Use default loop for main thread; create a new one for others. */
+    tc->loop = instance->main_thread ? uv_loop_new() : uv_default_loop();
 
     /* Initialize random number generator state. */
     MVM_proc_seed(tc, (MVM_platform_now() / 10000) * MVM_proc_getpid(tc));

src/gc/roots.c

@@ -45,6 +45,8 @@ void MVM_gc_root_add_instance_roots_to_worklist(MVMThreadContext *tc, MVMGCWorkl
     MVM_gc_worklist_add(tc, worklist, &tc->instance->compiler_registry);
     MVM_gc_worklist_add(tc, worklist, &tc->instance->hll_syms);
     MVM_gc_worklist_add(tc, worklist, &tc->instance->clargs);
+    MVM_gc_worklist_add(tc, worklist, &tc->instance->event_loop_todo_queue);
+    MVM_gc_worklist_add(tc, worklist, &tc->instance->event_loop_active);
 
     /* okay, so this makes the weak hash slightly less weak.. for certain
      * keys of it anyway... */

src/io/eventloop.c

@@ -0,0 +1,74 @@
+#include "moar.h"
+
+/* Asynchronous I/O, timers, file system notifications and signal handlers
+ * have their callbacks processed by this event loop. Its job is mostly to
+ * fire off work, receive the callbacks, and put stuff into the concurrent
+ * work queue of some scheduler or other. It's backed by a thread that is
+ * started in the usual way, but never actually ends up in interpreter;
+ * instead, it enters a libuv event loop "forever", until program exit.
+ *
+ * Work is sent to the event loop by 
+ */
+
+/* Sets up an async task to be done on the loop. */
+MVMint64 setup_work(MVMThreadContext *tc) {
+    MVMObject *task_obj;
+    MVMint64 setup = 0;
+    while ((task_obj = MVM_concblockingqueue_poll(tc,
+            (MVMConcBlockingQueue *)tc->instance->event_loop_todo_queue)) != NULL) {
+        MVMAsyncTask *task = (MVMAsyncTask *)task_obj;
+        task->body.ops->setup(tc, tc->loop, task->body.data);
+        setup = 1;
+    }
+    return setup;
+}
+
+/* Sees if we have an event loop processing thread set up already, and
+ * sets it up if not. */
+void idle_handler(uv_idle_t *handle, int status) {
+    MVMThreadContext *tc = (MVMThreadContext *)handle->data;
+    GC_SYNC_POINT(tc);
+    setup_work(tc);
+}
+void enter_loop(MVMThreadContext *tc, MVMCallsite *callsite, MVMRegister *args) {
+    uv_idle_t idle;
+    if (uv_idle_init(tc->loop, &idle) != 0)
+        MVM_panic(1, "Unable to initialize idle worker for event loop");
+    idle.data = tc;
+    if (uv_idle_start(&idle, idle_handler) != 0)
+        MVM_panic(1, "Unable to start idle worker for event loop");
+    uv_run(tc->loop, UV_RUN_DEFAULT);
+    MVM_panic(1, "Supposedly unending event loop thread ended");
+}
+static uv_loop_t *get_or_vivify_loop(MVMThreadContext *tc) {
+    if (!tc->instance->event_loop_thread) {
+        /* Grab starting mutex and ensure we didn't lose the race. */
+        uv_mutex_lock(&tc->instance->mutex_event_loop_start);
+        if (!tc->instance->event_loop_thread) {
+            /* Start the event loop thread, which will call a C function that
+             * sits in the uv loop, never leaving. */
+            MVMObject *thread, *loop_runner;
+            loop_runner = MVM_repr_alloc_init(tc, tc->instance->boot_types.BOOTCCode);
+            ((MVMCFunction *)loop_runner)->body.func = enter_loop;
+            MVMROOT(tc, loop_runner, {
+                thread = MVM_thread_new(tc, loop_runner, 1);
+                MVM_thread_run(tc, thread);
+            });
+            tc->instance->event_loop_thread     = ((MVMThread *)thread)->body.tc;
+            tc->instance->event_loop_todo_queue = MVM_repr_alloc_init(tc,
+                tc->instance->boot_types.BOOTQueue);
+            tc->instance->event_loop_active     = MVM_repr_alloc_init(tc,
+                tc->instance->boot_types.BOOTArray);
+        }
+        uv_mutex_unlock(&tc->instance->mutex_event_loop_start);
+    }
+    return tc->instance->event_loop_thread->loop;
+}
+
+/* Adds a work item into the event loop work queue. */
+void MVM_io_eventloop_queue_work(MVMThreadContext *tc, MVMObject *work) {
+    MVMROOT(tc, work, {
+        uv_loop_t *loop = get_or_vivify_loop(tc);
+        MVM_repr_push_o(tc, tc->instance->event_loop_todo_queue, work);
+    });
+}