1/**
2 * Controls how runnable Effect fiber tasks are dispatched.
3 *
4 * A scheduler decides how tasks are queued, when queued tasks run, and when a
5 * fiber should pause so other work can continue. This module includes the
6 * scheduler service reference, the default `MixedScheduler`, dispatcher types
7 * for queued tasks, and references for tuning or disabling automatic scheduler
8 * yields.
9 *
10 * @since 2.0.0
11 */
12import * as Context from "./Context.ts"
13import type * as Fiber from "./Fiber.ts"
14
15/**
16 * A scheduler manages the execution of Effect fibers by controlling when queued
17 * tasks run.
18 *
19 * **When to use**
20 *
21 * Use to define or provide custom runtime scheduling behavior for Effect fibers.
22 *
23 * **Details**
24 *
25 * A scheduler determines the execution mode, schedules tasks with different
26 * priorities, and decides when fibers should yield control after consuming
27 * their operation budget.
28 *
29 * @category services
30 * @since 2.0.0
31 */
32export interface Scheduler {
33 readonly executionMode: "sync" | "async"
34 shouldYield(fiber: Fiber.Fiber<unknown, unknown>): boolean
35 makeDispatcher(): SchedulerDispatcher
36}
37
38/**
39 * A dispatcher created by a `Scheduler` for enqueuing tasks and forcing queued
40 * tasks to run.
41 *
42 * **When to use**
43 *
44 * Use when implementing or testing scheduler-created dispatchers that enqueue
45 * prioritized runtime tasks and flush queued work deterministically.
46 *
47 * **Details**
48 *
49 * `scheduleTask` queues a task with a priority. `flush` drains pending work
50 * synchronously, which is useful when callers need deterministic completion of
51 * already scheduled tasks. Lower priority numbers run first, and equal
52 * priorities run in FIFO order.
53 *
54 * @category models
55 * @since 4.0.0
56 */
57export interface SchedulerDispatcher {
58 scheduleTask(task: () => void, priority: number): void
59 flush(): void
60}
61
62/**
63 * Context reference for the scheduler used by the Effect runtime.
64 *
65 * **When to use**
66 *
67 * Use when you need to replace scheduling behavior globally in tests or runtime
68 * setup, such as forcing deterministic task dispatch.
69 *
70 * **Details**
71 *
72 * The default value creates a `MixedScheduler`. Provide this service to
73 * customize execution mode, task dispatching, or yield behavior.
74 *
75 * @category services
76 * @since 2.0.0
77 */
78export const Scheduler: Context.Reference<Scheduler> = Context.Reference<Scheduler>("effect/Scheduler", {
79 fiberCached: true,
80 defaultValue: () => new MixedScheduler()
81})
82
83const setMicrotask = (f: () => void) => {
84 let cancelled = false
85 Promise.resolve().then(() => {
86 if (!cancelled) f()
87 })
88 return (): void => {
89 cancelled = true
90 }
91}
92
93const setTimer: (f: () => void) => () => void = "setImmediate" in globalThis
94 ? (f) => {
95 // @ts-ignore
96 const timer = globalThis.setImmediate(f)
97 // @ts-ignore
98 return (): void => globalThis.clearImmediate(timer)
99 }
100 : (f) => {
101 const timer = setTimeout(f, 0)
102 return (): void => clearTimeout(timer)
103 }
104
105// Some runtimes (e.g. Cloudflare Workers) throw when a timer is set in global
106// scope. Fall back to a microtask so effects can still yield at module load.
107const setImmediate = (f: () => void) => {
108 try {
109 return setTimer(f)
110 } catch {
111 return setMicrotask(f)
112 }
113}
114
115class PriorityBuckets {
116 buckets: Array<[priority: number, tasks: Array<() => void>]> = []
117
118 scheduleTask(task: () => void, priority: number): void {
119 const buckets = this.buckets
120 const len = buckets.length
121 let bucket: [number, Array<() => void>] | undefined
122 let index = 0
123 for (; index < len; index++) {
124 if (buckets[index][0] > priority) break
125 bucket = buckets[index]
126 }
127 if (bucket && bucket[0] === priority) {
128 bucket[1].push(task)
129 } else if (index === len) {
130 buckets.push([priority, [task]])
131 } else {
132 buckets.splice(index, 0, [priority, [task]])
133 }
134 }
135
136 drain() {
137 const buckets = this.buckets
138 this.buckets = []
139 return buckets
140 }
141}
142
143/**
144 * Provides a scheduler implementation that batches queued tasks and dispatches them by
145 * priority.
146 *
147 * **When to use**
148 *
149 * Use when you need the default runtime scheduler directly, including a
150 * scheduler that batches queued work by priority and preserves FIFO order within
151 * each priority.
152 *
153 * **Details**
154 *
155 * `MixedScheduler` supports synchronous and asynchronous execution modes, uses
156 * operation counts to decide when fibers should yield, and is the default
157 * scheduler implementation.
158 *
159 * @category models
160 * @since 2.0.0
161 */
162export class MixedScheduler implements Scheduler {
163 readonly executionMode: "sync" | "async"
164 readonly setImmediate: (f: () => void) => () => void
165
166 constructor(
167 executionMode: "sync" | "async" = "async",
168 setImmediateFn?: (f: () => void) => () => void
169 ) {
170 this.executionMode = executionMode
171 this.setImmediate = setImmediateFn ?? (executionMode === "sync" ? setMicrotask : setImmediate)
172 }
173
174 /**
175 * Returns whether the fiber has reached its operation budget and should yield.
176 *
177 * **When to use**
178 *
179 * Use to decide whether a fiber should yield after consuming its current
180 * operation budget.
181 *
182 * @since 2.0.0
183 */
184 shouldYield(fiber: Fiber.Fiber<unknown, unknown>) {
185 return fiber.currentOpCount >= fiber.cache.maxOpsBeforeYield
186 }
187
188 /**
189 * Creates a dispatcher that schedules work through this scheduler.
190 *
191 * **When to use**
192 *
193 * Use when you need a standalone dispatcher from a scheduler instance, for
194 * example in tests that enqueue tasks and then flush them deterministically.
195 *
196 * @since 4.0.0
197 */
198 makeDispatcher() {
199 return new MixedSchedulerDispatcher(this.setImmediate)
200 }
201}
202
203class MixedSchedulerDispatcher implements SchedulerDispatcher {
204 private tasks = new PriorityBuckets()
205 private running: (() => void) | undefined = undefined
206 readonly setImmediate: (f: () => void) => () => void
207
208 constructor(
209 setImmediateFn: (f: () => void) => () => void = setImmediate
210 ) {
211 this.setImmediate = setImmediateFn
212 }
213
214 /**
215 * @since 2.0.0
216 */
217 scheduleTask(task: () => void, priority: number) {
218 this.tasks.scheduleTask(task, priority)
219 if (this.running === undefined) {
220 this.running = this.setImmediate(this.afterScheduled)
221 }
222 }
223
224 /**
225 * @since 2.0.0
226 */
227 afterScheduled = () => {
228 this.running = undefined
229 this.runTasks()
230 }
231
232 /**
233 * @since 2.0.0
234 */
235 runTasks() {
236 const buckets = this.tasks.drain()
237 for (let i = 0; i < buckets.length; i++) {
238 const toRun = buckets[i][1]
239 for (let j = 0; j < toRun.length; j++) {
240 toRun[j]()
241 }
242 }
243 }
244
245 /**
246 * @since 2.0.0
247 */
248 flush() {
249 while (this.tasks.buckets.length > 0) {
250 if (this.running !== undefined) {
251 this.running()
252 this.running = undefined
253 }
254 this.runTasks()
255 }
256 }
257}
258
259/**
260 * Context reference that controls the maximum number of operations a fiber
261 * can perform before yielding control back to the scheduler.
262 *
263 * **When to use**
264 *
265 * Use to tune scheduler fairness for CPU-bound fibers by changing the scheduler
266 * operation budget that triggers a yield.
267 *
268 * **Details**
269 *
270 * The default value is `2048` operations, which balances performance and
271 * fairness by helping prevent long-running fibers from monopolizing the
272 * execution thread.
273 *
274 * @see {@link PreventSchedulerYield} for bypassing scheduler yield checks entirely rather than tuning the operation budget
275 *
276 * @category services
277 * @since 4.0.0
278 */
279export const MaxOpsBeforeYield = Context.Reference<number>("effect/Scheduler/MaxOpsBeforeYield", {
280 fiberCached: true,
281 defaultValue: () => 2048
282})
283
284/**
285 * Context reference that controls whether the runtime should bypass scheduler
286 * yield checks. When set to `true`, the fiber run loop won't call
287 * `Scheduler.shouldYield`.
288 *
289 * **When to use**
290 *
291 * Use to bypass scheduler yield checks for controlled runtime workloads where
292 * cooperative yielding should be disabled.
293 *
294 * **Gotchas**
295 *
296 * Setting this reference to `true` can let long-running fibers monopolize the
297 * JavaScript thread.
298 *
299 * @see {@link MaxOpsBeforeYield} for tuning yield frequency without disabling yield checks
300 * @see {@link Scheduler} for providing custom scheduler yield behavior
301 *
302 * @category services
303 * @since 4.0.0
304 */
305export const PreventSchedulerYield = Context.Reference<boolean>("effect/Scheduler/PreventSchedulerYield", {
306 fiberCached: true,
307 defaultValue: () => false
308})