Skip to content

Commit 07de758

Browse files
committed
Implement Pthread Manager Worker for synchronous thread creation
This change introduces a dedicated 'Pthread Manager' worker that acts as an intermediary for managing the lifecycle of pthread workers. By moving the responsibility of spawning and managing workers to a dedicated manager worker, we enable synchronous pthread creation on the main browser thread even when it is blocked (e.g. during a join or futex wait). This is because the manager worker and its nested child workers can be started by the browser independently of the main thread's event loop. Key changes: - Add PTHREAD_MANAGER setting to enable this new mode. - Update shell.js and parseTools.mjs to detect and support the manager environment. - Update libpthread.js to spawn the manager worker and proxy thread creation requests to it. - Update runtime_pthread.js with logic to handle the manager worker's responsibilities: spawning, terminating, and relaying messages for child pthreads. - Ensure the manager worker is fully initialized before the main application starts using addRunDependency. Fixes: #18633
1 parent bc4ee8b commit 07de758

5 files changed

Lines changed: 117 additions & 11 deletions

File tree

src/lib/libpthread.js

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,8 +113,28 @@ var LibraryPThread = {
113113
}
114114
},
115115
initMainThread() {
116+
#if PTHREAD_MANAGER
117+
if ({{{ ENVIRONMENT_IS_MAIN_THREAD() }}}) {
118+
#if ASSERTIONS
119+
dbg('PThread: initializing manager worker');
120+
#endif
121+
PThread.allocateUnusedWorker();
122+
PThread.managerWorker = PThread.unusedWorkers.pop();
123+
addOnPreRun(async () => {
124+
var managerReady = PThread.loadWasmModuleToWorker(PThread.managerWorker);
125+
addRunDependency('manager-worker');
126+
await managerReady;
127+
PThread.managerWorker.postMessage({ cmd: 'makeManager' });
128+
removeRunDependency('manager-worker');
129+
});
130+
}
131+
#endif
116132
#if PTHREAD_POOL_SIZE
133+
#if PTHREAD_MANAGER
134+
var pthreadPoolSize = 0;
135+
#else
117136
var pthreadPoolSize = {{{ PTHREAD_POOL_SIZE }}};
137+
#endif
118138
// Start loading up the Worker pool, if requested.
119139
while (pthreadPoolSize--) {
120140
PThread.allocateUnusedWorker();
@@ -284,6 +304,11 @@ var LibraryPThread = {
284304

285305
if (cmd === 'checkMailbox') {
286306
checkMailbox();
307+
} else if (cmd === 'relay') {
308+
var targetWorker = PThread.pthreads[d.thread];
309+
if (targetWorker && targetWorker.onmessage) {
310+
targetWorker.onmessage({ data: d.data });
311+
}
287312
} else if (cmd === 'spawnThread') {
288313
spawnThread(d);
289314
} else if (cmd === 'cleanupThread') {
@@ -679,6 +704,23 @@ var LibraryPThread = {
679704
assert(threadParams.pthread_ptr, 'spawnThread called with null pthread ptr');
680705
#endif
681706

707+
#if PTHREAD_MANAGER
708+
if ({{{ ENVIRONMENT_IS_MAIN_THREAD() }}}) {
709+
var workerStub = {
710+
pthread_ptr: threadParams.pthread_ptr,
711+
postMessage: (msg, transfer) => {
712+
PThread.managerWorker.postMessage({ cmd: 'relay', thread: threadParams.pthread_ptr, data: msg }, transfer);
713+
},
714+
terminate: () => {
715+
PThread.managerWorker.postMessage({ cmd: 'terminate', thread: threadParams.pthread_ptr });
716+
}
717+
};
718+
PThread.pthreads[threadParams.pthread_ptr] = workerStub;
719+
PThread.managerWorker.postMessage({ cmd: 'spawnThread', threadParams: threadParams }, threadParams.transferList);
720+
return 0;
721+
}
722+
#endif
723+
682724
var worker = PThread.getNewWorker();
683725
if (!worker) {
684726
// No available workers in the PThread pool.

src/parseTools.mjs

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1121,7 +1121,10 @@ function ENVIRONMENT_IS_MAIN_THREAD() {
11211121
function ENVIRONMENT_IS_WORKER_THREAD() {
11221122
assert(PTHREADS || WASM_WORKERS);
11231123
var envs = [];
1124-
if (PTHREADS) envs.push('ENVIRONMENT_IS_PTHREAD');
1124+
if (PTHREADS) {
1125+
envs.push('ENVIRONMENT_IS_PTHREAD');
1126+
envs.push('ENVIRONMENT_IS_PTHREAD_MANAGER');
1127+
}
11251128
if (WASM_WORKERS) envs.push('ENVIRONMENT_IS_WASM_WORKER');
11261129
return '(' + envs.join('||') + ')';
11271130
}
@@ -1169,7 +1172,7 @@ function wasmWorkerDetection() {
11691172

11701173
function pthreadDetection() {
11711174
if (ASSERTIONS) {
1172-
return "globalThis.name?.startsWith('em-pthread')";
1175+
return "globalThis.name?.startsWith('em-pthread') && globalThis.name != 'em-pthread-manager'";
11731176
} else {
11741177
return "globalThis.name == 'em-pthread'";
11751178
}

src/runtime_pthread.js

Lines changed: 62 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ var sharedModules = {};
2222

2323
var startWorker;
2424

25-
if (ENVIRONMENT_IS_PTHREAD) {
25+
if (ENVIRONMENT_IS_PTHREAD || ENVIRONMENT_IS_PTHREAD_MANAGER) {
2626
// Thread-local guard variable for one-time init of the JS state
2727
var initializedJS = false;
2828

@@ -42,11 +42,53 @@ if (ENVIRONMENT_IS_PTHREAD) {
4242
// notified about them.
4343
self.onunhandledrejection = (e) => { throw e.reason || e; };
4444

45+
function handleManagerMessage(e) {
46+
var d = e.data;
47+
var cmd = d.cmd;
48+
if (cmd === 'spawnThread') {
49+
var threadParams = d.threadParams;
50+
var worker = PThread.getNewWorker();
51+
PThread.runningWorkers.push(worker);
52+
PThread.pthreads[threadParams.pthread_ptr] = worker;
53+
worker.pthread_ptr = threadParams.pthread_ptr;
54+
worker.onmessage = (e) => {
55+
postMessage({ cmd: 'relay', thread: worker.pthread_ptr, data: e.data });
56+
};
57+
var msg = {
58+
cmd: 'run',
59+
start_routine: threadParams.startRoutine,
60+
arg: threadParams.arg,
61+
pthread_ptr: threadParams.pthread_ptr,
62+
};
63+
worker.postMessage(msg, threadParams.transferList);
64+
} else if (cmd === 'relay') {
65+
var worker = PThread.pthreads[d.thread];
66+
if (worker) {
67+
worker.postMessage(d.data, d.transferList);
68+
}
69+
} else if (cmd === 'terminate') {
70+
var worker = PThread.pthreads[d.thread];
71+
if (worker) {
72+
worker.terminate();
73+
delete PThread.pthreads[d.thread];
74+
}
75+
} else if (cmd === 'checkMailbox') {
76+
if (initializedJS) {
77+
checkMailbox();
78+
}
79+
}
80+
}
81+
4582
{{{ asyncIf(ASYNCIFY == 2) }}}function handleMessage(e) {
4683
try {
4784
var msgData = e['data'];
48-
//dbg('msgData: ' + Object.keys(msgData));
4985
var cmd = msgData.cmd;
86+
if (cmd === 'makeManager') {
87+
ENVIRONMENT_IS_PTHREAD = false;
88+
ENVIRONMENT_IS_PTHREAD_MANAGER = true;
89+
self.onmessage = handleManagerMessage;
90+
return;
91+
}
5092
if (cmd === 'load') { // Preload command that is called once per worker to parse and load the Emscripten code.
5193
#if ASSERTIONS
5294
workerID = msgData.workerID;
@@ -63,12 +105,23 @@ if (ENVIRONMENT_IS_PTHREAD) {
63105
startWorker = () => {
64106
// Notify the main thread that this thread has loaded.
65107
postMessage({ cmd: 'loaded' });
66-
// Process any messages that were queued before the thread was ready.
67-
for (let msg of messageQueue) {
68-
handleMessage(msg);
108+
109+
if (ENVIRONMENT_IS_PTHREAD_MANAGER) {
110+
initializedJS = true;
111+
// Process any messages that were queued before the thread was ready.
112+
for (let msg of messageQueue) {
113+
handleManagerMessage(msg);
114+
}
115+
// Restore the real message handler.
116+
self.onmessage = handleManagerMessage;
117+
} else {
118+
// Process any messages that were queued before the thread was ready.
119+
for (let msg of messageQueue) {
120+
handleMessage(msg);
121+
}
122+
// Restore the real message handler.
123+
self.onmessage = handleMessage;
69124
}
70-
// Restore the real message handler.
71-
self.onmessage = handleMessage;
72125
};
73126

74127
#if MAIN_MODULE
@@ -192,11 +245,11 @@ if (ENVIRONMENT_IS_PTHREAD) {
192245
err(`worker: onmessage() captured an uncaught exception: ${ex}`);
193246
if (ex?.stack) err(ex.stack);
194247
#endif
195-
__emscripten_thread_crashed();
248+
if (typeof __emscripten_thread_crashed !== 'undefined') __emscripten_thread_crashed();
196249
throw ex;
197250
}
198251
};
199252

200253
self.onmessage = handleMessage;
201254

202-
} // ENVIRONMENT_IS_PTHREAD
255+
} // ENVIRONMENT_IS_PTHREAD || ENVIRONMENT_IS_PTHREAD_MANAGER

src/settings.js

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1617,6 +1617,11 @@ var WEBAUDIO_DEBUG = 0;
16171617
// repeatedly yield back to the JS event loop in order for the thread to
16181618
// actually start.
16191619
// If your application needs to be able to synchronously create new threads,
1620+
// If true, a dedicated worker is used to manage pthread lifecycles.
1621+
// This allows synchronous thread creation even when the main thread is
1622+
// blocked.
1623+
var PTHREAD_MANAGER = false;
1624+
16201625
// you can pre-create a pthread pool by specifying -sPTHREAD_POOL_SIZE=x,
16211626
// in which case the specified number of Workers will be preloaded into a pool
16221627
// before the application starts, and that many threads can then be available

src/shell.js

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,9 @@ var ENVIRONMENT_IS_SHELL = !ENVIRONMENT_IS_WEB && !ENVIRONMENT_IS_NODE && !ENVIR
9797
// The way we signal to a worker that it is hosting a pthread is to construct
9898
// it with a specific name.
9999
var ENVIRONMENT_IS_PTHREAD = ENVIRONMENT_IS_WORKER && {{{ pthreadDetection() }}}
100+
#if PTHREADS
101+
var ENVIRONMENT_IS_PTHREAD_MANAGER = ENVIRONMENT_IS_WORKER && globalThis.name == 'em-pthread-manager';
102+
#endif
100103

101104
#if MODULARIZE && ASSERTIONS
102105
if (ENVIRONMENT_IS_PTHREAD) {

0 commit comments

Comments
 (0)