perf: avoid branching on pool type
[poolifier.git] / src / pools / thread / fixed.ts
... / ...
CommitLineData
1import {
2 type MessageChannel,
3 type MessagePort,
4 type TransferListItem,
5 type Worker,
6 isMainThread
7} from 'node:worker_threads'
8import type { MessageValue } from '../../utility-types'
9import { AbstractPool } from '../abstract-pool'
10import { type PoolOptions, type PoolType, PoolTypes } from '../pool'
11import { type WorkerType, WorkerTypes } from '../worker'
12
13/**
14 * Options for a poolifier thread pool.
15 */
16export type ThreadPoolOptions = PoolOptions<Worker>
17
18/**
19 * A thread pool with a fixed number of threads.
20 *
21 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
22 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
23 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
24 * @since 0.0.1
25 */
26export class FixedThreadPool<
27 Data = unknown,
28 Response = unknown
29> extends AbstractPool<Worker, Data, Response> {
30 /**
31 * Constructs a new poolifier fixed thread pool.
32 *
33 * @param numberOfThreads - Number of threads for this pool.
34 * @param filePath - Path to an implementation of a `ThreadWorker` file, which can be relative or absolute.
35 * @param opts - Options for this fixed thread pool.
36 */
37 public constructor (
38 numberOfThreads: number,
39 filePath: string,
40 opts: ThreadPoolOptions = {},
41 maximumNumberOfThreads?: number
42 ) {
43 super(numberOfThreads, filePath, opts, maximumNumberOfThreads)
44 }
45
46 /** @inheritDoc */
47 protected isMain (): boolean {
48 return isMainThread
49 }
50
51 /** @inheritDoc */
52 protected sendToWorker (
53 workerNodeKey: number,
54 message: MessageValue<Data>,
55 transferList?: TransferListItem[]
56 ): void {
57 this.workerNodes[workerNodeKey].messageChannel?.port1?.postMessage(
58 { ...message, workerId: this.getWorkerInfo(workerNodeKey).id },
59 transferList
60 )
61 }
62
63 /** @inheritDoc */
64 protected sendStartupMessageToWorker (workerNodeKey: number): void {
65 const workerNode = this.workerNodes[workerNodeKey]
66 const port2: MessagePort = (workerNode.messageChannel as MessageChannel)
67 .port2
68 workerNode.worker.postMessage(
69 {
70 ready: false,
71 workerId: this.getWorkerInfo(workerNodeKey).id,
72 port: port2
73 },
74 [port2]
75 )
76 }
77
78 /** @inheritDoc */
79 protected registerWorkerMessageListener<Message extends Data | Response>(
80 workerNodeKey: number,
81 listener: (message: MessageValue<Message>) => void
82 ): void {
83 this.workerNodes[workerNodeKey].messageChannel?.port1?.on(
84 'message',
85 listener
86 )
87 }
88
89 /** @inheritDoc */
90 protected registerOnceWorkerMessageListener<Message extends Data | Response>(
91 workerNodeKey: number,
92 listener: (message: MessageValue<Message>) => void
93 ): void {
94 this.workerNodes[workerNodeKey].messageChannel?.port1?.once(
95 'message',
96 listener
97 )
98 }
99
100 /** @inheritDoc */
101 protected deregisterWorkerMessageListener<Message extends Data | Response>(
102 workerNodeKey: number,
103 listener: (message: MessageValue<Message>) => void
104 ): void {
105 this.workerNodes[workerNodeKey].messageChannel?.port1?.off(
106 'message',
107 listener
108 )
109 }
110
111 /** @inheritDoc */
112 protected shallCreateDynamicWorker (): boolean {
113 return false
114 }
115
116 /** @inheritDoc */
117 protected checkAndEmitDynamicWorkerCreationEvents (): void {}
118
119 /** @inheritDoc */
120 protected get type (): PoolType {
121 return PoolTypes.fixed
122 }
123
124 /** @inheritDoc */
125 protected get worker (): WorkerType {
126 return WorkerTypes.thread
127 }
128
129 /** @inheritDoc */
130 protected get busy (): boolean {
131 return this.internalBusy()
132 }
133}