feat(daemon): Preliminary worker
MODIFY
packages/daemon/index.d.ts
packages/daemon/index.d.ts
@@ -8,3 +8,9 @@ export async function stop(locator?: Locator);
export async function restart(locator?: Locator);
export async function terminate(locator?: Locator);
export async function clean(locator?: Locator);
+export async function makeEndoClient<TBootstrap>(
+ name: string,
+ sockPath: string,
+ cancelled: Promise<void>,
+ bootstrap?: TBootstrap,
+);

MODIFY
packages/daemon/index.js
packages/daemon/index.js
@@ -10,13 +10,18 @@ import { E } from '@endo/eventual-send';
import { whereEndoState, whereEndoSock, whereEndoCache } from '@endo/where';
import { makeEndoClient } from './src/client.js';
+// Reexports:
+export { makeEndoClient } from './src/client.js';
+
const defaultLocator = {
statePath: whereEndoState(process.platform, process.env),
sockPath: whereEndoSock(process.platform, process.env),
cachePath: whereEndoCache(process.platform, process.env),
};
-const endoDaemonPath = url.fileURLToPath(new URL('src/daemon.js', import.meta.url));
+const endoDaemonPath = url.fileURLToPath(
+ new URL('src/daemon.js', import.meta.url),
+);
export const terminate = async (locator = defaultLocator) => {
const { getBootstrap, finalize } = await makeEndoClient(
@@ -24,7 +29,7 @@ export const terminate = async (locator = defaultLocator) => {
locator.sockPath,
);
const bootstrap = getBootstrap();
- await E(E.get(bootstrap).privateFacet).terminate(locator);
+ await E(E.get(bootstrap).privateFacet).terminate();
finalize();
};

MODIFY
packages/daemon/jsconfig.json
packages/daemon/jsconfig.json
"strictNullChecks": true,
"moduleResolution": "node"
},
- "include": ["src/**/*.js", "index.d.ts"]
+ "include": ["**/*.js", "**/*.ts"]
}

MODIFY
packages/daemon/src/connection.js
packages/daemon/src/connection.js
@@ -61,7 +61,12 @@ const bytesToMessage = bytes => {
* @param {import('stream').Readable} nodeReader
* @param {TBootstrap} bootstrap
*/
-export const makeNodeNetstringCapTP = (name, nodeWriter, nodeReader, bootstrap) => {
+export const makeNodeNetstringCapTP = (
+ name,
+ nodeWriter,
+ nodeReader,
+ bootstrap,
+) => {
const writer = mapWriter(
makeNetstringWriter(makeNodeWriter(nodeWriter)),
messageToBytes,

MODIFY
packages/daemon/src/daemon.js
packages/daemon/src/daemon.js
// @ts-check
/// <reference types="ses"/>
-/* global process */
+/* global process, setTimeout */
// Establish a perimeter:
import '@agoric/babel-standalone';
@@ -8,9 +8,14 @@ import 'ses';
import '@endo/eventual-send/shim.js';
import '@endo/lockdown/commit.js';
+import crypto from 'crypto';
import net from 'net';
import fs from 'fs';
-import { Far } from '@endo/far';
+import path from 'path';
+import popen from 'child_process';
+import url from 'url';
+
+import { E, Far } from '@endo/far';
import { makePromiseKit } from '@endo/promise-kit';
import { makeNodeNetstringCapTP } from './connection.js';
@@ -18,21 +23,89 @@ const { quote: q } = assert;
const { promise: cancelled, reject: cancel } = makePromiseKit();
+// TODO thread through command arguments.
+const gracePeriodMs = 100;
+
+const grace = cancelled.catch(async () => {
+ await new Promise(resolve => setTimeout(resolve, gracePeriodMs));
+});
+
+const endoWorkerPath = url.fileURLToPath(new URL('worker.js', import.meta.url));
+
/** @param {Error} error */
const sinkError = error => {
console.error(error);
};
+/**
+ * @param {import('../index.js').Locator} locator
+ */
+const makeWorker = async locator => {
+ // @ts-ignore Node.js crypto does in fact have randomUUID.
+ const uuid = await crypto.randomUUID();
+ const workerCachePath = path.join(locator.cachePath, uuid);
+ await fs.promises.mkdir(workerCachePath, { recursive: true });
+ const logPath = path.join(workerCachePath, 'worker.log');
+ const output = fs.openSync(logPath, 'w');
+ const child = popen.fork(endoWorkerPath, [uuid, workerCachePath], {
+ stdio: ['ignore', output, output, 'pipe', 'ipc'],
+ });
+ console.error(`Endo worker started PID ${child.pid} UUID ${uuid}`);
+ const stream = /** @type {import('stream').Duplex} */ (child.stdio[3]);
+ assert(stream);
+ const { getBootstrap, drained, finalize } = makeNodeNetstringCapTP(
+ `Worker ${uuid}`,
+ stream,
+ stream,
+ undefined,
+ );
+
+ const bootstrap = getBootstrap();
+
+ const exited = new Promise(resolve => {
+ child.on('exit', () => {
+ console.error(`Endo worker stopped PID ${child.pid} UUID ${uuid}`);
+ resolve(undefined);
+ });
+ });
+
+ const terminated = Promise.all([exited, drained]);
+
+ const { resolve: cancelWorker, promise: workerCancelled } = makePromiseKit();
+
+ cancelled.catch(async () => cancelWorker());
+
+ workerCancelled.then(async () => {
+ const responded = E(bootstrap).terminate();
+ await Promise.race([grace, terminated, responded]);
+ child.kill();
+ });
+
+ const terminate = () => {
+ cancelWorker();
+ };
+
+ return harden({
+ actions: Far('EndoWorkerActions', {
+ terminate,
+ }),
+ terminated,
+ });
+};
+
/**
* @param {import('../index.js').Locator} locator
*/
const makeEndoFacets = locator => {
- const publicFacet = Far('Endo public facet', {});
+ const publicFacet = Far('EndoPublicFacet', {});
- const privateFacet = Far('Endo private facet', {
- async shutdown() {
- console.error('Endo received shutdown request');
- cancel(new Error('Shutdown'));
+ const privateFacet = Far('EndoPrivateFacet', {
+ async terminate() {
+ console.error('Endo received terminate request');
+ cancel(new Error('Terminate'));
+ },
+ async makeWorker() {
+ return makeWorker(locator);
},
});
@@ -45,8 +118,9 @@ const makeEndoFacets = locator => {
};
export const main = async () => {
+ console.error(`Endo starting on PID ${process.pid}`);
process.once('exit', () => {
- console.error('Endo exiting');
+ console.error(`Endo stopping on PID ${process.pid}`);
});
if (process.argv.length < 5) {
@@ -65,7 +139,12 @@ export const main = async () => {
const endoFacets = makeEndoFacets(locator);
- await fs.promises.mkdir(statePath, { recursive: true });
+ const statePathP = fs.promises.mkdir(statePath, { recursive: true });
+ const cachePathP = fs.promises.mkdir(cachePath, { recursive: true });
+ await Promise.all([statePathP, cachePathP]);
+
+ const pidPath = path.join(cachePath, 'endo.pid');
+ await fs.promises.writeFile(pidPath, `${process.pid}\n`);
const server = net.createServer();
@@ -74,7 +153,9 @@ export const main = async () => {
path: sockPath,
},
() => {
- console.log(`Listening on ${q(sockPath)} ${new Date().toISOString()}`);
+ console.log(
+ `Endo listening on ${q(sockPath)} ${new Date().toISOString()}`,
+ );
// Inform parent that we have an open unix domain socket, if we were
// spawned with IPC.
if (process.send) {

ADD
packages/daemon/src/worker.js
+// @ts-check
+/// <reference types="ses"/>
+/* global process */
+
+// Establish a perimeter:
+import '@agoric/babel-standalone';
+import 'ses';
+import '@endo/eventual-send/shim.js';
+import '@endo/lockdown/commit.js';
+
+import fs from 'fs';
+
+import { Far } from '@endo/far';
+import { makePromiseKit } from '@endo/promise-kit';
+import { makeNodeNetstringCapTP } from './connection.js';
+
+/** @param {Error} error */
+const sinkError = error => {
+ console.error(error);
+};
+
+const { promise: cancelled, reject: cancel } = makePromiseKit();
+
+const makeWorkerFacet = () => {
+ return Far('EndoWorkerFacet', {
+ async shutdown() {
+ console.error('Endo worker received shutdown request');
+ cancel(new Error('Shutdown'));
+ },
+ });
+};
+
+export const main = async () => {
+ process.once('exit', () => {
+ console.error('Endo Worker exiting');
+ });
+
+ if (process.argv.length < 2) {
+ throw new Error(
+ `worker.js requires arguments [uuid] [workerCachePath], got ${process.argv.join(
+ ', ',
+ )}`,
+ );
+ }
+
+ // const uuid = process.argv[2];
+ // const workerCachePath = process.argv[3];
+
+ // @ts-ignore This is in fact how you open a file descriptor.
+ const reader = fs.createReadStream(null, { fd: 3 });
+ // @ts-ignore This is in fact how you open a file descriptor.
+ const writer = fs.createWriteStream(null, { fd: 3 });
+
+ const workerFacet = makeWorkerFacet();
+
+ const { drained, finalize } = makeNodeNetstringCapTP(
+ 'Endo',
+ writer,
+ reader,
+ workerFacet,
+ );
+
+ cancelled.catch(async () => {
+ finalize();
+ writer.close();
+ });
+
+ drained.catch(sinkError);
+};
+
+main().catch(sinkError);

MODIFY
packages/daemon/test/test-endo.js
packages/daemon/test/test-endo.js
// @ts-check
/* global process */
+
+// Establish a perimeter:
+import '@agoric/babel-standalone';
+import 'ses';
+import '@endo/eventual-send/shim.js';
+import '@endo/lockdown/commit-debug.js';
+
import test from 'ava';
import url from 'url';
import path from 'path';
-import { start, stop, restart, clean } from '../index.js';
+import { E } from '@endo/far';
+import { start, stop, restart, clean, makeEndoClient } from '../index.js';
const { raw } = String;
-const dirname = url.fileURLToPath(new URL('.', import.meta.url)).toString();
+const dirname = url.fileURLToPath(new URL('..', import.meta.url)).toString();
const locator = {
- statePath: path.join(dirname, 'endo'),
- cachePath: dirname,
+ statePath: path.join(dirname, 'tmp/state'),
+ cachePath: path.join(dirname, 'tmp/cache'),
sockPath:
process.platform === 'win32'
? raw`\\?\pipe\endo-test.sock`
- : path.join(dirname, 'endo.sock'),
+ : path.join(dirname, 'tmp/endo.sock'),
};
test.serial('lifecycle', async t => {
@@ -23,6 +31,17 @@ test.serial('lifecycle', async t => {
await start(locator);
await stop(locator);
await restart(locator);
+
+ const { getBootstrap, finalize, drained } = await makeEndoClient(
+ 'client',
+ locator.sockPath,
+ );
+ const bootstrap = getBootstrap();
+ const worker = await E(E.get(bootstrap).privateFacet).makeWorker();
+ await E(E.get(worker).actions).terminate();
+ finalize();
+ await drained;
+
await stop(locator);
t.pass();
});

feat(daemon): Preliminary worker