worker-f

worker-f

Runs a function in a separate thread in a web browser or Node.js.

Demo

Why

• The code is universal and could be run in any environment: in a web browser or Node.js. The easy-to-use API hides the complexity of dealing with each particular environment.

• Doesn't require moving code to a separate file for it to be able to run in a worker. This prevents introducing the unnecessary and redundant concept of being able to access a "filesystem" or having to run a "web server". The code must not concern itself with such things that're completely irrelevant to its purpose. It shouldn't even know that a concept of a "file path" exists. Running a function in a separate thread should be as simple as it is in other programming languages.

• Because it removes the unnecessary requirement of being able to access a "filesystem" or running a "web server", it becomes possible to use workers in "libraries" ("packages"), not just end-user applications. For example, fflate uses workers to utilize multiple CPU cores (and to prevent freezing of the page) when zipping or unzipping large archives.

• Why use workers at all? In a web browser, it's about not freezing the app while doing "heavy computation". In Node.js, it's about not freezing the server while doing "heavy computation". And while freezing an app in a web browser is somewhat bearable, the main appeal of Node.js from the point of its creation has been that it doesn't "fork" the process for each incoming HTTP request, outperforming "classic" web servers by using an "event loop" instead, which quickly backstabs if that "event loop" is accidentally blocked by a "heavy computation".

Install

npm install worker-f

Alternatively, it could be included on a web page directly via a <script/> tag.

Use

Basic usage:

const workerFn = workerFunction((a, b) => a + b)

await workerFn.callOnce(1, 2) === 3

Advanced usage scenarios like "calling multiple times", "streaming", "caching", "transferring data" are described further in this document.

Import

This package provides a separate import path for each different environment, as described below.

Browser

import workerFunction from 'worker-f/browser'

const workerFn = workerFunction((a, b) => a + b)

Node

import workerFunction from 'worker-f/node'

const workerFn = workerFunction((a, b) => a + b)

API

Call Once

Use this when the function will only be called once. Attempting to call the function second time will throw an error.

The function could be synchronous or asynchronous.

import workerFunction from 'worker-f/node'

const workerFn = workerFunction((a, b) => a + b)

await workerFn.callOnce(1, 2) === 3

Call

When the function will be called multiple times, it should be started, then called as many times as needed, then stopped.

import workerFunction from 'worker-f/node'

const workerFn = workerFunction((a, b) => a + b)

workerFn.start()

await workerFn.call(1, 2) === 3
await workerFn.call(4, 5) === 9

workerFn.stop()

If the function rejects or throws an error, it will automatically stop.

If a developer forgets to stop a worker function that is no longer used, it will still stop automatically when the code no longer holds any "reference" to it. It will also stop automatically when the web browser tab is closed, or the Node.js process is killed. But until stopped, it will keep holding its memory.

Stream

If a function is designed to produce multiple outputs over time, it should use "stream" API rather than "call" API.

To use "stream" API, add /stream postfix to the import path.

import workerFunctionStream, { type Send } from 'worker-f/node/stream' // or 'worker-f/browser/stream'

const workerFn = workerFunctionStream(
  // Use the supplied `send()` function to "stream" output data
  // from the worker function to the main thread.
  // If you're using TypeScript, then you must declare the `send()` function's type
  // to be the exported `Send` type with a "generic" that describes the argument
  // that you will be passing to the `send()` function.
  (send: Send<[number, number]>) => {
    // This code will only be run once on `workerFn.start()`
    // and any variables declared here will exist until `workerFn.stop()`,
    // so this is good place to declare any of the function's state variables.
    let c = 0
    // Return an "input handler" function.
    // The "input handler" function is supposed to process the input in some way
    // and then (optionally) use the `send()` function to output some result
    // back to the main thread. Here's an example of an "input handler" function
    // that calculates a sum of two numbers and sends it back to the main thread.
    return ([a, b]: [number, number]) => {
      // Just an example of updating the function's state — a simple counter.
      c++
      // Calculate a sum of the two input numbers and send the result back
      // to the main thread along with the counter.
      send([a + b, c])
    }
  }
)

// Stores the outputs of the function
const sums = []

// How many outputs it still expects from the function
let pending = 0

// Calling `.onOutput()` sets up a listener that will handle any output from the function.
// Calling `.onOutput()` multiple times will just overwrite the listener.
workerFn.onOutput(([sum, c]) => {
  sums[c - 1] = sum
  pending--
  // When all outputs have been received
  if (pending === 0) {
    // Validate the end result
    sums.length === 3
    sums[0] === 3
    sums[1] === 7
    sums[1] === 11
    // Stop the worker function
    workerFn.stop()
  }
})

// (optional)
//
// Calling `.onError()` sets up a listener that will intercept any errors
// that're thrown from the worker function.
//
// Calling `.onError()` multiple times will just overwrite the error listener.
//
// Even if an error is intercepted, the worker function will still automatically stop.
// By default, if no `.onError()` listener is added, any error thrown from a worker function
// will just propagate to the main thread as an "unhandled rejection", crashing the app.
//
workerFn.onError((error) => {
  console.log('Worker function exited with an error', error)
})

workerFn.start()

pending++
workerFn.send([1, 2])

pending++
workerFn.send([3, 4])

pending++
workerFn.send([5, 6])

External Dependencies

An "isolated" function only uses the stuff that is defined within its body, and it never references anything from outside of its body. This is basically a requirement for any function that is to be extrated to a separate (worker) thread.

But in practice that's not always the case. More likely, a worker function is going to reference some variables or classes or other functions that're declared outside of its body. In that case, those variables or classes or functions must be specified as the worker function's "dependencies", or else it would throw a ReferenceError: <name> is not defined.

// External variable
const c = 3
// External function
const d = () => 4

// This worker function references `c` and `d` which are outside of its body
const workerFn = workerFunction((a, b) => a + b + c + d())

// Without declaring `c` and `d` as "dependencies",
// it throws: "ReferenceError: c is not defined"
await workerFn.callOnce(1, 2)

// How to fix the error:
workerFn.addDependencies(() => [c, d])
// Now it works
await workerFn.callOnce(1, 2) === 10

Any external dependencies that're not functions (or classes) must be clonable, i.e. they can't be arrays of functions (or classes) or JSON objects with values that're functions (or classes), etc. Otherwise, it would throw a DataCloneError.

Any external dependencies that're functions (or classes) must either be "isolated", or specify their own external dependencies in an .addDependencies(...) call, in which case the loop continues until the very last sub-sub-sub-dependency function (or class) is finally "isolated".

// External function that is "isolated" because it doesn't reference anything outside of its body
const d = () => 4

// External function that references `d` which is outside of its body
const c = () => 3 + d()

// This worker function references `c` which is outside of its body
const workerFn = workerFunction((a, b) => a + b + c())

// Specifying just `c` is not enough because `d` is also an external sub-dependency
workerFn.addDependencies(() => [c])
// throws: "ReferenceError: d is not defined"
await workerFn.callOnce(1, 2)

// How to fix the error:
workerFn.addDependencies(() => [c, d])
// Now it works
await workerFn.callOnce(1, 2) === 10

To reduce the number of external dependencies, one could put everyting into a single large "isolated" function that doesn't reference anything outside of its body. Or if it does reference anything outside of its body, those references themselves must be "isolated" functions (or classes) that don't reference anything outside of their body, etc.

With that in mind, one could see how specifying all the dependencies correctly could become a tedious task in a typical modular application where the code is spread over countless smaller modules, each of them importing other smaller modules, etc. So a better approach would be to just move everything — the function itself and most of its dependencies — into a single big "wrapper" function, as if we're back in 2000s, and then create a worker from it.

// A "wrapper" function that has the same arguments as the original function
export function fn_(a, b) {
  // Any dependencies are put right here, inside the wrapper function body
  const d = () => 4
  const c = () => 3 + d()

  // The original function is also put here
  const fn = (a, b) => a + b + c()

  // Call the original function with the arguments
  return fn(a, b)
}

// Create a worker from the "wrapper" function.
// No need to specify any external dependencies because there're none. Simple.
const workerFn = workerFunction(fn_)

Needless to say that after a worker function has been started, none of its dependencies should change because those changes won't be reflected inside the worker function's thread, i.e. it won't "see" any changes.

Caching

Every time a new worker function is created, it has to stringify the function body in order to generate the worker's code. If the application plans on creating many workers from same function, and that function has no external dependencies or those dependencies are constant, then it would make sense to only generate the function's source code once and then reuse it every time a new worker is created from this function.

To "cache" a function's source code for creating future workers, call .alias() method on the worker function.

const c = () => 1
const sum = (a, b) => a + b + c()

const sumFn = workerFunction(sum)
sumFn.addDependencies(() => [c])

// Calling `.alias()` creates a snapshot of this worker function.
// After assigning an alias, one could instantiate this type of worker function from the snapshot.
// The snapshot will include both the `sum` function body and any of its dependencies.
//
// Creating a new worker function from an alias will be slightly faster than from a function body
// because it won't have to redo the stringification of the function body and any of its dependencies.
// Does it really matter performance-wise? I didn't bother checking.
//
sumFn.alias('sum')

// (optional)
await sumFn.callOnce(2, 3) === 6

// Create a new worker function by the alias.
const sumFn2 = workerFunction<Args, Result>('sum')
await sumFn2.callOnce(4, 5) === 10

// Create a new worker function by the alias.
const sumFn3 = workerFunction<Args, Result>('sum')
await sumFn3.callOnce(6, 7) === 14

Performance

By default, any input passed to a worker function is cloned behind the scenes. And same goes for any output.

Because of how seamless the "cloning" is, developers don't even have to bother knowing that it takes place.

Yet, in some situations, the data being passed between the main thread and the worker thread might become large-enough to justify tinkering with potential optimization.

How large is "large-enough"?

• For JSON objects, the deeper the object is, the more costly it is to "serialize" and "deserialize" it back. There're some benchmarks from 2019 where it shows how "serializing"/"deserializing" a 10 MB JSON object with 6 levels of nesting is about 50 ms on a desktop or 100 ms on a phone.

• For ArrayBuffers, "cloning" is said to be "incredibly quick" without any further details.

So the short answer is: "I personally don't really know or care". The rule of thumb is to keep the data being sent between the main thread and the worker thread to a minimum.

To assess input/output "cloning" performance, every worker function exposes two properties — inputLatency and outputLatency (in milliseconds).

const workerFn = workerFunction((a, b) => a + b)

workerFn.start()

await workerFn.call(1, 2) === 3

// `inputLatency` is how long it took to send numbers `1` and `2` to the worker thread
console.log(workerFn.inputLatency)

// `outputLatency` is how long it took to send number `3` to the main thread
console.log(workerFn.outputLatency)

await workerFn.call(3, 4) === 7

// `inputLatency` is how long it took to send numbers `3` and `4` to the worker thread
console.log(workerFn.inputLatency)

// `outputLatency` is how long it took to send number `7` to the main thread
console.log(workerFn.outputLatency)

workerFn.stop()

import workerFunctionStream, { type Send } from 'worker-f/node/stream'

// Sums input numbers.
const workerFn = workerFunctionStream(
  (send: Send<number>) => {
    // Input numbers will be accumulated here.
    let numbers = []
    // A worker thread could send some output to the main thread
    // even before in receives any input from the main thread.
    send(0)
    // Calculates a sum of accumulated numbers.
    return (number: number) => {
      // Append this number to the list of accumulated numbers.
      numbers.push(number)
      // As soon as it has accumulated 3 numbers.
      if (numbers.length === 3) {
        // Calculate a sum of the 3 numbers and send the value to the main thread.
        send(numbers.reduce((sum, number) => sum + number))
        numbers = []
      } else {
        // Not enough numbers accumulated yet.
        // This is an example of how there might not necessarily be an output for a given input.
      }
    }
  }
)

workerFn.onOutput((sum) => {
  // `inputLatency` is the "latency" of the latest input
  // `outputLatency` is the "latency" of the latest output
  //
  // Specifically:
  //
  // * The first received output is `0`
  //   * `inputLatency` is `undefined`
  //   * `outputLatency` is the "latency" of this output
  // * The second received output is `6`
  //   * `inputLatency` is the "latency" of the `3` input
  //   * `outputLatency` is the "latency" of this output
  //
  console.log(workerFn.inputLatency)
  console.log(workerFn.outputLatency)
})

workerFn.start()

// An output of `0` will be received shortly after the function has started.

// Send the first number to the worker — no output.
workerFn.send(1)

// Send the second number to the worker — no output.
workerFn.send(2)

// Send the third number to the worker — outputs `1 + 2 + 3 === 6`
workerFn.send(3)

// (later) When the application is being closed
workerFn.stop()

Transfer

When passing ArrayBuffers, there's an optional feature called "transfer". When it "transfers" a buffer, it doesn't clone it, but instead it simply "transfers" the ownership of the buffer from the "main thread" to the "worker thread", and vice versa, which is a "free" operation. Although note that after a buffer has been "transferred", it's no longer usable in the code that "transferred" it.

To enable "transfer" for certain input/output buffers, call inputTransferList() / outputTransferList() methods on a worker function.

// A worker function with some input and output.
const parser = workerFunction((arrayBuffer, dataType) => {
  // ... Parse the data from the buffer according to the data type ...
  return data
})

// Pass a function that returns a `transferList` for the input of the worker function.
// By default, an empty `transferList` is used for the input of the worker function.
parser.inputTransferList((arrayBuffer, dataType) => [arrayBuffer])

// Pass a function that returns a `transferList` for the output of the worker function.
// By default, an empty `transferList` is used for the output of the worker function.
parser.outputTransferList((data) => [])

// Now, when passing an `arrayBuffer` to the worker function,
// it will be "transferred" from the main thread to the worker thread.
const data = await parser.callOnce(arrayBuffer, 'document')

Errors

Any errors thrown within a worker function will "bubble" to the main thread:

• When using "call" API, if the function rejects or throws an error, the returned Promise will be rejected and the worker function will automatically stop.

• When using "stream" API, the optional .onError(listener) listener will be called and the worker function will automatically stop.

But there's a catch: an error that "bubbles" from a worker function to the main thread will have its class rewritten with generic Error, and it will no longer be an instance of the original error class. This is because any error has to be "serialized" and then "de-serialized" in order to be sent from the worker thread to the main thread. So a "bubbled" error will still retain all the properties of the original error, including name, but it will no longer be an instance of the original error class. Consider this when your application detects error type using instanceof operator, and perhaps resort to comparing error.name property instead.

Also, if instead of throwing an error, a worker function sends it to the main thread as a usual value, then the error will neither retain its original class nor any of its properties: the class will be reset to generic Error, name will be rewritten with "Error", and any other properties will just be discarded.

Transpilers

This package can't be used with any transpilers such as Babel. This is because they insert their own helper functions at the top level of the transpiled code, so those helper functions become undeclared external dependencies that aren't possible to be declared beforehand in the original source code. As a result, when running such transpiled code, it throws an error: <helper-function-name> is not defined.

CDN

To include this library directly via a <script/> tag on a page, one can use any npm CDN service, e.g. unpkg.com or jsdelivr.com

<script src="https://unpkg.com/worker-f@0.1.x/bundle/worker-f.min.js"></script>

<script src="https://unpkg.com/worker-f@0.1.x/bundle/worker-f-stream.min.js"></script>

<script>
  const workerFn = workerFunction((a, b) => a + b)

  const workerFnStream = workerFunctionStream((send) => {
    return (a, b) => {
      send(a + b)
    }
  })
</script>

Development

npm install
npm test

It uses vitest to run unit tests, which also comes with a bug on Windows when it doesn't know how to properly handle lowercase drive letter. The error message is Error: Vitest failed to find the current suite. One of the following is possible. The workaround is to cd into same directory but with an uppercase drive letter.