Merge dependabot/npm_and_yarn/examples/typescript/smtp-client-pool/poolifier-4.0...
[poolifier.git] / tests / pools / cluster / dynamic.test.mjs
1 import { expect } from 'expect'
2
3 import {
4 DynamicClusterPool,
5 PoolEvents,
6 WorkerChoiceStrategies
7 } from '../../../lib/index.cjs'
8 import { TaskFunctions } from '../../test-types.cjs'
9 import { sleep, waitPoolEvents, waitWorkerEvents } from '../../test-utils.cjs'
10
11 describe('Dynamic cluster pool test suite', () => {
12 const min = 1
13 const max = 3
14 const pool = new DynamicClusterPool(
15 min,
16 max,
17 './tests/worker-files/cluster/testWorker.cjs',
18 {
19 errorHandler: e => console.error(e)
20 }
21 )
22
23 it('Verify that the function is executed in a worker cluster', async () => {
24 let result = await pool.execute({
25 function: TaskFunctions.fibonacci
26 })
27 expect(result).toBe(354224848179262000000)
28 result = await pool.execute({
29 function: TaskFunctions.factorial
30 })
31 expect(result).toBe(9.33262154439441e157)
32 })
33
34 it('Verify that new workers are created when required, max size is not exceeded and that after a while new workers will die', async () => {
35 let poolBusy = 0
36 pool.emitter.on(PoolEvents.busy, () => ++poolBusy)
37 for (let i = 0; i < max * 2; i++) {
38 pool.execute()
39 }
40 expect(pool.workerNodes.length).toBeLessThanOrEqual(max)
41 expect(pool.workerNodes.length).toBeGreaterThan(min)
42 expect(poolBusy).toBe(1)
43 const numberOfExitEvents = await waitWorkerEvents(pool, 'exit', max - min)
44 expect(numberOfExitEvents).toBe(max - min)
45 expect(pool.workerNodes.length).toBe(min)
46 })
47
48 it('Verify scale worker up and down is working', async () => {
49 for (let i = 0; i < max * 2; i++) {
50 pool.execute()
51 }
52 expect(pool.workerNodes.length).toBeGreaterThan(min)
53 await waitWorkerEvents(pool, 'exit', max - min)
54 expect(pool.workerNodes.length).toBe(min)
55 for (let i = 0; i < max * 2; i++) {
56 pool.execute()
57 }
58 expect(pool.workerNodes.length).toBeGreaterThan(min)
59 await waitWorkerEvents(pool, 'exit', max - min)
60 expect(pool.workerNodes.length).toBe(min)
61 })
62
63 it('Shutdown test', async () => {
64 const exitPromise = waitWorkerEvents(pool, 'exit', min)
65 expect(pool.emitter.eventNames()).toStrictEqual([PoolEvents.busy])
66 let poolDestroy = 0
67 pool.emitter.on(PoolEvents.destroy, () => ++poolDestroy)
68 expect(pool.emitter.eventNames()).toStrictEqual([
69 PoolEvents.busy,
70 PoolEvents.destroy
71 ])
72 await pool.destroy()
73 const numberOfExitEvents = await exitPromise
74 expect(pool.started).toBe(false)
75 expect(pool.emitter.eventNames()).toStrictEqual([
76 PoolEvents.busy,
77 PoolEvents.destroy
78 ])
79 expect(pool.readyEventEmitted).toBe(false)
80 expect(pool.workerNodes.length).toBe(0)
81 expect(numberOfExitEvents).toBe(min)
82 expect(poolDestroy).toBe(1)
83 })
84
85 it('Validation of inputs test', () => {
86 expect(() => new DynamicClusterPool(min)).toThrow(
87 'The worker file path must be specified'
88 )
89 })
90
91 it('Should work even without opts in input', async () => {
92 const pool = new DynamicClusterPool(
93 min,
94 max,
95 './tests/worker-files/cluster/testWorker.cjs'
96 )
97 const result = await pool.execute()
98 expect(result).toStrictEqual({ ok: 1 })
99 // We need to clean up the resources after our test
100 await pool.destroy()
101 })
102
103 it('Verify scale processes up and down is working when long executing task is used:hard', async () => {
104 const longRunningPool = new DynamicClusterPool(
105 min,
106 max,
107 './tests/worker-files/cluster/longRunningWorkerHardBehavior.cjs',
108 {
109 errorHandler: e => console.error(e),
110 onlineHandler: () => console.info('long executing worker is online'),
111 exitHandler: () => console.info('long executing worker exited')
112 }
113 )
114 expect(longRunningPool.workerNodes.length).toBe(min)
115 for (let i = 0; i < max * 2; i++) {
116 longRunningPool.execute()
117 }
118 expect(longRunningPool.workerNodes.length).toBe(max)
119 await waitWorkerEvents(longRunningPool, 'exit', max - min)
120 expect(longRunningPool.workerNodes.length).toBe(min)
121 expect(
122 longRunningPool.workerChoiceStrategiesContext.workerChoiceStrategies.get(
123 longRunningPool.workerChoiceStrategiesContext
124 .defaultWorkerChoiceStrategy
125 ).nextWorkerNodeKey
126 ).toBeLessThan(longRunningPool.workerNodes.length)
127 // We need to clean up the resources after our test
128 await longRunningPool.destroy()
129 })
130
131 it('Verify scale processes up and down is working when long executing task is used:soft', async () => {
132 const longRunningPool = new DynamicClusterPool(
133 min,
134 max,
135 './tests/worker-files/cluster/longRunningWorkerSoftBehavior.cjs',
136 {
137 errorHandler: e => console.error(e),
138 onlineHandler: () => console.info('long executing worker is online'),
139 exitHandler: () => console.info('long executing worker exited')
140 }
141 )
142 expect(longRunningPool.workerNodes.length).toBe(min)
143 for (let i = 0; i < max * 2; i++) {
144 longRunningPool.execute()
145 }
146 expect(longRunningPool.workerNodes.length).toBe(max)
147 await sleep(1000)
148 // Here we expect the workerNodes to be at the max size since the task is still executing
149 expect(longRunningPool.workerNodes.length).toBe(max)
150 // We need to clean up the resources after our test
151 await longRunningPool.destroy()
152 })
153
154 it('Verify that a pool with zero worker can be instantiated', async () => {
155 const pool = new DynamicClusterPool(
156 0,
157 max,
158 './tests/worker-files/cluster/testWorker.cjs'
159 )
160 expect(pool).toBeInstanceOf(DynamicClusterPool)
161 // We need to clean up the resources after our test
162 await pool.destroy()
163 })
164
165 it('Verify that a pool with zero worker works', async () => {
166 for (const workerChoiceStrategy of Object.values(WorkerChoiceStrategies)) {
167 const pool = new DynamicClusterPool(
168 0,
169 max,
170 './tests/worker-files/cluster/testWorker.cjs',
171 {
172 workerChoiceStrategy
173 }
174 )
175 expect(pool.starting).toBe(false)
176 expect(pool.readyEventEmitted).toBe(false)
177 for (let run = 0; run < 2; run++) {
178 run % 2 !== 0 && pool.enableTasksQueue(true)
179 const maxMultiplier = 4
180 const promises = new Set()
181 expect(pool.workerNodes.length).toBe(pool.info.minSize)
182 for (let i = 0; i < max * maxMultiplier; i++) {
183 promises.add(pool.execute())
184 }
185 await Promise.all(promises)
186 expect(pool.readyEventEmitted).toBe(true)
187 expect(pool.workerNodes.length).toBeGreaterThan(pool.info.minSize)
188 expect(pool.workerNodes.length).toBeLessThanOrEqual(pool.info.maxSize)
189 await waitPoolEvents(pool, PoolEvents.empty, 1)
190 expect(pool.readyEventEmitted).toBe(false)
191 expect(pool.workerNodes.length).toBe(pool.info.minSize)
192 }
193 // We need to clean up the resources after our test
194 await pool.destroy()
195 }
196 })
197 })