1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
|
'use strict';
// In worker threads, execute the script sent through the
// message port.
const {
ObjectDefineProperty,
} = primordials;
const {
patchProcessObject,
setupCoverageHooks,
setupInspectorHooks,
setupWarningHandler,
setupDebugEnv,
initializeDeprecations,
initializeCJSLoader,
initializeESMLoader,
initializeFrozenIntrinsics,
initializeReport,
loadPreloadModules,
setupTraceCategoryState
} = require('internal/bootstrap/pre_execution');
const {
threadId,
getEnvMessagePort
} = internalBinding('worker');
const workerIo = require('internal/worker/io');
const {
messageTypes: {
// Messages that may be received by workers
LOAD_SCRIPT,
// Messages that may be posted from workers
UP_AND_RUNNING,
ERROR_MESSAGE,
COULD_NOT_SERIALIZE_ERROR,
// Messages that may be either received or posted
STDIO_PAYLOAD,
STDIO_WANTS_MORE_DATA,
},
kStdioWantsMoreDataCallback
} = workerIo;
const {
onGlobalUncaughtException
} = require('internal/process/execution');
const publicWorker = require('worker_threads');
const debug = require('internal/util/debuglog').debuglog('worker');
const assert = require('internal/assert');
patchProcessObject();
setupInspectorHooks();
setupDebugEnv();
setupWarningHandler();
// Since worker threads cannot switch cwd, we do not need to
// overwrite the process.env.NODE_V8_COVERAGE variable.
if (process.env.NODE_V8_COVERAGE) {
setupCoverageHooks(process.env.NODE_V8_COVERAGE);
}
debug(`[${threadId}] is setting up worker child environment`);
// Set up the message port and start listening
const port = getEnvMessagePort();
// If the main thread is spawned with env NODE_CHANNEL_FD, it's probably
// spawned by our child_process module. In the work threads, mark the
// related IPC properties as unavailable.
if (process.env.NODE_CHANNEL_FD) {
const workerThreadSetup = require('internal/process/worker_thread_only');
ObjectDefineProperty(process, 'channel', {
enumerable: false,
get: workerThreadSetup.unavailable('process.channel')
});
ObjectDefineProperty(process, 'connected', {
enumerable: false,
get: workerThreadSetup.unavailable('process.connected')
});
process.send = workerThreadSetup.unavailable('process.send()');
process.disconnect =
workerThreadSetup.unavailable('process.disconnect()');
}
port.on('message', (message) => {
if (message.type === LOAD_SCRIPT) {
const {
argv,
cwdCounter,
filename,
doEval,
workerData,
publicPort,
manifestSrc,
manifestURL,
hasStdin
} = message;
setupTraceCategoryState();
initializeReport();
if (manifestSrc) {
require('internal/process/policy').setup(manifestSrc, manifestURL);
}
initializeDeprecations();
initializeCJSLoader();
initializeESMLoader();
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;
// The counter is only passed to the workers created by the main thread, not
// to workers created by other workers.
let cachedCwd = '';
let lastCounter = -1;
const originalCwd = process.cwd;
process.cwd = function() {
const currentCounter = Atomics.load(cwdCounter, 0);
if (currentCounter === lastCounter)
return cachedCwd;
lastCounter = currentCounter;
cachedCwd = originalCwd();
return cachedCwd;
};
workerIo.sharedCwdCounter = cwdCounter;
if (!hasStdin)
process.stdin.push(null);
debug(`[${threadId}] starts worker script ${filename} ` +
`(eval = ${eval}) at cwd = ${process.cwd()}`);
port.unref();
port.postMessage({ type: UP_AND_RUNNING });
if (doEval) {
const { evalScript } = require('internal/process/execution');
const name = '[worker eval]';
// This is necessary for CJS module compilation.
// TODO: pass this with something really internal.
ObjectDefineProperty(process, '_eval', {
configurable: true,
enumerable: true,
value: filename,
});
process.argv.splice(1, 0, name);
evalScript(name, filename);
} else {
// script filename
// 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);
CJSLoader.Module.runMain(filename);
}
} else if (message.type === STDIO_PAYLOAD) {
const { stream, chunk, encoding } = message;
process[stream].push(chunk, encoding);
} else {
assert(
message.type === STDIO_WANTS_MORE_DATA,
`Unknown worker message type ${message.type}`
);
const { stream } = message;
process[stream][kStdioWantsMoreDataCallback]();
}
});
function workerOnGlobalUncaughtException(error, fromPromise) {
debug(`[${threadId}] gets uncaught exception`);
let handled = false;
try {
handled = onGlobalUncaughtException(error, fromPromise);
} catch (e) {
error = e;
}
debug(`[${threadId}] uncaught exception handled = ${handled}`);
if (handled) {
return true;
}
let serialized;
try {
const { serializeError } = require('internal/error-serdes');
serialized = serializeError(error);
} catch {}
debug(`[${threadId}] uncaught exception serialized = ${!!serialized}`);
if (serialized)
port.postMessage({
type: ERROR_MESSAGE,
error: serialized
});
else
port.postMessage({ type: COULD_NOT_SERIALIZE_ERROR });
const { clearAsyncIdStack } = require('internal/async_hooks');
clearAsyncIdStack();
process.exit();
}
// Patch the global uncaught exception handler so it gets picked up by
// node::errors::TriggerUncaughtException().
process._fatalException = workerOnGlobalUncaughtException;
markBootstrapComplete();
port.start();
|
