ThreadsafeFunction

Threadsafe Function is a complex concept in Node.js. As we all know, Node.js is single-threaded, so you can't access napi_env, napi_value, and napi_ref on another thread.

TIP

napi_env, napi_value, and napi_ref are low level concepts in Node-API, which the #[napi] macro of NAPI-RS is built on top of. NAPI-RS also provides a low level API to access the original Node-API.

Node-API provides complex Threadsafe Function APIs to call JavaScript functions on other threads. It's very complex, so many developers don't understand how to use it correctly. NAPI-RS provides a limited version of Threadsafe Function APIs to make it easier to use:

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

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

Return type

The return type of the ThreadsafeFunction is the same as the return type of the JavaScript callback, you can define the return type in the second generic parameter of 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

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

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

CallJsBackArgs

Sometimes the args passed to the ThreadsafeFunction are not the same as the args passed to the JavaScript callback. You can build the ThreadsafeFunction from Function with the CallJsBackArgs to achieve this:

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

The callback argument and return types stored by a ThreadsafeFunction must be 'static, because the callback can run after the exported Rust function has returned. Do not use scoped values such as Unknown<'env>, Object<'env>, or Function<'env, ...> as CallJsBackArgs. Convert to owned Rust data such as String, Buffer, or a plain owned struct before crossing the thread boundary. An explicit scoped lifetime here produces E0521 because the borrowed JavaScript value would escape its callback scope; see napi-rs#3383.

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

index.ts

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

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

Error Status

The error status of the ThreadsafeFunction is the same as the error status of the JavaScript callback. You can define the error status in the fourth generic parameter of 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` error behavior

There are two different error-handling strategies for Threadsafe Function. The strategy can be defined in the fifth generic parameter of ThreadsafeFunction:

lib.rs

rust

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

`CalleeHandled: true` (default behavior)

Err from Rust code will be passed into the first argument of the JavaScript callback. This behavior follows the async callback conventions from Node.js: https://nodejs.org/en/learn/asynchronous-work/javascript-asynchronous-programming-and-callbacks#handling-errors-in-callbacks. Many async APIs in Node.js are designed this way, like fs.read.

With CalleeHandled: true, you must call the ThreadsafeFunction with the Result type, so that the Error will be handled and passed back to the JavaScript callback:

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

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

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

`CalleeHandled: false`

No Error will be passed back to the JavaScript side. You can use this strategy to avoid the Ok wrapping on the Rust side if your code will never return Err.

With this strategy, ThreadsafeFunction doesn't need to be called with Result<T>, and the first argument of JavaScript callback is the value from the Rust, not Error | null.

WARNING

With the CalleeHandled: false strategy, the ThreadsafeFunction will not be able to handle the error in the Rust threads, so you can't pass the Error back to the JavaScript side.

The plain call method has no error channel back to Rust. A synchronous throw in the JavaScript callback is routed through napi_fatal_exception, and a returned Promise is not awaited automatically. If Rust needs the callback's result, set a concrete Return type and use call_async_catch, or use call_with_return_value and handle the Result passed to its completion callback.

Use this mode only when native failures are handled before call and the JavaScript callback cannot throw.

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

export declare function callThreadsafeFunction(
  tsfn: (arg: number) => void,
): void

`Weak` ThreadsafeFunction

By default, the ThreadsafeFunction will cause the event loop on the thread on which it is created to remain alive until the ThreadsafeFunction is destroyed. See Deciding whether to keep the process running.

If you don't want to keep the Node.js process/event loop alive, you can define the Weak parameter of ThreadsafeFunction to 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(())
}

If you call this function like this:

index.ts

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

// Weak mode does not keep the event loop alive by itself.
callThreadsafeFunction((n) => console.log(n))

If nothing else keeps the event loop alive, Node.js may exit before some or all queued callbacks run. Other active handles or work can keep the process alive long enough to deliver them. Weak mode does not guarantee callback delivery or suppress callbacks; it only removes this ThreadsafeFunction as a reason to keep the loop alive.

`MaxQueueSize`

You can set the MaxQueueSize parameter of ThreadsafeFunction to limit the number of messages in the queue.

INFO

MaxQueueSize sets the queue capacity in both call modes. When that capacity is reached, Blocking waits for space; NonBlocking returns immediately with Status::QueueFull. See napi_call_threadsafe_function for more details.

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(())
}

When you call this function, and add heavy work in the callback, you will see the QueueFull status return from the tsfn.call:

index.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)
})

An illustrative run might produce output like this:

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

The exact number and order of Ok and QueueFull results depends on when the JavaScript thread drains the queue relative to the producer thread. A capacity of one guarantees the backpressure behavior, not a fixed output sequence.

# ThreadsafeFunction

# Return type

# CallJsBackArgs

# Error Status

# CalleeHandled error behavior

# CalleeHandled: true (default behavior)

# CalleeHandled: false

# Weak ThreadsafeFunction

# MaxQueueSize

ThreadsafeFunction

Return type

CallJsBackArgs

Error Status

`CalleeHandled` error behavior

`CalleeHandled: true` (default behavior)

`CalleeHandled: false`

`Weak` ThreadsafeFunction

`MaxQueueSize`