!@thi.ng/fibers

!npm downloads

> [!NOTE]
> This is one of 214 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.

$3

``ts tangle:export/readme-1.ts
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
`

$3

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

$3

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

`ts tangle:export/readme-2.ts
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
`

$3

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:

`ts
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

`ts tangle:export/pingpong.ts
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();
const pong = channel();

// 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:

`ts
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

`bash
yarn add @thi.ng/fibers
`

ESM import:

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

Browser ESM import:

`html

`

JSDelivr documentation

For Node.js REPL:

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

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

Dependencies

- @thi.ng/api
- @thi.ng/arrays
- @thi.ng/buffers
- @thi.ng/checks
- @thi.ng/errors
- @thi.ng/idgen
- @thi.ng/logger
- @thi.ng/random
- @thi.ng/timestamp

Note: @thi.ng/api is in _most_ cases a type-only import (not used at runtime)

Usage examples

Eight 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 |
| | Iterating the unique edges of a tessellation | 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:

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

License

© 2023 - 2026 Karsten Schmidt // Apache License 2.0