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

chore(napi): reduce Mutex usage while loading addon

Browse source
This commit is contained in:
LongYinan 2022-12-15 23:24:22 +08:00
parent 6ad1740dac
commit a9f9dbb232
No known key found for this signature in database
GPG key ID: C3666B7FC82ADAD7
10 changed files with 123 additions and 276 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
crates/napi/src/bindgen_runtime/js_values/arraybuffer.rs
View file

@ -7,8 +7,6 @@ use std::sync::{
Arc,
};
#[cfg(feature = "napi4")]
use crate::bindgen_prelude::{CUSTOM_GC_TSFN, CUSTOM_GC_TSFN_CLOSED, MAIN_THREAD_ID};
pub use crate::js_values::TypedArrayType;
use crate::{check_status, sys, Error, Result, Status};
@ -66,31 +64,6 @@ macro_rules! impl_typed_array {
fn drop(&mut self) {
if Arc::strong_count(&self.drop_in_vm) == 1 {
if let Some((ref_, env)) = self.raw {
#[cfg(feature = "napi4")]
{
if CUSTOM_GC_TSFN_CLOSED.load(std::sync::atomic::Ordering::SeqCst) {
return;
}
if !MAIN_THREAD_ID
.get()
.map(|id| &std::thread::current().id() == id)
.unwrap_or(false)
{
let status = unsafe {
sys::napi_call_threadsafe_function(
CUSTOM_GC_TSFN.load(std::sync::atomic::Ordering::SeqCst),
ref_ as *mut c_void,
1,
)
};
assert!(
status == sys::Status::napi_ok,
"Call custom GC in ArrayBuffer::drop failed {:?}",
Status::from(status)
);
return;
}
}
crate::check_status_or_throw!(
env,
unsafe { sys::napi_reference_unref(env, ref_, &mut 0) },

33
crates/napi/src/bindgen_runtime/js_values/buffer.rs
View file

@ -1,4 +1,4 @@
#[cfg(debug_assertions)]
#[cfg(all(debug_assertions, not(windows)))]
use std::collections::HashSet;
use std::ffi::c_void;
use std::mem;
@ -6,12 +6,12 @@ use std::ops::{Deref, DerefMut};
use std::ptr::{self, NonNull};
use std::slice;
use std::sync::Arc;
#[cfg(debug_assertions)]
#[cfg(all(debug_assertions, not(windows)))]
use std::sync::Mutex;
use crate::{bindgen_prelude::*, check_status, sys, Result, ValueType};
#[cfg(debug_assertions)]
#[cfg(all(debug_assertions, not(windows)))]
thread_local! {
pub (crate) static BUFFER_DATA: Mutex<HashSet<*mut u8>> = Default::default();
}
@ -32,31 +32,6 @@ impl Drop for Buffer {
fn drop(&mut self) {
if Arc::strong_count(&self.ref_count) == 1 {
if let Some((ref_, env)) = self.raw {
#[cfg(feature = "napi4")]
{
if CUSTOM_GC_TSFN_CLOSED.load(std::sync::atomic::Ordering::SeqCst) {
return;
}
if !MAIN_THREAD_ID
.get()
.map(|id| &std::thread::current().id() == id)
.unwrap_or(false)
{
let status = unsafe {
sys::napi_call_threadsafe_function(
CUSTOM_GC_TSFN.load(std::sync::atomic::Ordering::SeqCst),
ref_ as *mut c_void,
1,
)
};
assert!(
status == sys::Status::napi_ok,
"Call custom GC in Buffer::drop failed {:?}",
Status::from(status)
);
return;
}
}
let mut ref_count = 0;
check_status_or_throw!(
env,
@ -94,7 +69,7 @@ impl Clone for Buffer {
impl From<Vec<u8>> for Buffer {
fn from(mut data: Vec<u8>) -> Self {
let inner_ptr = data.as_mut_ptr();
#[cfg(debug_assertions)]
#[cfg(all(debug_assertions, not(windows)))]
{
let is_existed = BUFFER_DATA.with(|buffer_data| {
let buffer = buffer_data.lock().expect("Unlock buffer data failed");

2
crates/napi/src/bindgen_runtime/js_values/promise.rs
View file

@ -77,7 +77,7 @@ impl<T: FromNapiValue> ValidateNapiValue for Promise<T> {
}
}
unsafe impl<T: FromNapiValue> Send for Promise<T> {}
unsafe impl<T: FromNapiValue + Send> Send for Promise<T> {}
impl<T: FromNapiValue> FromNapiValue for Promise<T> {
unsafe fn from_napi_value(env: sys::napi_env, napi_val: sys::napi_value) -> crate::Result<Self> {

11
crates/napi/src/bindgen_runtime/js_values/value_ref.rs
View file

@ -6,7 +6,7 @@ use std::rc::{Rc, Weak};
use once_cell::sync::Lazy;
use crate::{
bindgen_runtime::{PersistedSingleThreadHashMap, ToNapiValue},
bindgen_runtime::{PersistedPerInstanceHashMap, ToNapiValue},
check_status, Env, Error, Result, Status,
};
@ -16,9 +16,16 @@ type RefInformation = (
*const Cell<*mut dyn FnOnce()>,
);
pub(crate) static REFERENCE_MAP: Lazy<PersistedSingleThreadHashMap<*mut c_void, RefInformation>> =
pub(crate) static REFERENCE_MAP: Lazy<PersistedPerInstanceHashMap<*mut c_void, RefInformation>> =
Lazy::new(Default::default);
#[ctor::dtor]
fn de_init() {
REFERENCE_MAP.borrow_mut(|reference_map| {
std::mem::take(reference_map);
});
}
/// ### Experimental feature
///
/// Create a `reference` from `Class` instance.

2
crates/napi/src/bindgen_runtime/mod.rs
View file

@ -76,7 +76,7 @@ pub unsafe extern "C" fn drop_buffer(
#[allow(unused)] finalize_data: *mut c_void,
finalize_hint: *mut c_void,
) {
#[cfg(debug_assertions)]
#[cfg(all(debug_assertions, not(windows)))]
{
js_values::BUFFER_DATA.with(|buffer_data| {
let mut buffer = buffer_data.lock().expect("Unlock Buffer data failed");

273
crates/napi/src/bindgen_runtime/module_register.rs
View file

@ -1,21 +1,10 @@
use std::collections::{HashMap, HashSet};
#[cfg(feature = "napi4")]
use std::ffi::c_void;
use std::ffi::CStr;
use std::mem;
#[cfg(feature = "napi4")]
use std::os::raw::c_char;
use std::ffi::{c_void, CStr};
use std::ptr;
use std::sync::{
atomic::{AtomicBool, AtomicPtr, Ordering},
Mutex,
};
#[cfg(feature = "napi4")]
use std::thread::ThreadId;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::{AtomicBool, AtomicPtr, Ordering};
use once_cell::sync::Lazy;
#[cfg(feature = "napi4")]
use once_cell::sync::OnceCell;
use crate::{
check_status, check_status_or_throw, sys, Env, JsError, JsFunction, Property, Result, Value,
@ -26,97 +15,118 @@ pub type ExportRegisterCallback = unsafe fn(sys::napi_env) -> Result<sys::napi_v
pub type ModuleExportsCallback =
unsafe fn(env: sys::napi_env, exports: sys::napi_value) -> Result<()>;
#[cfg(feature = "napi4")]
pub(crate) static CUSTOM_GC_TSFN: AtomicPtr<sys::napi_threadsafe_function__> =
AtomicPtr::new(ptr::null_mut());
#[cfg(feature = "napi4")]
// CustomGC ThreadsafeFunction may be deleted during the process exit.
// And there may still some Buffer alive after that.
pub(crate) static CUSTOM_GC_TSFN_CLOSED: AtomicBool = AtomicBool::new(false);
#[cfg(feature = "napi4")]
pub(crate) static MAIN_THREAD_ID: OnceCell<ThreadId> = OnceCell::new();
struct PersistedSingleThreadVec<T> {
inner: Mutex<Vec<T>>,
struct PersistedPerInstanceVec<T> {
inner: AtomicPtr<T>,
length: AtomicUsize,
}
impl<T> Default for PersistedSingleThreadVec<T> {
impl<T> Default for PersistedPerInstanceVec<T> {
fn default() -> Self {
PersistedSingleThreadVec {
inner: Mutex::new(Vec::new()),
}
let mut vec: Vec<T> = Vec::with_capacity(1);
let ret = Self {
inner: AtomicPtr::new(vec.as_mut_ptr()),
length: AtomicUsize::new(0),
};
std::mem::forget(vec);
ret
}
}
impl<T> PersistedSingleThreadVec<T> {
impl<T> PersistedPerInstanceVec<T> {
#[allow(clippy::mut_from_ref)]
fn borrow_mut<F>(&self, f: F)
where
F: FnOnce(&mut [T]),
{
let mut locked = self
.inner
.lock()
.expect("Acquire persisted thread vec lock failed");
f(&mut locked);
let length = self.length.load(Ordering::Relaxed);
if length == 0 {
f(&mut []);
} else {
let inner = self.inner.load(Ordering::Relaxed);
let mut temp = unsafe { Vec::from_raw_parts(inner, length, length) };
f(temp.as_mut_slice());
// Inner Vec has been reallocated, so we need to update the pointer
if temp.as_mut_ptr() != inner {
self.inner.store(temp.as_mut_ptr(), Ordering::Relaxed);
}
self.length.store(temp.len(), Ordering::Relaxed);
std::mem::forget(temp);
}
}
fn push(&self, item: T) {
let mut locked = self
.inner
.lock()
.expect("Acquire persisted thread vec lock failed");
locked.push(item);
let length = self.length.load(Ordering::Relaxed);
let inner = self.inner.load(Ordering::Relaxed);
let mut temp = unsafe { Vec::from_raw_parts(inner, length, length) };
temp.push(item);
// Inner Vec has been reallocated, so we need to update the pointer
if temp.as_mut_ptr() != inner {
self.inner.store(temp.as_mut_ptr(), Ordering::Relaxed);
}
std::mem::forget(temp);
self.length.fetch_add(1, Ordering::Relaxed);
}
}
unsafe impl<T> Send for PersistedSingleThreadVec<T> {}
unsafe impl<T> Sync for PersistedSingleThreadVec<T> {}
unsafe impl<T: Send> Send for PersistedPerInstanceVec<T> {}
unsafe impl<T: Sync> Sync for PersistedPerInstanceVec<T> {}
pub(crate) struct PersistedSingleThreadHashMap<K, V>(Mutex<HashMap<K, V>>);
pub(crate) struct PersistedPerInstanceHashMap<K, V>(*mut HashMap<K, V>);
impl<K, V> PersistedSingleThreadHashMap<K, V> {
impl<K, V> PersistedPerInstanceHashMap<K, V> {
#[allow(clippy::mut_from_ref)]
pub(crate) fn borrow_mut<F, R>(&self, f: F) -> R
where
F: FnOnce(&mut HashMap<K, V>) -> R,
{
let mut lock = self
.0
.lock()
.expect("Acquire persisted thread hash map lock failed");
f(&mut *lock)
f(unsafe { Box::leak(Box::from_raw(self.0)) })
}
}
impl<K, V> Default for PersistedSingleThreadHashMap<K, V> {
impl<K, V> Default for PersistedPerInstanceHashMap<K, V> {
fn default() -> Self {
PersistedSingleThreadHashMap(Mutex::new(Default::default()))
let map = Default::default();
Self(Box::into_raw(Box::new(map)))
}
}
type ModuleRegisterCallback =
PersistedSingleThreadVec<(Option<&'static str>, (&'static str, ExportRegisterCallback))>;
PersistedPerInstanceVec<(Option<&'static str>, (&'static str, ExportRegisterCallback))>;
type ModuleClassProperty = PersistedSingleThreadHashMap<
type ModuleClassProperty = PersistedPerInstanceHashMap<
&'static str,
HashMap<Option<&'static str>, (&'static str, Vec<Property>)>,
>;
unsafe impl<K, V> Send for PersistedSingleThreadHashMap<K, V> {}
unsafe impl<K, V> Sync for PersistedSingleThreadHashMap<K, V> {}
unsafe impl<K, V> Send for PersistedPerInstanceHashMap<K, V> {}
unsafe impl<K, V> Sync for PersistedPerInstanceHashMap<K, V> {}
type FnRegisterMap =
PersistedSingleThreadHashMap<ExportRegisterCallback, (sys::napi_callback, &'static str)>;
PersistedPerInstanceHashMap<ExportRegisterCallback, (sys::napi_callback, &'static str)>;
static MODULE_REGISTER_CALLBACK: Lazy<ModuleRegisterCallback> = Lazy::new(Default::default);
static MODULE_CLASS_PROPERTIES: Lazy<ModuleClassProperty> = Lazy::new(Default::default);
static MODULE_REGISTER_LOCK: Lazy<Mutex<()>> = Lazy::new(|| Mutex::new(()));
static REGISTERED: Lazy<AtomicBool> = Lazy::new(|| AtomicBool::new(false));
static REGISTERED: AtomicBool = AtomicBool::new(false);
static REGISTERED_CLASSES: Lazy<thread_local::ThreadLocal<AtomicPtr<RegisteredClasses>>> =
Lazy::new(thread_local::ThreadLocal::new);
static FN_REGISTER_MAP: Lazy<FnRegisterMap> = Lazy::new(Default::default);
#[ctor::dtor]
fn destroy() {
{
let ptr = MODULE_REGISTER_CALLBACK.inner.load(Ordering::Relaxed);
let len = MODULE_REGISTER_CALLBACK.length.load(Ordering::Relaxed);
unsafe { Vec::from_raw_parts(ptr, len, len) };
}
{
unsafe { Box::from_raw(MODULE_CLASS_PROPERTIES.0) };
}
{
unsafe { Box::from_raw(FN_REGISTER_MAP.0) };
}
}
#[inline]
fn wait_first_thread_registered() {
while !REGISTERED.load(Ordering::SeqCst) {
@ -130,7 +140,7 @@ type RegisteredClasses =
#[cfg(feature = "compat-mode")]
// compatibility for #[module_exports]
static MODULE_EXPORTS: Lazy<PersistedSingleThreadVec<ModuleExportsCallback>> =
static MODULE_EXPORTS: Lazy<PersistedPerInstanceVec<ModuleExportsCallback>> =
Lazy::new(Default::default);
#[doc(hidden)]
@ -271,20 +281,21 @@ pub fn get_c_callback(raw_fn: ExportRegisterCallback) -> Result<crate::Callback>
})
}
#[cfg(windows)]
#[ctor::ctor]
fn load_host() {
unsafe {
sys::setup();
}
}
#[no_mangle]
unsafe extern "C" fn napi_register_module_v1(
env: sys::napi_env,
exports: sys::napi_value,
) -> sys::napi_value {
#[cfg(windows)]
unsafe {
sys::setup();
}
crate::__private::___CALL_FROM_FACTORY.get_or_default();
let registered_classes_ptr = REGISTERED_CLASSES.get_or_default();
let lock = MODULE_REGISTER_LOCK
.lock()
.expect("Failed to acquire module register lock");
let mut exports_objects: HashSet<String> = HashSet::default();
MODULE_REGISTER_CALLBACK.borrow_mut(|inner| {
inner
@ -462,29 +473,27 @@ unsafe extern "C" fn napi_register_module_v1(
})
});
#[cfg(all(feature = "tokio_rt", feature = "napi4"))]
#[cfg(feature = "napi3")]
{
let _ = crate::tokio_runtime::RT.clone();
crate::tokio_runtime::TOKIO_RT_REF_COUNT.fetch_add(1, Ordering::SeqCst);
assert_eq!(
unsafe {
sys::napi_add_env_cleanup_hook(
env,
Some(crate::shutdown_tokio_rt),
env as *mut std::ffi::c_void,
)
},
sys::Status::napi_ok
);
unsafe {
sys::napi_add_env_cleanup_hook(env, Some(remove_registered_classes), env as *mut c_void)
};
}
mem::drop(lock);
#[cfg(feature = "napi4")]
create_custom_gc(env);
REGISTERED.store(true, Ordering::SeqCst);
exports
}
unsafe extern "C" fn remove_registered_classes(env: *mut c_void) {
let env = env as sys::napi_env;
if let Some(s) = REGISTERED_CLASSES.get() {
let registered_classes = unsafe { Box::from_raw(s.load(Ordering::Relaxed)) };
registered_classes.iter().for_each(|(_, v)| {
unsafe { sys::napi_delete_reference(env, *v) };
});
s.store(ptr::null_mut(), Ordering::Relaxed);
}
}
pub(crate) unsafe extern "C" fn noop(
env: sys::napi_env,
_info: sys::napi_callback_info,
@ -502,99 +511,3 @@ pub(crate) unsafe extern "C" fn noop(
}
ptr::null_mut()
}
#[cfg(feature = "napi4")]
fn create_custom_gc(env: sys::napi_env) {
let mut custom_gc_fn = ptr::null_mut();
check_status_or_throw!(
env,
unsafe {
sys::napi_create_function(
env,
"custom_gc".as_ptr() as *const c_char,
9,
Some(empty),
ptr::null_mut(),
&mut custom_gc_fn,
)
},
"Create Custom GC Function in napi_register_module_v1 failed"
);
let mut async_resource_name = ptr::null_mut();
check_status_or_throw!(
env,
unsafe {
sys::napi_create_string_utf8(
env,
"CustomGC".as_ptr() as *const c_char,
8,
&mut async_resource_name,
)
},
"Create async resource string in napi_register_module_v1 napi_register_module_v1"
);
let mut custom_gc_tsfn = ptr::null_mut();
check_status_or_throw!(
env,
unsafe {
sys::napi_create_threadsafe_function(
env,
custom_gc_fn,
ptr::null_mut(),
async_resource_name,
0,
1,
ptr::null_mut(),
Some(custom_gc_finalize),
ptr::null_mut(),
Some(custom_gc),
&mut custom_gc_tsfn,
)
},
"Create Custom GC ThreadsafeFunction in napi_register_module_v1 failed"
);
check_status_or_throw!(
env,
unsafe { sys::napi_unref_threadsafe_function(env, custom_gc_tsfn) },
"Unref Custom GC ThreadsafeFunction in napi_register_module_v1 failed"
);
CUSTOM_GC_TSFN.store(custom_gc_tsfn, Ordering::SeqCst);
MAIN_THREAD_ID.get_or_init(|| std::thread::current().id());
}
#[cfg(feature = "napi4")]
#[allow(unused)]
unsafe extern "C" fn empty(env: sys::napi_env, info: sys::napi_callback_info) -> sys::napi_value {
ptr::null_mut()
}
#[cfg(feature = "napi4")]
#[allow(unused)]
unsafe extern "C" fn custom_gc_finalize(
env: sys::napi_env,
finalize_data: *mut c_void,
finalize_hint: *mut c_void,
) {
CUSTOM_GC_TSFN_CLOSED.store(true, Ordering::SeqCst);
}
#[cfg(feature = "napi4")]
// recycle the ArrayBuffer/Buffer Reference if the ArrayBuffer/Buffer is not dropped on the main thread
extern "C" fn custom_gc(
env: sys::napi_env,
_js_callback: sys::napi_value,
_context: *mut c_void,
data: *mut c_void,
) {
let mut ref_count = 0;
check_status_or_throw!(
env,
unsafe { sys::napi_reference_unref(env, data as sys::napi_ref, &mut ref_count) },
"Failed to unref Buffer reference in Custom GC"
);
check_status_or_throw!(
env,
unsafe { sys::napi_delete_reference(env, data as sys::napi_ref) },
"Failed to delete Buffer reference in Custom GC"
);
}

2
crates/napi/src/lib.rs
View file

@ -109,8 +109,6 @@ pub use error::*;
pub use js_values::*;
pub use status::Status;
pub use task::Task;
#[cfg(all(feature = "tokio_rt", feature = "napi4"))]
pub use tokio_runtime::shutdown_tokio_rt;
pub use value_type::*;
pub use version::NodeVersion;
#[cfg(feature = "serde-json")]

28
crates/napi/src/tokio_runtime.rs
View file

@ -1,7 +1,5 @@
use std::ffi::c_void;
use std::future::Future;
use std::ptr;
use std::sync::atomic::{AtomicUsize, Ordering};
use once_cell::sync::Lazy;
use tokio::{
@ -29,27 +27,17 @@ pub(crate) static RT: Lazy<(Handle, mpsc::Sender<()>)> = Lazy::new(|| {
.expect("Create tokio runtime failed")
});
pub(crate) static TOKIO_RT_REF_COUNT: AtomicUsize = AtomicUsize::new(0);
#[doc(hidden)]
#[inline(never)]
pub unsafe extern "C" fn shutdown_tokio_rt(arg: *mut c_void) {
if TOKIO_RT_REF_COUNT.fetch_sub(1, Ordering::SeqCst) == 0 {
let sender = &RT.1;
if let Err(e) = sender.clone().try_send(()) {
match e {
TrySendError::Closed(_) => {}
TrySendError::Full(_) => {
panic!("Send shutdown signal to tokio runtime failed, queue is full");
}
#[ctor::dtor]
fn shutdown_tokio() {
let sender = &RT.1;
if let Err(e) = sender.clone().try_send(()) {
match e {
TrySendError::Closed(_) => {}
TrySendError::Full(_) => {
panic!("Send shutdown signal to tokio runtime failed, queue is full");
}
}
}
unsafe {
let env: sys::napi_env = arg as *mut sys::napi_env__;
sys::napi_remove_env_cleanup_hook(env, Some(shutdown_tokio_rt), arg);
}
}
/// Spawns a future onto the Tokio runtime.

4
crates/sys/src/functions.rs
View file

@ -738,7 +738,7 @@ pub use napi7::*;
pub use napi8::*;
#[cfg(windows)]
pub(super) unsafe fn load() -> Result<(), libloading::Error> {
pub(super) unsafe fn load() -> Result<libloading::Library, libloading::Error> {
let host = match libloading::os::windows::Library::this() {
Ok(lib) => lib.into(),
Err(err) => {
@ -764,5 +764,5 @@ pub(super) unsafe fn load() -> Result<(), libloading::Error> {
napi8::load(&host)?;
#[cfg(feature = "experimental")]
experimental::load(&host)?;
Ok(())
Ok(host)
}

17
crates/sys/src/lib.rs
View file

@ -77,28 +77,21 @@ macro_rules! generate {
};
}
#[cfg(windows)]
use std::sync::Once;
mod functions;
mod types;
pub use functions::*;
pub use types::*;
#[cfg(windows)]
static SETUP: Once = Once::new();
/// Loads N-API symbols from host process.
/// Must be called at least once before using any functions in bindings or
/// they will panic.
/// Safety: `env` must be a valid `napi_env` for the current thread
#[cfg(windows)]
#[allow(clippy::missing_safety_doc)]
pub unsafe fn setup() {
SETUP.call_once(|| {
if let Err(err) = load() {
panic!("{}", err);
}
});
pub unsafe fn setup() -> libloading::Library {
match load() {
Err(err) => panic!("{}", err),
Ok(l) => l,
}
}