ferrum_types/
execution_policy.rs

1//! Typed execution-policy and admission authority contracts.
2
3use serde::{Deserialize, Serialize};
4
5/// Current CUDA native-adaptive boundary between single- and multi-partition
6/// addressed decode. The compiled provider and effective-config report consume
7/// this same constant.
8pub const CUDA_NATIVE_ADAPTIVE_V1_MAX_SEQUENCE_TOKENS: u64 = 512;
9
10/// Declares the authoritative owner of request-lifetime accelerator resources.
11///
12/// `PlanRuntime` owns admission, allocation, fences, backpressure, and release.
13/// `LegacyEngine` remains only for executors that have not migrated.
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "snake_case")]
16pub enum ExecutionResourceAuthority {
17    LegacyEngine,
18    PlanRuntime,
19}
20
21impl Default for ExecutionResourceAuthority {
22    fn default() -> Self {
23        Self::LegacyEngine
24    }
25}
26
27/// Product policy for selecting an attention provider implementation.
28///
29/// The policy selects a provider family, not a fixed kernel. An adaptive
30/// provider may select different physical variants for each invocation based
31/// on the admitted shape and sequence frontier.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(rename_all = "kebab-case")]
34pub enum AttentionExecutionPolicy {
35    /// Resolve to the best compiled provider family at composition time.
36    Auto,
37    /// Use the portable backend implementation and avoid optional native
38    /// accelerator libraries.
39    Portable,
40    /// Use a compiled native provider with invocation-shape selection.
41    NativeAdaptive,
42}
43
44impl AttentionExecutionPolicy {
45    pub const fn as_runtime_value(self) -> &'static str {
46        match self {
47            Self::Auto => "auto",
48            Self::Portable => "portable",
49            Self::NativeAdaptive => "native-adaptive",
50        }
51    }
52
53    pub fn parse_runtime_value(raw: &str) -> Result<Self, String> {
54        match raw.trim().to_ascii_lowercase().replace('_', "-").as_str() {
55            "auto" => Ok(Self::Auto),
56            "portable" => Ok(Self::Portable),
57            "native-adaptive" => Ok(Self::NativeAdaptive),
58            _ => Err(format!(
59                "expected auto, portable, or native-adaptive; got {raw:?}"
60            )),
61        }
62    }
63
64    /// Resolves the product request into the policy sealed into the runtime
65    /// policy and provider descriptor.
66    pub fn resolve(self, native_adaptive_supported: bool) -> Result<Self, String> {
67        match self {
68            Self::Auto if native_adaptive_supported => Ok(Self::NativeAdaptive),
69            Self::Auto => Ok(Self::Portable),
70            Self::NativeAdaptive if !native_adaptive_supported => Err(
71                "native-adaptive attention was requested but the selected backend composition does not provide it"
72                    .to_owned(),
73            ),
74            resolved => Ok(resolved),
75        }
76    }
77
78    pub const fn is_resolved(self) -> bool {
79        !matches!(self, Self::Auto)
80    }
81}
82
83impl Default for AttentionExecutionPolicy {
84    fn default() -> Self {
85        Self::Auto
86    }
87}
88
89/// Immutable admission ceilings compiled into one model executor.
90#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
91pub struct ExecutorAdmissionLimits {
92    maximum_active_sequences: u32,
93    maximum_scheduled_tokens: u64,
94}
95
96impl ExecutorAdmissionLimits {
97    pub fn new(
98        maximum_active_sequences: u32,
99        maximum_scheduled_tokens: u64,
100    ) -> Result<Self, String> {
101        if maximum_active_sequences == 0 {
102            return Err("maximum_active_sequences must be non-zero".to_owned());
103        }
104        if maximum_scheduled_tokens == 0 {
105            return Err("maximum_scheduled_tokens must be non-zero".to_owned());
106        }
107        Ok(Self {
108            maximum_active_sequences,
109            maximum_scheduled_tokens,
110        })
111    }
112
113    pub const fn maximum_active_sequences(self) -> u32 {
114        self.maximum_active_sequences
115    }
116
117    pub const fn maximum_scheduled_tokens(self) -> u64 {
118        self.maximum_scheduled_tokens
119    }
120}
121
122/// Runtime-owned admission state exposed to product health and validators.
123///
124/// This snapshot is deliberately separate from startup sizing estimates. Its
125/// phase counts come from one scheduler-owned request index, so one request
126/// cannot appear in both prefill and decode in the same observation.
127#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
128pub struct ExecutorAdmissionSnapshot {
129    schema_version: u32,
130    resource_authority: ExecutionResourceAuthority,
131    #[serde(flatten)]
132    limits: ExecutorAdmissionLimits,
133    waiting_requests: u32,
134    active_sequences: u32,
135    active_prefill_sequences: u32,
136    active_decode_sequences: u32,
137    current_batch_size: Option<u32>,
138    capacity_blocked_requests: Option<u32>,
139}
140
141impl ExecutorAdmissionSnapshot {
142    pub fn new(
143        resource_authority: ExecutionResourceAuthority,
144        limits: ExecutorAdmissionLimits,
145        waiting_requests: u32,
146        active_prefill_sequences: u32,
147        active_decode_sequences: u32,
148        current_batch_size: Option<u32>,
149        capacity_blocked_requests: Option<u32>,
150    ) -> Result<Self, String> {
151        let active_sequences = active_prefill_sequences
152            .checked_add(active_decode_sequences)
153            .ok_or_else(|| "active sequence phase counts overflow u32".to_owned())?;
154        if active_sequences > limits.maximum_active_sequences {
155            return Err(format!(
156                "active_sequences {active_sequences} exceeds maximum_active_sequences {}",
157                limits.maximum_active_sequences
158            ));
159        }
160        if current_batch_size.is_some_and(|batch| batch > active_sequences) {
161            return Err(format!(
162                "current_batch_size {} exceeds active_sequences {active_sequences}",
163                current_batch_size.unwrap_or_default()
164            ));
165        }
166        let observed_requests = waiting_requests
167            .checked_add(active_sequences)
168            .ok_or_else(|| "observed admission request counts overflow u32".to_owned())?;
169        if capacity_blocked_requests.is_some_and(|blocked| blocked > observed_requests) {
170            return Err(format!(
171                "capacity_blocked_requests {} exceeds observed waiting plus active requests {observed_requests}",
172                capacity_blocked_requests.unwrap_or_default()
173            ));
174        }
175        Ok(Self {
176            schema_version: 2,
177            resource_authority,
178            limits,
179            waiting_requests,
180            active_sequences,
181            active_prefill_sequences,
182            active_decode_sequences,
183            current_batch_size,
184            capacity_blocked_requests,
185        })
186    }
187
188    pub const fn resource_authority(&self) -> ExecutionResourceAuthority {
189        self.resource_authority
190    }
191
192    pub const fn maximum_active_sequences(&self) -> u32 {
193        self.limits.maximum_active_sequences
194    }
195
196    pub const fn maximum_scheduled_tokens(&self) -> u64 {
197        self.limits.maximum_scheduled_tokens
198    }
199
200    pub const fn active_sequences(&self) -> u32 {
201        self.active_sequences
202    }
203
204    pub const fn waiting_requests(&self) -> u32 {
205        self.waiting_requests
206    }
207
208    pub const fn active_prefill_sequences(&self) -> u32 {
209        self.active_prefill_sequences
210    }
211
212    pub const fn active_decode_sequences(&self) -> u32 {
213        self.active_decode_sequences
214    }
215
216    pub const fn current_batch_size(&self) -> Option<u32> {
217        self.current_batch_size
218    }
219
220    pub const fn capacity_blocked_requests(&self) -> Option<u32> {
221        self.capacity_blocked_requests
222    }
223}
224
225#[cfg(test)]
226mod tests {
227    use super::{
228        AttentionExecutionPolicy, ExecutionResourceAuthority, ExecutorAdmissionLimits,
229        ExecutorAdmissionSnapshot,
230    };
231
232    #[test]
233    fn attention_policy_resolves_auto_without_hiding_native_requirements() {
234        assert_eq!(
235            AttentionExecutionPolicy::Auto.resolve(true).unwrap(),
236            AttentionExecutionPolicy::NativeAdaptive
237        );
238        assert_eq!(
239            AttentionExecutionPolicy::Auto.resolve(false).unwrap(),
240            AttentionExecutionPolicy::Portable
241        );
242        assert!(AttentionExecutionPolicy::NativeAdaptive
243            .resolve(false)
244            .is_err());
245    }
246
247    #[test]
248    fn admission_snapshot_rejects_runtime_over_admission_ceiling() {
249        let limits = ExecutorAdmissionLimits::new(16, 2048).unwrap();
250        assert!(ExecutorAdmissionSnapshot::new(
251            ExecutionResourceAuthority::PlanRuntime,
252            limits,
253            0,
254            8,
255            9,
256            None,
257            None,
258        )
259        .is_err());
260    }
261
262    #[test]
263    fn admission_snapshot_reconciles_single_source_phase_counts() {
264        let snapshot = ExecutorAdmissionSnapshot::new(
265            ExecutionResourceAuthority::PlanRuntime,
266            ExecutorAdmissionLimits::new(32, 4096).unwrap(),
267            3,
268            7,
269            11,
270            Some(8),
271            Some(2),
272        )
273        .unwrap();
274        assert_eq!(snapshot.active_sequences(), 18);
275        assert_eq!(snapshot.active_prefill_sequences(), 7);
276        assert_eq!(snapshot.active_decode_sequences(), 11);
277        assert_eq!(snapshot.current_batch_size(), Some(8));
278    }
279}
ferrum_types/execution_policy.rs

ferrum_types/
execution_policy.rs
