1import { assert, describe, it } from "@effect/vitest"
2import { Duration, Effect, Fiber, Option, PartitionedSemaphore, Semaphore } from "effect"
3import * as Scheduler from "effect/Scheduler"
4import { TestClock } from "effect/testing"
5
6describe("Semaphore", () => {
7 it.effect("basic single partition operation", () =>
8 Effect.gen(function*() {
9 const sem = yield* PartitionedSemaphore.make({ permits: 4 })
10 const messages: Array<string> = []
11
12 yield* Effect.forkChild(
13 Effect.all(
14 [0, 1, 2, 3].map((n) =>
15 sem.withPermits("partition-1", 2)(
16 Effect.delay(
17 Effect.sync(() => messages.push(`process: ${n}`)),
18 Duration.seconds(2)
19 )
20 )
21 ),
22 { concurrency: "unbounded", discard: true }
23 )
24 )
25
26 yield* TestClock.adjust(Duration.seconds(3))
27 assert.strictEqual(messages.length, 2)
28
29 yield* TestClock.adjust(Duration.seconds(3))
30 assert.strictEqual(messages.length, 4)
31 }))
32
33 it.effect("multiple partitions share total permits", () =>
34 Effect.gen(function*() {
35 const sem = yield* PartitionedSemaphore.make({ permits: 4 })
36 const messages: Array<string> = []
37
38 yield* Effect.forkChild(
39 Effect.all([
40 sem.withPermits("partition-1", 2)(
41 Effect.delay(
42 Effect.sync(() => messages.push("p1-task1")),
43 Duration.seconds(2)
44 )
45 ),
46 sem.withPermits("partition-2", 2)(
47 Effect.delay(
48 Effect.sync(() => messages.push("p2-task1")),
49 Duration.seconds(2)
50 )
51 ),
52 sem.withPermits("partition-1", 2)(
53 Effect.delay(
54 Effect.sync(() => messages.push("p1-task2")),
55 Duration.seconds(2)
56 )
57 )
58 ], { concurrency: "unbounded", discard: true })
59 )
60
61 yield* TestClock.adjust(Duration.seconds(3))
62 assert.strictEqual(messages.length, 2)
63
64 yield* TestClock.adjust(Duration.seconds(3))
65 assert.strictEqual(messages.length, 3)
66 }))
67
68 it.effect("round-robin fairness across partitions", () =>
69 Effect.gen(function*() {
70 const sem = yield* PartitionedSemaphore.make({ permits: 2 })
71 const messages: Array<string> = []
72
73 yield* Effect.forkChild(
74 sem.withPermits("partition-1", 2)(
75 Effect.delay(
76 Effect.sync(() => messages.push("p1-initial")),
77 Duration.seconds(2)
78 )
79 )
80 )
81
82 yield* TestClock.adjust(Duration.millis(1))
83
84 yield* Effect.forkChild(
85 Effect.all([
86 sem.withPermits("partition-1", 1)(Effect.sync(() => messages.push("p1-task1"))),
87 sem.withPermits("partition-1", 1)(Effect.sync(() => messages.push("p1-task2"))),
88 sem.withPermits("partition-1", 1)(Effect.sync(() => messages.push("p1-task3")))
89 ], { concurrency: "unbounded", discard: true })
90 )
91
92 yield* TestClock.adjust(Duration.millis(1))
93
94 yield* Effect.forkChild(
95 Effect.all([
96 sem.withPermits("partition-2", 1)(Effect.sync(() => messages.push("p2-task1"))),
97 sem.withPermits("partition-2", 1)(Effect.sync(() => messages.push("p2-task2"))),
98 sem.withPermits("partition-2", 1)(Effect.sync(() => messages.push("p2-task3")))
99 ], { concurrency: "unbounded", discard: true })
100 )
101
102 yield* TestClock.adjust(Duration.seconds(3))
103
104 assert.deepStrictEqual(messages, [
105 "p1-initial",
106 "p1-task1",
107 "p2-task1",
108 "p1-task2",
109 "p2-task2",
110 "p1-task3",
111 "p2-task3"
112 ])
113 }))
114
115 it.effect("requesting more permits than total returns never", () =>
116 Effect.gen(function*() {
117 const sem = yield* PartitionedSemaphore.make({ permits: 4 })
118
119 const fiber = yield* Effect.forkChild(
120 sem.withPermits("partition-1", 5)(Effect.succeed(42))
121 )
122
123 yield* TestClock.adjust(Duration.seconds(10))
124
125 assert.isUndefined(fiber.pollUnsafe())
126 }))
127
128 it.effect("single permit operations", () =>
129 Effect.gen(function*() {
130 const sem = yield* PartitionedSemaphore.make({ permits: 1 })
131 const messages: Array<string> = []
132
133 yield* Effect.forkChild(
134 Effect.all(
135 [0, 1, 2].map((n) =>
136 sem.withPermits("partition-1", 1)(
137 Effect.delay(
138 Effect.sync(() => messages.push(`task: ${n}`)),
139 Duration.seconds(1)
140 )
141 )
142 ),
143 { concurrency: "unbounded", discard: true }
144 )
145 )
146
147 yield* TestClock.adjust(Duration.seconds(1.5))
148 assert.strictEqual(messages.length, 1)
149
150 yield* TestClock.adjust(Duration.seconds(1))
151 assert.strictEqual(messages.length, 2)
152
153 yield* TestClock.adjust(Duration.seconds(1))
154 assert.strictEqual(messages.length, 3)
155 }))
156
157 it.effect("different permit sizes on same partition", () =>
158 Effect.gen(function*() {
159 const sem = yield* PartitionedSemaphore.make({ permits: 5 })
160 const messages: Array<string> = []
161
162 yield* Effect.forkChild(
163 Effect.all([
164 sem.withPermits("partition-1", 3)(
165 Effect.delay(
166 Effect.sync(() => messages.push("large")),
167 Duration.seconds(2)
168 )
169 ),
170 sem.withPermits("partition-1", 1)(
171 Effect.delay(
172 Effect.sync(() => messages.push("small-1")),
173 Duration.seconds(2)
174 )
175 ),
176 sem.withPermits("partition-1", 1)(
177 Effect.delay(
178 Effect.sync(() => messages.push("small-2")),
179 Duration.seconds(2)
180 )
181 )
182 ], { concurrency: "unbounded", discard: true })
183 )
184
185 yield* TestClock.adjust(Duration.seconds(3))
186 assert.strictEqual(messages.length, 3)
187 }))
188
189 it.effect("interruption releases permits", () =>
190 Effect.gen(function*() {
191 const sem = yield* PartitionedSemaphore.make({ permits: 2 })
192 const messages: Array<string> = []
193
194 const fiber = yield* Effect.forkChild(
195 sem.withPermits("partition-1", 2)(
196 Effect.delay(
197 Effect.sync(() => messages.push("long-task")),
198 Duration.seconds(10)
199 )
200 )
201 )
202
203 yield* TestClock.adjust(Duration.seconds(1))
204 yield* Fiber.interrupt(fiber)
205
206 yield* sem.withPermits("partition-1", 2)(
207 Effect.sync(() => messages.push("after-interrupt"))
208 )
209
210 assert.strictEqual(messages.length, 1)
211 assert.strictEqual(messages[0], "after-interrupt")
212 }))
213
214 it.effect("interruption with partial permit acquisition releases all taken permits", () =>
215 Effect.gen(function*() {
216 const sem = yield* PartitionedSemaphore.make({ permits: 3 })
217 const messages: Array<string> = []
218
219 const fiber1 = yield* Effect.forkChild(
220 sem.withPermits("partition-1", 2)(
221 Effect.delay(
222 Effect.sync(() => messages.push("first")),
223 Duration.seconds(5)
224 )
225 )
226 )
227
228 yield* TestClock.adjust(Duration.millis(1))
229
230 const fiber2 = yield* Effect.forkChild(
231 sem.withPermits("partition-2", 3)(
232 Effect.sync(() => messages.push("second"))
233 )
234 )
235
236 yield* TestClock.adjust(Duration.millis(1))
237 yield* Fiber.interrupt(fiber2)
238
239 yield* sem.withPermits("partition-3", 1)(
240 Effect.sync(() => messages.push("can-acquire-one"))
241 )
242
243 assert.deepStrictEqual(messages, ["can-acquire-one"])
244
245 yield* TestClock.adjust(Duration.seconds(5))
246 yield* Fiber.join(fiber1)
247
248 assert.deepStrictEqual(messages, ["can-acquire-one", "first"])
249
250 yield* sem.withPermits("partition-4", 3)(
251 Effect.sync(() => messages.push("all-three"))
252 )
253
254 assert.deepStrictEqual(messages, ["can-acquire-one", "first", "all-three"])
255 }))
256
257 it.effect("take interruption does not leak permits", () =>
258 Effect.gen(function*() {
259 const tasks: Array<() => void> = []
260 let shouldYield = false
261 const scheduler: Scheduler.Scheduler = {
262 executionMode: "async",
263 makeDispatcher() {
264 return {
265 scheduleTask(task, _priority) {
266 tasks.push(task)
267 },
268 flush() {
269 }
270 }
271 },
272 shouldYield: () => {
273 if (shouldYield) {
274 shouldYield = false
275 return true
276 }
277 return false
278 }
279 }
280 const step = Effect.sync(() => {
281 const task = tasks.shift()
282 if (task !== undefined) {
283 task()
284 }
285 })
286
287 const sem = yield* Semaphore.make(0)
288 const waiter = yield* sem.take(1).pipe(
289 Effect.provideService(Scheduler.Scheduler, scheduler),
290 Effect.forkChild
291 )
292
293 yield* Effect.yieldNow
294 yield* sem.release(1).pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
295 assert.isUndefined(waiter.pollUnsafe())
296
297 shouldYield = true
298 yield* step
299 assert.isUndefined(waiter.pollUnsafe())
300
301 yield* Fiber.interrupt(waiter)
302 yield* step
303
304 const result = yield* sem.withPermitsIfAvailable(1)(Effect.void)
305 assert.isTrue(Option.isSome(result))
306 }))
307
308 it.effect("withPermits interruption does not leak permits", () =>
309 Effect.gen(function*() {
310 const sem = yield* Semaphore.make(1)
311 let acquired = false
312 const waiter = yield* sem.withPermits(2)(
313 Effect.sync(() => {
314 acquired = true
315 })
316 ).pipe(Effect.forkChild)
317
318 yield* Effect.yieldNow
319 assert.isUndefined(waiter.pollUnsafe())
320
321 yield* Fiber.interrupt(waiter)
322 assert.isFalse(acquired)
323
324 const result = yield* sem.withPermitsIfAvailable(1)(Effect.void)
325 assert.isTrue(Option.isSome(result))
326
327 yield* sem.withPermits(1)(
328 Effect.sync(() => {
329 acquired = true
330 })
331 )
332 assert.isTrue(acquired)
333 }))
334
335 it.effect("withPermits interrupted at any operation does not leak permits", () =>
336 Effect.gen(function*() {
337 let operations = 0
338 const counted = new Scheduler.MixedScheduler()
339 const baseline = yield* (yield* Semaphore.make(1)).withPermits(1)(Effect.void).pipe(
340 Effect.provideService(Scheduler.Scheduler, {
341 executionMode: counted.executionMode,
342 makeDispatcher: () => counted.makeDispatcher(),
343 shouldYield: (fiber) => {
344 operations++
345 return counted.shouldYield(fiber)
346 }
347 }),
348 Effect.forkChild
349 )
350 yield* Fiber.await(baseline)
351
352 const recovered: Array<boolean> = []
353 let ranUnderPermits = false
354
355 for (let at = 1; at <= operations; at++) {
356 const sem = yield* Semaphore.make(1)
357 const base = new Scheduler.MixedScheduler()
358 let seen = 0
359 const scheduler: Scheduler.Scheduler = {
360 executionMode: base.executionMode,
361 makeDispatcher: () => base.makeDispatcher(),
362 shouldYield: (fiber) => {
363 if (++seen === at) fiber.interruptUnsafe()
364 return base.shouldYield(fiber)
365 }
366 }
367
368 const holder = yield* sem.withPermits(1)(
369 Effect.sync(() => {
370 ranUnderPermits = true
371 })
372 ).pipe(
373 Effect.provideService(Scheduler.Scheduler, scheduler),
374 Effect.forkChild
375 )
376 yield* Fiber.await(holder)
377
378 const result = yield* sem.withPermitsIfAvailable(1)(Effect.void)
379 recovered.push(Option.isSome(result))
380 }
381
382 assert.deepStrictEqual(recovered, Array.from({ length: operations }, () => true))
383 assert.isTrue(ranUnderPermits)
384 }))
385
386 it.effect("queued withPermits interrupted at any operation does not leak permits", () =>
387 Effect.gen(function*() {
388 const queuedScheduler = (
389 tasks: Array<() => void>,
390 onOperation: (fiber: Fiber.Fiber<unknown, unknown>) => void
391 ) => {
392 const scheduler: Scheduler.Scheduler = {
393 executionMode: "async",
394 makeDispatcher: () => ({
395 scheduleTask(task) {
396 tasks.push(task)
397 },
398 flush() {}
399 }),
400 shouldYield: (fiber) => {
401 onOperation(fiber)
402 return false
403 }
404 }
405 return scheduler
406 }
407
408 let operations = 0
409 const baselineSem = yield* Semaphore.make(1)
410 const baselineTasks: Array<() => void> = []
411 const baselineHolder = yield* Effect.forkChild(baselineSem.withPermits(1)(Effect.never))
412 yield* Effect.yieldNow
413 yield* Effect.forkChild(
414 baselineSem.withPermits(1)(Effect.void).pipe(
415 Effect.provideService(Scheduler.Scheduler, queuedScheduler(baselineTasks, () => operations++))
416 ),
417 { startImmediately: true }
418 )
419 yield* Fiber.interrupt(baselineHolder)
420 yield* Effect.yieldNow
421 while (baselineTasks.length > 0) baselineTasks.shift()!()
422
423 const recovered: Array<boolean> = []
424 let ranUnderPermits = false
425
426 for (let at = 1; at <= operations; at++) {
427 const sem = yield* Semaphore.make(1)
428 const tasks: Array<() => void> = []
429 let seen = 0
430 const scheduler = queuedScheduler(tasks, (fiber) => {
431 if (++seen === at) fiber.interruptUnsafe()
432 })
433
434 const holder = yield* Effect.forkChild(sem.withPermits(1)(Effect.never))
435 yield* Effect.yieldNow
436
437 yield* Effect.forkChild(
438 sem.withPermits(1)(
439 Effect.sync(() => {
440 ranUnderPermits = true
441 })
442 ).pipe(Effect.provideService(Scheduler.Scheduler, scheduler)),
443 { startImmediately: true }
444 )
445
446 yield* Fiber.interrupt(holder)
447 yield* Effect.yieldNow
448 while (tasks.length > 0) tasks.shift()!()
449
450 const result = yield* sem.withPermitsIfAvailable(1)(Effect.void)
451 recovered.push(Option.isSome(result))
452 }
453
454 assert.deepStrictEqual(recovered, Array.from({ length: operations }, () => true))
455 assert.isTrue(ranUnderPermits)
456 }))
457
458 it.effect("takeIfAvailable acquires permits when they are available", () =>
459 Effect.gen(function*() {
460 const sem = yield* Semaphore.make(2)
461
462 const acquired = yield* Semaphore.takeIfAvailable(sem, 2)
463 assert.isTrue(acquired)
464
465 const unavailable = yield* sem.takeIfAvailable(1)
466 assert.isFalse(unavailable)
467
468 const released = yield* sem.release(2)
469 assert.strictEqual(released, 2)
470 }))
471
472 it.effect("takeIfAvailable returns immediately when permits are unavailable", () =>
473 Effect.gen(function*() {
474 const sem = yield* Semaphore.make(1)
475
476 yield* sem.take(1)
477
478 const acquired = yield* Semaphore.takeIfAvailable(1)(sem)
479 assert.isFalse(acquired)
480
481 yield* sem.release(1)
482 assert.isTrue(yield* sem.takeIfAvailable(1))
483 }))
484
485 it.effect("module-level combinators delegate to the instance api", () =>
486 Effect.gen(function*() {
487 const sem = yield* Semaphore.make(1)
488
489 const taken = yield* Semaphore.take(sem, 1)
490 assert.strictEqual(taken, 1)
491
492 const released = yield* Semaphore.release(sem, 1)
493 assert.strictEqual(released, 1)
494
495 yield* Semaphore.resize(sem, 2)
496
497 const value = yield* Semaphore.withPermit(sem, Effect.succeed(1))
498 assert.strictEqual(value, 1)
499
500 const value2 = yield* Semaphore.withPermits(sem, 2, Effect.succeed(2))
501 assert.strictEqual(value2, 2)
502
503 const available = yield* Semaphore.withPermitsIfAvailable(sem, 1, Effect.succeed("ok"))
504 assert.deepStrictEqual(available, Option.some("ok"))
505
506 const piped = yield* Effect.succeed(3).pipe(Semaphore.withPermit(sem))
507 assert.strictEqual(piped, 3)
508
509 const piped2 = yield* Effect.succeed(4).pipe(Semaphore.withPermits(sem, 1))
510 assert.strictEqual(piped2, 4)
511
512 const pipedAvailable = yield* Effect.succeed("pipe").pipe(Semaphore.withPermitsIfAvailable(sem, 1))
513 assert.deepStrictEqual(pipedAvailable, Option.some("pipe"))
514
515 const releasedAll = yield* Semaphore.releaseAll(sem)
516 assert.strictEqual(releasedAll, 2)
517 }))
518
519 it.effect("exact permit match", () =>
520 Effect.gen(function*() {
521 const sem = yield* PartitionedSemaphore.make({ permits: 4 })
522 const messages: Array<string> = []
523
524 yield* Effect.forkChild(
525 Effect.all([
526 sem.withPermits("partition-1", 4)(
527 Effect.delay(
528 Effect.sync(() => messages.push("exact-match")),
529 Duration.seconds(2)
530 )
531 ),
532 sem.withPermits("partition-2", 1)(
533 Effect.delay(
534 Effect.sync(() => messages.push("waiting")),
535 Duration.seconds(2)
536 )
537 )
538 ], { concurrency: "unbounded", discard: true })
539 )
540
541 yield* TestClock.adjust(Duration.seconds(3))
542 assert.strictEqual(messages.length, 1)
543 assert.strictEqual(messages[0], "exact-match")
544
545 yield* TestClock.adjust(Duration.seconds(3))
546 assert.strictEqual(messages.length, 2)
547 }))
548
549 it.effect("many partitions", () =>
550 Effect.gen(function*() {
551 const sem = yield* PartitionedSemaphore.make({ permits: 3 })
552 const messages: Array<string> = []
553
554 yield* Effect.forkChild(
555 Effect.all(
556 Array.from({ length: 10 }, (_, i) =>
557 sem.withPermits(`partition-${i % 5}`, 1)(
558 Effect.delay(
559 Effect.sync(() => messages.push(`p${i % 5}-task`)),
560 Duration.seconds(1)
561 )
562 )),
563 { concurrency: "unbounded", discard: true }
564 )
565 )
566
567 yield* TestClock.adjust(Duration.seconds(1.5))
568 assert.strictEqual(messages.length, 3)
569
570 yield* TestClock.adjust(Duration.seconds(1))
571 assert.strictEqual(messages.length, 6)
572
573 yield* TestClock.adjust(Duration.seconds(1))
574 assert.strictEqual(messages.length, 9)
575
576 yield* TestClock.adjust(Duration.seconds(1))
577 assert.strictEqual(messages.length, 10)
578 }))
579
580 it.effect("partial permit allocation", () =>
581 Effect.gen(function*() {
582 const sem = yield* PartitionedSemaphore.make({ permits: 3 })
583 const messages: Array<string> = []
584
585 yield* Effect.forkChild(
586 sem.withPermits("partition-1", 2)(
587 Effect.delay(
588 Effect.sync(() => messages.push("first")),
589 Duration.seconds(2)
590 )
591 )
592 )
593
594 yield* TestClock.adjust(Duration.millis(100))
595
596 const fiber = yield* Effect.forkChild(
597 sem.withPermits("partition-2", 2)(
598 Effect.sync(() => messages.push("second"))
599 )
600 )
601
602 yield* TestClock.adjust(Duration.millis(100))
603
604 assert.isUndefined(fiber.pollUnsafe())
605 assert.strictEqual(messages.length, 0)
606
607 yield* TestClock.adjust(Duration.seconds(3))
608
609 assert.strictEqual(messages.length, 2)
610 assert.deepStrictEqual(messages, ["first", "second"])
611 }))
612
613 it.effect("zero permits requested", () =>
614 Effect.gen(function*() {
615 const sem = yield* PartitionedSemaphore.make({ permits: 2 })
616 let executed = false
617
618 yield* sem.withPermits("partition-1", 0)(
619 Effect.sync(() => {
620 executed = true
621 })
622 )
623
624 assert.isTrue(executed)
625 }))
626
627 it.effect("sequential tasks in same partition", () =>
628 Effect.gen(function*() {
629 const sem = yield* PartitionedSemaphore.make({ permits: 2 })
630 const messages: Array<string> = []
631
632 yield* sem.withPermits("partition-1", 2)(
633 Effect.sync(() => messages.push("task-1"))
634 )
635
636 yield* sem.withPermits("partition-1", 2)(
637 Effect.sync(() => messages.push("task-2"))
638 )
639
640 yield* sem.withPermits("partition-1", 2)(
641 Effect.sync(() => messages.push("task-3"))
642 )
643
644 assert.deepStrictEqual(messages, ["task-1", "task-2", "task-3"])
645 }))
646})