X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fthread%2Ffixed.ts;h=a61c10b2346c8a44c4094605dfe6aa25787b8f09;hb=deb85c12b77faf6974551cefcd9676e62a392086;hp=36eddace5ec149c41b8daafbd4ada0b91babb32d;hpb=325f50bc1777ea44abc9736ce9d780ec0c8f90e2;p=poolifier.git diff --git a/src/pools/thread/fixed.ts b/src/pools/thread/fixed.ts index 36eddace..a61c10b2 100644 --- a/src/pools/thread/fixed.ts +++ b/src/pools/thread/fixed.ts @@ -1,141 +1,84 @@ import { isMainThread, MessageChannel, SHARE_ENV, Worker } from 'worker_threads' import type { Draft, MessageValue } from '../../utility-types' +import type { PoolOptions } from '../abstract-pool' +import { AbstractPool } from '../abstract-pool' -export type WorkerWithMessageChannel = Worker & Draft - -export interface FixedThreadPoolOptions { - /** - * A function that will listen for error event on each worker thread. - */ - errorHandler?: (this: Worker, e: Error) => void - /** - * A function that will listen for online event on each worker thread. - */ - onlineHandler?: (this: Worker) => void - /** - * A function that will listen for exit event on each worker thread. - */ - exitHandler?: (this: Worker, code: number) => void - /** - * This is just to avoid not useful warnings message, is used to set `maxListeners` on event emitters (workers are event emitters). - * - * @default 1000 - */ - maxTasks?: number -} +/** + * A thread worker with message channels for communication between main thread and thread worker. + */ +export type ThreadWorkerWithMessageChannel = Worker & Draft /** - * A thread pool with a static number of threads, is possible to execute tasks in sync or async mode as you prefer. + * A thread pool with a fixed number of threads. * - * This pool will select the worker thread in a round robin fashion. + * It is possible to perform tasks in sync or asynchronous mode as you prefer. + * + * This pool selects the threads in a round robin fashion. + * + * @template Data Type of data sent to the worker. This can only be serializable data. + * @template Response Type of response of execution. This can only be serializable data. * * @author [Alessandro Pio Ardizio](https://github.com/pioardi) * @since 0.0.1 */ -// eslint-disable-next-line @typescript-eslint/no-explicit-any -export class FixedThreadPool { - public readonly workers: WorkerWithMessageChannel[] = [] - public nextWorker: number = 0 - - // threadId as key and an integer value - public readonly tasks: Map = new Map< - WorkerWithMessageChannel, - number - >() - - protected id: number = 0 - +export class FixedThreadPool< + Data = unknown, + Response = unknown +> extends AbstractPool { /** - * @param numThreads Num of threads for this worker pool. - * @param filePath A file path with implementation of `ThreadWorker` class, relative path is fine. - * @param opts An object with possible options for example `errorHandler`, `onlineHandler`. Default: `{ maxTasks: 1000 }` + * Constructs a new poolifier fixed thread pool. + * + * @param numberOfThreads Number of threads for this pool. + * @param filePath Path to an implementation of a `ThreadWorker` file, which can be relative or absolute. + * @param opts Options for this fixed thread pool. Default: `{ maxTasks: 1000 }` */ public constructor ( - public readonly numThreads: number, - public readonly filePath: string, - public readonly opts: FixedThreadPoolOptions = { maxTasks: 1000 } + numberOfThreads: number, + filePath: string, + opts: PoolOptions = { maxTasks: 1000 } ) { - if (!isMainThread) { - throw new Error('Cannot start a thread pool from a worker thread !!!') - } - // TODO christopher 2021-02-07: Improve this check e.g. with a pattern or blank check - if (!this.filePath) { - throw new Error('Please specify a file with a worker implementation') - } + super(numberOfThreads, filePath, opts) + } - for (let i = 1; i <= this.numThreads; i++) { - this.newWorker() - } + protected isMain (): boolean { + return isMainThread } - public async destroy (): Promise { - for (const worker of this.workers) { - await worker.terminate() - } + protected async destroyWorker ( + worker: ThreadWorkerWithMessageChannel + ): Promise { + this.sendToWorker(worker, { kill: 1 }) + await worker.terminate() } - /** - * Execute the task specified into the constructor with the data parameter. - * - * @param data The input for the task specified. - * @returns Promise that is resolved when the task is done. - */ - public execute (data: Data): Promise { - // configure worker to handle message with the specified task - const worker = this.chooseWorker() - const previousWorkerIndex = this.tasks.get(worker) - if (previousWorkerIndex !== undefined) { - this.tasks.set(worker, previousWorkerIndex + 1) - } else { - throw Error('Worker could not be found in tasks map') - } - const id = ++this.id - const res = this.internalExecute(worker, id) - worker.postMessage({ data: data || {}, id: id }) - return res + protected sendToWorker ( + worker: ThreadWorkerWithMessageChannel, + message: MessageValue + ): void { + worker.postMessage(message) } - protected internalExecute ( - worker: WorkerWithMessageChannel, - id: number - ): Promise { - return new Promise((resolve, reject) => { - const listener: (message: MessageValue) => void = message => { - if (message.id === id) { - worker.port2?.removeListener('message', listener) - const previousWorkerIndex = this.tasks.get(worker) - if (previousWorkerIndex !== undefined) { - this.tasks.set(worker, previousWorkerIndex + 1) - } else { - throw Error('Worker could not be found in tasks map') - } - if (message.error) reject(message.error) - else resolve(message.data as Response) - } - } - worker.port2?.on('message', listener) - }) + protected registerWorkerMessageListener ( + messageChannel: ThreadWorkerWithMessageChannel, + listener: (message: MessageValue) => void + ): void { + messageChannel.port2?.on('message', listener) } - protected chooseWorker (): WorkerWithMessageChannel { - if (this.workers.length - 1 === this.nextWorker) { - this.nextWorker = 0 - return this.workers[this.nextWorker] - } else { - this.nextWorker++ - return this.workers[this.nextWorker] - } + protected unregisterWorkerMessageListener ( + messageChannel: ThreadWorkerWithMessageChannel, + listener: (message: MessageValue) => void + ): void { + messageChannel.port2?.removeListener('message', listener) } - protected newWorker (): WorkerWithMessageChannel { - const worker: WorkerWithMessageChannel = new Worker(this.filePath, { + protected createWorker (): ThreadWorkerWithMessageChannel { + return new Worker(this.filePath, { env: SHARE_ENV }) - worker.on('error', this.opts.errorHandler ?? (() => {})) - worker.on('online', this.opts.onlineHandler ?? (() => {})) - // TODO handle properly when a thread exit - worker.on('exit', this.opts.exitHandler ?? (() => {})) - this.workers.push(worker) + } + + protected afterWorkerSetup (worker: ThreadWorkerWithMessageChannel): void { const { port1, port2 } = new MessageChannel() worker.postMessage({ parent: port1 }, [port1]) worker.port1 = port1 @@ -143,8 +86,5 @@ export class FixedThreadPool { // we will attach a listener for every task, // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size worker.port2.setMaxListeners(this.opts.maxTasks ?? 1000) - // init tasks map - this.tasks.set(worker, 0) - return worker } }