-
-
Notifications
You must be signed in to change notification settings - Fork 29
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
6286b81
commit 7872f5f
Showing
15 changed files
with
768 additions
and
208 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
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,110 @@ | ||
| import { stderr, stdout } from 'src/utils' | ||
| import { | ||
| ReadyMessage, | ||
| RequestMessage, | ||
| ResponseMessage, | ||
| StartupMessage, | ||
| TinypoolWorkerMessage, | ||
| } from '../common' | ||
| import { getHandler, throwInNextTick } from './utils' | ||
|
|
||
| type Message = | ||
| | (StartupMessage & TinypoolWorkerMessage & { source: 'channel' }) | ||
| | (RequestMessage & TinypoolWorkerMessage & { source: 'port' }) | ||
|
|
||
| process.__tinypool_state__ = { | ||
| isChildProcess: true, | ||
| isTinyWorker: true, | ||
| workerData: null, | ||
| workerId: process.pid, | ||
| } | ||
|
|
||
| process.on('message', (message: Message) => { | ||
| if (!message.__tinypool_worker_message__) { | ||
| // Message was not for port or channel | ||
| // It's likely end-users own communication between main and worker | ||
| return | ||
| } | ||
|
|
||
| if (message.source === 'channel') { | ||
| const { filename, name } = message | ||
|
|
||
| ;(async function () { | ||
| if (filename !== null) { | ||
| await getHandler(filename, name) | ||
| } | ||
|
|
||
| const readyMessage: ReadyMessage & | ||
| TinypoolWorkerMessage & { target: 'channel' } = { | ||
| ready: true, | ||
| target: 'channel', | ||
| __tinypool_worker_message__: true, | ||
| } | ||
| process.send!(readyMessage) | ||
| })().catch(throwInNextTick) | ||
|
|
||
| return | ||
| } | ||
|
|
||
| if (message.source === 'port') { | ||
| return onMessage(message).catch(throwInNextTick) | ||
| } | ||
|
|
||
| return | ||
| }) | ||
|
|
||
| async function onMessage(message: RequestMessage) { | ||
| const { taskId, task, filename, name } = message | ||
| let response: ResponseMessage & TinypoolWorkerMessage & { target: 'port' } | ||
|
|
||
| try { | ||
| const handler = await getHandler(filename, name) | ||
| if (handler === null) { | ||
| throw new Error(`No handler function exported from ${filename}`) | ||
| } | ||
| const result = await handler(task) | ||
| response = { | ||
| target: 'port', | ||
| __tinypool_worker_message__: true, | ||
| taskId, | ||
| result: result, | ||
| error: null, | ||
| usedMemory: process.memoryUsage().heapUsed, | ||
| } | ||
|
|
||
| // If the task used e.g. console.log(), wait for the stream to drain | ||
| // before potentially entering the `Atomics.wait()` loop, and before | ||
| // returning the result so that messages will always be printed even | ||
| // if the process would otherwise be ready to exit. | ||
| if (stdout()?.writableLength! > 0) { | ||
| await new Promise((resolve) => process.stdout.write('', resolve)) | ||
| } | ||
| if (stderr()?.writableLength! > 0) { | ||
| await new Promise((resolve) => process.stderr.write('', resolve)) | ||
| } | ||
| } catch (error) { | ||
| response = { | ||
| target: 'port', | ||
| __tinypool_worker_message__: true, | ||
| taskId, | ||
| result: null, | ||
| error: serializeError(error), | ||
| usedMemory: process.memoryUsage().heapUsed, | ||
| } | ||
| } | ||
|
|
||
| process.send!(response) | ||
| } | ||
|
|
||
| function serializeError(error: unknown) { | ||
| if (error instanceof Error) { | ||
| return { | ||
| ...error, | ||
| name: error.name, | ||
| stack: error.stack, | ||
| message: error.message, | ||
| } | ||
| } | ||
|
|
||
| return String(error) | ||
| } |
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,74 @@ | ||
| import { pathToFileURL } from 'url' | ||
|
|
||
| // Get `import(x)` as a function that isn't transpiled to `require(x)` by | ||
| // TypeScript for dual ESM/CJS support. | ||
| // Load this lazily, so that there is no warning about the ESM loader being | ||
| // experimental (on Node v12.x) until we actually try to use it. | ||
| let importESMCached: (specifier: string) => Promise<any> | undefined | ||
|
|
||
| function getImportESM() { | ||
| if (importESMCached === undefined) { | ||
| // eslint-disable-next-line no-new-func | ||
| importESMCached = new Function( | ||
| 'specifier', | ||
| 'return import(specifier)' | ||
| ) as typeof importESMCached | ||
| } | ||
| return importESMCached | ||
| } | ||
|
|
||
| const handlerCache: Map<string, Function> = new Map() | ||
|
|
||
| // Look up the handler function that we call when a task is posted. | ||
| // This is either going to be "the" export from a file, or the default export. | ||
| export async function getHandler( | ||
| filename: string, | ||
| name: string | ||
| ): Promise<Function | null> { | ||
| let handler = handlerCache.get(`${filename}/${name}`) | ||
| if (handler !== undefined) { | ||
| return handler | ||
| } | ||
|
|
||
| try { | ||
| // With our current set of TypeScript options, this is transpiled to | ||
| // `require(filename)`. | ||
| const handlerModule = await import(filename) | ||
|
|
||
| // Check if the default export is an object, because dynamic import | ||
| // resolves with `{ default: { default: [Function] } }` for CJS modules. | ||
| handler = | ||
| (typeof handlerModule.default !== 'function' && handlerModule.default) || | ||
| handlerModule | ||
|
|
||
| if (typeof handler !== 'function') { | ||
| handler = await (handler as any)[name] | ||
| } | ||
| } catch {} | ||
| if (typeof handler !== 'function') { | ||
| handler = await getImportESM()(pathToFileURL(filename).href) | ||
| if (typeof handler !== 'function') { | ||
| handler = await (handler as any)[name] | ||
| } | ||
| } | ||
| if (typeof handler !== 'function') { | ||
| return null | ||
| } | ||
|
|
||
| // Limit the handler cache size. This should not usually be an issue and is | ||
| // only provided for pathological cases. | ||
| if (handlerCache.size > 1000) { | ||
| // @ts-ignore | ||
| const [[key]] = handlerCache | ||
| handlerCache.delete(key) | ||
| } | ||
|
|
||
| handlerCache.set(`${filename}/${name}`, handler) | ||
| return handler | ||
| } | ||
|
|
||
| export function throwInNextTick(error: Error) { | ||
| process.nextTick(() => { | ||
| throw error | ||
| }) | ||
| } |
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
Oops, something went wrong.