naskya/napi-rs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #295 from napi-rs/tsfn

Browse source 
feat(napi): impl Clone and Drop for ThreadSafeFunction, return Status  from ThreadSafeFunction::call
This commit is contained in:
LongYinan 2020-11-11 18:38:59 +08:00 • committed by GitHub
commit eb5d1439af
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 368 additions and 307 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
napi/src/async_work.rs
View file

@ -107,16 +107,16 @@ unsafe extern "C" fn complete<T: Task>(
match check_status(status).and_then(move |_| value) { match check_status(status).and_then(move |_| value) {
Ok(v) => { Ok(v) => {
let status = sys::napi_resolve_deferred(env, deferred, v.raw()); let status = sys::napi_resolve_deferred(env, deferred, v.raw());
debug_assert!(status == sys::napi_status::napi_ok, "Reject promise failed"); debug_assert!(status == sys::Status::napi_ok, "Reject promise failed");
} }
Err(e) => { Err(e) => {
let status = sys::napi_reject_deferred(env, deferred, e.into_raw(env)); let status = sys::napi_reject_deferred(env, deferred, e.into_raw(env));
debug_assert!(status == sys::napi_status::napi_ok, "Reject promise failed"); debug_assert!(status == sys::Status::napi_ok, "Reject promise failed");
} }
}; };
let delete_status = sys::napi_delete_async_work(env, napi_async_work); let delete_status = sys::napi_delete_async_work(env, napi_async_work);
debug_assert!( debug_assert!(
delete_status == sys::napi_status::napi_ok, delete_status == sys::Status::napi_ok,
"Delete async work failed" "Delete async work failed"
); );
} }

2
napi/src/env.rs
View file

@ -597,7 +597,7 @@ impl Env {
R: 'static + Send + FnMut(ThreadSafeCallContext<T>) -> Result<Vec<V>>, R: 'static + Send + FnMut(ThreadSafeCallContext<T>) -> Result<Vec<V>>,
>( >(
&self, &self,
func: JsFunction, func: &JsFunction,
max_queue_size: usize, max_queue_size: usize,
callback: R, callback: R,
) -> Result<ThreadsafeFunction<T>> { ) -> Result<ThreadsafeFunction<T>> {

7
napi/src/error.rs
View file

@ -71,12 +71,9 @@ impl Error {
s.len() as _, s.len() as _,
&mut err_reason, &mut err_reason,
); );
debug_assert!( debug_assert!(status == sys::Status::napi_ok, "Create error reason failed");
status == sys::napi_status::napi_ok,
"Create error reason failed"
);
let status = sys::napi_create_error(env, ptr::null_mut(), err_reason, &mut err); let status = sys::napi_create_error(env, ptr::null_mut(), err_reason, &mut err);
debug_assert!(status == sys::napi_status::napi_ok, "Create error failed"); debug_assert!(status == sys::Status::napi_ok, "Create error failed");
}; };
err err
} }

68
napi/src/js_values/arraybuffer.rs
View file

@ -1,5 +1,3 @@
use std::convert::TryFrom;
use std::convert::TryInto;
use std::mem; use std::mem;
use std::ops::Deref; use std::ops::Deref;
use std::os::raw::c_void; use std::os::raw::c_void;
@ -7,7 +5,7 @@ use std::ptr;
use super::{Value, ValueType}; use super::{Value, ValueType};
use crate::error::check_status; use crate::error::check_status;
use crate::{sys, Error, JsUnknown, NapiValue, Ref, Result}; use crate::{sys, JsUnknown, NapiValue, Ref, Result};
#[repr(transparent)] #[repr(transparent)]
#[derive(Debug)] #[derive(Debug)]
@ -44,10 +42,10 @@ pub struct JsDataViewValue {
pub length: u64, pub length: u64,
} }
#[repr(u8)] #[repr(i32)]
#[derive(Debug)] #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
pub enum TypedArrayType { pub enum TypedArrayType {
Int8, Int8 = 0,
Uint8, Uint8,
Uint8Clamped, Uint8Clamped,
Int16, Int16,
@ -56,51 +54,35 @@ pub enum TypedArrayType {
Uint32, Uint32,
Float32, Float32,
Float64, Float64,
#[cfg(feature = "napi6")]
BigInt64, BigInt64,
#[cfg(feature = "napi6")]
BigUint64, BigUint64,
/// compatible with higher versions
Unknown = 1024,
} }
impl TryFrom<sys::napi_typedarray_type> for TypedArrayType { impl From<sys::napi_typedarray_type> for TypedArrayType {
type Error = Error; fn from(value: sys::napi_typedarray_type) -> Self {
fn try_from(value: sys::napi_typedarray_type) -> Result<Self> {
match value { match value {
sys::napi_typedarray_type::napi_int8_array => Ok(Self::Int8), sys::TypedarrayType::napi_int8_array => Self::Int8,
sys::napi_typedarray_type::napi_uint8_array => Ok(Self::Uint8), sys::TypedarrayType::napi_uint8_array => Self::Uint8,
sys::napi_typedarray_type::napi_uint8_clamped_array => Ok(Self::Uint8Clamped), sys::TypedarrayType::napi_uint8_clamped_array => Self::Uint8Clamped,
sys::napi_typedarray_type::napi_int16_array => Ok(Self::Int16), sys::TypedarrayType::napi_int16_array => Self::Int16,
sys::napi_typedarray_type::napi_uint16_array => Ok(Self::Uint16), sys::TypedarrayType::napi_uint16_array => Self::Uint16,
sys::napi_typedarray_type::napi_int32_array => Ok(Self::Int32), sys::TypedarrayType::napi_int32_array => Self::Int32,
sys::napi_typedarray_type::napi_uint32_array => Ok(Self::Uint32), sys::TypedarrayType::napi_uint32_array => Self::Uint32,
sys::napi_typedarray_type::napi_float32_array => Ok(Self::Float32), sys::TypedarrayType::napi_float32_array => Self::Float32,
sys::napi_typedarray_type::napi_float64_array => Ok(Self::Float64), sys::TypedarrayType::napi_float64_array => Self::Float64,
#[cfg(feature = "napi6")] sys::TypedarrayType::napi_bigint64_array => Self::BigInt64,
sys::napi_typedarray_type::napi_bigint64_array => Ok(Self::BigInt64), sys::TypedarrayType::napi_biguint64_array => Self::BigUint64,
#[cfg(feature = "napi6")] _ => Self::Unknown,
sys::napi_typedarray_type::napi_biguint64_array => Ok(Self::BigUint64),
} }
} }
} }
impl From<TypedArrayType> for sys::napi_typedarray_type { impl From<TypedArrayType> for sys::napi_typedarray_type {
fn from(value: TypedArrayType) -> Self { fn from(value: TypedArrayType) -> sys::napi_typedarray_type {
match value { value as _
TypedArrayType::Int8 => sys::napi_typedarray_type::napi_int8_array,
TypedArrayType::Uint8 => sys::napi_typedarray_type::napi_uint8_array,
TypedArrayType::Uint8Clamped => sys::napi_typedarray_type::napi_uint8_clamped_array,
TypedArrayType::Int16 => sys::napi_typedarray_type::napi_int16_array,
TypedArrayType::Uint16 => sys::napi_typedarray_type::napi_uint16_array,
TypedArrayType::Int32 => sys::napi_typedarray_type::napi_int32_array,
TypedArrayType::Uint32 => sys::napi_typedarray_type::napi_uint32_array,
TypedArrayType::Float32 => sys::napi_typedarray_type::napi_float32_array,
TypedArrayType::Float64 => sys::napi_typedarray_type::napi_float64_array,
#[cfg(feature = "napi6")]
TypedArrayType::BigInt64 => sys::napi_typedarray_type::napi_bigint64_array,
#[cfg(feature = "napi6")]
TypedArrayType::BigUint64 => sys::napi_typedarray_type::napi_biguint64_array,
}
} }
} }
@ -221,7 +203,7 @@ impl JsTypedArray {
/// ///
/// ***Warning***: Use caution while using this API since the underlying data buffer is managed by the VM. /// ***Warning***: Use caution while using this API since the underlying data buffer is managed by the VM.
pub fn into_value(self) -> Result<JsTypedArrayValue> { pub fn into_value(self) -> Result<JsTypedArrayValue> {
let mut typedarray_type = sys::napi_typedarray_type::napi_int8_array; let mut typedarray_type = 0;
let mut len = 0u64; let mut len = 0u64;
let mut data = ptr::null_mut(); let mut data = ptr::null_mut();
let mut arraybuffer_value = ptr::null_mut(); let mut arraybuffer_value = ptr::null_mut();
@ -242,7 +224,7 @@ impl JsTypedArray {
data, data,
length: len, length: len,
byte_offset, byte_offset,
typedarray_type: typedarray_type.try_into()?, typedarray_type: typedarray_type.into(),
arraybuffer: JsArrayBuffer::from_raw_unchecked(self.0.env, arraybuffer_value), arraybuffer: JsArrayBuffer::from_raw_unchecked(self.0.env, arraybuffer_value),
}) })
} }

2
napi/src/js_values/function.rs
View file

@ -34,7 +34,7 @@ impl JsFunction {
.map(|u| u.raw()) .map(|u| u.raw())
}) })
.ok_or(Error::new( .ok_or(Error::new(
Status::Unknown, Status::GenericFailure,
"Get raw this failed".to_owned(), "Get raw this failed".to_owned(),
))?; ))?;
let raw_args = args let raw_args = args

4
napi/src/js_values/mod.rs
View file

@ -76,9 +76,9 @@ pub struct JsExternal(pub(crate) Value);
#[inline] #[inline]
pub(crate) fn type_of(env: sys::napi_env, raw_value: sys::napi_value) -> Result<ValueType> { pub(crate) fn type_of(env: sys::napi_env, raw_value: sys::napi_value) -> Result<ValueType> {
unsafe { unsafe {
let mut value_type = sys::napi_valuetype::napi_undefined; let mut value_type = 0;
check_status(sys::napi_typeof(env, raw_value, &mut value_type))?; check_status(sys::napi_typeof(env, raw_value, &mut value_type))?;
Ok(ValueType::from(value_type)) ValueType::try_from(value_type).or_else(|_| Ok(ValueType::Unknown))
} }
} }

55
napi/src/js_values/value_type.rs
View file

@ -1,8 +1,8 @@
use std::convert::TryInto; use std::fmt::{Display, Formatter, Result};
use crate::{sys, Error, Result, Status}; use crate::sys;
#[repr(u8)] #[repr(i32)]
#[derive(Copy, Clone, Debug, PartialEq, PartialOrd, Ord, Eq, Hash)] #[derive(Copy, Clone, Debug, PartialEq, PartialOrd, Ord, Eq, Hash)]
pub enum ValueType { pub enum ValueType {
Undefined = 0, Undefined = 0,
@ -16,44 +16,31 @@ pub enum ValueType {
External = 8, External = 8,
#[cfg(feature = "napi6")] #[cfg(feature = "napi6")]
Bigint = 9, Bigint = 9,
Unknown = 255, Unknown = 1024,
} }
impl TryInto<sys::napi_valuetype> for ValueType { impl Display for ValueType {
type Error = Error; fn fmt(&self, f: &mut Formatter<'_>) -> Result {
let status_string = format!("{:?}", self);
fn try_into(self) -> Result<sys::napi_valuetype> { write!(f, "{}", status_string)
match self {
ValueType::Unknown => Err(Error::from_status(Status::Unknown)),
#[cfg(feature = "napi6")]
ValueType::Bigint => Ok(sys::napi_valuetype::napi_bigint),
ValueType::Boolean => Ok(sys::napi_valuetype::napi_boolean),
ValueType::External => Ok(sys::napi_valuetype::napi_external),
ValueType::Function => Ok(sys::napi_valuetype::napi_function),
ValueType::Null => Ok(sys::napi_valuetype::napi_null),
ValueType::Number => Ok(sys::napi_valuetype::napi_number),
ValueType::Object => Ok(sys::napi_valuetype::napi_object),
ValueType::String => Ok(sys::napi_valuetype::napi_string),
ValueType::Symbol => Ok(sys::napi_valuetype::napi_symbol),
ValueType::Undefined => Ok(sys::napi_valuetype::napi_undefined),
}
} }
} }
impl From<sys::napi_valuetype> for ValueType { impl From<i32> for ValueType {
fn from(value: sys::napi_valuetype) -> Self { fn from(value: i32) -> ValueType {
match value { match value {
#[cfg(feature = "napi6")] #[cfg(feature = "napi6")]
sys::napi_valuetype::napi_bigint => ValueType::Bigint, sys::ValueType::napi_bigint => ValueType::Bigint,
sys::napi_valuetype::napi_boolean => ValueType::Boolean, sys::ValueType::napi_boolean => ValueType::Boolean,
sys::napi_valuetype::napi_external => ValueType::External, sys::ValueType::napi_external => ValueType::External,
sys::napi_valuetype::napi_function => ValueType::Function, sys::ValueType::napi_function => ValueType::Function,
sys::napi_valuetype::napi_null => ValueType::Null, sys::ValueType::napi_null => ValueType::Null,
sys::napi_valuetype::napi_number => ValueType::Number, sys::ValueType::napi_number => ValueType::Number,
sys::napi_valuetype::napi_object => ValueType::Object, sys::ValueType::napi_object => ValueType::Object,
sys::napi_valuetype::napi_string => ValueType::String, sys::ValueType::napi_string => ValueType::String,
sys::napi_valuetype::napi_symbol => ValueType::Symbol, sys::ValueType::napi_symbol => ValueType::Symbol,
sys::napi_valuetype::napi_undefined => ValueType::Undefined, sys::ValueType::napi_undefined => ValueType::Undefined,
_ => ValueType::Unknown,
} }
} }
} }

7
napi/src/promise.rs
View file

@ -105,14 +105,11 @@ unsafe extern "C" fn call_js_cb<T, V: NapiValue>(
match js_value_to_resolve { match js_value_to_resolve {
Ok(v) => { Ok(v) => {
let status = sys::napi_resolve_deferred(raw_env, deferred, v.raw()); let status = sys::napi_resolve_deferred(raw_env, deferred, v.raw());
debug_assert!( debug_assert!(status == sys::Status::napi_ok, "Resolve promise failed");
status == sys::napi_status::napi_ok,
"Resolve promise failed"
);
} }
Err(e) => { Err(e) => {
let status = sys::napi_reject_deferred(raw_env, deferred, e.into_raw(raw_env)); let status = sys::napi_reject_deferred(raw_env, deferred, e.into_raw(raw_env));
debug_assert!(status == sys::napi_status::napi_ok, "Reject promise failed"); debug_assert!(status == sys::Status::napi_ok, "Reject promise failed");
} }
}; };
check_status(sys::napi_release_threadsafe_function( check_status(sys::napi_release_threadsafe_function(

103
napi/src/status.rs
View file

@ -1,8 +1,11 @@
use crate::sys::napi_status; use std::fmt::{Display, Formatter, Result};
use crate::sys;
#[repr(i32)]
#[derive(Eq, PartialEq, Debug, Clone, Copy)] #[derive(Eq, PartialEq, Debug, Clone, Copy)]
pub enum Status { pub enum Status {
Ok, Ok = 0,
InvalidArg, InvalidArg,
ObjectExpected, ObjectExpected,
StringExpected, StringExpected,
@ -17,71 +20,51 @@ pub enum Status {
EscapeCalledTwice, EscapeCalledTwice,
HandleScopeMismatch, HandleScopeMismatch,
CallbackScopeMismatch, CallbackScopeMismatch,
#[cfg(feature = "napi4")] /// ThreadSafeFunction queue is full
QueueFull, QueueFull,
#[cfg(feature = "napi4")] /// ThreadSafeFunction closed
Closing, Closing,
#[cfg(feature = "napi6")]
BigintExpected, BigintExpected,
Unknown, DateExpected,
ArrayBufferExpected,
DetachableArraybufferExpected,
WouldDeadlock,
Unknown = 1024, // unknown status. for example, using napi3 module in napi7 NodeJS, and generate an invalid napi3 status
} }
impl From<napi_status> for Status { impl Display for Status {
fn from(code: napi_status) -> Self { fn fmt(&self, f: &mut Formatter<'_>) -> Result {
use Status::*; let status_string = format!("{:?}", self);
write!(f, "{}", status_string)
}
}
impl From<i32> for Status {
fn from(code: i32) -> Self {
match code { match code {
napi_status::napi_ok => Ok, sys::Status::napi_ok => Status::Ok,
napi_status::napi_invalid_arg => InvalidArg, sys::Status::napi_invalid_arg => Status::InvalidArg,
napi_status::napi_object_expected => ObjectExpected, sys::Status::napi_object_expected => Status::ObjectExpected,
napi_status::napi_string_expected => StringExpected, sys::Status::napi_string_expected => Status::StringExpected,
napi_status::napi_name_expected => NameExpected, sys::Status::napi_name_expected => Status::NameExpected,
napi_status::napi_function_expected => FunctionExpected, sys::Status::napi_function_expected => Status::FunctionExpected,
napi_status::napi_number_expected => NumberExpected, sys::Status::napi_number_expected => Status::NumberExpected,
napi_status::napi_boolean_expected => BooleanExpected, sys::Status::napi_boolean_expected => Status::BooleanExpected,
napi_status::napi_array_expected => ArrayExpected, sys::Status::napi_array_expected => Status::ArrayExpected,
napi_status::napi_generic_failure => GenericFailure, sys::Status::napi_generic_failure => Status::GenericFailure,
napi_status::napi_pending_exception => PendingException, sys::Status::napi_pending_exception => Status::PendingException,
napi_status::napi_cancelled => Cancelled, sys::Status::napi_cancelled => Status::Cancelled,
napi_status::napi_escape_called_twice => EscapeCalledTwice, sys::Status::napi_escape_called_twice => Status::EscapeCalledTwice,
napi_status::napi_handle_scope_mismatch => HandleScopeMismatch, sys::Status::napi_handle_scope_mismatch => Status::HandleScopeMismatch,
napi_status::napi_callback_scope_mismatch => CallbackScopeMismatch, sys::Status::napi_callback_scope_mismatch => Status::CallbackScopeMismatch,
#[cfg(feature = "napi4")] sys::Status::napi_queue_full => Status::QueueFull,
napi_status::napi_queue_full => QueueFull, sys::Status::napi_closing => Status::Closing,
#[cfg(feature = "napi4")] sys::Status::napi_bigint_expected => Status::BigintExpected,
napi_status::napi_closing => Closing, sys::Status::napi_date_expected => Status::DateExpected,
#[cfg(feature = "napi6")] sys::Status::napi_arraybuffer_expected => Status::ArrayBufferExpected,
napi_status::napi_bigint_expected => BigintExpected, sys::Status::napi_detachable_arraybuffer_expected => Status::DetachableArraybufferExpected,
_ => Unknown, sys::Status::napi_would_deadlock => Status::WouldDeadlock,
} _ => Status::Unknown,
}
}
impl Into<self::napi_status> for Status {
fn into(self) -> napi_status {
match self {
Self::Ok => napi_status::napi_ok,
Self::InvalidArg => napi_status::napi_invalid_arg,
Self::ObjectExpected => napi_status::napi_object_expected,
Self::StringExpected => napi_status::napi_string_expected,
Self::NameExpected => napi_status::napi_name_expected,
Self::FunctionExpected => napi_status::napi_function_expected,
Self::NumberExpected => napi_status::napi_number_expected,
Self::BooleanExpected => napi_status::napi_boolean_expected,
Self::ArrayExpected => napi_status::napi_array_expected,
Self::GenericFailure => napi_status::napi_generic_failure,
Self::PendingException => napi_status::napi_pending_exception,
Self::Cancelled => napi_status::napi_cancelled,
Self::EscapeCalledTwice => napi_status::napi_escape_called_twice,
Self::HandleScopeMismatch => napi_status::napi_handle_scope_mismatch,
Self::CallbackScopeMismatch => napi_status::napi_callback_scope_mismatch,
#[cfg(feature = "napi4")]
Self::QueueFull => napi_status::napi_queue_full,
#[cfg(feature = "napi4")]
Self::Closing => napi_status::napi_closing,
#[cfg(feature = "napi6")]
Self::BigintExpected => napi_status::napi_bigint_expected,
Self::Unknown => napi_status::napi_generic_failure,
} }
} }
} }

131
napi/src/threadsafe_function.rs
View file

@ -2,13 +2,16 @@ use std::convert::Into;
use std::marker::PhantomData; use std::marker::PhantomData;
use std::os::raw::{c_char, c_void}; use std::os::raw::{c_char, c_void};
use std::ptr; use std::ptr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use crate::error::check_status; use crate::error::check_status;
use crate::{sys, Env, JsFunction, NapiValue, Result}; use crate::{sys, Env, Error, JsFunction, NapiValue, Result, Status};
use sys::napi_threadsafe_function_call_mode; use sys::napi_threadsafe_function_call_mode;
use sys::napi_threadsafe_function_release_mode;
/// ThreadSafeFunction Context object
/// the `value` is the value passed to `call` method
pub struct ThreadSafeCallContext<T: 'static> { pub struct ThreadSafeCallContext<T: 'static> {
pub env: Env, pub env: Env,
pub value: T, pub value: T,
@ -20,12 +23,6 @@ pub enum ThreadsafeFunctionCallMode {
Blocking, Blocking,
} }
#[repr(u8)]
pub enum ThreadsafeFunctionReleaseMode {
Release,
Abort,
}
impl Into<napi_threadsafe_function_call_mode> for ThreadsafeFunctionCallMode { impl Into<napi_threadsafe_function_call_mode> for ThreadsafeFunctionCallMode {
fn into(self) -> napi_threadsafe_function_call_mode { fn into(self) -> napi_threadsafe_function_call_mode {
match self { match self {
@ -39,19 +36,6 @@ impl Into<napi_threadsafe_function_call_mode> for ThreadsafeFunctionCallMode {
} }
} }
impl Into<napi_threadsafe_function_release_mode> for ThreadsafeFunctionReleaseMode {
fn into(self) -> napi_threadsafe_function_release_mode {
match self {
ThreadsafeFunctionReleaseMode::Release => {
napi_threadsafe_function_release_mode::napi_tsfn_release
}
ThreadsafeFunctionReleaseMode::Abort => {
napi_threadsafe_function_release_mode::napi_tsfn_abort
}
}
}
}
/// Communicate with the addon's main thread by invoking a JavaScript function from other threads. /// Communicate with the addon's main thread by invoking a JavaScript function from other threads.
/// ///
/// ## Example /// ## Example
@ -74,20 +58,17 @@ impl Into<napi_threadsafe_function_release_mode> for ThreadsafeFunctionReleaseMo
/// pub fn test_threadsafe_function(ctx: CallContext) -> Result<JsUndefined> { /// pub fn test_threadsafe_function(ctx: CallContext) -> Result<JsUndefined> {
/// let func = ctx.get::<JsFunction>(0)?; /// let func = ctx.get::<JsFunction>(0)?;
/// let tsfn = /// let tsfn = ctx.env
/// ctx
/// .env
/// .create_threadsafe_function(func, 0, |ctx: ThreadSafeCallContext<Vec<u32>>| { /// .create_threadsafe_function(func, 0, |ctx: ThreadSafeCallContext<Vec<u32>>| {
/// ctx /// ctx.value
/// .value
/// .iter() /// .iter()
/// .map(|v| ctx.env.create_uint32(*v)) /// .map(|v| ctx.env.create_uint32(*v))]
/// .collect::<Result<Vec<JsNumber>>>() /// .collect::<Result<Vec<JsNumber>>>()
/// })?; /// })?;
/// thread::spawn(move || { /// thread::spawn(move || {
/// let output: Vec<u32> = vec![42, 1, 2, 3]; /// let output: Vec<u32> = vec![42, 1, 2, 3];
/// /// It's okay to call a threadsafe function multiple times. /// // It's okay to call a threadsafe function multiple times.
/// tsfn.call(Ok(output.clone()), ThreadsafeFunctionCallMode::Blocking); /// tsfn.call(Ok(output.clone()), ThreadsafeFunctionCallMode::Blocking);
/// tsfn.call(Ok(output.clone()), ThreadsafeFunctionCallMode::NonBlocking); /// tsfn.call(Ok(output.clone()), ThreadsafeFunctionCallMode::NonBlocking);
/// tsfn.release(ThreadsafeFunctionReleaseMode::Release); /// tsfn.release(ThreadsafeFunctionReleaseMode::Release);
@ -98,6 +79,7 @@ impl Into<napi_threadsafe_function_release_mode> for ThreadsafeFunctionReleaseMo
/// ``` /// ```
pub struct ThreadsafeFunction<T: 'static> { pub struct ThreadsafeFunction<T: 'static> {
raw_tsfn: sys::napi_threadsafe_function, raw_tsfn: sys::napi_threadsafe_function,
aborted: Arc<AtomicBool>,
_phantom: PhantomData<T>, _phantom: PhantomData<T>,
} }
@ -118,7 +100,7 @@ impl<T: 'static> ThreadsafeFunction<T> {
R: 'static + Send + FnMut(ThreadSafeCallContext<T>) -> Result<Vec<V>>, R: 'static + Send + FnMut(ThreadSafeCallContext<T>) -> Result<Vec<V>>,
>( >(
env: sys::napi_env, env: sys::napi_env,
func: JsFunction, func: &JsFunction,
max_queue_size: usize, max_queue_size: usize,
callback: R, callback: R,
) -> Result<Self> { ) -> Result<Self> {
@ -155,59 +137,92 @@ impl<T: 'static> ThreadsafeFunction<T> {
Ok(ThreadsafeFunction { Ok(ThreadsafeFunction {
raw_tsfn, raw_tsfn,
aborted: Arc::new(AtomicBool::new(false)),
_phantom: PhantomData, _phantom: PhantomData,
}) })
} }
/// See [napi_call_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_call_threadsafe_function) /// See [napi_call_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_call_threadsafe_function)
/// for more information. /// for more information.
pub fn call(&self, value: Result<T>, mode: ThreadsafeFunctionCallMode) { pub fn call(&self, value: Result<T>, mode: ThreadsafeFunctionCallMode) -> Status {
let status = unsafe { if self.aborted.load(Ordering::Acquire) {
return Status::Closing;
}
unsafe {
sys::napi_call_threadsafe_function( sys::napi_call_threadsafe_function(
self.raw_tsfn, self.raw_tsfn,
Box::into_raw(Box::new(value)) as *mut _, Box::into_raw(Box::new(value)) as *mut _,
mode.into(), mode.into(),
) )
};
debug_assert!(
status == sys::napi_status::napi_ok,
"Threadsafe Function call failed"
);
} }
.into()
/// See [napi_acquire_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_acquire_threadsafe_function)
/// for more information.
pub fn acquire(&self) {
let status = unsafe { sys::napi_acquire_threadsafe_function(self.raw_tsfn) };
debug_assert!(
status == sys::napi_status::napi_ok,
"Threadsafe Function acquire failed"
);
}
/// See [napi_release_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_release_threadsafe_function)
/// for more information.
pub fn release(self, mode: ThreadsafeFunctionReleaseMode) {
let status = unsafe { sys::napi_release_threadsafe_function(self.raw_tsfn, mode.into()) };
debug_assert!(
status == sys::napi_status::napi_ok,
"Threadsafe Function call failed"
);
} }
/// See [napi_ref_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_ref_threadsafe_function) /// See [napi_ref_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_ref_threadsafe_function)
/// for more information. /// for more information.
/// ///
/// "ref" is a keyword so that we use "refer" here. /// "ref" is a keyword so that we use "refer" here.
pub fn refer(&self, env: &Env) -> Result<()> { pub fn refer(&mut self, env: &Env) -> Result<()> {
check_status(unsafe { sys::napi_ref_threadsafe_function(env.0, self.raw_tsfn) }) check_status(unsafe { sys::napi_ref_threadsafe_function(env.0, self.raw_tsfn) })
} }
/// See [napi_unref_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_unref_threadsafe_function) /// See [napi_unref_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_unref_threadsafe_function)
/// for more information. /// for more information.
pub fn unref(&self, env: &Env) -> Result<()> { pub fn unref(&mut self, env: &Env) -> Result<()> {
check_status(unsafe { sys::napi_unref_threadsafe_function(env.0, self.raw_tsfn) }) check_status(unsafe { sys::napi_unref_threadsafe_function(env.0, self.raw_tsfn) })
} }
pub fn aborted(&self) -> bool {
self.aborted.load(Ordering::Acquire)
}
pub fn abort(self) -> Result<()> {
check_status(unsafe {
sys::napi_release_threadsafe_function(
self.raw_tsfn,
sys::napi_threadsafe_function_release_mode::napi_tsfn_abort,
)
})?;
self.aborted.store(true, Ordering::Release);
Ok(())
}
pub fn try_clone(&self) -> Result<Self> {
if self.aborted.load(Ordering::Acquire) {
return Err(Error::new(
Status::Closing,
format!("Thread safe function already aborted"),
));
}
check_status(unsafe { sys::napi_acquire_threadsafe_function(self.raw_tsfn) })?;
Ok(Self {
raw_tsfn: self.raw_tsfn,
aborted: Arc::clone(&self.aborted),
_phantom: PhantomData,
})
}
/// Get the raw `ThreadSafeFunction` pointer
pub fn raw(&self) -> sys::napi_threadsafe_function {
self.raw_tsfn
}
}
impl<T: 'static> Drop for ThreadsafeFunction<T> {
fn drop(&mut self) {
if !self.aborted.load(Ordering::Acquire) {
let release_status = unsafe {
sys::napi_release_threadsafe_function(
self.raw_tsfn,
sys::napi_threadsafe_function_release_mode::napi_tsfn_release,
)
};
assert!(
release_status == sys::Status::napi_ok,
"Threadsafe Function release failed"
);
}
}
} }
unsafe extern "C" fn thread_finalize_cb<T: 'static, V: NapiValue>( unsafe extern "C" fn thread_finalize_cb<T: 'static, V: NapiValue>(
@ -271,5 +286,5 @@ unsafe extern "C" fn call_js_cb<T: 'static, V: NapiValue>(
); );
} }
} }
debug_assert!(status == sys::napi_status::napi_ok, "CallJsCB failed"); debug_assert!(status == sys::Status::napi_ok, "CallJsCB failed");
} }

112
sys/src/lib.rs
View file

@ -68,71 +68,63 @@ pub enum napi_property_attributes {
napi_static = 1 << 10, napi_static = 1 << 10,
} }
#[repr(C)] pub type napi_valuetype = i32;
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
pub enum napi_valuetype { pub mod ValueType {
napi_undefined, pub const napi_undefined: i32 = 0;
napi_null, pub const napi_null: i32 = 1;
napi_boolean, pub const napi_boolean: i32 = 2;
napi_number, pub const napi_number: i32 = 3;
napi_string, pub const napi_string: i32 = 4;
napi_symbol, pub const napi_symbol: i32 = 5;
napi_object, pub const napi_object: i32 = 6;
napi_function, pub const napi_function: i32 = 7;
napi_external, pub const napi_external: i32 = 8;
#[cfg(feature = "napi6")] #[cfg(feature = "napi6")]
napi_bigint, pub const napi_bigint: i32 = 9;
} }
#[repr(C)] pub type napi_typedarray_type = i32;
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
pub enum napi_typedarray_type { pub mod TypedarrayType {
napi_int8_array, pub const napi_int8_array: i32 = 0;
napi_uint8_array, pub const napi_uint8_array: i32 = 1;
napi_uint8_clamped_array, pub const napi_uint8_clamped_array: i32 = 2;
napi_int16_array, pub const napi_int16_array: i32 = 3;
napi_uint16_array, pub const napi_uint16_array: i32 = 4;
napi_int32_array, pub const napi_int32_array: i32 = 5;
napi_uint32_array, pub const napi_uint32_array: i32 = 6;
napi_float32_array, pub const napi_float32_array: i32 = 7;
napi_float64_array, pub const napi_float64_array: i32 = 8;
#[cfg(feature = "napi6")] pub const napi_bigint64_array: i32 = 9;
napi_bigint64_array, pub const napi_biguint64_array: i32 = 10;
#[cfg(feature = "napi6")]
napi_biguint64_array,
} }
#[repr(C)] pub type napi_status = i32;
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
pub enum napi_status { pub mod Status {
napi_ok, pub const napi_ok: i32 = 0;
napi_invalid_arg, pub const napi_invalid_arg: i32 = 1;
napi_object_expected, pub const napi_object_expected: i32 = 2;
napi_string_expected, pub const napi_string_expected: i32 = 3;
napi_name_expected, pub const napi_name_expected: i32 = 4;
napi_function_expected, pub const napi_function_expected: i32 = 5;
napi_number_expected, pub const napi_number_expected: i32 = 6;
napi_boolean_expected, pub const napi_boolean_expected: i32 = 7;
napi_array_expected, pub const napi_array_expected: i32 = 8;
napi_generic_failure, pub const napi_generic_failure: i32 = 9;
napi_pending_exception, pub const napi_pending_exception: i32 = 10;
napi_cancelled, pub const napi_cancelled: i32 = 11;
napi_escape_called_twice, pub const napi_escape_called_twice: i32 = 12;
napi_handle_scope_mismatch, pub const napi_handle_scope_mismatch: i32 = 13;
napi_callback_scope_mismatch, pub const napi_callback_scope_mismatch: i32 = 14;
#[cfg(feature = "napi4")] pub const napi_queue_full: i32 = 15;
napi_queue_full, pub const napi_closing: i32 = 16;
#[cfg(feature = "napi4")] pub const napi_bigint_expected: i32 = 17;
napi_closing, pub const napi_date_expected: i32 = 18;
#[cfg(feature = "napi6")] pub const napi_arraybuffer_expected: i32 = 19;
napi_bigint_expected, pub const napi_detachable_arraybuffer_expected: i32 = 20;
#[cfg(feature = "napi6")] pub const napi_would_deadlock: i32 = 21; // unused
napi_date_expected,
#[cfg(feature = "napi7")]
napi_arraybuffer_expected,
#[cfg(feature = "napi7")]
napi_detachable_arraybuffer_expected,
napi_would_deadlock, // unused
} }
pub type napi_callback = ::std::option::Option< pub type napi_callback = ::std::option::Option<

30
test_module/__test__/napi4/threadsafe_function.spec.ts
View file

@ -16,7 +16,11 @@ test('should get js function called from a thread', async (t) => {
bindings.testThreadsafeFunction((...args: any[]) => { bindings.testThreadsafeFunction((...args: any[]) => {
called += 1 called += 1
try { try {
t.deepEqual(args, [null, 42, 1, 2, 3]) if (args[1] === 0) {
t.deepEqual(args, [null, 0, 1, 2, 3])
} else {
t.deepEqual(args, [null, 3, 2, 1, 0])
}
} catch (err) { } catch (err) {
reject(err) reject(err)
} }
@ -27,3 +31,27 @@ test('should get js function called from a thread', async (t) => {
}) })
}) })
}) })
test('should be able to abort tsfn', (t) => {
if (napiVersion < 4) {
t.is(bindings.testAbortThreadsafeFunction, undefined)
return
}
t.true(bindings.testAbortThreadsafeFunction(() => {}))
})
test('should be able to abort independent tsfn', (t) => {
if (napiVersion < 4) {
t.is(bindings.testAbortIndependentThreadsafeFunction, undefined)
return
}
t.false(bindings.testAbortIndependentThreadsafeFunction(() => {}))
})
test('should return Closing while calling aborted tsfn', (t) => {
if (napiVersion < 4) {
t.is(bindings.testCallAbortedThreadsafeFunction, undefined)
return
}
t.notThrows(() => bindings.testCallAbortedThreadsafeFunction(() => {}))
})

12
test_module/src/napi4/mod.rs
View file

@ -8,5 +8,17 @@ pub fn register_js(module: &mut Module) -> Result<()> {
module.create_named_method("testThreadsafeFunction", test_threadsafe_function)?; module.create_named_method("testThreadsafeFunction", test_threadsafe_function)?;
module.create_named_method("testTsfnError", test_tsfn_error)?; module.create_named_method("testTsfnError", test_tsfn_error)?;
module.create_named_method("testTokioReadfile", test_tokio_readfile)?; module.create_named_method("testTokioReadfile", test_tokio_readfile)?;
module.create_named_method(
"testAbortThreadsafeFunction",
test_abort_threadsafe_function,
)?;
module.create_named_method(
"testAbortIndependentThreadsafeFunction",
test_abort_independent_threadsafe_function,
)?;
module.create_named_method(
"testCallAbortedThreadsafeFunction",
test_call_aborted_threadsafe_function,
)?;
Ok(()) Ok(())
} }

90
test_module/src/napi4/tsfn.rs
View file

@ -2,10 +2,8 @@ use std::path::Path;
use std::thread; use std::thread;
use napi::{ use napi::{
threadsafe_function::{ threadsafe_function::{ThreadSafeCallContext, ThreadsafeFunctionCallMode},
ThreadSafeCallContext, ThreadsafeFunctionCallMode, ThreadsafeFunctionReleaseMode, CallContext, Error, JsBoolean, JsFunction, JsNumber, JsString, JsUndefined, Result, Status,
},
CallContext, Error, JsFunction, JsNumber, JsString, JsUndefined, Result, Status,
}; };
use tokio; use tokio;
@ -16,7 +14,7 @@ pub fn test_threadsafe_function(ctx: CallContext) -> Result<JsUndefined> {
let tsfn = let tsfn =
ctx ctx
.env .env
.create_threadsafe_function(func, 0, |ctx: ThreadSafeCallContext<Vec<u32>>| { .create_threadsafe_function(&func, 0, |ctx: ThreadSafeCallContext<Vec<u32>>| {
ctx ctx
.value .value
.iter() .iter()
@ -24,14 +22,80 @@ pub fn test_threadsafe_function(ctx: CallContext) -> Result<JsUndefined> {
.collect::<Result<Vec<JsNumber>>>() .collect::<Result<Vec<JsNumber>>>()
})?; })?;
let tsfn_cloned = tsfn.try_clone()?;
thread::spawn(move || { thread::spawn(move || {
let output: Vec<u32> = vec![42, 1, 2, 3]; let output: Vec<u32> = vec![0, 1, 2, 3];
// It's okay to call a threadsafe function multiple times. // It's okay to call a threadsafe function multiple times.
tsfn.call(Ok(output.clone()), ThreadsafeFunctionCallMode::Blocking); tsfn.call(Ok(output.clone()), ThreadsafeFunctionCallMode::Blocking);
tsfn.call(Ok(output.clone()), ThreadsafeFunctionCallMode::NonBlocking);
tsfn.release(ThreadsafeFunctionReleaseMode::Release);
}); });
thread::spawn(move || {
let output: Vec<u32> = vec![3, 2, 1, 0];
// It's okay to call a threadsafe function multiple times.
tsfn_cloned.call(Ok(output.clone()), ThreadsafeFunctionCallMode::NonBlocking);
});
ctx.env.get_undefined()
}
#[js_function(1)]
pub fn test_abort_threadsafe_function(ctx: CallContext) -> Result<JsBoolean> {
let func = ctx.get::<JsFunction>(0)?;
let tsfn =
ctx
.env
.create_threadsafe_function(&func, 0, |ctx: ThreadSafeCallContext<Vec<u32>>| {
ctx
.value
.iter()
.map(|v| ctx.env.create_uint32(*v))
.collect::<Result<Vec<JsNumber>>>()
})?;
let tsfn_cloned = tsfn.try_clone()?;
tsfn_cloned.abort()?;
ctx.env.get_boolean(tsfn.aborted())
}
#[js_function(1)]
pub fn test_abort_independent_threadsafe_function(ctx: CallContext) -> Result<JsBoolean> {
let func = ctx.get::<JsFunction>(0)?;
let tsfn = ctx
.env
.create_threadsafe_function(&func, 0, |ctx: ThreadSafeCallContext<u32>| {
ctx.env.create_uint32(ctx.value).map(|v| vec![v])
})?;
let tsfn_other =
ctx
.env
.create_threadsafe_function(&func, 0, |ctx: ThreadSafeCallContext<u32>| {
ctx.env.create_uint32(ctx.value).map(|v| vec![v])
})?;
tsfn_other.abort()?;
ctx.env.get_boolean(tsfn.aborted())
}
#[js_function(1)]
pub fn test_call_aborted_threadsafe_function(ctx: CallContext) -> Result<JsUndefined> {
let func = ctx.get::<JsFunction>(0)?;
let tsfn = ctx
.env
.create_threadsafe_function(&func, 0, |ctx: ThreadSafeCallContext<u32>| {
ctx.env.create_uint32(ctx.value).map(|v| vec![v])
})?;
let tsfn_clone = tsfn.try_clone()?;
tsfn_clone.abort()?;
let call_status = tsfn.call(Ok(1), ThreadsafeFunctionCallMode::NonBlocking);
assert!(call_status == Status::Closing);
ctx.env.get_undefined() ctx.env.get_undefined()
} }
@ -40,15 +104,14 @@ pub fn test_tsfn_error(ctx: CallContext) -> Result<JsUndefined> {
let func = ctx.get::<JsFunction>(0)?; let func = ctx.get::<JsFunction>(0)?;
let tsfn = ctx let tsfn = ctx
.env .env
.create_threadsafe_function(func, 0, |ctx: ThreadSafeCallContext<()>| { .create_threadsafe_function(&func, 0, |ctx: ThreadSafeCallContext<()>| {
ctx.env.get_undefined().map(|v| vec![v]) ctx.env.get_undefined().map(|v| vec![v])
})?; })?;
thread::spawn(move || { thread::spawn(move || {
tsfn.call( tsfn.call(
Err(Error::new(Status::Unknown, "invalid".to_owned())), Err(Error::new(Status::GenericFailure, "invalid".to_owned())),
ThreadsafeFunctionCallMode::Blocking, ThreadsafeFunctionCallMode::Blocking,
); );
tsfn.release(ThreadsafeFunctionReleaseMode::Release);
}); });
ctx.env.get_undefined() ctx.env.get_undefined()
@ -57,7 +120,7 @@ pub fn test_tsfn_error(ctx: CallContext) -> Result<JsUndefined> {
async fn read_file_content(filepath: &Path) -> Result<Vec<u8>> { async fn read_file_content(filepath: &Path) -> Result<Vec<u8>> {
tokio::fs::read(filepath) tokio::fs::read(filepath)
.await .await
.map_err(|e| Error::new(Status::Unknown, format!("{}", e))) .map_err(|e| Error::new(Status::GenericFailure, format!("{}", e)))
} }
#[js_function(2)] #[js_function(2)]
@ -69,7 +132,7 @@ pub fn test_tokio_readfile(ctx: CallContext) -> Result<JsUndefined> {
let tsfn = let tsfn =
ctx ctx
.env .env
.create_threadsafe_function(js_func, 0, |ctx: ThreadSafeCallContext<Vec<u8>>| { .create_threadsafe_function(&js_func, 0, |ctx: ThreadSafeCallContext<Vec<u8>>| {
ctx ctx
.env .env
.create_buffer_with_data(ctx.value) .create_buffer_with_data(ctx.value)
@ -81,7 +144,6 @@ pub fn test_tokio_readfile(ctx: CallContext) -> Result<JsUndefined> {
rt.block_on(async move { rt.block_on(async move {
let ret = read_file_content(&Path::new(&path_str)).await; let ret = read_file_content(&Path::new(&path_str)).await;
tsfn.call(ret, ThreadsafeFunctionCallMode::Blocking); tsfn.call(ret, ThreadsafeFunctionCallMode::Blocking);
tsfn.release(ThreadsafeFunctionReleaseMode::Release);
}); });
ctx.env.get_undefined() ctx.env.get_undefined()

12
test_module/src/tokio_rt/read_file.rs
View file

@ -7,8 +7,14 @@ pub fn test_execute_tokio_readfile(ctx: CallContext) -> Result<JsObject> {
let js_filepath = ctx.get::<JsString>(0)?; let js_filepath = ctx.get::<JsString>(0)?;
let path_str = js_filepath.into_utf8()?.to_owned()?; let path_str = js_filepath.into_utf8()?.to_owned()?;
ctx.env.execute_tokio_future( ctx.env.execute_tokio_future(
tokio::fs::read(path_str) tokio::fs::read(path_str).map(|v| {
.map(|v| v.map_err(|e| Error::new(Status::Unknown, format!("failed to read file, {}", e)))), v.map_err(|e| {
Error::new(
Status::GenericFailure,
format!("failed to read file, {}", e),
)
})
}),
|&mut env, data| env.create_buffer_with_data(data).map(|v| v.into_raw()), |&mut env, data| env.create_buffer_with_data(data).map(|v| v.into_raw()),
) )
} }
@ -22,7 +28,7 @@ pub fn error_from_tokio_future(ctx: CallContext) -> Result<JsObject> {
.map_err(Error::from) .map_err(Error::from)
.and_then(|_| async move { .and_then(|_| async move {
Err::<Vec<u8>, Error>(Error::new( Err::<Vec<u8>, Error>(Error::new(
Status::Unknown, Status::GenericFailure,
"Error from tokio future".to_owned(), "Error from tokio future".to_owned(),
)) ))
}), }),