1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
// SPDX-License-Identifier: MIT OR Apache-2.0
//! 2026-09-25: The pre-step tick phases: drain, admission, settling a step that
//! runs ahead, the snapshot, quiescent LoRA rotations, the shutdown check and
//! the admission-time swap-out.
//!
//! Owner: scheduler.
//! Invariants: none beyond the types.
use super::*;
impl SchedulerCore {
pub(super) fn plan_tick(&mut self) -> TickPlan<Lane> {
let Self {
ctx: sched,
pending,
active,
prefilling,
policy,
swapped,
preempted,
..
} = self;
let max_batch_size = self.max_batch_size;
let admit_watermark = self.admit_watermark;
let block_size = self.block_size;
// 2026-09-25: Each `t_loop` read opens one tick section, recorded as a `Loop*`
// phase (`mtp_timing::Phase::LoopDrain` and the rest).
let t_loop = sched.io.clock.now();
let new_reqs = drain_pending_requests(
&sched.io,
pending,
active,
prefilling,
&**policy,
&sched.levers,
max_batch_size,
!(swapped.is_empty() && preempted.is_empty()),
);
// 2026-09-25: Hold back what does not fit the KV pool or the adapter cohort
// (see `admission`).
let new_reqs = admission::gate_admissions(
sched,
sched.io.dev.model(),
pending,
new_reqs,
active,
prefilling,
swapped,
preempted,
admit_watermark,
sched.limits.max_seq_len,
block_size,
);
sched.io.tel.mark(mtp_timing::Phase::LoopDrain, t_loop);
// 2026-09-25: Settle a step running ahead of the host before this tick starts
// or continues a prefill, or may resume a parked sequence.
if self.pipeline.has_inflight()
&& (!new_reqs.is_empty()
|| !self.prefilling.is_empty()
|| !self.swapped.is_empty()
|| !self.preempted.is_empty())
{
self.drain_pipeline();
}
let Self {
ctx: sched,
pending,
active,
prefilling,
mtp_gate,
session_manager,
snapshot_steps,
swapped,
preempted,
..
} = self;
*snapshot_steps += 1;
let t_loop = sched.io.clock.now();
{
let (mtp_mode, delivered_tps) = match mtp_gate.as_ref() {
Some(g) => g.observe(),
None => (metrale_speculative::snapshot::MtpModeSnap::Off, 0.0),
};
sched
.io
.tel
.publish(metrale_speculative::snapshot::SchedulerSnapshot {
active_seqs: active.len() as u32,
prefilling_seqs: prefilling.len() as u32,
swapped_seqs: (swapped.len() + preempted.len()) as u32,
pending_len: new_reqs.len() as u32,
kv_blocks_free: sched.io.dev.model().num_free_blocks() as u32,
kv_blocks_total: sched.io.dev.model().num_total_blocks() as u32,
// 2026-09-25: The model's SSM snapshot pool occupancy.
// `SessionSsmManager::save_snapshot` has no production caller,
// so the session manager's counters are only a fallback for a
// model without a pool.
ssm_slots_used: sched
.io
.dev
.model()
.ssm_snapshot_occupancy()
.map(|(u, _)| u)
.unwrap_or(session_manager.session_count() as u32),
ssm_slots_total: sched
.io
.dev
.model()
.ssm_snapshot_occupancy()
.map(|(_, t)| t)
.unwrap_or(session_manager.total_slots() as u32),
mtp_mode,
delivered_tps,
steps_total: *snapshot_steps,
published_at: sched.io.clock.now(),
});
}
sched.io.tel.mark(mtp_timing::Phase::LoopSnapshot, t_loop);
// 2026-09-25: Apply queued LoRA rotations only when no sequence is active,
// prefilling, newly drained, spilled or requeued; otherwise they stay
// queued. A spilled sequence keeps the adapter slot and id it was
// prefilled under (`SwappedSeq::adapter_slot`/`adapter_id`), which a
// rotation could re-point before it resumes.
let mut rotated = false;
if active.is_empty()
&& prefilling.is_empty()
&& new_reqs.is_empty()
&& swapped.is_empty()
&& preempted.is_empty()
{
let rotations = std::mem::take(&mut pending.rotations);
rotated = !rotations.is_empty();
for (cmd, ack) in rotations {
let res = sched.io.dev.lora(cmd);
sched.io.req.lora_ack(ack, res);
}
}
// 2026-10-03: A tick that applied a rotation is not idle. Its ack can be what makes a
// client close the inbox, and whether that close lands before the check below would
// otherwise decide on which tick the loop stops; the next idle tick stops instead.
if !rotated && new_reqs.is_empty() && active.is_empty() && prefilling.is_empty() {
// 2026-09-25: Idle tick: stop once the inbox has closed.
pending.absorb(sched.io.req.recv(io::WaitPolicy::NoWait));
if pending.closed {
return TickPlan {
lanes: Vec::new(),
shutdown: true,
};
}
}
// 2026-09-25: Admission-time swap-out: for each new request, spill the active
// sequence holding the most KV blocks (grammar-active ones excluded) until
// the request's prompt fits or nothing more can be spilled.
if let Some(spill) = sched.io.spill.as_deref() {
for req in &new_reqs {
let prompt_len = req.prompt_len();
// 2026-09-25: Reclaim prefix-cache blocks before spilling a live sequence.
// `num_free_blocks()` does not count blocks the cache holds, so it
// can read near zero while blocks are reclaimable (see
// `Model::reclaim_prefix_blocks`).
// 2026-09-30: The target leaves out the cached prefix the prefill will
// reuse, and that prefix is never evicted to make room for its own request
// (`Model::reclaim_prefix_blocks_for`). A prompt-logprobs request reuses
// nothing.
let prompt = req.prompt_tokens_arc();
let reused: &[u32] = if req.prompt_logprobs().is_some() {
&[]
} else {
&prompt
};
let blocks_needed = sched
.io
.dev
.model()
.reclaim_prefix_blocks_for(
reused,
req.adapter_slot(),
prompt_len / block_size + 1,
)
.target;
while sched.io.dev.model().num_free_blocks() < blocks_needed && !active.is_empty() {
let victim_idx = active
.iter()
.enumerate()
.filter(|(_, a)| a.grammar_state.is_none())
.max_by_key(|(_, a)| a.seq.block_table.len())
.map(|(i, _)| i);
let Some(victim_idx) = victim_idx else {
tracing::warn!("No swappable sequences (all grammar-active)");
break;
};
match swap_out_sequence(&sched.io, active, victim_idx, spill) {
Ok(s) => {
tracing::info!(
"Swap-out: evicted seq (seq_len={}, blocks={}) to disk",
s.seq_len,
s.num_blocks,
);
swapped.push(s);
}
Err(e) => {
tracing::error!("Swap-out failed: {e:#}");
break;
}
}
}
}
}
TickPlan {
lanes: vec![
Lane::StartPrefills(new_reqs),
Lane::ContinuePrefills,
Lane::Decode,
],
shutdown: false,
}
}
}