-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Experimentally implement shared workers
- Loading branch information
1 parent
2cee0af
commit f19aa09
Showing
81 changed files
with
1,560 additions
and
37 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,9 @@ | ||
| const skipTests = []; | ||
| if (process.versions.node < '12.17.0') { | ||
| skipTests.push('!test/shared-workers/!(requires-newish-node)/**'); | ||
| } | ||
|
|
||
| export default { | ||
| files: ['test/**', '!test/**/{fixtures,helpers}/**'], | ||
| files: ['test/**', '!test/**/{fixtures,helpers}/**', ...skipTests], | ||
| ignoredByWatcher: ['{coverage,docs,media,test-d,test-tap}/**'] | ||
| }; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,237 @@ | ||
| const {EventEmitter, on} = require('events'); | ||
| const {workerData, parentPort} = require('worker_threads'); // eslint-disable-line node/no-unsupported-features/node-builtins | ||
| const pkg = require('../../package.json'); | ||
|
|
||
| // Every subscription adds a listener, this is expected. | ||
| parentPort.setMaxListeners(Number.MAX_SAFE_INTEGER); | ||
|
|
||
| // Map of active test workers, used in receiveMessages() to get a reference to | ||
| // the TestWorker instance, and relevant release functions. | ||
| const activeTestWorkers = new Map(); | ||
|
|
||
| class TestWorker { | ||
| constructor(id, file) { | ||
| this.id = id; | ||
| this.file = file; | ||
| } | ||
|
|
||
| defer(fn) { | ||
| let released = false; | ||
| const release = async () => { | ||
| if (released) { | ||
| return; | ||
| } | ||
|
|
||
| released = true; | ||
| if (activeTestWorkers.has(this.id)) { | ||
| activeTestWorkers.get(this.id).releaseFns.delete(release); | ||
| } | ||
|
|
||
| await fn(); | ||
| }; | ||
|
|
||
| activeTestWorkers.get(this.id).releaseFns.add(release); | ||
|
|
||
| return release; | ||
| } | ||
|
|
||
| publish(data) { | ||
| return publishMessage(this, data); | ||
| } | ||
|
|
||
| async * subscribe() { | ||
| yield * receiveMessages(this); | ||
| } | ||
| } | ||
|
|
||
| class ReceivedMessage { | ||
| constructor(testWorker, id, data) { | ||
| this.testWorker = testWorker; | ||
| this.id = id; | ||
| this.data = data; | ||
| } | ||
|
|
||
| reply(data) { | ||
| return publishMessage(this.testWorker, data, this.id); | ||
| } | ||
| } | ||
|
|
||
| // Ensure that, no matter how often it's received, we have a stable message | ||
| // object. | ||
| const messageCache = new WeakMap(); | ||
|
|
||
| async function * receiveMessages(fromTestWorker, replyTo) { | ||
| for await (const [message] of on(parentPort, 'message')) { | ||
| if (fromTestWorker !== undefined) { | ||
| if (message.type === 'deregister-test-worker' && message.id === fromTestWorker.id) { | ||
| return; | ||
| } | ||
|
|
||
| if (message.type === 'message' && message.testWorkerId !== fromTestWorker.id) { | ||
| continue; | ||
| } | ||
| } | ||
|
|
||
| if (message.type !== 'message') { | ||
| continue; | ||
| } | ||
|
|
||
| if (replyTo === undefined && message.replyTo !== undefined) { | ||
| continue; | ||
| } | ||
|
|
||
| if (replyTo !== undefined && message.replyTo !== replyTo) { | ||
| continue; | ||
| } | ||
|
|
||
| const active = activeTestWorkers.get(message.testWorkerId); | ||
| if (active === undefined) { | ||
| return; | ||
| } | ||
|
|
||
| let received = messageCache.get(message); | ||
| if (received === undefined) { | ||
| received = new ReceivedMessage(active.instance, message.messageId, message.data); | ||
| messageCache.set(message, received); | ||
| } | ||
|
|
||
| yield received; | ||
| } | ||
| } | ||
|
|
||
| let messageCounter = 0; | ||
| const messageIdPrefix = `${workerData.id}/message`; | ||
| const nextMessageId = () => `${messageIdPrefix}/${++messageCounter}`; | ||
|
|
||
| function publishMessage(testWorker, data, replyTo) { | ||
| const id = nextMessageId(); | ||
| parentPort.postMessage({ | ||
| type: 'message', | ||
| messageId: id, | ||
| testWorkerId: testWorker.id, | ||
| data, | ||
| replyTo | ||
| }); | ||
|
|
||
| return { | ||
| id, | ||
| async * replies() { | ||
| yield * receiveMessages(testWorker, id); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| function broadcastMessage(data) { | ||
| const id = nextMessageId(); | ||
| parentPort.postMessage({ | ||
| type: 'broadcast', | ||
| messageId: id, | ||
| data | ||
| }); | ||
|
|
||
| return { | ||
| id, | ||
| async * replies() { | ||
| yield * receiveMessages(undefined, id); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| const loadFactory = async () => { | ||
| try { | ||
| const mod = require(workerData.filename); | ||
| if (typeof mod === 'function') { | ||
| return mod; | ||
| } | ||
|
|
||
| return mod.default; | ||
| } catch (error) { | ||
| if (error && (error.code === 'ERR_REQUIRE_ESM' || (error.code === 'MODULE_NOT_FOUND' && workerData.filename.startsWith('file://')))) { | ||
| const {default: factory} = await import(workerData.filename); // eslint-disable-line node/no-unsupported-features/es-syntax | ||
| return factory; | ||
| } | ||
|
|
||
| throw error; | ||
| } | ||
| }; | ||
|
|
||
| let signalAvailable = () => { | ||
| parentPort.postMessage({type: 'available'}); | ||
| signalAvailable = () => {}; | ||
| }; | ||
|
|
||
| let fatal; | ||
| loadFactory(workerData.filename).then(factory => { | ||
| if (typeof factory !== 'function') { | ||
| throw new TypeError(`Missing default factory function export for shared worker plugin at ${workerData.filename}`); | ||
| } | ||
|
|
||
| factory({ | ||
| negotiateProtocol(supported) { | ||
| if (!supported.includes('experimental')) { | ||
| fatal = new Error(`This version of AVA (${pkg.version}) is not compatible with shared worker plugin at ${workerData.filename}`); | ||
| throw fatal; | ||
| } | ||
|
|
||
| const events = new EventEmitter(); | ||
| const produceTestWorker = instance => events.emit('testWorker', instance); | ||
|
|
||
| parentPort.on('message', async message => { | ||
| if (message.type === 'register-test-worker') { | ||
| const {id, file} = message; | ||
| const instance = new TestWorker(id, file); | ||
|
|
||
| activeTestWorkers.set(id, {instance, releaseFns: new Set()}); | ||
|
|
||
| produceTestWorker(instance); | ||
| } | ||
|
|
||
| if (message.type === 'deregister-test-worker') { | ||
| const {id} = message; | ||
| const {releaseFns} = activeTestWorkers.get(id); | ||
| activeTestWorkers.delete(id); | ||
|
|
||
| // Run possibly asynchronous release functions serially, in reverse | ||
| // order. Any error will crash the worker. | ||
| for await (const fn of [...releaseFns].reverse()) { | ||
| await fn(); | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| return { | ||
| initialData: workerData.initialData, | ||
| protocol: 'experimental', | ||
|
|
||
| ready() { | ||
| signalAvailable(); | ||
| return this; | ||
| }, | ||
|
|
||
| broadcast(data) { | ||
| return broadcastMessage(data); | ||
| }, | ||
|
|
||
| async * subscribe() { | ||
| yield * receiveMessages(); | ||
| }, | ||
|
|
||
| async * testWorkers() { | ||
| for await (const [worker] of on(events, 'testWorker')) { | ||
| yield worker; | ||
| } | ||
| } | ||
| }; | ||
| } | ||
| }); | ||
| }).catch(error => { | ||
| if (fatal === undefined) { | ||
| fatal = error; | ||
| } | ||
| }).finally(() => { | ||
| if (fatal !== undefined) { | ||
| process.nextTick(() => { | ||
| throw fatal; | ||
| }); | ||
| } | ||
| }); |
Oops, something went wrong.