Task API for running background tasks in the libuv thread pool

crates/neon-runtime/src/lib.rs

@@ -16,3 +16,4 @@ pub mod mem;
 pub mod fun;
 pub mod convert;
 pub mod class;
+pub mod task;

crates/neon-runtime/src/neon.cc

@@ -6,6 +6,7 @@
 #include "neon.h"
 #include "neon_string.h"
 #include "neon_class_metadata.h"
+#include "neon_task.h"
 
 extern "C" void Neon_Call_SetReturn(v8::FunctionCallbackInfo<v8::Value> *info, v8::Local<v8::Value> value) {
   info->GetReturnValue().Set(value);
@@ -527,3 +528,9 @@ extern "C" void Neon_Error_ThrowSyntaxErrorFromCString(const char *msg) {
 extern "C" bool Neon_Mem_SameHandle(v8::Local<v8::Value> v1, v8::Local<v8::Value> v2) {
   return v1 == v2;
 }
+
+extern "C" void Neon_Task_Schedule(void *task, Neon_TaskPerformCallback perform, Neon_TaskCompleteCallback complete, v8::Local<v8::Function> callback) {
+  v8::Isolate *isolate = v8::Isolate::GetCurrent();
+  neon::Task *internal_task = new neon::Task(isolate, task, perform, complete, callback);
+  neon::queue_task(internal_task);
+}

crates/neon-runtime/src/neon.h

@@ -151,6 +151,11 @@ extern "C" {
   void Neon_Error_ThrowSyntaxErrorFromCString(const char *msg);
 
   bool Neon_Mem_SameHandle(v8::Local<v8::Value> v1, v8::Local<v8::Value> v2);
+
+  typedef void* (*Neon_TaskPerformCallback)(void *);
+  typedef void (*Neon_TaskCompleteCallback)(void *, void *, v8::Local<v8::Value> *out);
+
+  void Neon_Task_Schedule(void *task, Neon_TaskPerformCallback perform, Neon_TaskCompleteCallback complete, v8::Local<v8::Function> callback);
 }
 
 #endif

crates/neon-runtime/src/neon_task.h

@@ -0,0 +1,101 @@
+#ifndef NEON_TASK_H_
+#define NEON_TASK_H_
+
+#include <uv.h>
+#include "neon.h"
+#include "v8.h"
+
+namespace neon {
+
+class Task {
+public:
+  Task(v8::Isolate *isolate,
+       void *rust_task,
+       Neon_TaskPerformCallback perform,
+       Neon_TaskCompleteCallback complete,
+       v8::Local<v8::Function> callback)
+    : isolate_(isolate),
+      rust_task_(rust_task),
+      perform_(perform),
+      complete_(complete)
+  {
+    request_.data = this;
+    result_ = nullptr;
+    // Save the callback to be invoked when the task completes.
+    callback_.Reset(isolate, callback);
+    // Save the context (aka realm) to be used when invoking the callback.
+    context_.Reset(isolate, isolate->GetCurrentContext());
+  }
+
+  void execute() {
+    result_ = perform_(rust_task_);
+  }
+
+  void complete() {
+    // Ensure that we have all the proper scopes installed on the C++ stack before
+    // invoking the callback, and use the context (i.e. realm) we saved with the task.
+    v8::Isolate::Scope isolate_scope(isolate_);
+    v8::HandleScope handle_scope(isolate_);
+    v8::Local<v8::Context> context = v8::Local<v8::Context>::New(isolate_, context_);
+    v8::Context::Scope context_scope(context);
+
+    v8::Local<v8::Value> argv[2];
+
+    argv[0] = v8::Null(isolate_);
+    argv[1] = v8::Undefined(isolate_);
+
+    {
+      v8::TryCatch trycatch(isolate_);
+
+      v8::Local<v8::Value> completion;
+
+      complete_(rust_task_, result_, &completion);
+
+      if (trycatch.HasCaught()) {
+        argv[0] = trycatch.Exception();
+      } else {
+        argv[1] = completion;
+      }
+    }
+
+    v8::Local<v8::Function> callback = v8::Local<v8::Function>::New(isolate_, callback_);
+    callback->Call(context, v8::Null(isolate_), 2, argv);
+  }
+
+  void *get_result() {
+    return result_;
+  }
+
+  uv_work_t request_;
+
+private:
+  v8::Isolate *isolate_;
+  void *rust_task_;
+  Neon_TaskPerformCallback perform_;
+  Neon_TaskCompleteCallback complete_;
+  void *result_;
+  v8::Persistent<v8::Function> callback_;
+  v8::Persistent<v8::Context> context_;
+};
+
+void execute_task(uv_work_t *request) {
+  Task *task = static_cast<Task*>(request->data);
+  task->execute();
+}
+
+void complete_task(uv_work_t *request) {
+  Task *task = static_cast<Task*>(request->data);
+  task->complete();
+  delete task;
+}
+
+void queue_task(Task *task) {
+  uv_queue_work(uv_default_loop(),
+                &task->request_,
+                execute_task,
+                (uv_after_work_cb)complete_task);
+}
+
+}
+
+#endif

crates/neon-runtime/src/task.rs

@@ -0,0 +1,15 @@
+//! Facilities for running background tasks in the libuv thread pool.
+
+use raw::Local;
+use std::os::raw::c_void;
+
+extern "C" {
+
+    /// Schedules a background task.
+    #[link_name = "Neon_Task_Schedule"]
+    pub fn schedule(task: *mut c_void,
+                    perform: unsafe extern fn(*mut c_void) -> *mut c_void,
+                    complete: unsafe extern fn(*mut c_void, *mut c_void, &mut Local),
+                    callback: Local);
+
+}

src/internal/vm.rs

@@ -41,6 +41,7 @@ pub trait IsolateInternal {
     fn to_raw(self) -> *mut raw::Isolate;
     fn from_raw(ptr: *mut raw::Isolate) -> Self;
     fn class_map(&mut self) -> &mut ClassMap;
+    fn current() -> Isolate;
 }
 
 pub struct ClassMap {
@@ -86,6 +87,12 @@ impl IsolateInternal for Isolate {
         }
         unsafe { mem::transmute(ptr) }
     }
+
+    fn current() -> Isolate {
+        unsafe {
+            mem::transmute(neon_runtime::call::current_isolate())
+        }
+    }
 }
 
 extern "C" fn drop_class_map(map: Box<ClassMap>) {

src/lib.rs

@@ -8,6 +8,7 @@ pub mod mem;
 pub mod vm;
 pub mod scope;
 pub mod js;
+pub mod task;
 
 #[doc(hidden)]
 pub mod macro_internal;

src/task.rs

@@ -0,0 +1,53 @@
+use std::marker::{Send, Sized};
+use std::mem;
+use std::os::raw::c_void;
+
+use js::{Value, JsFunction};
+use mem::Handle;
+use internal::mem::Managed;
+use internal::scope::{Scope, RootScope, RootScopeInternal};
+use internal::vm::{JsResult, Isolate, IsolateInternal};
+use neon_runtime;
+use neon_runtime::raw;
+
+pub trait Task: Send + Sized {
+    type Output: Send;
+    type Error: Send;
+    type JsEvent: Value;
+
+    fn perform(&self) -> Result<Self::Output, Self::Error>;
+
+    fn complete<'a, T: Scope<'a>>(self, scope: &'a mut T, result: Result<Self::Output, Self::Error>) -> JsResult<Self::JsEvent>;
+
+    fn schedule(self, callback: Handle<JsFunction>) {
+        let boxed_self = Box::new(self);
+        let self_raw = Box::into_raw(boxed_self);
+        let callback_raw = callback.to_raw();
+        unsafe {
+            neon_runtime::task::schedule(mem::transmute(self_raw),
+                                         perform_task::<Self>,
+                                         complete_task::<Self>,
+                                         callback_raw);
+        }
+    }
+}
+
+unsafe extern "C" fn perform_task<T: Task>(task: *mut c_void) -> *mut c_void {
+    let task: Box<T> = Box::from_raw(mem::transmute(task));
+    let result = task.perform();
+    Box::into_raw(task);
+    mem::transmute(Box::into_raw(Box::new(result)))
+}
+
+unsafe extern "C" fn complete_task<T: Task>(task: *mut c_void, result: *mut c_void, out: &mut raw::Local) {
+    let result: Result<T::Output, T::Error> = *Box::from_raw(mem::transmute(result));
+    let task: Box<T> = Box::from_raw(mem::transmute(task));
+
+    // The neon::Task::complete() method installs an outer v8::HandleScope
+    // that is responsible for managing the out pointer, so it's safe to
+    // create the RootScope here without creating a local v8::HandleScope.
+    let mut scope = RootScope::new(Isolate::current());
+    if let Ok(result) = task.complete(&mut scope, result) {
+        *out = result.to_raw();
+    }
+}

tests/lib/tasks.js

@@ -0,0 +1,30 @@
+var addon = require('../native');
+var assert = require('chai').assert;
+
+describe('Task', function() {
+  it('completes a successful task', function (done) {
+    addon.perform_async_task((err, n) => {
+      if (err) {
+        done(err);
+      } else if (n === 17) {
+        done();
+      } else {
+        done(new Error("not 17 but: " + n));
+      }
+    });
+  });
+
+  it('completes a failing task', function (done) {
+    addon.perform_failing_task((err, n) => {
+      if (err) {
+        if (err.message === 'I am a failing task') {
+          done();
+        } else {
+          done(new Error("expected error message 'I am a failing task', got: " + err.message));
+        }
+      } else {
+        done(new Error("expected task to fail, got: " + n));
+      }
+    });
+  });
+});

tests/native/src/js/tasks.rs

@@ -0,0 +1,49 @@
+use neon::vm::{Call, JsResult};
+use neon::scope::{Scope};
+use neon::js::{JsUndefined, JsNumber, JsFunction};
+use neon::js::error::{Kind, JsError};
+use neon::task::Task;
+
+struct SuccessTask;
+
+impl Task for SuccessTask {
+    type Output = i32;
+    type Error = String;
+    type JsEvent = JsNumber;
+
+    fn perform(&self) -> Result<Self::Output, Self::Error> {
+        Ok(17)
+    }
+
+    fn complete<'a, T: Scope<'a>>(self, scope: &'a mut T, result: Result<Self::Output, Self::Error>) -> JsResult<Self::JsEvent> {
+        Ok(JsNumber::new(scope, result.unwrap() as f64))
+    }
+}
+
+pub fn perform_async_task(call: Call) -> JsResult<JsUndefined> {
+    let f = call.arguments.require(call.scope, 0)?.check::<JsFunction>()?;
+    SuccessTask.schedule(f);
+    Ok(JsUndefined::new())
+}
+
+struct FailureTask;
+
+impl Task for FailureTask {
+    type Output = i32;
+    type Error = String;
+    type JsEvent = JsNumber;
+
+    fn perform(&self) -> Result<Self::Output, Self::Error> {
+        Err(format!("I am a failing task"))
+    }
+
+    fn complete<'a, T: Scope<'a>>(self, _: &'a mut T, result: Result<Self::Output, Self::Error>) -> JsResult<Self::JsEvent> {
+        JsError::throw(Kind::Error, &result.unwrap_err())
+    }
+}
+
+pub fn perform_failing_task(call: Call) -> JsResult<JsUndefined> {
+    let f = call.arguments.require(call.scope, 0)?.check::<JsFunction>()?;
+    FailureTask.schedule(f);
+    Ok(JsUndefined::new())
+}