src/vs/base/test/common/virtualScheduling/processor.ts
384 LOC · 370 covered · 14 uncovered · 56 ranges · 2121 concepts · 13 introducers · 1304 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
>
/*---------------------------------------------------------------------------------------------
processor.ts ×17
2
>
* Copyright (c) Microsoft Corporation. All rights reserved.
3
>
* Licensed under the MIT License. See License.txt in the project root for license information.
4
>
*--------------------------------------------------------------------------------------------*/
5
>
6
>
import { CancellationToken } from '../../../common/cancellation.js';
7
>
import { Disposable, DisposableStore, IDisposable } from '../../../common/lifecycle.js';
8
>
import { Embedding, nextMacrotask } from './embedding.js';
9
>
import { TimeApi } from './timeApi.js';
10
>
import { ROOT_TRACE, TraceContext } from './trace.js';
11
>
import { EventSource, VirtualClock, VirtualEvent, VirtualTime } from './virtualClock.js';
12
>
13
>
// ============================================================================
14
>
// Termination policy
15
>
// ============================================================================
16
>
17
>
/**
18
>
* When a {@link Run} should terminate.
19
>
*
20
>
* Greenfield design choice: termination is *always* explicit. There is no
21
>
* "bare run()" that terminates on first empty queue, because that creates a
22
>
* race with the caller's microtask chain (the run can resolve before the
23
>
* caller's `.then` has had a chance to schedule).
24
>
*/
25
>
export type TerminationPolicy =
26
>
/** Resolve as soon as the virtual queue is empty. */
27
>
| { readonly kind: 'idle' }
28
>
/** Resolve when the token is cancelled AND the queue is empty. */
29
>
| { readonly kind: 'token'; readonly token: CancellationToken }
30
>
/** Resolve when virtual time has reached `time` and all events scheduled
31
>
* at or before `time` have been processed. A sentinel event at `time`
32
>
* is scheduled by the processor so virtual time always reaches it. */
33
>
| { readonly kind: 'time'; readonly time: VirtualTime };
34
>
35
>
export const untilIdle: TerminationPolicy = { kind: 'idle' };
36
>
export function untilToken(token: CancellationToken): TerminationPolicy { return { kind: 'token', token }; }
37
>
export function untilTime(time: VirtualTime): TerminationPolicy { return { kind: 'time', time }; }
38
>
39
>
export interface RunOptions {
40
>
readonly until: TerminationPolicy;
41
>
/** Maximum number of virtual events this run will execute. Default: 100. */
42
>
readonly maxEvents?: number;
43
>
/** Maximum causal-trace depth this run will tolerate. Useful for catching
44
>
* runaway self-rescheduling timers. */
45
>
readonly maxTraceDepth?: number;
46
>
}
47
>
48
>
// ============================================================================
49
>
// Run — internal state for a single processor.run() invocation
50
>
// ============================================================================
51
>
52
>
type RunStatus = 'continue' | 'done' | { readonly error: Error };
53
>
54
>
class Run {
55
>
private static _idCounter = 0;
56
>
public readonly id = ++Run._idCounter;
57
>
58
>
public readonly promise: Promise<void>;
59
>
private _resolve!: () => void;
60
>
private _reject!: (e: Error) => void;
61
>
private _settled = false;
62
>
public get settled(): boolean { return this._settled; }
63
>
64
>
constructor(
66
>
public readonly executedAtStart: number,
67
>
public readonly maxEvents: number,
68
>
) {
69
>
this.promise = new Promise<void>((res, rej) => { this._resolve = res; this._reject = rej; });
70
>
}
72
>
settle(error?: Error): void {
74
>
this._settled = true;
75
>
if (error) { this._reject(error); } else { this._resolve(); }
76
>
}
78
>
evaluate(clock: VirtualClock, executedTotal: number, makeOverflow: () => Error): RunStatus {
80
>
if (local >= this.maxEvents && clock.hasEvents) {
82
>
}
84
>
const u = this.options.until;
85
>
switch (u.kind) {
86
>
case 'idle':
89
>
return u.token.isCancellationRequested && !clock.hasEvents ? 'done' : 'continue';
processor.ts ×1
92
>
// The sentinel guarantees the queue is non-empty until at
93
>
// least the deadline is reached, so we never resolve "early"
94
>
// just because nothing has been scheduled yet.
95
>
const next = clock.peekNext();
96
>
return next === undefined || next.time > u.time ? 'done' : 'continue';
97
>
}
99
>
}
101
>
102
>
// ============================================================================
103
>
// Step outcome — what the pure state machine tells the trampoline
104
>
// ============================================================================
105
>
106
>
type StepOutcome =
107
>
/** Either a virtual event was executed, or a run was rejected for a
108
>
* bookkeeping reason (depth/event overflow). The trampoline should let
109
>
* the embedding decide how to reach the next step. */
110
>
| 'progress'
111
>
/** No actionable event under any active deadline. The trampoline should
112
>
* park until something wakes the processor. */
113
>
| 'park'
114
>
/** No active runs. The trampoline should stop driving. */
115
>
| 'quiesce';
116
>
117
>
// ============================================================================
118
>
// VirtualTimeProcessor
119
>
// ============================================================================
120
>
121
>
export interface VirtualTimeProcessorOptions {
122
>
readonly defaultMaxEvents?: number;
123
>
}
124
>
125
>
/**
126
>
* # VirtualTimeProcessor
127
>
*
128
>
* Drives a {@link VirtualClock} from the host event loop. This is the
129
>
* **embedding** of a small virtual event loop into the host event loop.
130
>
*
131
>
* ## Responsibilities, separated
132
>
*
133
>
* - {@link _step} is a *pure* state-machine advance. It reads the clock,
134
>
* decides what to do, optionally executes one virtual event, and returns
135
>
* a {@link StepOutcome}. It never touches host time.
136
>
*
137
>
* - {@link _drive} is the *trampoline*. It calls `_step` and lets the
138
>
* {@link Embedding} decide whether to loop in place (`'continueSync'`)
139
>
* or schedule the next iteration on the host (`'cbScheduled'`). It is
140
>
* the only code that touches host time.
141
>
*
142
>
* - {@link Run} carries the user's termination predicate. Runs are pure
143
>
* over `_step`'s observations; they never schedule.
144
>
*
145
>
* ## Invariants
146
>
*
147
>
* 1. **Single driver.** At any moment at most one `_drive` invocation is
148
>
* active per processor (the `_inDrive` guard).
149
>
*
150
>
* 2. **Step is pure w.r.t. host time.** `_step` only reads the clock,
151
>
* mutates the run set via settling, and synchronously runs at most one
152
>
* virtual event. It never calls into a host time API.
153
>
*
154
>
* 3. **Embedding chooses the host primitive.** Whether the next step runs
155
>
* inline, after a microtask drain, or on a paint frame is entirely the
156
>
* embedding's decision — *per event*.
157
>
*
158
>
* 4. **Park is breakable.** While parked, the processor wakes on
159
>
* {@link VirtualClock.onEventScheduled}, on a token cancellation, and
160
>
* on a new run being added.
161
>
*
162
>
* 5. **Disposal is terminal.** After dispose, all runs are rejected and
163
>
* `_step`/`_drive` short-circuit to `'quiesce'`.
164
>
*
165
>
* ## On the trace-reset sink
166
>
*
167
>
* The trace context's deferred reset (see {@link TraceContext.runAsHandler})
168
>
* needs a "fire after the microtask closure" primitive. The processor passes
169
>
* its *own* {@link nextMacrotask} as that sink, so the reset goes through
170
>
* the same primitive the embedding uses for its own host hops. This removes
171
>
* any race between the processor's hops and the trace-reset timer.
172
>
*/
173
>
export class VirtualTimeProcessor extends Disposable {
174
>
175
>
private readonly _runs = new Map<Run, IDisposable>();
176
>
private readonly _history: VirtualEvent[] = [];
177
>
private _executedTotal = 0;
178
>
private _disposed = false;
179
>
180
>
private _inDrive = false;
181
>
private _parkCleanup: IDisposable | undefined;
182
>
183
>
private readonly _defaultMaxEvents: number;
184
>
185
>
public get history(): readonly VirtualEvent[] { return this._history; }
186
>
public get executedTotal(): number { return this._executedTotal; }
187
>
188
>
constructor(
190
>
private readonly _embedding: Embedding,
191
>
private readonly _realApi: TimeApi,
192
>
opts: VirtualTimeProcessorOptions = {},
193
>
) {
194
>
super();
195
>
this._defaultMaxEvents = opts.defaultMaxEvents ?? 100;
196
>
this._register({ dispose: () => this._onDispose() });
197
>
}
199
>
// ---- Public API -----------------------------------------------------
200
>
201
>
/** Start a run with the given termination policy. */
202
>
run(options: RunOptions): Promise<void> {
203
>
const run = new Run(options, this._executedTotal, options.maxEvents ?? this._defaultMaxEvents);
processor.ts ×13
204
>
const cleanup = new DisposableStore();
205
>
206
>
// Wake the loop on token cancellation so the run can re-evaluate.
207
>
if (options.until.kind === 'token') {
209
>
}
211
>
// For time-based termination, schedule a sentinel event at the
212
>
// deadline. This guarantees virtual time reaches the deadline even if
213
>
// the user never schedules anything else, and that the run does not
214
>
// resolve early just because the queue happens to be empty *now*.
215
>
if (options.until.kind === 'time' && options.until.time > this._clock.now) {
216
>
const source: EventSource = { toString: () => `<deadline of run #${run.id}>` };
processor.ts ×2
217
>
cleanup.add(this._clock.schedule({
218
>
time: options.until.time,
219
>
source,
220
>
run: () => { /* sentinel: no-op */ },
221
>
}));
222
>
}
224
>
this._runs.set(run, cleanup);
225
>
this._wake();
226
>
return run.promise;
227
>
}
229
>
// ---- The pure step --------------------------------------------------
230
>
231
>
private _step(): StepOutcome {
234
>
this._settleFinishedRuns();
235
>
if (this._runs.size === 0) { return 'quiesce'; }
237
>
const next = this._clock.peekNext();
238
>
if (next === undefined) { return 'park'; }
239
>
240
>
// Per-run trace-depth check: reject any run whose limit this event
241
>
// would exceed, before executing.
243
>
let depthOverflow = false;
244
>
for (const run of [...this._runs.keys()]) {
246
>
if (limit !== undefined && traceDepth > limit) {
247
this._settleRun(run, this._buildDepthOverflow(run, traceDepth));
248
depthOverflow = true;
249
}
251
>
if (depthOverflow) { return 'progress'; }
252
>
253
>
this._executeOne(next);
254
>
return 'progress';
257
>
private _executeOne(event: VirtualEvent): void {
259
>
TraceContext.instance.runAsHandler(
260
>
event.trace ?? ROOT_TRACE,
261
>
() => {
262
>
const e = this._clock.runNext();
263
>
if (e) {
264
>
this._history.push(e);
265
>
this._executedTotal++;
266
>
}
267
>
},
268
>
{
269
>
// Route the trace-reset through the same host primitive
270
>
// the embedding uses, so there is no race between this
271
>
// timer and the embedding's next hop.
272
>
afterMicrotaskClosure: cb => nextMacrotask(this._realApi, cb),
273
>
},
274
>
);
275
>
} catch (e) {
276
const err = e instanceof Error ? e : new Error(String(e));
277
// We can't tell which run "owned" the throwing event. Reject all
278
// active runs so the failure is observed exactly once per caller.
279
for (const run of [...this._runs.keys()]) { this._settleRun(run, err); }
280
}
283
>
// ---- The trampoline -------------------------------------------------
284
>
285
>
private readonly _drive = (): void => {
286
>
if (this._inDrive) { return; }
287
>
this._inDrive = true;
288
>
try {
289
>
while (true) {
290
>
const outcome = this._step();
291
>
if (outcome === 'quiesce') { return; }
293
>
294
>
// 'progress': read the next event so the embedding can pick a
295
>
// per-event primitive. If there is none, loop and let the next
296
>
// `_step` decide between 'park' and 'quiesce'.
297
>
const next = this._clock.peekNext();
298
>
if (next === undefined) { continue; }
300
>
const choice = this._embedding(next, this._drive);
301
>
if (choice === 'cbScheduled') { return; }
303
>
}
304
>
} finally {
305
>
this._inDrive = false;
306
>
}
307
>
};
308
>
309
>
// ---- Park & wake ----------------------------------------------------
310
>
311
>
private _park(): void {
313
>
const store = new DisposableStore();
314
>
store.add(this._clock.onEventScheduled(() => this._wake()));
315
>
this._parkCleanup = store;
316
>
}
318
>
private _unpark(): void {
320
>
this._parkCleanup = undefined;
321
>
}
323
>
private _wake(): void {
325
>
this._unpark();
326
>
// Re-enter the trampoline on a host macrotask, NOT a microtask. This:
327
>
// - coalesces multiple wake() calls in the same tick,
328
>
// - keeps the driver off the caller's stack frame, and
329
>
// - lets the entire pending microtask closure (including microtasks
330
>
// enqueued AFTER this `_wake` call within the same outer microtask
331
>
// -- e.g. `queueMicrotask(...)` calls inside an `AsyncIterable`
332
>
// constructor that runs after `clock.schedule` triggered the wake)
333
>
// drain before the next `_step`. A microtask hop here would queue
334
>
// the driver in FIFO order with those subsequent microtasks, so the
335
>
// driver could run a virtual event before the consumer-side promise
336
>
// chain that depends on it has settled.
337
>
nextMacrotask(this._realApi, this._drive);
338
>
}
340
>
// ---- Run lifecycle --------------------------------------------------
341
>
342
>
private _settleFinishedRuns(): void {
344
>
if (run.settled) { continue; }
345
>
const status = run.evaluate(this._clock, this._executedTotal, () => this._buildOverflow(run));
346
>
if (status === 'done') {
350
>
}
352
>
}
354
>
private _settleRun(run: Run, error?: Error): void {
356
>
if (!cleanup) { return; }
357
>
this._runs.delete(run);
358
>
cleanup.dispose();
359
>
run.settle(error);
360
>
}
362
>
private _buildOverflow(run: Run): Error {
364
>
return new Error(
365
>
`[VirtualTimeProcessor] Run #${run.id} exceeded maxEvents (${run.maxEvents}) — ` +
366
>
`executed ${local} virtual event(s) and the queue is still not empty.`
367
>
);
368
>
}
370
>
private _buildDepthOverflow(run: Run, depth: number): Error {
371
return new Error(
372
`[VirtualTimeProcessor] Run #${run.id} exceeded maxTraceDepth (${run.options.maxTraceDepth}) — ` +
373
`next event has trace depth ${depth}. ` +
374
`This usually indicates a runaway self-rescheduling timer.`
375
);
376
}
378
>
private _onDispose(): void {
380
>
this._unpark();
381
>
const err = new Error('VirtualTimeProcessor disposed');
382
>
for (const run of [...this._runs.keys()]) { this._settleRun(run, err); }
383
>
}