1import { assert, describe, it } from "@effect/vitest"
2import { Effect, Exit, Fiber, Option, PartitionedSemaphore, Scheduler } from "effect"
3
4describe("PartitionedSemaphore", () => {
5 it.effect("module-level combinators delegate to the instance api", () =>
6 Effect.gen(function*() {
7 const sem = yield* PartitionedSemaphore.make<string>({ permits: 2 })
8
9 assert.strictEqual(PartitionedSemaphore.capacity(sem), 2)
10 assert.strictEqual(yield* PartitionedSemaphore.available(sem), 2)
11
12 yield* PartitionedSemaphore.take(sem, "a", 1)
13 assert.strictEqual(yield* PartitionedSemaphore.available(sem), 1)
14
15 const value = yield* PartitionedSemaphore.withPermit(sem, "a", Effect.succeed(1))
16 assert.strictEqual(value, 1)
17
18 const released = yield* PartitionedSemaphore.release(sem, 1)
19 assert.strictEqual(released, 2)
20
21 const value2 = yield* PartitionedSemaphore.withPermits(sem, "b", 2, Effect.succeed(2))
22 assert.strictEqual(value2, 2)
23
24 const available = yield* PartitionedSemaphore.withPermitsIfAvailable(sem, 1, Effect.succeed("ok"))
25 assert.deepStrictEqual(available, Option.some("ok"))
26
27 const piped = yield* Effect.succeed(3).pipe(PartitionedSemaphore.withPermit(sem, "c"))
28 assert.strictEqual(piped, 3)
29
30 const piped2 = yield* Effect.succeed(4).pipe(PartitionedSemaphore.withPermits(sem, "c", 1))
31 assert.strictEqual(piped2, 4)
32
33 const pipedAvailable = yield* Effect.succeed("pipe").pipe(PartitionedSemaphore.withPermitsIfAvailable(sem, 1))
34 assert.deepStrictEqual(pipedAvailable, Option.some("pipe"))
35 }))
36
37 it.effect("zero permits run immediately", () =>
38 Effect.gen(function*() {
39 const sem = yield* PartitionedSemaphore.make<string>({ permits: 1 })
40 let executed = false
41
42 yield* PartitionedSemaphore.withPermits(
43 sem,
44 "a",
45 0,
46 Effect.sync(() => {
47 executed = true
48 })
49 )
50
51 assert.isTrue(executed)
52 assert.strictEqual(yield* PartitionedSemaphore.available(sem), 1)
53 }))
54
55 it.effect("withPermitsIfAvailable does not block or run when unavailable", () =>
56 Effect.gen(function*() {
57 const sem = yield* PartitionedSemaphore.make<string>({ permits: 1 })
58 let executed = false
59
60 yield* PartitionedSemaphore.take(sem, "a", 1)
61
62 const result = yield* PartitionedSemaphore.withPermitsIfAvailable(
63 sem,
64 1,
65 Effect.sync(() => {
66 executed = true
67 return "ok"
68 })
69 )
70
71 assert.deepStrictEqual(result, Option.none())
72 assert.isFalse(executed)
73 }))
74
75 it.effect("interrupting a partially satisfied waiter releases all acquired permits", () =>
76 Effect.gen(function*() {
77 const sem = yield* PartitionedSemaphore.make<string>({ permits: 4 })
78
79 yield* PartitionedSemaphore.take(sem, "a", 3)
80 const waiter = yield* PartitionedSemaphore.take(sem, "b", 3).pipe(Effect.forkChild)
81 yield* Effect.yieldNow
82
83 yield* PartitionedSemaphore.release(sem, 1)
84 assert.strictEqual(yield* PartitionedSemaphore.available(sem), 0)
85 yield* Fiber.interrupt(waiter)
86 assert.strictEqual(yield* PartitionedSemaphore.available(sem), 2)
87 yield* PartitionedSemaphore.release(sem, 2)
88
89 assert.strictEqual(yield* PartitionedSemaphore.available(sem), 4)
90 yield* PartitionedSemaphore.take(sem, "c", 4)
91 }))
92
93 it.effect("interrupting an immediately satisfied take before completion restores permits", () =>
94 Effect.gen(function*() {
95 const sem = yield* PartitionedSemaphore.make<string>({ permits: 1 })
96 const tasks: Array<() => void> = []
97 let paused = false
98 const scheduler: Scheduler.Scheduler = {
99 executionMode: "async",
100 makeDispatcher: () => ({
101 scheduleTask: (task) => {
102 tasks.push(task)
103 },
104 flush() {}
105 }),
106 shouldYield: () => !paused && (paused = Effect.runSync(sem.available) === 0)
107 }
108
109 const taker = yield* sem.take("a", 1).pipe(
110 Effect.provideService(Scheduler.Scheduler, scheduler),
111 Effect.forkChild({ startImmediately: true })
112 )
113 assert.isTrue(paused)
114 taker.interruptUnsafe()
115 while (tasks.length > 0) tasks.shift()!()
116 const exit = yield* Fiber.await(taker)
117
118 assert.isTrue(Exit.isFailure(exit))
119 assert.strictEqual(yield* sem.available, 1)
120 }))
121
122 it.effect("competing takes cannot acquire the same permit across scheduler yields", () =>
123 Effect.gen(function*() {
124 for (let yieldPoint = 0; yieldPoint < 20; yieldPoint++) {
125 const sem = yield* PartitionedSemaphore.make<string>({ permits: 1 })
126 const tasks: Array<() => void> = []
127 let checks = 0
128 const scheduler: Scheduler.Scheduler = {
129 executionMode: "async",
130 makeDispatcher: () => ({
131 scheduleTask: (task) => {
132 tasks.push(task)
133 },
134 flush() {}
135 }),
136 shouldYield: () => ++checks === yieldPoint
137 }
138
139 const first = yield* sem.take("first", 1).pipe(
140 Effect.provideService(Scheduler.Scheduler, scheduler),
141 Effect.forkChild({ startImmediately: true })
142 )
143 const second = yield* sem.take("second", 1).pipe(Effect.forkChild({ startImmediately: true }))
144 assert.isAtLeast(yield* sem.available, 0)
145 while (tasks.length > 0) tasks.shift()!()
146
147 const firstExit = first.pollUnsafe()
148 const secondExit = second.pollUnsafe()
149 const firstSucceeded = firstExit !== undefined && Exit.isSuccess(firstExit)
150 const secondSucceeded = secondExit !== undefined && Exit.isSuccess(secondExit)
151 assert.isFalse(firstSucceeded && secondSucceeded, "yield point " + yieldPoint)
152 assert.isAtLeast(yield* sem.available, 0, "yield point " + yieldPoint)
153
154 yield* Fiber.interrupt(first)
155 yield* Fiber.interrupt(second)
156 yield* sem.release(Number(firstSucceeded) + Number(secondSucceeded))
157 assert.strictEqual(yield* sem.available, 1, "yield point " + yieldPoint)
158 }
159 }))
160
161 it.effect("interrupting withPermitsIfAvailable before cleanup is installed restores permits", () =>
162 Effect.gen(function*() {
163 const sem = yield* PartitionedSemaphore.make<string>({ permits: 1 })
164 const tasks: Array<() => void> = []
165 let paused = false
166 let ran = false
167 const scheduler: Scheduler.Scheduler = {
168 executionMode: "async",
169 makeDispatcher: () => ({
170 scheduleTask: (task) => {
171 tasks.push(task)
172 },
173 flush() {}
174 }),
175 shouldYield: () => !paused && (paused = Effect.runSync(sem.available) === 0)
176 }
177
178 const user = Effect.sync(() => {
179 ran = true
180 })
181 const fiber = yield* sem.withPermitsIfAvailable(1)(user).pipe(
182 Effect.provideService(Scheduler.Scheduler, scheduler),
183 Effect.forkChild({ startImmediately: true })
184 )
185 assert.isTrue(paused)
186 fiber.interruptUnsafe()
187 while (tasks.length > 0) tasks.shift()!()
188 const exit = yield* Fiber.await(fiber)
189
190 assert.isTrue(Exit.isFailure(exit))
191 assert.isFalse(ran)
192 assert.strictEqual(yield* sem.available, 1)
193 }))
194
195 for (const key of ["other", "same"]) {
196 it.effect(`interrupting a resumed waiter preserves the next ${key}-key waiter`, () =>
197 Effect.gen(function*() {
198 const tasks: Array<() => void> = []
199 let shouldYield = false
200 const scheduler: Scheduler.Scheduler = {
201 executionMode: "async",
202 makeDispatcher() {
203 return {
204 scheduleTask(task) {
205 tasks.push(task)
206 },
207 flush() {}
208 }
209 },
210 shouldYield: () => {
211 if (!shouldYield) return false
212 shouldYield = false
213 return true
214 }
215 }
216
217 const sem = yield* PartitionedSemaphore.make<string>({ permits: 1 })
218 yield* sem.take("holder", 1)
219 const first = yield* sem.take("same", 1).pipe(
220 Effect.provideService(Scheduler.Scheduler, scheduler),
221 Effect.forkChild({ startImmediately: true })
222 )
223
224 // Pause the selected waiter before its take effect finishes.
225 shouldYield = true
226 yield* sem.release(1)
227 const next = yield* sem.take(key, 1).pipe(Effect.forkChild({ startImmediately: true }))
228 yield* Fiber.interrupt(first)
229 while (tasks.length > 0) tasks.shift()!()
230
231 const exit = next.pollUnsafe()
232 yield* Fiber.interrupt(next)
233
234 assert.deepStrictEqual(exit, Exit.void)
235 }))
236 }
237})