Skip to content

Commit 8db6b8a

Browse files
joyeecheungtargos
joyeecheung
authored and
targos
committed
worker: move worker thread setup code into the main script
This patch directly inlines `createMessageHandler()` and `createWorkerFatalExeception()` in the new `lib/internal/main/worker_thread.js` since the implementation of the two methods are related to the execution flow of workers. PR-URL: #25667 Reviewed-By: Anna Henningsen <anna@addaleax.net> Backport-PR-URL: #26036
1 parent e6a4fb6 commit 8db6b8a

File tree

2 files changed

+103
-120
lines changed

2 files changed

+103
-120
lines changed

‎lib/internal/main/worker_thread.js

+102-14
Original file line numberDiff line numberDiff line change
@@ -10,30 +10,118 @@ const {
1010
} = require('internal/bootstrap/pre_execution');
1111

1212
const {
13-
getEnvMessagePort,
14-
threadId
13+
threadId,
14+
getEnvMessagePort
1515
} = internalBinding('worker');
1616

1717
const {
18-
createMessageHandler,
19-
createWorkerFatalExeception
20-
} = require('internal/process/worker_thread_only');
18+
messageTypes: {
19+
// Messages that may be received by workers
20+
LOAD_SCRIPT,
21+
// Messages that may be posted from workers
22+
UP_AND_RUNNING,
23+
ERROR_MESSAGE,
24+
COULD_NOT_SERIALIZE_ERROR,
25+
// Messages that may be either received or posted
26+
STDIO_PAYLOAD,
27+
STDIO_WANTS_MORE_DATA,
28+
},
29+
kStdioWantsMoreDataCallback
30+
} = require('internal/worker/io');
2131

32+
const {
33+
fatalException: originalFatalException
34+
} = require('internal/process/execution');
35+
36+
const publicWorker = require('worker_threads');
2237
const debug = require('util').debuglog('worker');
23-
debug(`[${threadId}] is setting up worker child environment`);
2438

25-
function prepareUserCodeExecution() {
26-
initializeClusterIPC();
27-
initializeESMLoader();
28-
loadPreloadModules();
29-
}
39+
debug(`[${threadId}] is setting up worker child environment`);
3040

3141
// Set up the message port and start listening
3242
const port = getEnvMessagePort();
33-
port.on('message', createMessageHandler(port, prepareUserCodeExecution));
34-
port.start();
43+
44+
port.on('message', (message) => {
45+
if (message.type === LOAD_SCRIPT) {
46+
const {
47+
filename,
48+
doEval,
49+
workerData,
50+
publicPort,
51+
manifestSrc,
52+
manifestURL,
53+
hasStdin
54+
} = message;
55+
if (manifestSrc) {
56+
require('internal/process/policy').setup(manifestSrc, manifestURL);
57+
}
58+
initializeClusterIPC();
59+
initializeESMLoader();
60+
loadPreloadModules();
61+
publicWorker.parentPort = publicPort;
62+
publicWorker.workerData = workerData;
63+
64+
if (!hasStdin)
65+
process.stdin.push(null);
66+
67+
debug(`[${threadId}] starts worker script ${filename} ` +
68+
`(eval = ${eval}) at cwd = ${process.cwd()}`);
69+
port.unref();
70+
port.postMessage({ type: UP_AND_RUNNING });
71+
if (doEval) {
72+
const { evalScript } = require('internal/process/execution');
73+
evalScript('[worker eval]', filename);
74+
} else {
75+
process.argv[1] = filename; // script filename
76+
require('module').runMain();
77+
}
78+
return;
79+
} else if (message.type === STDIO_PAYLOAD) {
80+
const { stream, chunk, encoding } = message;
81+
process[stream].push(chunk, encoding);
82+
return;
83+
} else if (message.type === STDIO_WANTS_MORE_DATA) {
84+
const { stream } = message;
85+
process[stream][kStdioWantsMoreDataCallback]();
86+
return;
87+
}
88+
89+
require('assert').fail(`Unknown worker message type ${message.type}`);
90+
});
3591

3692
// Overwrite fatalException
37-
process._fatalException = createWorkerFatalExeception(port);
93+
process._fatalException = (error) => {
94+
debug(`[${threadId}] gets fatal exception`);
95+
let caught = false;
96+
try {
97+
caught = originalFatalException.call(this, error);
98+
} catch (e) {
99+
error = e;
100+
}
101+
debug(`[${threadId}] fatal exception caught = ${caught}`);
102+
103+
if (!caught) {
104+
let serialized;
105+
try {
106+
const { serializeError } = require('internal/error-serdes');
107+
serialized = serializeError(error);
108+
} catch {}
109+
debug(`[${threadId}] fatal exception serialized = ${!!serialized}`);
110+
if (serialized)
111+
port.postMessage({
112+
type: ERROR_MESSAGE,
113+
error: serialized
114+
});
115+
else
116+
port.postMessage({ type: COULD_NOT_SERIALIZE_ERROR });
117+
118+
const { clearAsyncIdStack } = require('internal/async_hooks');
119+
clearAsyncIdStack();
120+
121+
process.exit();
122+
}
123+
};
38124

39125
markBootstrapComplete();
126+
127+
port.start();

‎lib/internal/process/worker_thread_only.js

+1-106
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,10 @@
33
// This file contains process bootstrappers that can only be
44
// run in the worker thread.
55
const {
6-
getEnvMessagePort,
7-
threadId
6+
getEnvMessagePort
87
} = internalBinding('worker');
98

109
const {
11-
messageTypes,
12-
kStdioWantsMoreDataCallback,
1310
kWaitingStreams,
1411
ReadableWorkerStdio,
1512
WritableWorkerStdio
@@ -18,15 +15,6 @@ const {
1815
const {
1916
codes: { ERR_WORKER_UNSUPPORTED_OPERATION }
2017
} = require('internal/errors');
21-
22-
let debuglog;
23-
function debug(...args) {
24-
if (!debuglog) {
25-
debuglog = require('util').debuglog('worker');
26-
}
27-
return debuglog(...args);
28-
}
29-
3018
const workerStdio = {};
3119

3220
function initializeWorkerStdio() {
@@ -43,97 +31,6 @@ function initializeWorkerStdio() {
4331
};
4432
}
4533

46-
function createMessageHandler(port, prepareUserCodeExecution) {
47-
const publicWorker = require('worker_threads');
48-
49-
return function(message) {
50-
if (message.type === messageTypes.LOAD_SCRIPT) {
51-
const {
52-
filename,
53-
doEval,
54-
workerData,
55-
publicPort,
56-
manifestSrc,
57-
manifestURL,
58-
hasStdin
59-
} = message;
60-
if (manifestSrc) {
61-
require('internal/process/policy').setup(manifestSrc, manifestURL);
62-
}
63-
prepareUserCodeExecution();
64-
publicWorker.parentPort = publicPort;
65-
publicWorker.workerData = workerData;
66-
67-
if (!hasStdin)
68-
workerStdio.stdin.push(null);
69-
70-
debug(`[${threadId}] starts worker script ${filename} ` +
71-
`(eval = ${eval}) at cwd = ${process.cwd()}`);
72-
port.unref();
73-
port.postMessage({ type: messageTypes.UP_AND_RUNNING });
74-
if (doEval) {
75-
const { evalScript } = require('internal/process/execution');
76-
evalScript('[worker eval]', filename);
77-
} else {
78-
process.argv[1] = filename; // script filename
79-
require('module').runMain();
80-
}
81-
return;
82-
} else if (message.type === messageTypes.STDIO_PAYLOAD) {
83-
const { stream, chunk, encoding } = message;
84-
workerStdio[stream].push(chunk, encoding);
85-
return;
86-
} else if (message.type === messageTypes.STDIO_WANTS_MORE_DATA) {
87-
const { stream } = message;
88-
workerStdio[stream][kStdioWantsMoreDataCallback]();
89-
return;
90-
}
91-
92-
require('assert').fail(`Unknown worker message type ${message.type}`);
93-
};
94-
}
95-
96-
// XXX(joyeecheung): this has to be returned as an anonymous function
97-
// wrapped in a closure, see the comment of the original
98-
// process._fatalException in lib/internal/process/execution.js
99-
function createWorkerFatalExeception(port) {
100-
const {
101-
fatalException: originalFatalException
102-
} = require('internal/process/execution');
103-
104-
return (error) => {
105-
debug(`[${threadId}] gets fatal exception`);
106-
let caught = false;
107-
try {
108-
caught = originalFatalException.call(this, error);
109-
} catch (e) {
110-
error = e;
111-
}
112-
debug(`[${threadId}] fatal exception caught = ${caught}`);
113-
114-
if (!caught) {
115-
let serialized;
116-
try {
117-
const { serializeError } = require('internal/error-serdes');
118-
serialized = serializeError(error);
119-
} catch {}
120-
debug(`[${threadId}] fatal exception serialized = ${!!serialized}`);
121-
if (serialized)
122-
port.postMessage({
123-
type: messageTypes.ERROR_MESSAGE,
124-
error: serialized
125-
});
126-
else
127-
port.postMessage({ type: messageTypes.COULD_NOT_SERIALIZE_ERROR });
128-
129-
const { clearAsyncIdStack } = require('internal/async_hooks');
130-
clearAsyncIdStack();
131-
132-
process.exit();
133-
}
134-
};
135-
}
136-
13734
// The execution of this function itself should not cause any side effects.
13835
function wrapProcessMethods(binding) {
13936
function umask(mask) {
@@ -150,7 +47,5 @@ function wrapProcessMethods(binding) {
15047

15148
module.exports = {
15249
initializeWorkerStdio,
153-
createMessageHandler,
154-
createWorkerFatalExeception,
15550
wrapProcessMethods
15651
};

0 commit comments

Comments
 (0)