X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fthread-worker.ts;h=bf8647c53b68f972c1dbe2cf806689d7889d2d2f;hb=refs%2Fheads%2Fmaster;hp=d0af904bfb16d9fd2ba396684a12fd2220d3c89e;hpb=325f50bc1777ea44abc9736ce9d780ec0c8f90e2;p=poolifier.git diff --git a/src/worker/thread-worker.ts b/src/worker/thread-worker.ts index d0af904b..8057d6f3 100644 --- a/src/worker/thread-worker.ts +++ b/src/worker/thread-worker.ts @@ -1,97 +1,98 @@ -import { AsyncResource } from 'async_hooks' -import { isMainThread, parentPort } from 'worker_threads' -import type { MessageValue } from '../utility-types' -import type { WorkerOptions } from './worker-options' +import { + isMainThread, + type MessagePort, + parentPort, + threadId, +} from 'node:worker_threads' + +import type { MessageValue } from '../utility-types.js' +import { AbstractWorker } from './abstract-worker.js' +import type { TaskFunction, TaskFunctions } from './task-functions.js' +import type { WorkerOptions } from './worker-options.js' /** - * An example worker that will be always alive, you just need to **extend** this class if you want a static pool. + * A thread worker used by a poolifier `ThreadPool`. * - * When this worker is inactive for more than 1 minute, it will send this info to the main thread, - * if you are using DynamicThreadPool, the workers created after will be killed, the min num of thread will be guaranteed. + * When this worker is inactive for more than the given `maxInactiveTime`, + * it will send a termination request to its main thread. * + * If you use a `DynamicThreadPool` the extra workers that were created will be terminated, + * but the minimum number of workers will be guaranteed. + * @typeParam Data - Type of data this worker receives from pool's execution. This can only be structured-cloneable data. + * @typeParam Response - Type of response the worker sends back to the main thread. This can only be structured-cloneable data. * @author [Alessandro Pio Ardizio](https://github.com/pioardi) * @since 0.0.1 */ -// eslint-disable-next-line @typescript-eslint/no-explicit-any -export class ThreadWorker extends AsyncResource { - protected readonly maxInactiveTime: number - protected readonly async: boolean - protected lastTask: number - protected readonly interval?: NodeJS.Timeout - protected parent?: MessagePort +export class ThreadWorker< + Data = unknown, + Response = unknown +> extends AbstractWorker { + /** + * Message port used to communicate with the main worker. + */ + private port?: MessagePort + /** + * Constructs a new poolifier thread worker. + * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked. + * @param opts - Options for the worker. + */ public constructor ( - fn: (data: Data) => Response, - public readonly opts: WorkerOptions = {} + taskFunctions: TaskFunction | TaskFunctions, + opts: WorkerOptions = {} ) { - super('worker-thread-pool:pioardi') + super(isMainThread, parentPort, taskFunctions, opts) + } - this.maxInactiveTime = this.opts.maxInactiveTime ?? 1000 * 60 - this.async = !!this.opts.async - this.lastTask = Date.now() - if (!fn) throw new Error('Fn parameter is mandatory') - // keep the worker active - if (!isMainThread) { - this.interval = setInterval( - this.checkAlive.bind(this), - this.maxInactiveTime / 2 - ) - this.checkAlive.bind(this)() - } - parentPort?.on('message', (value: MessageValue) => { - if (value?.data && value.id) { - // here you will receive messages - // console.log('This is the main thread ' + isMainThread) - if (this.async) { - this.runInAsyncScope(this.runAsync.bind(this), this, fn, value) - } else { - this.runInAsyncScope(this.run.bind(this), this, fn, value) - } - } else if (value.parent) { - // save the port to communicate with the main thread - // this will be received once - this.parent = value.parent - } else if (value.kill) { - // here is time to kill this thread, just clearing the interval - if (this.interval) clearInterval(this.interval) - this.emitDestroy() + /** @inheritDoc */ + protected handleReadyMessage (message: MessageValue): void { + if ( + message.workerId === this.id && + message.ready === false && + message.port != null + ) { + try { + this.port = message.port + this.port.on('message', this.messageListener.bind(this)) + this.sendToMainWorker({ + ready: true, + taskFunctionsProperties: this.listTaskFunctionsProperties(), + }) + } catch { + this.sendToMainWorker({ + ready: false, + taskFunctionsProperties: this.listTaskFunctionsProperties(), + }) } - }) + } } - protected checkAlive (): void { - if (Date.now() - this.lastTask > this.maxInactiveTime) { - this.parent?.postMessage({ kill: 1 }) - } + /** @inheritDoc */ + protected handleKillMessage (message: MessageValue): void { + super.handleKillMessage(message) + this.port?.unref() + this.port?.close() } - protected run ( - fn: (data?: Data) => Response, - value: MessageValue - ): void { - try { - const res = fn(value.data) - this.parent?.postMessage({ data: res, id: value.id }) - this.lastTask = Date.now() - } catch (e) { - this.parent?.postMessage({ error: e, id: value.id }) - this.lastTask = Date.now() - } + /** @inheritDoc */ + protected get id (): number { + return threadId + } + + /** @inheritDoc */ + protected readonly sendToMainWorker = ( + message: MessageValue + ): void => { + this.port?.postMessage({ + ...message, + workerId: this.id, + } satisfies MessageValue) } - protected runAsync ( - fn: (data?: Data) => Promise, - value: MessageValue - ): void { - fn(value.data) - .then(res => { - this.parent?.postMessage({ data: res, id: value.id }) - this.lastTask = Date.now() - return null - }) - .catch(e => { - this.parent?.postMessage({ error: e, id: value.id }) - this.lastTask = Date.now() - }) + /** + * @inheritDoc + */ + protected handleError (error: Error | string): string { + return error as string } }