Skip to content

Commit

Permalink
module: move esm loader hooks to worker thread
Browse files Browse the repository at this point in the history
  • Loading branch information
bfarias-godaddy authored and jkrems committed Jan 7, 2020
1 parent 2551a21 commit a2a6f44
Show file tree
Hide file tree
Showing 11 changed files with 537 additions and 90 deletions.
38 changes: 32 additions & 6 deletions lib/internal/bootstrap/pre_execution.js
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,8 @@ function prepareMainThreadExecution(expandArgv1 = false) {
}

setupDebugEnv();
// Load policy from disk and parse it.
initializePolicy();

// Process initial diagnostic reporting configuration, if present.
initializeReport();
Expand All @@ -50,9 +52,6 @@ function prepareMainThreadExecution(expandArgv1 = false) {
// process.disconnect().
setupChildProcessIpcChannel();

// Load policy from disk and parse it.
initializePolicy();

// If this is a worker in cluster mode, start up the communication
// channel. This needs to be done before any user code gets executed
// (including preload modules).
Expand All @@ -61,7 +60,32 @@ function prepareMainThreadExecution(expandArgv1 = false) {
initializeDeprecations();
initializeWASI();
initializeCJSLoader();
initializeESMLoader();

function startLoaders() {
const loaderHREF = getOptionValue('--experimental-loader');
if (!loaderHREF) return null;
const { InternalWorker } = require('internal/worker');
const { MessageChannel } = require('internal/worker/io');
// DO NOT ADD CODE ABOVE THIS LINE
// THIS SHOULD POST THE PORTS ASAP TO THE PARENT
const {
port1: outsideBelowPort,
port2: insideBelowPort
} = new MessageChannel();
outsideBelowPort.name = 'outsideBelowPort';
InternalWorker('internal/modules/esm/worker', {
// stdout: true,
// strerr: true,
transferList: [ insideBelowPort ],
workerData: {
loaderHREF,
insideBelowPort,
insideAbovePort: null
}
}).unref();
return outsideBelowPort;
}
initializeESMLoader(startLoaders());

const CJSLoader = require('internal/modules/cjs/loader');
assert(!CJSLoader.hasLoadedAnyUserCJSModule);
Expand Down Expand Up @@ -336,12 +360,12 @@ function initializeClusterIPC() {
}
}

const { pathToFileURL, URL } = require('url');
function initializePolicy() {
const experimentalPolicy = getOptionValue('--experimental-policy');
if (experimentalPolicy) {
process.emitWarning('Policies are experimental.',
'ExperimentalWarning');
const { pathToFileURL, URL } = require('url');
// URL here as it is slightly different parsing
// no bare specifiers for now
let manifestURL;
Expand Down Expand Up @@ -400,14 +424,16 @@ function initializeCJSLoader() {
require('internal/modules/run_main').executeUserEntryPoint;
}

function initializeESMLoader() {
function initializeESMLoader(bottomLoader) {
// Create this WeakMap in js-land because V8 has no C++ API for WeakMap.
internalBinding('module_wrap').callbackMap = new SafeWeakMap();

const {
setImportModuleDynamicallyCallback,
setInitializeImportMetaObjectCallback
} = internalBinding('module_wrap');
const esmLoader = require('internal/modules/esm/loader');
esmLoader.initUserLoaders(bottomLoader);
const esm = require('internal/process/esm_loader');
// Setup per-isolate callbacks that locate data or callbacks that we keep
// track of for different ESM modules.
Expand Down
26 changes: 17 additions & 9 deletions lib/internal/main/worker_thread.js
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,9 @@ port.on('message', (message) => {
cwdCounter,
filename,
doEval,
internal,
workerData,
loaderPort,
publicPort,
manifestSrc,
manifestURL,
Expand All @@ -112,14 +114,16 @@ port.on('message', (message) => {
initializeDeprecations();
initializeWASI();
initializeCJSLoader();
initializeESMLoader();

const CJSLoader = require('internal/modules/cjs/loader');
assert(!CJSLoader.hasLoadedAnyUserCJSModule);
loadPreloadModules();
initializeFrozenIntrinsics();
if (argv !== undefined) {
process.argv = process.argv.concat(argv);
initializeESMLoader(loaderPort);

if (!internal) {
const CJSLoader = require('internal/modules/cjs/loader');
assert(!CJSLoader.hasLoadedAnyUserCJSModule);
loadPreloadModules();
initializeFrozenIntrinsics();
if (argv !== undefined) {
process.argv = process.argv.concat(argv);
}
}
publicWorker.parentPort = publicPort;
publicWorker.workerData = workerData;
Expand Down Expand Up @@ -147,7 +151,10 @@ port.on('message', (message) => {
`(eval = ${eval}) at cwd = ${process.cwd()}`);
port.unref();
port.postMessage({ type: UP_AND_RUNNING });
if (doEval) {
if (internal) {
debug(`WHTY ${publicPort}`);
require(filename);
} else if (doEval) {
const { evalScript } = require('internal/process/execution');
const name = '[worker eval]';
// This is necessary for CJS module compilation.
Expand All @@ -164,6 +171,7 @@ port.on('message', (message) => {
// runMain here might be monkey-patched by users in --require.
// XXX: the monkey-patchability here should probably be deprecated.
process.argv.splice(1, 0, filename);
const CJSLoader = require('internal/modules/cjs/loader');
CJSLoader.Module.runMain(filename);
}
} else if (message.type === STDIO_PAYLOAD) {
Expand Down
139 changes: 139 additions & 0 deletions lib/internal/modules/esm/ipc_types.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
/* eslint-disable */
let nextId = 1;
function getNewId() {
const id = nextId;
nextId++;
return id;
}
class RPCIncomingBridge {
port;
constructor(port, handler) {
this.port = port;
port.on('message', async (msg) => {
const { id, body } = msg;
try {
const result = await handler(body);
port.postMessage({
id,
result,
hadThrow: false
});
} catch (e) {
port.postMessage({
id,
result: e,
hadThrow: true
});
}
});
}
}
class RPCOutgoingBridge {
pending = new Map();
port;
constructor(port) {
this.port = port;
port.on('message', async (msg) => {
// console.dir({RPCOutgoingBridge_onmessage: msg})
const { id, result, hadThrow } = msg;
if (this.pending.has(id)) {
const handler = this.pending.get(id);
this.pending.delete(id);
if (this.pending.size === 0) {
this.port.unref();
}

if (hadThrow) {
handler.throw(result);
} else {
handler.return(result);
}
}
});
this.port.unref();
}
send(body, transferList) {
const id = getNewId();
// console.dir({RPCOutgoingBridge_send: body})
return new Promise((f, r) => {
this.pending.set(id, {
return: f,
throw: r
});
if (this.pending.size === 1) {
this.port.ref();
}
this.port.postMessage({
id,
body
}, transferList);
});
}
}
class ResolveRequest {
static type = 'resolve.request';
type = ResolveRequest.type;
specifier;
base;
clientId;
conditions;
constructor({ specifier, base, clientId, conditions }) {
this.specifier = specifier;
this.base = base;
this.clientId = clientId;
this.conditions = conditions;
}
static fromOrNull(o) {
if (o.type !== ResolveRequest.type) return null;
return new ResolveRequest(o);
}
}
class ResolveResponse {
static type = 'resolve.response';
type = ResolveResponse.type;
url;
format;
constructor({ url, format }) {
this.url = url;
this.format = format;
}
static fromOrNull(o) {
if (o.type !== ResolveResponse.type) return null;
return new ResolveResponse(o);
}
}
class GetFormatRequest {
static type = 'getFormat.request';
type = GetFormatRequest.type;
url;
clientId;
constructor({ url, clientId }) {
this.url = url;
this.clientId = clientId;
}
static fromOrNull(o) {
if (o.type !== GetFormatRequest.type) return null;
return new GetFormatRequest(o);
}
}
class GetFormatResponse {
static type = 'getFormat.response';
type = GetFormatResponse.type;
format;
constructor({ format }) {
this.format = format;
}
static fromOrNull(o) {
if (o.type !== GetFormatResponse.type) return null;
return new GetFormatResponse(o);
}
}
module.exports = {
GetFormatRequest,
GetFormatResponse,
ResolveRequest,
ResolveResponse,
RPCIncomingBridge,
RPCOutgoingBridge,
getNewId
};
Loading

0 comments on commit a2a6f44

Please sign in to comment.