X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fworker-node.ts;h=03fc0df192f3ce6689306ba3f6eb38c12dabed40;hb=241cf15ea44259be2ec66970c3137f99071f4e41;hp=fbd00096a769d53cd5d3d9415fae64c428cfb3ee;hpb=1f0766e784d0391b6ed084f2fcbc5b7e09d25fa1;p=poolifier.git diff --git a/src/pools/worker-node.ts b/src/pools/worker-node.ts index fbd00096..03fc0df1 100644 --- a/src/pools/worker-node.ts +++ b/src/pools/worker-node.ts @@ -9,6 +9,8 @@ import { } from '../utils' import { Deque } from '../deque' import { + type BackPressureCallback, + type EmptyQueueCallback, type IWorker, type IWorkerNode, type WorkerInfo, @@ -30,18 +32,18 @@ implements IWorkerNode { /** @inheritdoc */ public readonly info: WorkerInfo /** @inheritdoc */ - public messageChannel?: MessageChannel - /** @inheritdoc */ public usage: WorkerUsage /** @inheritdoc */ + public messageChannel?: MessageChannel + /** @inheritdoc */ public tasksQueueBackPressureSize: number /** @inheritdoc */ - public onBackPressure?: (workerId: number) => void + public onBackPressure?: BackPressureCallback /** @inheritdoc */ - public onEmptyQueue?: (workerId: number) => void - private readonly taskFunctionsUsage: Map + public onEmptyQueue?: EmptyQueueCallback private readonly tasksQueue: Deque> private onEmptyQueueCount: number + private readonly taskFunctionsUsage: Map /** * Constructs a new worker node. @@ -75,14 +77,14 @@ implements IWorkerNode { } this.worker = worker this.info = this.initWorkerInfo(worker, workerType) + this.usage = this.initWorkerUsage() if (workerType === WorkerTypes.thread) { this.messageChannel = new MessageChannel() } - this.usage = this.initWorkerUsage() - this.taskFunctionsUsage = new Map() - this.tasksQueue = new Deque>() this.tasksQueueBackPressureSize = tasksQueueBackPressureSize + this.tasksQueue = new Deque>() this.onEmptyQueueCount = 0 + this.taskFunctionsUsage = new Map() } /** @inheritdoc */ @@ -180,13 +182,12 @@ implements IWorkerNode { private async startOnEmptyQueue (): Promise { if ( this.onEmptyQueueCount > 0 && - this.usage.tasks.executing > 0 && - this.tasksQueue.size > 0 + (this.usage.tasks.executing > 0 || this.tasksQueue.size > 0) ) { this.onEmptyQueueCount = 0 return } - (this.onEmptyQueue as (workerId: number) => void)(this.info.id as number) + (this.onEmptyQueue as EmptyQueueCallback)(this.info.id as number) ++this.onEmptyQueueCount await sleep(exponentialDelay(this.onEmptyQueueCount)) await this.startOnEmptyQueue()