X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker-node.ts;h=1a4df6cd4afd479607a267e0d5ada4744a312e9e;hb=5b95eb9bcafda56ce57003da834cf4e153bb0509;hp=9fa4ea3201e3ca0db3d3aa9f79153adc3f1625d5;hpb=799c9e089724b6402ae72e0a60719017055a6669;p=poolifier.git diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts index 9fa4ea32..1a4df6cd 100644 --- a/src/pools/worker-node.ts +++ b/src/pools/worker-node.ts @@ -1,9 +1,16 @@ -import { MessageChannel } from 'node:worker_threads' import { EventEmitter } from 'node:events' +import { MessageChannel } from 'node:worker_threads' + import { CircularArray } from '../circular-array.js' -import type { Task } from '../utility-types.js' -import { DEFAULT_TASK_NAME, getWorkerId, getWorkerType } from '../utils.js' import { Deque } from '../deque.js' +import type { Task } from '../utility-types.js' +import { DEFAULT_TASK_NAME } from '../utils.js' +import { + checkWorkerNodeArguments, + createWorker, + getWorkerId, + getWorkerType +} from './utils.js' import { type EventHandler, type IWorker, @@ -15,7 +22,6 @@ import { WorkerTypes, type WorkerUsage } from './worker.js' -import { checkWorkerNodeArguments, createWorker } from './utils.js' /** * Worker node. @@ -78,8 +84,7 @@ export class WorkerNode const tasksQueueSize = this.tasksQueue.push(task) if (this.hasBackPressure() && !this.onBackPressureStarted) { this.onBackPressureStarted = true - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - this.emit('backPressure', { workerId: this.info.id! }) + this.emit('backPressure', { workerId: this.info.id }) this.onBackPressureStarted = false } return tasksQueueSize @@ -90,8 +95,7 @@ export class WorkerNode const tasksQueueSize = this.tasksQueue.unshift(task) if (this.hasBackPressure() && !this.onBackPressureStarted) { this.onBackPressureStarted = true - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - this.emit('backPressure', { workerId: this.info.id! }) + this.emit('backPressure', { workerId: this.info.id }) this.onBackPressureStarted = false } return tasksQueueSize @@ -134,6 +138,7 @@ export class WorkerNode this.removeAllListeners() switch (this.info.type) { case WorkerTypes.thread: + this.worker.unref?.() await this.worker.terminate?.() break case WorkerTypes.cluster: