@thi.ng/fibers

[!NOTE] This is one of 192 standalone projects, maintained as part of the @thi.ng/umbrella monorepo and anti-framework.

🚀 Please help me to work full-time on these projects by sponsoring me on GitHub. Thank you! ❤️

• About
  • Basic usage
  • Fiber operators
  • Composition via transducers
  • CSP primitives (Communicating Sequential Processes)
    • Buffering behaviors
    • Channels
    • CSP ping/pong example
• Status
• Related packages
• Installation
• Dependencies
• Usage examples
• API
• Authors
• License

About

Process hierarchies & operators for cooperative multitasking.

This package provides a Fiber primitive acting as wrapper around ES6 generators (co-routines) and supplements them with nested child processes, cancellation, error & event handling and (optional) logging features. Additionally, there're a number of fiber-related utilities and higher order operators to construct and compose fibers.

Basic usage

import { fiber, wait } from "@thi.ng/fibers";
import { ConsoleLogger } from "@thi.ng/logger";

// wrap an endless generator as fiber
const app = fiber(
    function* () {
        while(true) {
            console.log("hello");
            // wait 0.25s
            yield* wait(250);
            console.log("fiber");
            // wait 1s
            yield* wait(1000);
        }
    }
);

// start processing it with default handlers
// see: https://docs.thi.ng/umbrella/fibers/classes/Fiber.html#run
app.run();

// create a child process which runs semi-independently
// (only executes if parent is still active)
// child fibers are auto-removed from parent when they terminate
const child = app.fork(
    // the `ctx` arg is the fiber wrapping this generator
    function* (ctx) {
        for(let i = 0; i < 3; i++) {
            ctx.logger?.debug("count", i);
            yield* wait(100);
        }
        // return value will be stored in fiber for future reference
        return 42;
    },
    // fiber options
    {
        // custom fiber ID (else autogenerated)
        id: "child-demo",
        // custom logger (default: none)
        logger: new ConsoleLogger("child")
    }
);

// hello
// [DEBUG] child: init child-demo
// [DEBUG] child: count 0
// [DEBUG] child: count 1
// [DEBUG] child: count 2
// fiber
// [DEBUG] child: done child-demo 42
// [DEBUG] child: deinit child-demo
// hello
// fiber
// hello
// ...

// once a fiber has completed, its value can be obtained
// e.g. here we create another fiber, which first waits for `child` to complete
app.fork(function* () {
    // wait for other fiber
    const result = yield* child;
    console.log("result", result);
    // alt way to obtain value
    console.log("deref", child.deref());
});
// result 42
// deref 42

Fiber operators

The following operators act as basic composition helpers to construct more elaborate fiber setups:

• all: wait for all given fibers to complete
• asPromise: wrap fiber as promise for use in async contexts
• first: wait for one of the given fibers to complete
• fork: create & attach a new child process
• forkAll: create & attach multiple child processes
• join: wait for all child processes to complete
• sequence: execute fibers in sequence
• shuffle: execute fibers in constantly randomized order
• timeSlice: execute fiber in batches of N milliseconds
• timeSliceIterable: consume iterable in batches of N milliseconds
• until: wait until predicate is truthy
• untilEvent: wait until event occurs
• untilPromise: wait until promise resolves/rejects
• untilState: stateful version of until
• wait: wait for N milliseconds or indefinitely
• waitFrames: wait for N frames/ticks
• withTimeout: wait for given fiber, but only max N milliseconds

Composition via transducers

The @thi.ng/transducers package can be very helpful to create complex fiber setups, for example:

import { sequence, wait, type MaybeFiber } from "@thi.ng/fibers";
import {
    cycle,
    interpose,
    map,
    partition,
    repeatedly,
} from "@thi.ng/transducers";

const defWorkItem = (id: number) =>
    function* () {
        console.log("part", id);
    };

const defWorkGroup = (items: MaybeFiber[]) =>
    function* () {
        // interject a short pause between given work items
        // then execute in order and wait until all done
        yield* sequence(interpose(() => wait(100), items));
        console.log("---");
        yield* wait(1000);
    };

// create fiber which executes given sub-processes in order
sequence(
    // generate 25 work items
    // partition into groups of 5
    // transform into iterable of work groups
    // repeat indefinitely
    cycle(map(defWorkGroup, partition(5, repeatedly(defWorkItem, 25))))
).run();

// part 0
// part 1
// part 2
// part 3
// part 4
// ---
// part 5
// part 6
// part 7

CSP primitives (Communicating Sequential Processes)

References:

• Wikipedia
• Communicating Sequential Processes, C.A.R. Hoare

In addition to the operators above, the basic fiber implementation can also be used to construct other types of primitives, like those required for channel-based communication between processes, as proposed by Tony Hoare. The package includes a fiber-based read/write channel primitive which can be customized with different buffer implementations to control blocking behaviors and backpressure handling (aka attempting to write faster to a channel than values are being read, essentially a memory management issue).

Buffering behaviors

The following channel buffer types/behaviors are included (from the thi.ng/buffers package), all accepting a max. capacity and all implementing the IReadWriteBuffer interface required by the channel:

• fifo: First in, first out ring buffer. Writes to the channel will start blocking once the buffer's capacity is reached, otherwise complete immediately. Likewise, channel reads are non-blocking whilst there're more buffered values available. Reads will only block if the buffer is empty.
• lifo: Last in, first out. Write behavior is the same as with fifo, reads are in reverse order (as the name indicates), i.e. the last value written will be the first value read (i.e. stack behavior).
• sliding: Sliding window ring buffer. Writes to the channel are never blocking! Whilst the buffer is at full capacity, new writes will first expunge the oldest buffered value (similar to LRU cache behavior). Read behavior is the same as for fifo.
• dropping: Dropping value ring buffer. Writes to the channel are never blocking! Whilst the buffer is at full capacity, new writes will be silently ignored. Read behavior is the same as for fifo.

Channels

As mentioned previously, channels and their read, write and close operations are the key building blocks for CSP. In this fiber-based implementation, all channel operations are executed in individual fibers to deal with the potential blocking behaviors. This is demonstrated in the simple example below.

In general, due to fibers not being true multi-threaded processes (all are executed in the single thread of the JS engine), any number of fibers can read or write to a channel.

Channels can be created like so:

import { channel, sliding } from "@thi.ng/fibers";
import { ConsoleLogger } from "@thi.ng/logger";

// create unbuffered channel with single value capacity
const chan1 = channel();

// create channel with a FIFO buffer, capacity: 2 values
const chan2 = channel(2);

// create channel with a sliding window buffer and custom ID & logger
const chan3 = channel(
  sliding(3),
  { id: "main", logger: new ConsoleLogger("chan") }
);

CSP ping/pong example

import { channel, fiber, wait } from "@thi.ng/fibers";
import { ConsoleLogger } from "@thi.ng/logger";

// create idle main fiber with custom options
const app = fiber(null, {
    id: "main",
    logger: new ConsoleLogger("app"),
    // if true, fiber automatically terminates once all child fibers are done
    terminate: true,
});

// create CSP channels (w/ default config)
const ping = channel<number>();
const pong = channel<number>();

// attach ping/pong child processes
app.forkAll(
    // ping
    function* () {
        while (ping.readable()) {
            // blocking read op
            // (waits until value is available in `ping` channel)
            const x = yield* ping.read();
            // check if channel was closed meanwhile
            if (x === undefined) break;
            console.log("PING", x);
            // possibly blocking (in general) write op to other channel
            yield* pong.write(x);
            // slowdown
            yield* wait(100);
        }
    },
    // pong (very similar)
    function* () {
        while (pong.readable()) {
            const x = yield* pong.read();
            if (x === undefined) break;
            console.log("PONG", x);
            // trigger next iteration
            yield* ping.write(x + 1);
        }
    },
    // channel managment
    function* () {
        // kickoff ping/pong
        yield* ping.write(0);
        yield* wait(1000);
        // wait for both channels to close
        yield* ping.close();
        yield* pong.close();
    }
);
app.run();

// [DEBUG] app: forking fib-0
// [DEBUG] app: forking fib-1
// [DEBUG] app: forking fib-2
// [DEBUG] app: running main...
// [DEBUG] app: init main
// [DEBUG] app: init fib-0
// [DEBUG] app: init fib-1
// [DEBUG] app: init fib-2
// PING 0
// PONG 0
// PING 1
// PONG 1
// ...
// PING 9
// PONG 9
// [DEBUG] app: done fib-2 undefined
// [DEBUG] app: deinit fib-2
// [DEBUG] app: done fib-1 undefined
// [DEBUG] app: deinit fib-1
// [DEBUG] app: done fib-0 undefined
// [DEBUG] app: deinit fib-0
// [DEBUG] app: cancel main
// [DEBUG] app: deinit main

Additional CSP operators are planned, but since everything here is based on fibers, the various channel operations can be already combined with the available fiber operators/combinators...

For example, a channel read or write op can be combined with a timeout:

import { withTimeout } from "@thi.ng/fibers";

// ...then, inside a fiber function...

const res = (yield* withTimeout(chan.read(), 1000)).deref();
if (res !== undefined) {
    console.log("read value", x);
} else {
    console.log("read timeout");
}

Status

ALPHA - bleeding edge / work-in-progress

Search or submit any issues for this package

Related packages

• @thi.ng/csp - Primitives & operators for Communicating Sequential Processes based on async/await and async iterables
• @thi.ng/transducers-async - Async versions of various highly composable transducers, reducers and iterators

Installation

yarn add @thi.ng/fibers

ESM import:

import * as fib from "@thi.ng/fibers";

Browser ESM import:

<script type="module" src="https://esm.run/@thi.ng/fibers"></script>

JSDelivr documentation

For Node.js REPL:

const fib = await import("@thi.ng/fibers");

Package sizes (brotli'd, pre-treeshake): ESM: 2.45 KB

Dependencies

• @thi.ng/api
• @thi.ng/arrays
• @thi.ng/bench
• @thi.ng/buffers
• @thi.ng/checks
• @thi.ng/errors
• @thi.ng/idgen
• @thi.ng/logger
• @thi.ng/random

Usage examples

Several projects in this repo's /examples directory are using this package:

Screenshot	Description	Live demo	Source
	ASCII art raymarching with thi.ng/shader-ast & thi.ng/text-canvas	Demo	Source
	Fiber-based cooperative multitasking basics	Demo	Source
	Barnsley fern IFS fractal renderer	Demo	Source
	Mastodon API feed reader with support for different media types, fullscreen media modal, HTML rewriting	Demo	Source
	Animated, iterative polygon subdivisions & visualization	Demo	Source
	Responsive image gallery with tag-based Jaccard similarity ranking	Demo	Source
	Generative audio synth offline renderer and WAV file export	Demo	Source

API

Generated API docs

TODO

Authors

• Karsten Schmidt

If this project contributes to an academic publication, please cite it as:

@misc{thing-fibers,
  title = "@thi.ng/fibers",
  author = "Karsten Schmidt",
  note = "https://thi.ng/fibers",
  year = 2023
}

License

© 2023 - 2024 Karsten Schmidt // Apache License 2.0