Skip to content

ThreadsafeFunction

在 Node.js 中，Threadsafe Function 是一个复杂的概念。众所周知，Node.js 是单线程的，因此不能从其他线程访问 napi_env、napi_value 和 napi_ref。

TIP

napi_env、napi_value 和 napi_ref 是 Node-API 中的底层概念，NAPI-RS 的 #[napi] 宏构建在这些概念之上。 NAPI-RS 也提供了底层 API，用于访问原始 Node-API。

Node-API 提供了复杂的 Threadsafe Function API，用于从其他线程调用 JavaScript 函数。这套 API 非常复杂，许多开发者并不了解如何正确使用。NAPI-RS 提供了功能受限的 Threadsafe Function API，使其更易于使用：

lib.rs
rust
use std::{sync::Arc, thread};

use napi::{
    bindgen_prelude::*,
    threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode},
};
use napi_derive::napi;

#[napi]
pub fn call_threadsafe_function(callback: ThreadsafeFunction<u32, ()>) -> Result<()> { 
  let tsfn = Arc::new(callback);
  for n in 0..100 {
    let tsfn = tsfn.clone();
    thread::spawn(move || {
      tsfn.call(Ok(n), ThreadsafeFunctionCallMode::Blocking);
    });
  }
  Ok(())
}

⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️

index.d.ts
ts
export function callThreadsafeFunction(
  callback: (err: null | Error, result: number) => void,
): void

返回类型

ThreadsafeFunction 的返回类型与 JavaScript 回调的返回类型相同。可以在 ThreadsafeFunction 的第二个泛型参数中定义返回类型：

lib.rs
rust
use std::thread;

use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode};
use napi_derive::napi;

#[napi]
pub fn call_threadsafe_function(callback: ThreadsafeFunction<u32, u32>) {
  thread::spawn(move || {
    callback.call_with_return_value(Ok(1), ThreadsafeFunctionCallMode::Blocking, |ret, _| {
      println!("ret: {:?}", ret); // Ok(101)
      Ok(())
    });
  });
}
index.ts
ts
import { callThreadsafeFunction } from './index.js'

callThreadsafeFunction((err, result) => {
  return result + 100
})

CallJsBackArgs

有时，传给 ThreadsafeFunction 的参数与传给 JavaScript 回调的参数并不相同。可以使用 CallJsBackArgs 从 Function 构建 ThreadsafeFunction 来实现这一点：

lib.rs
rust
use std::thread;

use napi::{
  bindgen_prelude::*,
  threadsafe_function::{ThreadsafeCallContext, ThreadsafeFunctionCallMode},
};
use napi_derive::napi;

struct Data {
  name: String,
}

#[napi]
pub fn call_threadsafe_function(callback: Function<String, ()>) -> Result<()> {
  let tsfn = callback
    .build_threadsafe_function()
    .build_callback(|ctx: ThreadsafeCallContext<Data>| Ok(format!("Hello {}", ctx.value.name)))?; 
  thread::spawn(move || {
    tsfn.call(
      Data {
        name: "John".to_string(),
      },
      ThreadsafeFunctionCallMode::NonBlocking,
    );
  });
  Ok(())
}

WARNING

ThreadsafeFunction 存储的回调参数和返回类型必须满足 'static，因为导出的 Rust 函数返回后，回调仍可能运行。不要把 Unknown<'env>、Object<'env> 或 Function<'env, ...> 等作用域值用作 CallJsBackArgs。跨越线程边界之前， 请先转换为拥有所有权的 Rust 数据，例如 String、Buffer 或普通的拥有所有权的结构体。 在这里显式指定作用域生命周期会产生 E0521，因为借用的 JavaScript 值会逃逸出其回调作用域； 参阅 napi-rs#3383。

⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️

index.ts
ts
import { callThreadsafeFunction } from './index.js'

callThreadsafeFunction((data) => {
  console.log(data) // Hello John
})

错误状态

ThreadsafeFunction 的错误状态与 JavaScript 回调的错误状态相同。可以在 ThreadsafeFunction 的第四个泛型参数中定义错误状态：

lib.rs
rust
use std::{sync::Arc, thread};

use napi::{
  bindgen_prelude::*,
  threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode},
};
use napi_derive::napi;

pub struct CustomErrorStatus(String);

impl AsRef<str> for CustomErrorStatus {
  fn as_ref(&self) -> &str {
    &self.0
  }
}

impl From<Status> for CustomErrorStatus {
  fn from(value: Status) -> Self {
    CustomErrorStatus(value.to_string())
  }
}

#[napi]
pub fn call_threadsafe_function(
  tsfn: Arc<ThreadsafeFunction<u32, u32, u32, CustomErrorStatus>>, 
) -> Result<()> {
  for n in 0..100 {
    let tsfn = tsfn.clone();
    thread::spawn(move || {
      tsfn.call(
        Err(Error::new(
          CustomErrorStatus("Custom".to_owned()),
          format!("Custom error: {}", n),
        )),
        ThreadsafeFunctionCallMode::Blocking,
      );
    });
  }
  Ok(())
}

CalleeHandled 错误行为

Threadsafe Function 有两种错误处理策略。可以在 ThreadsafeFunction 的第五个泛型参数中定义该策略：

lib.rs
rust
let tsfn: ThreadsafeFunction<u32, u32, u32, Status, false> = ...

CalleeHandled: true（默认行为）

Rust 代码中的 Err 会传给 JavaScript 回调的第一个参数。这一行为遵循 Node.js 的异步回调约定：https://nodejs.org/en/learn/asynchronous-work/javascript-asynchronous-programming-and-callbacks#handling-errors-in-callbacks 。Node.js 中的许多异步 API 都采用这种设计，例如 fs.read。

启用 CalleeHandled: true 时，必须使用 Result 类型调用 ThreadsafeFunction，这样 Error 才会被处理并传回 JavaScript 回调：

lib.rs
rust
use std::{sync::Arc, thread};

use napi::{
  bindgen_prelude::*,
  threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode},
};
use napi_derive::napi;

#[napi]
pub fn call_threadsafe_function(
  tsfn: Arc<ThreadsafeFunction<u32, (), u32, Status, true>>, 
) -> Result<()> {
  for n in 0..100 {
    let tsfn = tsfn.clone();
    thread::spawn(move || {
      tsfn.call( 
        Err(Error::new( 
          Status::GenericFailure, 
          format!("Error with: {n}"), 
        )), 
        ThreadsafeFunctionCallMode::Blocking, 
      ); 
    });
  }
  Ok(())
}

⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️

index.ts
ts
import { callThreadsafeFunction } from './index.js'

callThreadsafeFunction((err, result) => {
  if (err) {
    console.error(err) // [Error: Error with: 0] { code: 'GenericFailure' }
  }
  console.log(result)
})

CalleeHandled: false

不会把 Error 传回 JavaScript 端。如果代码永远不会返回 Err，可以使用此策略，省去 Rust 端的 Ok 包装。

使用此策略时，调用 ThreadsafeFunction 无需传入 Result<T>，并且 JavaScript 回调的第一个参数是来自 Rust 的值，而不是 Error | null。

WARNING

使用 CalleeHandled: false 策略时，ThreadsafeFunction 无法处理 Rust 线程中的错误，因此无法把 Error 传回 JavaScript 端。

普通的 call 方法没有把错误传回 Rust 的通道。JavaScript 回调同步抛出的错误会通过 napi_fatal_exception 处理，而返回的 Promise 不会被自动等待。如果 Rust 需要回调结果， 请设置具体的 Return 类型并使用 call_async_catch，或者使用 call_with_return_value 并处理其完成回调收到的 Result。

只有在调用 call 前已经处理原生错误，并且 JavaScript 回调不会抛出错误时，才使用此模式。

lib.rs
rust
use std::{sync::Arc, thread};

use napi::{
  bindgen_prelude::*,
  threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode},
};
use napi_derive::napi;

#[napi]
pub fn call_threadsafe_function(
  tsfn: Arc<ThreadsafeFunction<u32, (), u32, Status, false>>, 
) -> Result<()> {
  for n in 0..100 {
    let tsfn = tsfn.clone();
    thread::spawn(move || {
      tsfn.call(n, ThreadsafeFunctionCallMode::Blocking);
    });
  }
  Ok(())
}

⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️ ⬇️

index.d.ts
ts
export declare function callThreadsafeFunction(
  tsfn: (arg: number) => void,
): void

Weak ThreadsafeFunction

默认情况下，ThreadsafeFunction 会让创建它的线程上的 event loop 保持存活，直至 ThreadsafeFunction 被销毁。参阅 决定是否让进程保持运行。

如果不想让 Node.js 进程/event loop 保持存活，可以把 ThreadsafeFunction 的 Weak 参数设为 true：

lib.rs
rust
use std::{sync::Arc, thread};

use napi::{
  bindgen_prelude::*,
  threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode},
};
use napi_derive::napi;

#[napi]
pub fn call_threadsafe_function(
  tsfn: Arc<ThreadsafeFunction<u32, (), u32, Status, false, true>>, 
) -> Result<()> {
  for n in 0..100 {
    let tsfn = tsfn.clone();
    thread::spawn(move || {
      tsfn.call(n, ThreadsafeFunctionCallMode::Blocking);
    });
  }
  Ok(())
}

如果像这样调用该函数：

index.ts
ts
import { callThreadsafeFunction } from './index.js'

// Weak 模式本身不会让 event loop 保持存活。
callThreadsafeFunction((n) => console.log(n))

如果没有其他工作让 event loop 保持存活，Node.js 可能会在部分或全部排队回调执行前退出。其他活动句柄或任务也可能让这些回调得以执行。Weak 模式既不保证回调一定送达，也不会禁止回调；它只是不再让这个 ThreadsafeFunction 成为保持 event loop 存活的原因。

MaxQueueSize

可以设置 ThreadsafeFunction 的 MaxQueueSize 参数，限制队列中的消息数量。

INFO

MaxQueueSize 在两种调用模式中都会设置队列容量。达到容量时，Blocking 模式会等待空位；NonBlocking 模式会立即返回 Status::QueueFull。更多细节请参阅 napi_call_threadsafe_function。

lib.rs
rust
use std::{sync::Arc, thread};

use napi::{
  bindgen_prelude::*,
  threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode},
};
use napi_derive::napi;

#[napi]
pub fn call_threadsafe_function(
  tsfn: Arc<ThreadsafeFunction<u32, (), u32, Status, false, false, 1>>, 
) -> Result<()> {
  thread::spawn(move || {
    for n in 0..100 {
      let tsfn = tsfn.clone();
      let status = tsfn.call(n, ThreadsafeFunctionCallMode::NonBlocking); 
      println!("{}", status)
    }
  });
  Ok(())
}

调用该函数并在回调中加入繁重工作时，会看到 tsfn.call 返回 QueueFull 状态：

index.ts
ts
import { callThreadsafeFunction } from './index.js'

function fib(n: number): number {
  if (n <= 1) return n
  return fib(n - 1) + fib(n - 2)
}

callThreadsafeFunction(() => {
  fib(40)
})

某次运行可能产生如下输出：

Ok
Ok
QueueFull
QueueFull
QueueFull
QueueFull
QueueFull
QueueFull
QueueFull
QueueFull
...

Ok 与 QueueFull 的确切数量和顺序取决于 JavaScript 线程何时相对于生产线程清空队列。容量为一只保证背压语义，不保证固定输出序列。

在 GitHub 上编辑本页 →