1/**
2 * Limits how many effects can use a shared resource at the same time.
3 *
4 * A `Semaphore` owns a number of permits. Work can run only after acquiring the
5 * permits it needs, and those permits are returned when the work finishes. This
6 * module includes constructors, automatic wrappers that acquire and release
7 * permits around an effect, manual permit operations, a non-waiting variant for
8 * work that should only run immediately, and resizing support for an existing
9 * semaphore.
10 *
11 * @since 4.0.0
12 */
13import type * as Effect from "./Effect.ts"
14import type { Fiber } from "./Fiber.ts"
15import { dual } from "./Function.ts"
16import * as core from "./internal/core.ts"
17import * as internal from "./internal/effect.ts"
18import type * as Option from "./Option.ts"
19
20/**
21 * A counting semaphore that coordinates concurrent access with permits.
22 *
23 * **When to use**
24 *
25 * Use to coordinate concurrent effects that need bounded access to a shared
26 * resource.
27 *
28 * **Details**
29 *
30 * Effects can acquire permits, wait until enough permits are available,
31 * release permits, or run with permits that are automatically released when
32 * the effect exits.
33 *
34 * **Example** (Controlling concurrent access)
35 *
36 * ```ts import.meta.vitest
37 * import { Effect, Semaphore } from "effect"
38 *
39 * // Create and use a semaphore for controlling concurrent access
40 * const program = Effect.gen(function*() {
41 * const semaphore = yield* Semaphore.make(2)
42 *
43 * return yield* semaphore.withPermits(1)(
44 * Effect.succeed("Resource accessed")
45 * )
46 * })
47 *
48 * await Effect.runPromise(program) // => "Resource accessed"
49 * ```
50 *
51 * @see {@link make} for creating a semaphore inside Effect code
52 * @see {@link makeUnsafe} for creating a semaphore synchronously
53 *
54 * @category models
55 * @since 4.0.0
56 */
57export interface Semaphore {
58 /**
59 * Adjusts the number of permits available in the semaphore.
60 *
61 * **When to use**
62 *
63 * Use to change the total permit count of an existing semaphore.
64 */
65 resize(this: Semaphore, permits: number): Effect.Effect<void>
66
67 /**
68 * Runs an effect with the given number of permits and releases the permits
69 * when the effect completes.
70 *
71 * **When to use**
72 *
73 * Use to run an effect while holding a specified number of semaphore permits.
74 *
75 * **Details**
76 *
77 * This function acquires the specified number of permits before executing
78 * the provided effect. Once the effect finishes, the permits are released.
79 * If insufficient permits are available, the function will wait until they
80 * are released by other tasks.
81 */
82 withPermits(this: Semaphore, permits: number): <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
83
84 /**
85 * Runs an effect with the given number of permits and releases the permits
86 * when the effect completes.
87 *
88 * **When to use**
89 *
90 * Use to run an effect while holding exactly one semaphore permit.
91 *
92 * **Details**
93 *
94 * This function acquires the specified number of permits before executing
95 * the provided effect. Once the effect finishes, the permits are released.
96 * If insufficient permits are available, the function will wait until they
97 * are released by other tasks.
98 */
99 withPermit<A, E, R>(self: Effect.Effect<A, E, R>): Effect.Effect<A, E, R>
100
101 /**
102 * Runs an effect only if the specified number of permits are immediately
103 * available.
104 *
105 * **When to use**
106 *
107 * Use when guarded work should run only if the requested permits are
108 * immediately available.
109 *
110 * **Details**
111 *
112 * This function attempts to acquire the specified number of permits. If they
113 * are available, it runs the effect and releases the permits after the effect
114 * completes. If permits are not available, the effect does not execute, and
115 * the result is `Option.none`.
116 */
117 withPermitsIfAvailable(
118 this: Semaphore,
119 permits: number
120 ): <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<Option.Option<A>, E, R>
121
122 /**
123 * Acquires the specified number of permits and returns the acquired permit
124 * count, suspending the task if they are not yet available. Pending `take`
125 * calls are scanned in registration order, but a request is served only when
126 * enough permits are available, so a smaller later request may overtake a
127 * larger earlier request.
128 *
129 * **When to use**
130 *
131 * Use to manually acquire permits for lower-level coordination protocols.
132 */
133 take(this: Semaphore, permits: number): Effect.Effect<number>
134
135 /**
136 * Acquires the specified number of permits only if they are immediately
137 * available.
138 *
139 * **When to use**
140 *
141 * Use to manually acquire permits without waiting, paired with `release`.
142 */
143 takeIfAvailable(this: Semaphore, permits: number): Effect.Effect<boolean>
144
145 /**
146 * Releases the specified number of permits and returns the resulting
147 * available permits.
148 *
149 * **When to use**
150 *
151 * Use to manually return permits acquired by a lower-level coordination
152 * protocol.
153 */
154 release(this: Semaphore, permits: number): Effect.Effect<number>
155
156 /**
157 * Releases all permits held by this semaphore and returns the resulting available permits.
158 *
159 * **When to use**
160 *
161 * Use to return every currently taken permit to the semaphore at once.
162 *
163 * **Gotchas**
164 *
165 * This does not stop effects already running with `withPermit` or `withPermits`.
166 * Their permits become available immediately, but those effects still release
167 * their permits when they finish. The resulting available count can then
168 * exceed the semaphore's configured number of permits.
169 */
170 readonly releaseAll: Effect.Effect<number>
171}
172
173/**
174 * Creates a `Semaphore` synchronously with the specified total
175 * number of permits.
176 *
177 * **When to use**
178 *
179 * Use to construct a semaphore synchronously when an immediate value is
180 * required outside an Effect workflow.
181 *
182 * **Example** (Creating an unsafe semaphore)
183 *
184 * ```ts import.meta.vitest
185 * import { Effect, Semaphore } from "effect"
186 *
187 * const semaphore = Semaphore.makeUnsafe(3)
188 *
189 * const task = (id: number) =>
190 * semaphore.withPermits(1)(
191 * Effect.gen(function*() {
192 * yield* Effect.yieldNow
193 * return id
194 * })
195 * )
196 *
197 * // Only 3 tasks can run concurrently
198 * const program = Effect.all([
199 * task(1),
200 * task(2),
201 * task(3),
202 * task(4),
203 * task(5)
204 * ], { concurrency: "unbounded" })
205 *
206 * await Effect.runPromise(program) // => [1, 2, 3, 4, 5]
207 * ```
208 *
209 * @category constructors
210 * @since 4.0.0
211 */
212export const makeUnsafe = (permits: number): Semaphore => new SemaphoreImpl(permits)
213
214const waitForPermits = <A, E, R>(
215 self: SemaphoreImpl,
216 n: number,
217 effect: Effect.Effect<A, E, R>
218): Effect.Effect<A, E, R> =>
219 internal.callback((resume) => {
220 if (self.free >= n) return resume(effect)
221 const observer = () => {
222 if (self.free < n) return
223 self.waiters.delete(observer)
224 resume(effect)
225 }
226 self.waiters.add(observer)
227 return internal.sync(() => {
228 self.waiters.delete(observer)
229 })
230 })
231
232class SemaphoreImpl implements Semaphore {
233 public waiters = new Set<() => void>()
234 public taken = 0
235 public permits: number
236
237 constructor(permits: number) {
238 this.permits = permits
239 }
240
241 get free() {
242 return this.permits - this.taken
243 }
244
245 take(n: number): Effect.Effect<number> {
246 const take: Effect.Effect<number> = internal.suspend(() => {
247 if (this.free < n) {
248 return waitForPermits(this, n, take)
249 }
250 this.taken += n
251 return internal.succeed(n)
252 })
253 return take
254 }
255
256 takeIfAvailable(n: number): Effect.Effect<boolean> {
257 return internal.suspend(() => {
258 if (this.free < n) return internal.succeed(false)
259 this.taken += n
260 return internal.succeed(true)
261 })
262 }
263
264 releaseUnsafe(fiber: Fiber<any, any>, n: number): number {
265 this.taken -= n
266 if (this.waiters.size > 0) {
267 fiber.currentDispatcher.scheduleTask(() => {
268 for (const observer of this.waiters) {
269 if (this.free <= 0) break
270 observer()
271 }
272 }, 0)
273 }
274 return this.free
275 }
276
277 resize(permits: number) {
278 return core.withFiber((fiber) => {
279 this.permits = permits
280 if (this.free < 0) return internal.void
281 this.releaseUnsafe(fiber, 0)
282 return internal.void
283 })
284 }
285
286 release(n: number): Effect.Effect<number> {
287 return core.withFiber((fiber) => internal.succeed(this.releaseUnsafe(fiber, n)))
288 }
289
290 get releaseAll(): Effect.Effect<number> {
291 return core.withFiber((fiber) => internal.succeed(this.releaseUnsafe(fiber, this.taken)))
292 }
293
294 withPermits(n: number) {
295 return <A, E, R>(self: Effect.Effect<A, E, R>) =>
296 internal.uninterruptibleMask((restore) => {
297 const acquire: Effect.Effect<A, E, R> = internal.suspend(() => {
298 if (this.free < n) {
299 const wait = waitForPermits(this, n, internal.void)
300 return internal.flatMap(restore(wait), () => acquire)
301 }
302 this.taken += n
303 return internal.onExitPrimitive(
304 restore(self),
305 () => {
306 this.releaseUnsafe(internal.getCurrentFiber()!, n)
307 return undefined
308 },
309 true
310 )
311 })
312 return acquire
313 })
314 }
315
316 readonly withPermit = this.withPermits(1)
317
318 withPermitsIfAvailable(n: number) {
319 return <A, E, R>(self: Effect.Effect<A, E, R>) =>
320 internal.uninterruptibleMask((restore) => {
321 if (this.free < n) return internal.succeedNone
322 this.taken += n
323 return internal.onExitPrimitive(restore(internal.asSome(self)), () => {
324 this.releaseUnsafe(internal.getCurrentFiber()!, n)
325 return undefined
326 }, true)
327 })
328 }
329}
330
331/**
332 * Creates a `Semaphore` initialized with the specified total number of permits.
333 *
334 * **When to use**
335 *
336 * Use to create a semaphore inside Effect code for bounding concurrency with
337 * automatic or manual permit management.
338 *
339 * **Example** (Creating a semaphore)
340 *
341 * ```ts import.meta.vitest
342 * import { Effect, Semaphore } from "effect"
343 *
344 * const program = Effect.gen(function*() {
345 * const semaphore = yield* Semaphore.make(2)
346 *
347 * const task = (id: number) =>
348 * semaphore.withPermits(1)(
349 * Effect.gen(function*() {
350 * yield* Effect.yieldNow
351 * return id
352 * })
353 * )
354 *
355 * // Run 4 tasks, but only 2 can run concurrently
356 * return yield* Effect.all([task(1), task(2), task(3), task(4)])
357 * })
358 *
359 * await Effect.runPromise(program) // => [1, 2, 3, 4]
360 * ```
361 *
362 * @category constructors
363 * @since 4.0.0
364 */
365export const make = (permits: number): Effect.Effect<Semaphore> => internal.sync(() => new SemaphoreImpl(permits))
366
367/**
368 * Sets the total number of permits managed by the semaphore.
369 *
370 * **When to use**
371 *
372 * Use to change the concurrency limit of an existing semaphore while keeping
373 * current acquisitions in place.
374 *
375 * **Details**
376 *
377 * Existing acquisitions remain taken after resizing. If the new total is less
378 * than the currently taken permit count, new acquisitions wait until enough
379 * permits are released.
380 *
381 * @see {@link make} for creating a semaphore with an initial permit count
382 * @see {@link release} for returning permits without changing semaphore capacity
383 *
384 * @category combinators
385 * @since 4.0.0
386 */
387export const resize: {
388 (permits: number): (self: Semaphore) => Effect.Effect<void>
389 (self: Semaphore, permits: number): Effect.Effect<void>
390} = dual(2, (self: Semaphore, permits: number) => self.resize(permits))
391
392/**
393 * Runs an effect with the given number of permits and releases the permits when
394 * the effect completes.
395 *
396 * **When to use**
397 *
398 * Use to run an effect while holding a specified number of semaphore permits
399 * for the duration of that effect.
400 *
401 * **Details**
402 *
403 * The effect waits until enough permits are available. Acquired permits are
404 * released when the wrapped effect exits.
405 *
406 * @see {@link withPermit} for acquiring exactly one permit
407 * @see {@link withPermitsIfAvailable} for running only when permits are immediately available
408 * @see {@link take} for manually acquiring permits
409 * @see {@link release} for manually returning permits
410 *
411 * @category combinators
412 * @since 4.0.0
413 */
414export const withPermits: {
415 (self: Semaphore, permits: number): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
416 <A, E, R>(self: Semaphore, permits: number, effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R>
417} = ((self: Semaphore, permits: number, effect?: Effect.Effect<any, any, any>) => {
418 const withPermits = self.withPermits(permits)
419 return effect ? withPermits(effect) : withPermits
420}) as any
421
422/**
423 * Runs an effect with a single permit and releases the permit when the effect
424 * completes.
425 *
426 * **When to use**
427 *
428 * Use to guard an effect with exactly one semaphore permit while automatically
429 * releasing that permit when the effect exits.
430 *
431 * @see {@link withPermits} for acquiring more than one permit
432 * @see {@link withPermitsIfAvailable} for running only when permits are immediately available
433 * @see {@link take} for manually acquiring permits
434 * @see {@link release} for manually returning permits
435 *
436 * @category combinators
437 * @since 4.0.0
438 */
439export const withPermit: {
440 (self: Semaphore): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
441 <A, E, R>(self: Semaphore, effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R>
442} = ((self: Semaphore, effect?: Effect.Effect<any, any, any>) => {
443 if (!effect) return self.withPermit
444 return self.withPermit(effect)
445}) as any
446
447/**
448 * Runs an effect only if the specified number of permits are immediately
449 * available.
450 *
451 * **When to use**
452 *
453 * Use when guarded work should run only if the requested permits are
454 * immediately available.
455 *
456 * **Details**
457 *
458 * When the permits are unavailable, the effect is not run and the result is
459 * `Option.none`. When permits are available, the effect is run, its result is
460 * wrapped in `Option.some`, and the acquired permits are released when the
461 * effect exits.
462 *
463 * @see {@link withPermits} for the variant that waits until permits are available
464 *
465 * @category combinators
466 * @since 4.0.0
467 */
468export const withPermitsIfAvailable: {
469 (self: Semaphore, permits: number): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<Option.Option<A>, E, R>
470 <A, E, R>(
471 self: Semaphore,
472 permits: number,
473 effect: Effect.Effect<A, E, R>
474 ): Effect.Effect<Option.Option<A>, E, R>
475} = ((self: Semaphore, permits: number, effect?: Effect.Effect<any, any, any>) => {
476 const withPermits = self.withPermitsIfAvailable(permits)
477 return effect ? withPermits(effect) : withPermits
478}) as any
479
480/**
481 * Acquires the specified number of permits and returns the acquired permit
482 * count.
483 *
484 * **When to use**
485 *
486 * Use when you need manual permit acquisition for a lower-level protocol with
487 * explicit acquisition and release control.
488 *
489 * **Details**
490 *
491 * The effect waits until enough permits are available.
492 *
493 * @see {@link withPermit} for automatically acquiring and releasing one permit around an effect
494 * @see {@link withPermits} for automatically acquiring and releasing multiple permits around an effect
495 * @see {@link takeIfAvailable} for manually acquiring permits without waiting
496 * @see {@link release} for returning manually acquired permits
497 *
498 * @category combinators
499 * @since 4.0.0
500 */
501export const take: {
502 (permits: number): (self: Semaphore) => Effect.Effect<number>
503 (self: Semaphore, permits: number): Effect.Effect<number>
504} = dual(2, (self: Semaphore, permits: number) => self.take(permits))
505
506/**
507 * Acquires the specified number of permits only if they are immediately
508 * available.
509 *
510 * **When to use**
511 *
512 * Use when you need fail-fast manual permit acquisition for a lower-level
513 * protocol with explicit acquisition and release control.
514 *
515 * **Details**
516 *
517 * If enough permits are available, they are acquired and the effect returns
518 * `true`. Otherwise, the effect returns `false` immediately without acquiring
519 * any permits.
520 *
521 * @see {@link take} for the variant that waits until permits are available
522 * @see {@link release} for returning manually acquired permits
523 * @see {@link withPermitsIfAvailable} for automatic acquisition and release around an effect
524 *
525 * @category combinators
526 * @since 4.0.0
527 */
528export const takeIfAvailable: {
529 (permits: number): (self: Semaphore) => Effect.Effect<boolean>
530 (self: Semaphore, permits: number): Effect.Effect<boolean>
531} = dual(2, (self: Semaphore, permits: number) => self.takeIfAvailable(permits))
532
533/**
534 * Releases the specified number of permits and returns the resulting available
535 * permits.
536 *
537 * **When to use**
538 *
539 * Use when you need to return permits acquired with `take` in a lower-level
540 * permit protocol with explicit release control.
541 *
542 * **Details**
543 *
544 * Running the effect releases the requested permits, wakes waiting acquirers
545 * when permits become available, and returns the current available permit
546 * count.
547 *
548 * **Gotchas**
549 *
550 * Manual `take` / `release` usage must keep permit counts balanced. Prefer
551 * `withPermit` or `withPermits` when the acquisition can be scoped to one
552 * effect.
553 *
554 * @see {@link take} for manually acquiring permits
555 * @see {@link releaseAll} for returning every currently taken permit
556 * @see {@link withPermits} for automatic acquire and release around an effect
557 *
558 * @category combinators
559 * @since 4.0.0
560 */
561export const release: {
562 (permits: number): (self: Semaphore) => Effect.Effect<number>
563 (self: Semaphore, permits: number): Effect.Effect<number>
564} = dual(2, (self: Semaphore, permits: number) => self.release(permits))
565
566/**
567 * Releases all permits held by this semaphore and returns the resulting
568 * available permits.
569 *
570 * **When to use**
571 *
572 * Use to return every currently taken permit to a semaphore at once, typically
573 * during cleanup of manual `take` / `release` protocols.
574 *
575 * **Gotchas**
576 *
577 * This does not stop effects already running with `withPermit` or `withPermits`.
578 * Their permits become available immediately, but those effects still release
579 * their permits when they finish. The resulting available count can then
580 * exceed the semaphore's configured number of permits.
581 *
582 * @see {@link release} for releasing a known permit count
583 * @see {@link withPermits} for automatic acquire and release around an effect
584 *
585 * @category combinators
586 * @since 4.0.0
587 */
588export const releaseAll = (self: Semaphore): Effect.Effect<number> => self.releaseAll