1import { assert, describe, it } from "@effect/vitest"
2import { Effect, Exit, Fiber, Latch } from "effect"
3import * as Scheduler from "effect/Scheduler"
4
5describe("Latch", () => {
6 it.effect("release wakes current waiters and keeps the latch closed", () =>
7 Effect.gen(function*() {
8 const latch = yield* Latch.make(false)
9 const waiter = yield* Effect.forkChild(
10 Latch.await(latch),
11 { startImmediately: true }
12 )
13
14 assert.isUndefined(waiter.pollUnsafe())
15
16 yield* latch.release
17 yield* Effect.yieldNow
18 yield* Effect.yieldNow
19
20 assert.isDefined(waiter.pollUnsafe())
21 }))
22
23 it.effect("isOpen reflects the state of the latch", () =>
24 Effect.gen(function*() {
25 const latch = yield* Latch.make(false)
26
27 assert.isFalse(latch.isOpen())
28
29 yield* latch.open
30 yield* Effect.yieldNow
31
32 assert.isTrue(latch.isOpen())
33 }))
34
35 it.effect("open then close does not resume waiters registered after close", () =>
36 Effect.gen(function*() {
37 const latch = Latch.makeUnsafe(false)
38 const before = yield* Effect.forkChild(
39 Latch.await(latch),
40 { startImmediately: true }
41 )
42
43 yield* latch.open
44 yield* latch.close
45 const after = yield* Effect.forkChild(
46 Latch.await(latch),
47 { startImmediately: true }
48 )
49
50 yield* Effect.yieldNow
51 yield* Effect.yieldNow
52
53 assert.isDefined(before.pollUnsafe())
54 assert.isUndefined(after.pollUnsafe())
55 }))
56
57 it.effect("release while a flush is pending covers the new waiters", () =>
58 Effect.gen(function*() {
59 const latch = yield* Latch.make(false)
60 const tasks: Array<() => void> = []
61 const scheduler: Scheduler.Scheduler = {
62 executionMode: "async",
63 makeDispatcher() {
64 return {
65 scheduleTask(task, _priority) {
66 tasks.push(task)
67 },
68 flush() {
69 }
70 }
71 },
72 shouldYield: () => false
73 }
74
75 const first = yield* Effect.forkChild(
76 Latch.await(latch),
77 { startImmediately: true }
78 )
79 yield* latch.release.pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
80 assert.lengthOf(tasks, 1)
81
82 const second = yield* Effect.forkChild(
83 Latch.await(latch),
84 { startImmediately: true }
85 )
86 yield* latch.release.pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
87 assert.lengthOf(tasks, 1)
88
89 tasks.shift()!()
90 yield* Fiber.join(first)
91 yield* Fiber.join(second)
92 assert.isFalse(latch.isOpen())
93 }))
94
95 it.effect("release does not resume waiters registered after the release", () =>
96 Effect.gen(function*() {
97 const latch = yield* Latch.make(false)
98 const covered = yield* Effect.forkChild(
99 Latch.await(latch),
100 { startImmediately: true }
101 )
102
103 yield* latch.release
104 const late = yield* Effect.forkChild(
105 Latch.await(latch),
106 { startImmediately: true }
107 )
108
109 yield* Effect.yieldNow
110 yield* Effect.yieldNow
111
112 assert.isDefined(covered.pollUnsafe())
113 assert.isUndefined(late.pollUnsafe())
114 }))
115
116 it.effect("openUnsafe does not resume waiters registered after a reentrant close", () =>
117 Effect.gen(function*() {
118 const latch = Latch.makeUnsafe(false)
119 const tasks: Array<() => void> = []
120 const scheduler: Scheduler.Scheduler = {
121 executionMode: "async",
122 makeDispatcher() {
123 return {
124 scheduleTask(task, _priority) {
125 tasks.push(task)
126 },
127 flush() {
128 }
129 }
130 },
131 shouldYield: () => false
132 }
133
134 let second: Fiber.Fiber<void> | undefined
135 yield* Effect.forkChild(
136 Effect.gen(function*() {
137 yield* Latch.await(latch)
138 latch.closeUnsafe()
139 second = yield* Effect.forkChild(Latch.await(latch), { startImmediately: true })
140 return yield* Effect.never
141 }),
142 { startImmediately: true }
143 )
144
145 yield* latch.release.pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
146 latch.openUnsafe()
147 assert.isUndefined(second!.pollUnsafe())
148 assert.lengthOf(tasks, 1)
149 tasks.shift()!()
150 assert.isUndefined(second!.pollUnsafe())
151 }))
152
153 it.effect("interrupting a waiter removes it from a pending flush", () =>
154 Effect.gen(function*() {
155 const latch = yield* Latch.make(false)
156 const tasks: Array<() => void> = []
157 const scheduler: Scheduler.Scheduler = {
158 executionMode: "async",
159 makeDispatcher() {
160 return {
161 scheduleTask(task, _priority) {
162 tasks.push(task)
163 },
164 flush() {
165 }
166 }
167 },
168 shouldYield: () => false
169 }
170
171 const resumed: Array<string> = []
172 const waiter = yield* Effect.forkChild(
173 Effect.gen(function*() {
174 yield* Latch.await(latch)
175 resumed.push("waiter")
176 }),
177 { startImmediately: true }
178 )
179
180 yield* latch.release.pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
181 assert.lengthOf(tasks, 1)
182
183 yield* Fiber.interrupt(waiter)
184 const exit = yield* Fiber.await(waiter)
185 assert.isTrue(Exit.hasInterrupts(exit))
186
187 tasks.shift()!()
188 assert.deepStrictEqual(resumed, [])
189 assert.isFalse(latch.isOpen())
190 }))
191
192 it.effect("await is interruptible and cleans up interrupted waiters", () =>
193 Effect.gen(function*() {
194 const latch = yield* Latch.make(false)
195 const tasks: Array<() => void> = []
196 const scheduler: Scheduler.Scheduler = {
197 executionMode: "async",
198 makeDispatcher() {
199 return {
200 scheduleTask(task, _priority) {
201 tasks.push(task)
202 },
203 flush() {
204 }
205 }
206 },
207 shouldYield: () => false
208 }
209 const resumed: Array<string> = []
210 const interrupted = yield* Effect.forkChild(
211 Effect.gen(function*() {
212 yield* Latch.await(latch)
213 resumed.push("interrupted")
214 }),
215 { startImmediately: true }
216 )
217
218 assert.isUndefined(interrupted.pollUnsafe())
219
220 yield* Fiber.interrupt(interrupted)
221 const exit = yield* Fiber.await(interrupted)
222 assert.isTrue(Exit.hasInterrupts(exit))
223
224 yield* latch.release.pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
225 assert.lengthOf(tasks, 0)
226
227 const live = yield* Effect.forkChild(
228 Effect.gen(function*() {
229 yield* Latch.await(latch)
230 resumed.push("live")
231 }),
232 { startImmediately: true }
233 )
234
235 assert.isUndefined(live.pollUnsafe())
236
237 yield* latch.release.pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
238 assert.lengthOf(tasks, 1)
239 assert.isUndefined(live.pollUnsafe())
240
241 tasks.shift()!()
242 yield* Fiber.join(live)
243
244 assert.deepStrictEqual(resumed, ["live"])
245 assert.isFalse(latch.isOpen())
246 }))
247})