1// Host-side wrapper around the numbl worker. Only the CENTRAL peer creates
2// one; everyone else just receives the results through the p2p network.
3//
4// The wrapper runs hitandrun_demo.m once (which opens the uihtml "figure" —
5// intercepted here, never rendered) and then serves compute requests by
6// speaking the figure's own event protocol to the script:
7// resample {n, x, y, convex, local} -> 'samples' event {x, y, n}
8// newRegion {n, convex, local} -> 'data' event {region, samples, n, convex}
9// Timeouts surface as EngineError so the network layer can step down and let
10// another peer take over.
12import type {Points, Params} from '../types'
13import {PROJECT_FILES, MAIN_FILE} from './project'
14import type {ToWorker, FromWorker} from './protocol'
16const START_TIMEOUT_MS = 120_000
17const COMPUTE_TIMEOUT_MS = 60_000
19export class EngineError extends Error {}
21export interface EngineInit {
22 region: Points
23 samples: Points
24 n: number
25 convex: boolean
26}
28interface HitAndRunData {
29 type: string
30 region: Points
31 samples: Points
32 n: number
33 convex?: boolean
34}
36interface SamplesEvent {
37 x: number[]
38 y: number[]
39 n: number
40}
42// jsonencode collapses 1-element vectors to scalars; normalize.
43const asArray = (v: unknown): number[] =>
44 Array.isArray(v) ? (v as number[]) : typeof v === 'number' ? [v] : []
46const asPoints = (p: {x?: unknown; y?: unknown} | undefined): Points => ({
47 x: asArray(p?.x),
48 y: asArray(p?.y)
49})
51export class Engine {
52 private worker: Worker | null = null
53 private compId: string | null = null
54 private initialData: HitAndRunData | null = null
55 private runDone = false
56 private disposed = false
58 // One request at a time; the network layer serializes computes.
59 private waiter: {
60 event: string
61 resolve: (data: unknown) => void
62 reject: (err: Error) => void
63 } | null = null
64 private startWaiter: {
65 resolve: (init: EngineInit) => void
66 reject: (err: Error) => void
67 } | null = null
69 /** Boot the worker, run the script, resolve with the initial region+samples. */
70 start(): Promise<EngineInit> {
71 if (this.worker) throw new EngineError('engine already started')
72 this.worker = new Worker(new URL('./numbl.worker.ts', import.meta.url), {
73 type: 'module'
74 })
75 this.worker.onmessage = (e: MessageEvent<FromWorker>) =>
76 this.handleMessage(e.data)
77 this.worker.onerror = e => {
78 this.fail(new EngineError(`worker error: ${e.message || 'unknown'}`))
79 }
80 this.post({type: 'run', files: PROJECT_FILES, mainFileName: MAIN_FILE})
82 return new Promise<EngineInit>((resolve, reject) => {
83 this.startWaiter = {resolve, reject}
84 this.armTimeout(START_TIMEOUT_MS, 'engine start timed out')
85 })
86 }
88 /** Draw n fresh samples in the given (current) region. */
89 async resample(req: {
90 params: Params
91 region: Points
92 }): Promise<{samples: Points; n: number}> {
93 const {params, region} = req
94 const data = await this.request(
95 'resample',
96 {n: params.n, x: region.x, y: region.y, convex: params.convex, local: params.local},
97 'samples'
98 )
99 const s = data as SamplesEvent
100 return {
101 samples: {x: asArray(s.x), y: asArray(s.y)},
102 n: typeof s.n === 'number' ? s.n : params.n
103 }
104 }
106 /** Build a brand-new region (convex or not) and sample it. */
107 async newRegion(req: {params: Params}): Promise<EngineInit> {
108 const {params} = req
109 const data = await this.request(
110 'newRegion',
111 {n: params.n, convex: params.convex, local: params.local},
112 'data'
113 )
114 const d = data as HitAndRunData
115 return {
116 region: asPoints(d.region),
117 samples: asPoints(d.samples),
118 n: typeof d.n === 'number' ? d.n : params.n,
119 convex: d.convex !== false
120 }
121 }
123 dispose(): void {
124 if (this.disposed) return
125 this.disposed = true
126 this.fail(new EngineError('engine disposed'))
127 }
129 // ---- internals ---------------------------------------------------------
131 private timeoutId: ReturnType<typeof setTimeout> | null = null
133 private armTimeout(ms: number, message: string) {
134 this.clearTimeout()
135 this.timeoutId = setTimeout(() => this.fail(new EngineError(message)), ms)
136 }
138 private clearTimeout() {
139 if (this.timeoutId !== null) clearTimeout(this.timeoutId)
140 this.timeoutId = null
141 }
143 private post(msg: ToWorker) {
144 this.worker?.postMessage(msg)
145 }
147 private request(
148 name: 'resample' | 'newRegion',
149 payload: unknown,
150 expectEvent: 'samples' | 'data'
151 ): Promise<unknown> {
152 if (!this.worker || !this.runDone || !this.compId) {
153 return Promise.reject(new EngineError('engine not ready'))
154 }
155 if (this.waiter) {
156 return Promise.reject(new EngineError('engine busy'))
157 }
158 this.post({type: 'event', compId: this.compId, name, data: payload})
159 return new Promise<unknown>((resolve, reject) => {
160 this.waiter = {event: expectEvent, resolve, reject}
161 this.armTimeout(COMPUTE_TIMEOUT_MS, `'${name}' timed out`)
162 })
163 }
165 /** A hard failure: everything pending rejects and the worker is torn down. */
166 private fail(err: EngineError) {
167 this.clearTimeout()
168 this.worker?.terminate()
169 this.worker = null
170 this.runDone = false
171 const sw = this.startWaiter
172 const w = this.waiter
173 this.startWaiter = null
174 this.waiter = null
175 sw?.reject(err)
176 w?.reject(err)
177 }
179 private handleMessage(msg: FromWorker) {
180 switch (msg.type) {
181 case 'output':
182 console.log(`[numbl] ${msg.text.replace(/\n$/, '')}`)
183 break
185 case 'uihtml': {
186 // Track the most recent component; its Data is the initial payload.
187 this.compId = msg.compId
188 if (msg.dataJson) {
189 try {
190 this.initialData = JSON.parse(msg.dataJson) as HitAndRunData
191 } catch {
192 /* ignore malformed */
193 }
194 }
195 this.maybeResolveStart()
196 break
197 }
199 case 'runDone': {
200 this.runDone = true
201 if (!msg.hasSession || !this.compId) {
202 this.fail(new EngineError('script finished without a uihtml session'))
203 return
204 }
205 this.maybeResolveStart()
206 break
207 }
209 case 'runError':
210 this.fail(new EngineError(`script error: ${msg.message}`))
211 break
213 case 'eventError':
214 this.fail(new EngineError(`callback error: ${msg.message}`))
215 break
217 case 'hostEvent': {
218 if (!this.waiter || msg.name !== this.waiter.event) break
219 const w = this.waiter
220 this.waiter = null
221 this.clearTimeout()
222 try {
223 w.resolve(JSON.parse(msg.dataJson))
224 } catch (err) {
225 w.reject(new EngineError(`bad event payload: ${String(err)}`))
226 }
227 break
228 }
229 }
230 }
232 private maybeResolveStart() {
233 if (!this.startWaiter || !this.runDone) return
234 if (!this.initialData || this.initialData.type !== 'hitandrun') return
235 const d = this.initialData
236 const w = this.startWaiter
237 this.startWaiter = null
238 this.clearTimeout()
239 w.resolve({
240 region: asPoints(d.region),
241 samples: asPoints(d.samples),
242 n: typeof d.n === 'number' ? d.n : 10000,
243 convex: d.convex !== false
244 })
245 }
246}