feat(fibers): add shuffle() operator, update deps

thi-ng · Aug 9, 2023 · b3efa79 · b3efa79
1 parent e59063d
commit b3efa79
Show file tree

Hide file tree

Showing 3 changed files with 74 additions and 1 deletion.
diff --git a/packages/fibers/package.json b/packages/fibers/package.json
@@ -35,11 +35,13 @@
 	},
 	"dependencies": {
 		"@thi.ng/api": "^8.9.0",
+		"@thi.ng/arrays": "^2.5.15",
 		"@thi.ng/bench": "^3.4.1",
 		"@thi.ng/checks": "^3.4.0",
 		"@thi.ng/errors": "^2.3.0",
 		"@thi.ng/idgen": "^2.2.0",
-		"@thi.ng/logger": "^1.4.16"
+		"@thi.ng/logger": "^1.4.16",
+		"@thi.ng/random": "^3.5.1"
 	},
 	"devDependencies": {
 		"@microsoft/api-extractor": "^7.36.3",

diff --git a/packages/fibers/src/ops.ts b/packages/fibers/src/ops.ts
@@ -1,12 +1,16 @@
 import type { Fn0, Nullable, Predicate } from "@thi.ng/api";
+import { shuffle as $shuffle } from "@thi.ng/arrays/shuffle";
 import { now, timeDiff } from "@thi.ng/bench/now";
+import type { IRandom } from "@thi.ng/random";
+import { SYSTEM } from "@thi.ng/random/system";
 import {
 	STATE_ACTIVE,
 	STATE_DONE,
 	STATE_ERROR,
 	type FiberFactory,
 	type FiberOpts,
 	type MaybeFiber,
+	type State,
 } from "./api.js";
 import { Fiber, fiber } from "./fiber.js";
 
@@ -234,3 +238,68 @@ export const untilEvent = (
 		},
 	});
 };
+
+/**
+ * Custom fiber implementation for {@link shuffle}.
+ */
+export class Shuffle extends Fiber {
+	rnd: IRandom;
+
+	constructor(
+		fibers: Iterable<MaybeFiber>,
+		opts?: Partial<FiberOpts & { rnd: IRandom }>
+	) {
+		super((ctx) => ctx.join(), opts);
+		this.rnd = opts?.rnd || SYSTEM;
+		this.forkAll(...fibers);
+	}
+
+	next(): State {
+		if (!this.isActive()) return this.state;
+		$shuffle(this.children!, this.children!.length, this.rnd);
+		return super.next();
+	}
+}
+
+/**
+ * Higher-order fiber for creating a constantly randomized execution order of
+ * given `fibers`, e.g. for distributing workloads. Creates and returns a new
+ * fiber as parent of the given `fibers` which then shuffles their execution
+ * order on each {@link Fiber.next} invocation/update. The fiber terminates when
+ * all children are done.
+ *
+ * @remarks
+ * The `rnd` option can be used to customize the
+ * [`IRandom`](https://docs.thi.ng/umbrella/random/interfaces/IRandom.html)
+ * implementation used for shuffling. Defaults to
+ * [`SYSTEM`](https://docs.thi.ng/umbrella/random/variables/SYSTEM.html).
+ *
+ * @example
+ * ```ts
+ * import { repeatedly } from "@thi.ng/transducers";
+ *
+ * // create & run fiber with 4 children, executing in random order
+ * shuffle(
+ *   repeatedly(
+ *     (id) => function*() { while(true) { console.log(`worker #{id}`); yield; } },
+ *     4
+ *   )
+ * ).run()
+ *
+ * // worker #0
+ * // worker #1
+ * // worker #3
+ * // worker #3
+ * // worker #2
+ * // worker #0
+ * // worker #2
+ * // ...
+ * ```
+ *
+ * @param fibers
+ * @param opts
+ */
+export const shuffle = (
+	fibers: Iterable<MaybeFiber>,
+	opts?: Partial<FiberOpts & { rnd: IRandom }>
+) => new Shuffle(fibers, opts);
diff --git a/yarn.lock b/yarn.lock
@@ -3253,11 +3253,13 @@ __metadata:
   dependencies:
     "@microsoft/api-extractor": ^7.36.3
     "@thi.ng/api": ^8.9.0
+    "@thi.ng/arrays": ^2.5.15
     "@thi.ng/bench": ^3.4.1
     "@thi.ng/checks": ^3.4.0
     "@thi.ng/errors": ^2.3.0
     "@thi.ng/idgen": ^2.2.0
     "@thi.ng/logger": ^1.4.16
+    "@thi.ng/random": ^3.5.1
     "@thi.ng/testament": ^0.3.18
     rimraf: ^5.0.1
     tools: "workspace:^"