|
| 1 | +/* global harden */ |
| 2 | +// this file is loaded at the start of a new Worker, which makes it a new JS |
| 3 | +// environment (with it's own Realm), so we must install-ses too. |
| 4 | +import '@agoric/install-ses'; |
| 5 | +import { parentPort } from 'worker_threads'; |
| 6 | +import anylogger from 'anylogger'; |
| 7 | + |
| 8 | +import { assert } from '@agoric/assert'; |
| 9 | +import { importBundle } from '@agoric/import-bundle'; |
| 10 | +import { Remotable, getInterfaceOf } from '@agoric/marshal'; |
| 11 | +import { HandledPromise } from '@agoric/eventual-send'; |
| 12 | +import { waitUntilQuiescent } from '../../waitUntilQuiescent'; |
| 13 | +import { makeLiveSlots } from '../liveSlots'; |
| 14 | + |
| 15 | +// eslint-disable-next-line no-unused-vars |
| 16 | +function workerLog(first, ...args) { |
| 17 | + // console.error(`---worker: ${first}`, ...args); |
| 18 | +} |
| 19 | + |
| 20 | +workerLog(`supervisor started`); |
| 21 | + |
| 22 | +function makeConsole(tag) { |
| 23 | + const log = anylogger(tag); |
| 24 | + const cons = {}; |
| 25 | + for (const level of ['debug', 'log', 'info', 'warn', 'error']) { |
| 26 | + cons[level] = log[level]; |
| 27 | + } |
| 28 | + return harden(cons); |
| 29 | +} |
| 30 | + |
| 31 | +function runAndWait(f, errmsg) { |
| 32 | + Promise.resolve() |
| 33 | + .then(f) |
| 34 | + .then(undefined, err => workerLog(`doProcess: ${errmsg}:`, err)); |
| 35 | + return waitUntilQuiescent(); |
| 36 | +} |
| 37 | + |
| 38 | +function sendUplink(msg) { |
| 39 | + assert(msg instanceof Array, `msg must be an Array`); |
| 40 | + parentPort.postMessage(msg); |
| 41 | +} |
| 42 | + |
| 43 | +let dispatch; |
| 44 | + |
| 45 | +async function doProcess(dispatchRecord, errmsg) { |
| 46 | + const dispatchOp = dispatchRecord[0]; |
| 47 | + const dispatchArgs = dispatchRecord.slice(1); |
| 48 | + workerLog(`runAndWait`); |
| 49 | + await runAndWait(() => dispatch[dispatchOp](...dispatchArgs), errmsg); |
| 50 | + workerLog(`doProcess done`); |
| 51 | +} |
| 52 | + |
| 53 | +function doNotify(vpid, vp) { |
| 54 | + const errmsg = `vat.promise[${vpid}] ${vp.state} failed`; |
| 55 | + switch (vp.state) { |
| 56 | + case 'fulfilledToPresence': |
| 57 | + return doProcess(['notifyFulfillToPresence', vpid, vp.slot], errmsg); |
| 58 | + case 'redirected': |
| 59 | + throw new Error('not implemented yet'); |
| 60 | + case 'fulfilledToData': |
| 61 | + return doProcess(['notifyFulfillToData', vpid, vp.data], errmsg); |
| 62 | + case 'rejected': |
| 63 | + return doProcess(['notifyReject', vpid, vp.data], errmsg); |
| 64 | + default: |
| 65 | + throw Error(`unknown promise state '${vp.state}'`); |
| 66 | + } |
| 67 | +} |
| 68 | + |
| 69 | +let syscallLog; |
| 70 | +parentPort.on('message', ([type, ...margs]) => { |
| 71 | + workerLog(`received`, type); |
| 72 | + if (type === 'start') { |
| 73 | + // TODO: parent should send ['start', vatID] |
| 74 | + workerLog(`got start`); |
| 75 | + sendUplink(['gotStart']); |
| 76 | + } else if (type === 'setBundle') { |
| 77 | + const [bundle, vatParameters] = margs; |
| 78 | + const endowments = { |
| 79 | + console: makeConsole(`SwingSet:vatWorker`), |
| 80 | + HandledPromise, |
| 81 | + }; |
| 82 | + importBundle(bundle, { endowments }).then(vatNS => { |
| 83 | + workerLog(`got vatNS:`, Object.keys(vatNS).join(',')); |
| 84 | + sendUplink(['gotBundle']); |
| 85 | + |
| 86 | + function doSyscall(vatSyscallObject) { |
| 87 | + sendUplink(['syscall', ...vatSyscallObject]); |
| 88 | + } |
| 89 | + const syscall = harden({ |
| 90 | + send: (...args) => doSyscall(['send', ...args]), |
| 91 | + callNow: (..._args) => { |
| 92 | + throw Error(`nodeWorker cannot syscall.callNow`); |
| 93 | + }, |
| 94 | + subscribe: (...args) => doSyscall(['subscribe', ...args]), |
| 95 | + fulfillToData: (...args) => doSyscall(['fulfillToData', ...args]), |
| 96 | + fulfillToPresence: (...args) => |
| 97 | + doSyscall(['fulfillToPresence', ...args]), |
| 98 | + reject: (...args) => doSyscall(['reject', ...args]), |
| 99 | + }); |
| 100 | + |
| 101 | + const state = null; |
| 102 | + const vatID = 'demo-vatID'; |
| 103 | + // todo: maybe add transformTildot, makeGetMeter/transformMetering to |
| 104 | + // vatPowers, but only if options tell us they're wanted. Maybe |
| 105 | + // transformTildot should be async and outsourced to the kernel |
| 106 | + // process/thread. |
| 107 | + const vatPowers = { Remotable, getInterfaceOf }; |
| 108 | + dispatch = makeLiveSlots( |
| 109 | + syscall, |
| 110 | + state, |
| 111 | + vatNS.buildRootObject, |
| 112 | + vatID, |
| 113 | + vatPowers, |
| 114 | + vatParameters, |
| 115 | + ); |
| 116 | + workerLog(`got dispatch:`, Object.keys(dispatch).join(',')); |
| 117 | + sendUplink(['dispatchReady']); |
| 118 | + }); |
| 119 | + } else if (type === 'deliver') { |
| 120 | + if (!dispatch) { |
| 121 | + workerLog(`error: deliver before dispatchReady`); |
| 122 | + return; |
| 123 | + } |
| 124 | + const [dtype, ...dargs] = margs; |
| 125 | + if (dtype === 'message') { |
| 126 | + const [targetSlot, msg] = dargs; |
| 127 | + const errmsg = `vat[${targetSlot}].${msg.method} dispatch failed`; |
| 128 | + doProcess( |
| 129 | + ['deliver', targetSlot, msg.method, msg.args, msg.result], |
| 130 | + errmsg, |
| 131 | + ).then(() => { |
| 132 | + sendUplink(['deliverDone']); |
| 133 | + }); |
| 134 | + } else if (dtype === 'notify') { |
| 135 | + doNotify(...dargs).then(() => sendUplink(['deliverDone', syscallLog])); |
| 136 | + } else { |
| 137 | + throw Error(`bad delivery type ${dtype}`); |
| 138 | + } |
| 139 | + } else { |
| 140 | + workerLog(`unrecognized downlink message ${type}`); |
| 141 | + } |
| 142 | +}); |
0 commit comments