Skip to content

Latest commit

 

History

History
366 lines (305 loc) · 11.5 KB

tpl.readme.md

File metadata and controls

366 lines (305 loc) · 11.5 KB
Raw

About

{{pkg.description}}

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:

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");
}

{{meta.status}}

{{repo.supportPackages}}

{{repo.relatedPackages}}

{{meta.blogPosts}}

Installation

{{pkg.install}}

{{pkg.size}}

Dependencies

{{pkg.deps}}

{{repo.examples}}

API

{{pkg.docs}}

TODO