Merge pull request #375 from geovie/feature/threadsafe_callback

Add ThreadSafeCallback helper to schedule callbacks on the main thread

Author: dherman

Cargo.toml

@@ -31,6 +31,9 @@ default = ["legacy-runtime"]
 # error message formatting from version to version, so they're disabled by default.
 enable-static-tests = []
 
+# Enable the EventHandler API of RFC 25.
+event-handler-api = []
+
 # Enable the default panic hook. Useful for debugging neon itself.
 default-panic-hook = []
 

crates/neon-runtime/src/nan/handler.rs

@@ -0,0 +1,8 @@
+//! Facilities for running a callback in the libuv main thread.
+
+/// Creates a new event handler which can be used to execute a callback in the libuv main thread
+pub use neon_sys::Neon_EventHandler_New as new;
+/// Executes the given callback in the libuv main thread
+pub use neon_sys::Neon_EventHandler_Schedule as schedule;
+// Free the thread safe callback and any memory hold
+pub use neon_sys::Neon_EventHandler_Delete as delete;

crates/neon-runtime/src/nan/mod.rs

@@ -15,3 +15,4 @@ pub mod fun;
 pub mod convert;
 pub mod class;
 pub mod task;
+pub mod handler;

crates/neon-runtime/src/napi/handler.rs

@@ -0,0 +1,7 @@
+use raw::Local;
+use std::os::raw::c_void;
+
+pub unsafe extern "C" fn new(_isolate: *mut c_void, _this: Local, _callback: Local) -> *mut c_void { unimplemented!() }
+pub unsafe extern "C" fn schedule(_thread_safe_cb: *mut c_void, _rust_callback: *mut c_void,
+            _complete: unsafe extern fn(Local, Local, *mut c_void)) { unimplemented!() }
+pub unsafe extern "C" fn delete(_thread_safe_cb: *mut c_void) { unimplemented!() }

crates/neon-runtime/src/napi/mod.rs

@@ -14,3 +14,4 @@ pub mod scope;
 pub mod string;
 pub mod tag;
 pub mod task;
+pub mod handler;

crates/neon-sys/native/src/neon.cc

@@ -8,6 +8,7 @@
 #include "neon_string.h"
 #include "neon_class_metadata.h"
 #include "neon_task.h"
+#include "neon_event.h"
 
 extern "C" void Neon_Call_SetReturn(v8::FunctionCallbackInfo<v8::Value> *info, v8::Local<v8::Value> value) {
   info->GetReturnValue().Set(value);
@@ -520,3 +521,17 @@ extern "C" void Neon_Task_Schedule(void *task, Neon_TaskPerformCallback perform,
   neon::Task *internal_task = new neon::Task(isolate, task, perform, complete, callback);
   neon::queue_task(internal_task);
 }
+
+extern "C" void* Neon_EventHandler_New(v8::Isolate *isolate, v8::Local<v8::Value> self, v8::Local<v8::Function> callback) {
+  return new neon::EventHandler(isolate, self, callback);
+}
+
+extern "C" void Neon_EventHandler_Schedule(void *thread_safe_cb, void *rust_callback, Neon_EventHandler handler) {
+    neon::EventHandler *cb = static_cast<neon::EventHandler*>(thread_safe_cb);
+    cb->schedule(rust_callback, handler);
+}
+
+extern "C" void Neon_EventHandler_Delete(void * thread_safe_cb) {
+    neon::EventHandler *cb = static_cast<neon::EventHandler*>(thread_safe_cb);
+    cb->close();
+}

crates/neon-sys/native/src/neon.h

@@ -133,6 +133,12 @@ extern "C" {
   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);
+
+  typedef void (*Neon_EventHandler)(v8::Local<v8::Value> self, v8::Local<v8::Value> callback, void* arg_cb);
+
+  void* Neon_EventHandler_New(v8::Isolate *isolate, v8::Local<v8::Value> self, v8::Local<v8::Function> callback);
+  void Neon_EventHandler_Schedule(void* thread_safe_cb, void* rust_callback, Neon_EventHandler handler);
+  void Neon_EventHandler_Delete(void* thread_safe_cb);
 }
 
 #endif

crates/neon-sys/native/src/neon_event.h

@@ -0,0 +1,101 @@
+#ifndef NEON_EVENTHANDLER_H_
+#define NEON_EVENTHANDLER_H_
+
+#include <uv.h>
+#include "neon.h"
+#include "v8.h"
+#include <mutex>
+
+namespace neon {
+
+    // THe implementation of this class was adapted from
+    // https://github.com/mika-fischer/napi-thread-safe-callback
+    class EventHandler {
+    public:
+        EventHandler(v8::Isolate *isolate,
+                           v8::Local<v8::Value> self,
+                           v8::Local<v8::Function> callback): isolate_(isolate), close_(false)
+        {
+            async_.data = this;
+            uv_async_init(uv_default_loop(), &async_, async_complete);
+            // Save the this argument and the callback to be invoked.
+            self_.Reset(isolate, self);
+            callback_.Reset(isolate, callback);
+            // Save the context (aka realm) to be used when invoking the callback.
+            context_.Reset(isolate, isolate->GetCurrentContext());
+        }
+
+        void schedule(void *rust_callback, Neon_EventHandler handler) {
+            {
+                std::lock_guard<std::mutex> lock(mutex_);
+                handlers_.push_back({ rust_callback, handler });
+            }
+            uv_async_send(&async_);
+        }
+
+        void close() {
+            // close is called when the rust struct is dropped
+            // this guarantees that it's called only once and
+            // that no other method (call) will be called after it.
+            close_ = true;
+            uv_async_send(&async_);
+        }
+
+        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> self = v8::Local<v8::Value>::New(isolate_, self_);
+            v8::Local<v8::Function> callback = v8::Local<v8::Function>::New(isolate_, callback_);
+
+            while (true) {
+                std::vector<HandlerData> handlers;
+                {
+                    std::lock_guard<std::mutex> lock(mutex_);
+                    if (handlers_.empty()) {
+                        break;
+                    } else {
+                        handlers.swap(handlers_);
+                    }
+                }
+                for (const HandlerData &data : handlers) {
+                    data.handler(self, callback, data.rust_callback);
+                }
+            }
+
+            if (close_) {
+                uv_close(reinterpret_cast<uv_handle_t*>(&async_), [](uv_handle_t* handle) {
+                    delete static_cast<EventHandler*>(handle->data);
+                });
+            }
+        }
+
+    private:
+        static void async_complete(uv_async_t* handle) {
+            EventHandler* cb = static_cast<EventHandler*>(handle->data);
+            cb->complete();
+        }
+
+        uv_async_t async_;
+        v8::Isolate *isolate_;
+        v8::Persistent<v8::Value> self_;
+        v8::Persistent<v8::Function> callback_;
+        v8::Persistent<v8::Context> context_;
+
+        struct HandlerData {
+            void *rust_callback;
+            Neon_EventHandler handler;
+        };
+
+        std::mutex mutex_;
+        std::vector<HandlerData> handlers_;
+
+        bool close_;
+    };
+}
+
+#endif

crates/neon-sys/src/lib.rs

@@ -198,4 +198,8 @@ extern "C" {
                               complete: unsafe extern fn(*mut c_void, *mut c_void, &mut Local),
                               callback: Local);
 
+    pub fn Neon_EventHandler_New(isolate: Isolate, this: Local, callback: Local) -> *mut c_void;
+    pub fn Neon_EventHandler_Schedule(thread_safe_cb: *mut c_void, rust_callback: *mut c_void,
+                                        complete: unsafe extern fn(Local, Local, *mut c_void));
+    pub fn Neon_EventHandler_Delete(thread_safe_cb: *mut c_void);
 }

src/event/mod.rs

@@ -0,0 +1,73 @@
+//! Helper to run a callback in the libuv main thread.
+
+use std::mem;
+use std::os::raw::c_void;
+
+use types::*;
+use handle::{Handle, Managed};
+use neon_runtime;
+use neon_runtime::raw;
+use std::sync::Arc;
+use context::Context;
+
+type EventContext<'a> = crate::context::TaskContext<'a>;
+
+struct EventHandlerInner(*mut c_void);
+
+unsafe impl Send for EventHandlerInner {}
+unsafe impl Sync for EventHandlerInner {}
+
+impl Drop for EventHandlerInner {
+    fn drop(&mut self) {
+        unsafe {
+            neon_runtime::handler::delete(self.0);
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct EventHandler(Arc<EventHandlerInner>);
+
+impl EventHandler {
+    #[cfg(feature = "napi-runtime")]
+    pub fn new<'a, C: Context<'a>, T: Value>(cx: &C, this: Handle<T>, callback: Handle<JsFunction>) -> Self {
+        unimplemented!()
+    }
+
+    #[cfg(feature = "legacy-runtime")]
+    pub fn new<'a, C: Context<'a>, T: Value>(cx: &C, this: Handle<T>, callback: Handle<JsFunction>) -> Self {
+        let cb = unsafe {
+            neon_runtime::handler::new(cx.env().to_raw(), this.to_raw(), callback.to_raw())
+        };
+        EventHandler(Arc::new(EventHandlerInner(cb)))
+    }
+
+    pub fn schedule<T, F>(&self, arg_cb: F)
+        where T: Value,
+              F: for<'a> FnOnce(&mut EventContext<'a>) -> Vec<Handle<'a, T>>,
+              F: Send + 'static {
+        self.schedule_with(move |cx, this, callback| {
+            let args = arg_cb(cx);
+            let _result = callback.call(cx, this, args);
+        })
+    }
+
+    pub fn schedule_with<F>(&self, arg_cb: F)
+        where F: FnOnce(&mut EventContext, Handle<JsValue>, Handle<JsFunction>),
+              F: Send + 'static {
+        let callback = Box::into_raw(Box::new(arg_cb)) as *mut c_void;
+        unsafe {
+            neon_runtime::handler::schedule((*self.0).0, callback, handle_callback::<F>);
+        }
+    }
+}
+
+unsafe extern "C" fn handle_callback<F>(this: raw::Local, func: raw::Local, callback: *mut c_void)
+    where F: FnOnce(&mut EventContext, Handle<JsValue>, Handle<JsFunction>), F: Send + 'static {
+    EventContext::with(|mut cx: EventContext| {
+        let this = JsValue::new_internal(this);
+        let func: Handle<JsFunction> = Handle::new_internal(JsFunction::from_raw(func));
+        let callback: Box<F> = Box::from_raw(mem::transmute(callback));
+        callback(&mut cx, this, func);
+    })
+}