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 for a given input, or to produce output without any input at all, then it should use "stream" API rather than "call" API.

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

import createWorkerStream, { type Emit } from 'worker-f/node/stream' // or 'worker-f/browser/stream'

// In this example, we create an "adder" stream that receives two numbers
// and outputs their sum, along with a supplementary "counter" number.
type Input = [number, number]
type Output = [number, number]

// Create an "adder" stream.
const workerStream = createWorkerStream(
  // Anything that happens inside these parentheses happens in a worker thread.
  // In other words, you cannot reference anything outside of these parentheses
  // because these parentheses are an impenetrable barrier between two parallel worlds —
  // the world of the main thread and the world of a worker thread.
  // These two worlds can communicate by passing input or output across the barrier,
  // but they won't ever "see" each other.
  //
  // So here we define the code that runs in isolation in a worker thread.
  // The only things you've got here are: any kind of input that the main thread
  // decides to send to this worker stream, and a reference to an `emit()` function
  // that can be used to send any kind of output to the main thread. That's it.
  //
  // If you're using TypeScript, then you must also pre-define the type of data
  // that you'll be emitting using the `emit()` function.
  //
  (emit: Emit<Output>) => {
    // The code inside this function will only be run once when you call `workerStream.start()`,
    // and any variables declared here will exist until you call `workerStream.stop()`,
    // so this is a good place to declare any "state" variables.
    //
    // For example, here we declare a "counter" variable.
    // It will count the number of "sum" operations it did so far.
    //
    let count = 0

    // In the end, we return an "input handler" function.
    // This function should (optionally) process any input in some kind of way
    // and then (optionally) use the `emit()` function to output some kind of result
    // back to the main thread.
    //
    // In this example, the "input handler" function receives two numbers as an input,
    // calculates their sum, and outputs it to the main thread, along with the count
    // of "sum" oprations it did so far.
    //
    return ([a, b]: Input) => {
      // Increase the number of "sum" oprations it did so far.
      count++
      // Calculate a sum of the two input numbers and send the result back
      // to the main thread, along with the number of "sum" oprations it did so far.
      emit([a + b, count])
    }
  }
)

// Here, we're back in the main thread.

// Stores the results of adding two numbers.
const sums = []

// How many outputs it still expects from the "adder" stream.
let pendingOperations = 0

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

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

workerStream.start()

pendingOperations++
workerStream.push([1, 2])

pendingOperations++
workerStream.push([3, 4])

pendingOperations++
workerStream.push([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 you create a new worker function, it has to stringify the function's body, and all of its dependencies, in order to generate the worker's code. If you plan on creating many instances of same worker function (with same dependencies), you may assign it an "alias" by calling the .alias() method on the worker function. That will cache the worker's code under an alias, so that next time you create a new worker function using that alias, it could simply reuse the already-generated worker's code.

Performance-wise, will it even be noticeable? I suppose not, so perhaps you shouldn't even bother.

Because aliasing a function caches all of its dependencies, it is only viable for functions that either have no external dependencies or those dependencies are always identical no matter how or where you create a new instance of this worker function.

// `c` is a "global constant" that doesn't ever change.
const c = () => 1

// `sum` is a function that adds two numbers.
const sum = (a, b) => a + b + c()

// Define a worker function.
const sumFn = workerFunction(sum)
sumFn.addDependencies(() => [c])

// Call `.alias()` to create a snapshot of the worker function.
// Later, one could create a new instance of this worker function from this snapshot using the alias.
// The snapshot will include both the `sum` function body and any of its dependencies.
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` back 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` back to the main thread
console.log(workerFn.outputLatency)

workerFn.stop()

import createWorkerStream, { type Emit } from 'worker-f/node/stream'

// Sums input numbers.
const workerStream = createWorkerStream(
  (emit: Emit<number>) => {
    // Input numbers will be accumulated here.
    let numbers = []
    // A worker stream could send some output to the main thread
    // even before in receives any input from the main thread.
    emit(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.
        emit(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.
      }
    }
  }
)

workerStream.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(workerStream.inputLatency)
  console.log(workerStream.outputLatency)
})

workerStream.start()

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

// Send the first number to the worker — no output.
workerStream.push(1)

// Send the second number to the worker — no output.
workerStream.push(2)

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

// (later) When the application is being closed
workerStream.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.

Content Security Policy

There're some websites that choose to define a strict Content Security Policy, preventing the page from accessing non-pre-approved URLs. If your website does that, then you might get an error:

Content-Security-Policy: The page's settings blocked a worker script (worker-src) at blob:… because it violates the following directive: "…"

To fix it, you'll have to explicitly allow "blob": URLs in your Content-Security-Policy header:

Content-Security-Policy: worker-src 'self' blob:

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 workerStream = createWorkerStream((emit) => {
    return ([a, b]) => {
      emit(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.