1// A pool of the workers of worker.ts. It runs the eigenvalue jobs of
2// modes.ts through a JobRunner, and splits the modes of a model between the
3// workers for the synthesis: the main thread steps the contact (which
4// couples all modes and is cheap), each worker forms the parts of its
5// modes, and the main thread sums them.
7import { type LinearSystem } from "../model/assemble.ts";
8import { type Loss, type Plate, type SimParams } from "../params.ts";
9import { type Job, type JobResult, type JobRunner, type ModeSet, computeModes } from "./modes.ts";
10import {
11 type ContactResult,
12 type ModalModel,
13 type Synthesis,
14 attackPressure,
15 listenerWeights,
16 modalRates,
17 simulateContact,
18 sliceContact,
19 sliceModel,
20 strikeGains,
21 tailStart,
22} from "./synth.ts";
23import { type WorkerRequest, type WorkerResponse } from "./worker.ts";
25interface Pending {
26 resolve: (r: WorkerResponse) => void;
27 reject: (e: Error) => void;
28}
30class PoolWorker {
31 worker: Worker;
32 busy = false;
33 pending = new Map<number, Pending>();
34 ready: Pending | null = null;
35 constructor() {
36 this.worker = new Worker(new URL("./worker.ts", import.meta.url), { type: "module" });
37 this.worker.onmessage = (ev: MessageEvent<WorkerResponse>) => {
38 const r = ev.data;
39 if (r.type === "ready") { this.ready?.resolve(r); this.ready = null; return; }
40 const p = this.pending.get(r.id);
41 if (!p) return;
42 this.pending.delete(r.id);
43 if (r.type === "error") p.reject(new Error(r.message)); else p.resolve(r);
44 };
45 this.worker.onerror = (e) => {
46 for (const p of this.pending.values()) p.reject(new Error(e.message));
47 this.pending.clear();
48 };
49 }
50 send(msg: WorkerRequest, transfer: ArrayBuffer[] = []): Promise<WorkerResponse> {
51 return new Promise((resolve, reject) => {
52 if (msg.type === "system" || msg.type === "model") this.ready = { resolve, reject };
53 else this.pending.set(msg.id, { resolve, reject });
54 this.worker.postMessage(msg, transfer);
55 });
56 }
57}
59export class WorkerPool {
60 private workers: PoolWorker[] = [];
61 private queue: (() => void)[] = [];
62 private modelSplit: { from: number; to: number }[] = [];
63 private nextId = 1;
65 constructor(public readonly size: number) {
66 for (let i = 0; i < size; i++) this.workers.push(new PoolWorker());
67 }
69 static defaultSize(): number {
70 const hw = typeof navigator !== "undefined" ? navigator.hardwareConcurrency || 4 : 4;
71 return Math.max(2, Math.min(12, hw));
72 }
74 /** Waits for an idle worker. */
75 private acquire(): Promise<PoolWorker> {
76 return new Promise((resolve) => {
77 const tryTake = (): boolean => {
78 const w = this.workers.find((w) => !w.busy);
79 if (!w) return false;
80 w.busy = true;
81 resolve(w);
82 return true;
83 };
84 if (!tryTake()) this.queue.push(() => tryTake());
85 });
86 }
87 private release(w: PoolWorker): void {
88 w.busy = false;
89 const next = this.queue.shift();
90 if (next) next();
91 }
93 /** Sends the linear system to every worker. */
94 async setSystem(sys: LinearSystem): Promise<void> {
95 await Promise.all(this.workers.map((w) => w.send({ type: "system", n: sys.n, bw: sys.bw, band: sys.band, mass: sys.mass })));
96 }
98 runner(): JobRunner {
99 return {
100 run: async (job: Job): Promise<JobResult> => {
101 const w = await this.acquire();
102 try {
103 const r = await w.send({ type: "job", id: this.nextId++, job });
104 return (r as { result: JobResult }).result;
105 } finally {
106 this.release(w);
107 }
108 },
109 close: () => {},
110 };
111 }
113 /** The modes of the system, computed by the workers. */
114 async modes(sys: LinearSystem, fMax: number, onProgress?: (done: number, total: number) => void): Promise<ModeSet> {
115 await this.setSystem(sys);
116 return computeModes(sys, fMax, this.runner(), { onProgress });
117 }
119 /** Splits the modes of the model between the workers. */
120 async setModel(model: ModalModel): Promise<void> {
121 const n = this.workers.length;
122 this.modelSplit = [];
123 const per = Math.ceil(model.count / n);
124 await Promise.all(
125 this.workers.map((w, i) => {
126 const from = Math.min(model.count, i * per), to = Math.min(model.count, (i + 1) * per);
127 this.modelSplit.push({ from, to });
128 const slice = sliceModel(model, from, to);
129 return w.send({ type: "model", model: slice });
130 }),
131 );
132 }
134 /**
135 * Synthesizes a strike with the model last set. The audio comes first
136 * (through `onAudio`), then the movie frames; the returned promise
137 * resolves when both are done.
138 */
139 async synthesize(model: ModalModel, pl: Plate, loss: Loss, p: SimParams, onAudio?: (audio: Float32Array, contact: ContactResult) => void, withMovie = true): Promise<Synthesis> {
140 const sigma = modalRates(model, loss);
141 const g = strikeGains(model, pl, p);
142 const contact = simulateContact(model, sigma, g, p, p.fps);
143 const { weight, delay, maxDelay } = listenerWeights(pl, p);
144 const tStart = tailStart(contact, maxDelay);
145 const len = Math.min(contact.tableLen, Math.ceil(tStart * contact.tableRate) + 2);
146 const nSamples = Math.round(p.duration * p.fs);
147 const nFrames = withMovie ? Math.round(p.duration * p.fps) : 0;
148 const parts = await Promise.all(
149 this.workers.map((w, i) => {
150 const { from, to } = this.modelSplit[i];
151 if (to <= from) return Promise.resolve(null);
152 return w.send({
153 type: "synth", id: this.nextId++, sigma: sigma.slice(from, to), contact: sliceContact(contact, from, to),
154 weight, delay, tStart, len, fs: p.fs, nSamples, audio: true, movie: null,
155 });
156 }),
157 );
158 const audio = new Float32Array(nSamples);
159 const table = new Float32Array(model.nRows * len);
160 for (const r of parts) {
161 if (!r || r.type !== "synth") continue;
162 if (r.audio) for (let i = 0; i < nSamples; i++) audio[i] += r.audio[i];
163 if (r.table) for (let i = 0; i < table.length; i++) table[i] += r.table[i];
164 }
165 attackPressure(model, table, len, contact.tableRate, weight, delay, p.fs, tStart, audio);
166 onAudio?.(audio, contact);
167 const frames = new Float32Array(nFrames * model.nGrid);
168 if (withMovie) {
169 const mparts = await Promise.all(
170 this.workers.map((w, i) => {
171 const { from, to } = this.modelSplit[i];
172 if (to <= from) return Promise.resolve(null);
173 return w.send({
174 type: "synth", id: this.nextId++, sigma: sigma.slice(from, to), contact: sliceContact(contact, from, to),
175 weight, delay, tStart, len, fs: p.fs, nSamples, audio: false, movie: { fps: p.fps, nFrames },
176 });
177 }),
178 );
179 for (const r of mparts) if (r && r.type === "synth" && r.frames) for (let i = 0; i < frames.length; i++) frames[i] += r.frames[i];
180 }
181 return { audio, frames, nFrames, contact };
182 }
184 terminate(): void {
185 for (const w of this.workers) w.worker.terminate();
186 this.workers = [];
187 }
188}