Merge pull request #44 from Brooooooklyn/muti-thread

feat(napi): impl spawn for tasks need to be run in the other thread

napi/Cargo.toml

@@ -11,6 +11,9 @@ edition = "2018"
 
 [dependencies]
 futures = { version = "0.3", features = ["default", "thread-pool"] }
+num_cpus = "1.13"
+once_cell = "1.3"
+threadpool = "1.8"
 
 [target.'cfg(windows)'.build-dependencies]
 flate2 = "1.0"

napi/src/lib.rs

@@ -1,7 +1,8 @@
-extern crate futures;
-
 use core::fmt::Debug;
+use futures::channel::oneshot::channel;
 use futures::prelude::*;
+use num_cpus::get_physical;
+use once_cell::sync::OnceCell;
 use std::any::TypeId;
 use std::convert::{TryFrom, TryInto};
 use std::ffi::CString;
@@ -13,20 +14,27 @@ use std::ptr;
 use std::slice;
 use std::str;
 use std::string::String as RustString;
+use std::sync::Arc;
+use threadpool::ThreadPool;
 
 mod call_context;
 mod executor;
 mod promise;
 pub mod sys;
+mod task;
 mod version;
 
 pub use call_context::CallContext;
 pub use sys::{napi_valuetype, Status};
+pub use task::Task;
+use task::{NapiRSThreadPool, ThreadSafeTask};
 pub use version::NodeVersion;
 
 pub type Result<T> = std::result::Result<T, Error>;
 pub type Callback = extern "C" fn(sys::napi_env, sys::napi_callback_info) -> sys::napi_value;
 
+static THREAD_POOL: OnceCell<NapiRSThreadPool> = OnceCell::new();
+
 #[derive(Debug, Clone)]
 pub struct Error {
   pub status: Status,
@@ -526,7 +534,8 @@ impl Env {
     Ok(Value::from_raw_value(self, result, Object))
   }
 
-  pub fn perform_async_operation<
+  #[inline]
+  pub fn execute<
     T: 'static,
     V: 'static + ValueType,
     F: 'static + Future<Output = Result<T>>,
@@ -568,6 +577,43 @@ impl Env {
     Ok(Value::from_raw_value(self, raw_promise, Object))
   }
 
+  #[inline]
+  pub fn spawn<
+    V: 'static + ValueType,
+    O: Send + 'static,
+    T: 'static + Send + Task<Output = O, JsValue = V>,
+  >(
+    &self,
+    task: T,
+  ) -> Result<Value<Object>> {
+    let (sender, receiver) = channel::<Result<O>>();
+    let threadpool =
+      THREAD_POOL.get_or_init(|| NapiRSThreadPool(ThreadPool::new(get_physical() * 2)));
+    let inner_task = Arc::new(ThreadSafeTask::new(task));
+    let thread_task = Arc::clone(&inner_task);
+    let promise_task = Arc::clone(&inner_task);
+    threadpool.0.execute(move || {
+      let value = thread_task.borrow().compute();
+      match sender.send(value) {
+        Err(e) => panic!(e),
+        _ => {}
+      };
+    });
+    let rev_value = async {
+      let result = receiver.await;
+      result
+        .map_err(|_| Error {
+          status: Status::Cancelled,
+          reason: Some("Receiver cancelled".to_owned()),
+        })
+        .and_then(|v| v)
+    };
+
+    self.execute(rev_value, move |env, v| {
+      promise_task.borrow().resolve(env, v)
+    })
+  }
+
   pub fn get_node_version(&self) -> Result<NodeVersion> {
     let mut result = ptr::null();
     check_status(unsafe { sys::napi_get_node_version(self.0, &mut result) })?;

napi/src/task.rs

@@ -0,0 +1,40 @@
+use threadpool::ThreadPool;
+
+use crate::{Env, Result, Value, ValueType};
+
+pub struct NapiRSThreadPool(pub ThreadPool);
+
+unsafe impl Send for NapiRSThreadPool {}
+
+unsafe impl Sync for NapiRSThreadPool {}
+
+pub trait Task {
+  type Output: Send + 'static;
+  type JsValue: ValueType;
+
+  fn compute(&mut self) -> Result<Self::Output>;
+
+  fn resolve(&self, env: &mut Env, output: Self::Output) -> Result<Value<Self::JsValue>>;
+}
+
+pub struct ThreadSafeTask<T: Task>(pub *mut T);
+
+impl<T: Task> ThreadSafeTask<T> {
+  pub fn new(task: T) -> ThreadSafeTask<T> {
+    ThreadSafeTask(Box::into_raw(Box::new(task)))
+  }
+
+  #[inline]
+  pub fn borrow(&self) -> &'static mut T {
+    Box::leak(unsafe { Box::from_raw(self.0) })
+  }
+}
+
+impl<T: Task> Drop for ThreadSafeTask<T> {
+  fn drop(&mut self) {
+    unsafe { Box::from_raw(self.0) };
+  }
+}
+
+unsafe impl<T: Task> Send for ThreadSafeTask<T> {}
+unsafe impl<T: Task> Sync for ThreadSafeTask<T> {}

test_module/index.js

@@ -16,6 +16,11 @@ function testThrow() {
   }
 }
 
+function testSpawnThread(n) {
+  console.info('=== Test spawn task to threadpool')
+  return testModule.testSpawnThread(n)
+}
+
 const future = testSpawn()
 
 future
@@ -23,6 +28,11 @@ future
     console.info(`${value} from napi`)
     testThrow()
   })
+  .then(() => testSpawnThread(20))
+  .then((value) => {
+    console.assert(value === 6765)
+    console.info('=== fibonacci result', value)
+  })
   .catch((e) => {
     console.error(e)
     process.exit(1)

test_module/src/lib.rs

@@ -5,13 +5,18 @@ extern crate napi_rs_derive;
 
 extern crate futures;
 
-use napi::{Any, CallContext, Env, Error, Object, Result, Status, Value};
+use napi::{Any, CallContext, Env, Error, Number, Object, Result, Status, Task, Value};
+use std::convert::TryInto;
 
 register_module!(test_module, init);
 
 fn init(env: &Env, exports: &mut Value<Object>) -> Result<()> {
   exports.set_named_property("testSpawn", env.create_function("testSpawn", test_spawn)?)?;
   exports.set_named_property("testThrow", env.create_function("testThrow", test_throw)?)?;
+  exports.set_named_property(
+    "testSpawnThread",
+    env.create_function("testSpawnThread", test_spawn_thread)?,
+  )?;
   Ok(())
 }
 
@@ -35,7 +40,45 @@ fn test_spawn(ctx: CallContext) -> Result<Value<Object>> {
     Ok(results.len() as u32)
   };
 
-  env.perform_async_operation(fut_values, |&mut env, len| env.create_uint32(len))
+  env.execute(fut_values, |&mut env, len| env.create_uint32(len))
+}
+
+struct ComputeFib {
+  n: u32,
+}
+
+impl ComputeFib {
+  pub fn new(n: u32) -> ComputeFib {
+    ComputeFib { n }
+  }
+}
+
+impl Task for ComputeFib {
+  type Output = u32;
+  type JsValue = Number;
+
+  fn compute(&mut self) -> Result<Self::Output> {
+    Ok(fibonacci_native(self.n))
+  }
+
+  fn resolve(&self, env: &mut Env, output: Self::Output) -> Result<Value<Self::JsValue>> {
+    env.create_uint32(output)
+  }
+}
+
+#[inline]
+fn fibonacci_native(n: u32) -> u32 {
+  match n {
+    1 | 2 => 1,
+    _ => fibonacci_native(n - 1) + fibonacci_native(n - 2),
+  }
+}
+
+#[js_function(1)]
+fn test_spawn_thread(ctx: CallContext) -> Result<Value<Object>> {
+  let n = ctx.get::<Number>(0)?;
+  let task = ComputeFib::new(n.try_into()?);
+  ctx.env.spawn(task)
 }
 
 #[js_function]