-
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
b93cce1
commit aba2021
Showing
87 changed files
with
1,645 additions
and
42 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,251 @@ | ||
| 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'); | ||
|
|
||
| // Used to forward messages received over the `parentPort`. Every subscription | ||
| // adds a listener, so do not enforce any maximums. | ||
| const events = new EventEmitter().setMaxListeners(0); | ||
|
|
||
| // 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(events, '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); | ||
| // It is possible for a message to have been buffering for so long — perhaps | ||
| // due to the caller waiting before iterating to the next message — that the | ||
| // test worker has been deregistered. Ignore such messages. | ||
| // | ||
| // (This is really hard to write a test for, however!) | ||
| if (active === undefined) { | ||
| continue; | ||
| } | ||
|
|
||
| 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); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| async function loadFactory() { | ||
| 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 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(); | ||
| } | ||
|
|
||
| parentPort.postMessage({ | ||
| type: 'deregistered-test-worker', | ||
| id | ||
| }); | ||
| } | ||
|
|
||
| // Wait for a turn of the event loop, to allow new subscriptions to be | ||
| // set up in response to the previous message. | ||
| setImmediate(() => events.emit('message', message)); | ||
| }); | ||
|
|
||
| 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.