Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -255,7 +255,7 @@ APIs:

- [Estimator/FPE Python and Rust SDK](docs/core-api.md)
- [AIC-compatible modeled-power contract (semantics only)](docs/power-model.md)
- [FPM collection-to-prediction workflow](python/aisimulate/docs/fpm/end-to-end-workflow.md)
- [Self-benchmarking and FPM onboarding guide](python/aisimulate/docs/fpm/self-benchmarking-and-onboarding.md)
- [Replay SDK and artifact contract](crates/core/src/replay/README.md)
- [Sweeper SDK](docs/sweeper/overview.md)
- [AIConfigurator compatibility Python API](python/aisimulate/README.md#python-api)
Expand Down
2 changes: 1 addition & 1 deletion crates/core/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ pub use perfmodel::{
};
pub use perfmodel::{
CorrectionFactorBounds, CorrectionFeatureSpace, FpmInterpolationConfig, OpLevelConfig,
RegressionFitConfig, RegressionFitKind,
RegressionFitConfig, RegressionFitKind, UnrecordedFpmQuantMode,
};

#[cfg(feature = "python")]
Expand Down
2 changes: 2 additions & 0 deletions crates/core/src/perfmodel/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,8 @@ pub struct ParallelMapping {
/// re-derived from this field.
#[serde(default)]
pub cp_size: Option<u32>,
#[serde(default)]
pub dcp_size: Option<u32>,
}

/// Precision/quantization dtypes. Flattened into [`EngineConfig`]. Field
Expand Down
1 change: 1 addition & 0 deletions crates/core/src/perfmodel/engine/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1933,6 +1933,7 @@ mod tests {
forward_model: None,
kv_block_size: None,
parallel: ParallelMapping {
dcp_size: None,
Comment thread
simone-chen marked this conversation as resolved.
Resolved
tp_size: 8,
pp_size: 1,
attention_dp_size: Some(1),
Expand Down
14 changes: 13 additions & 1 deletion crates/core/src/perfmodel/engine/spec.rs
Original file line number Diff line number Diff line change
Expand Up @@ -769,6 +769,7 @@ mod tests {
forward_model: None,
kv_block_size: Some(64),
parallel: ParallelMapping {
dcp_size: None,
tp_size: 8,
pp_size: 1,
attention_dp_size: Some(8),
Expand Down Expand Up @@ -1166,9 +1167,20 @@ mod tests {
EngineSpec::from_bincode(&bytes),
Err(AicError::UnsupportedSchemaVersion {
got: 17,
expected: 18,
expected: ENGINE_SPEC_SCHEMA_VERSION,
..
})
));
}
#[test]
fn recorded_dcp_and_legacy_absence_round_trip() {
let mut spec = handshake_spec();
spec.engine.parallel.dcp_size = Some(2);
let decoded = EngineSpec::from_bincode(&spec.to_bincode().unwrap()).unwrap();
assert_eq!(decoded.engine.parallel.dcp_size, Some(2));
let mut legacy = serde_json::to_value(&spec.engine).unwrap();
legacy.as_object_mut().unwrap().remove("dcp_size");
let legacy: EngineConfig = serde_json::from_value(legacy).unwrap();
assert_eq!(legacy.parallel.dcp_size, None);
}
}
48 changes: 48 additions & 0 deletions crates/core/src/perfmodel/fpm/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,9 @@ pub struct ForwardPassPerfModelConfig {
pub pp: u32,
#[serde(default = "one", alias = "attention_dp_size")]
pub attention_dp: u32,
/// Recorded decode context parallelism within the TP group.
#[serde(default)]
pub dcp: Option<u32>,
#[serde(default)]
pub moe_tp_size: Option<u32>,
#[serde(default)]
Expand Down Expand Up @@ -155,6 +158,7 @@ impl ForwardPassPerfModelConfig {
tp: 1,
pp: 1,
attention_dp: 1,
dcp: None,
moe_tp_size: None,
moe_ep_size: None,
gemm_quant_mode: None,
Expand Down Expand Up @@ -217,6 +221,24 @@ impl ForwardPassPerfModelConfig {
if self.tp == 0 || self.pp == 0 || self.attention_dp == 0 {
return Err(invalid_config("tp, pp, and attention_dp must be positive"));
}
if self.dcp.is_some_and(|dcp| dcp == 0 || self.tp % dcp != 0) {
return Err(invalid_config("dcp must be positive and divide tp"));
}
for mode in &self
.estimator_config
.fpm_interpolation
.unrecorded_quant_modes
{
let explicit = match mode {
super::UnrecordedFpmQuantMode::Fmha => &self.fmha_quant_mode,
super::UnrecordedFpmQuantMode::Comm => &self.comm_quant_mode,
};
if explicit.is_some() {
return Err(invalid_config(
"an unrecorded FPM quant mode cannot have an explicit quantization override",
));
}
}
if self.moe_tp_size.is_some() != self.moe_ep_size.is_some() {
return Err(invalid_config(
"moe_tp_size and moe_ep_size must be configured together",
Expand Down Expand Up @@ -410,4 +432,30 @@ mod tests {
.is_err()
);
}
#[test]
fn recorded_dcp_is_validated_and_survives_round_trip() {
let cfg =
config(serde_json::json!({"tp": 8, "dcp": 8, "moe_tp_size": 8, "moe_ep_size": 1}));
cfg.validate().unwrap();
let restored: ForwardPassPerfModelConfig =
serde_json::from_str(&serde_json::to_string(&cfg).unwrap()).unwrap();
assert_eq!(restored.dcp, Some(8));
assert_eq!(config(serde_json::json!({})).dcp, None);
for value in [0, 3, 16] {
assert!(
config(serde_json::json!({"tp": 8, "dcp": value}))
.validate()
.is_err()
);
}
}

#[test]
fn unrecorded_profile_quantization_rejects_explicit_overrides() {
let mut cfg = config(serde_json::json!({"estimation_mode": "fpm_interpolation",
"estimator_config": {"fpm_interpolation": {"text_only": true, "unrecorded_quant_modes": ["fmha", "comm"]}}}));
cfg.validate().unwrap();
cfg.fmha_quant_mode = Some("bfloat16".into());
assert!(cfg.validate().is_err());
}
}
22 changes: 20 additions & 2 deletions crates/core/src/perfmodel/fpm/estimator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,26 @@ pub struct EstimatorConfig {
pub struct OpLevelConfig {}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FpmInterpolationConfig {}
#[serde(default, deny_unknown_fields)]
pub struct FpmInterpolationConfig {
/// The profile covers text prefill/decode; encoder weights remain resident.
#[serde(skip_serializing_if = "is_false")]
pub text_only: bool,
/// Match null profile identities only for these unspecified quant modes.
#[serde(skip_serializing_if = "Vec::is_empty")]
pub unrecorded_quant_modes: Vec<UnrecordedFpmQuantMode>,
}

fn is_false(value: &bool) -> bool {
!*value
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum UnrecordedFpmQuantMode {
Fmha,
Comm,
}

/// These weights currently affect regression. Native correction keeps its
/// established workload coordinates until a replacement is accuracy-qualified.
Expand Down
8 changes: 8 additions & 0 deletions crates/core/src/perfmodel/fpm/model.rs
Original file line number Diff line number Diff line change
Expand Up @@ -645,6 +645,14 @@ impl ForwardPassPerfModel {
fn build_native_candidate(
config: &ForwardPassPerfModelConfig,
) -> Result<(Engine, PathBuf), AicError> {
if config.dcp.is_some_and(|dcp| dcp > 1)
Comment thread
tedzhouhk marked this conversation as resolved.
Resolved
&& (config.estimation_mode != EstimationMode::FpmInterpolation
|| config.backend != crate::BackendKind::Vllm)
{
return Err(AicError::UnsupportedModel(
"DCP timing requires measured vLLM FPM interpolation".into(),
));
}
if config.estimation_mode == EstimationMode::FpmInterpolation && config.nextn != 0 {
return Err(AicError::UnsupportedModel(
"FPM interpolation does not support MTP".into(),
Expand Down
1 change: 1 addition & 0 deletions crates/core/src/perfmodel/fpm/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,7 @@ fn fixture_engine_config() -> EngineConfig {
forward_model: None,
kv_block_size: None,
parallel: ParallelMapping {
dcp_size: None,
tp_size: 8,
pp_size: 1,
attention_dp_size: Some(1),
Expand Down
1 change: 1 addition & 0 deletions crates/core/src/perfmodel/memory.rs
Original file line number Diff line number Diff line change
Expand Up @@ -509,6 +509,7 @@ mod tests {
forward_model: None,
kv_block_size: None,
parallel: ParallelMapping {
dcp_size: None,
tp_size: 1,
pp_size: 1,
attention_dp_size: Some(1),
Expand Down
2 changes: 1 addition & 1 deletion crates/core/src/perfmodel/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ pub use fpm::{
};
pub use fpm::{
CorrectionFactorBounds, CorrectionFeatureSpace, FpmInterpolationConfig, OpLevelConfig,
RegressionFitConfig, RegressionFitKind,
RegressionFitConfig, RegressionFitKind, UnrecordedFpmQuantMode,
};
// Forward-pass metrics telemetry types and schema version, plus the
// crate-internal validation helper. Re-exported at the crate root so existing
Expand Down
31 changes: 28 additions & 3 deletions crates/core/src/perfmodel/operators/fpm_forward.rs
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,20 @@ fn data_err(msg: String) -> AicError {
}

impl FpmForwardOp {
fn sol_at(&self, db: &PerfDatabase, coords: &[f64]) -> Result<f64, AicError> {
if self
.match_identity
.get(crate::perf_database::fpm_forward::FPM_CELL_MATCH_COLUMNS.len())
.and_then(|v| v.parse::<u32>().ok())
.is_some_and(|dcp| dcp > 1)
{
return Err(AicError::UnsupportedModel(
"DCP FPM supports measured interpolation, not op-level SOL transfer".into(),
));
}
sol_total(&self.sol_ops, self.phase, db, coords)
}

/// Mirror of Python `FPMForwardOp.query`: validate kwargs, map to
/// iteration-total coordinates, resolve against the selected cell.
pub fn query(
Expand Down Expand Up @@ -341,8 +355,8 @@ impl FpmForwardOp {
let candidate: Vec<f64> = std::iter::once(max as f64)
.chain(coords[1..].iter().copied())
.collect();
let true_sol = sol_total(&self.sol_ops, self.phase, db, coords);
let ceiling_sol = sol_total(&self.sol_ops, self.phase, db, &candidate);
let true_sol = self.sol_at(db, coords);
let ceiling_sol = self.sol_at(db, &candidate);
if let (Ok(t), Ok(c)) = (true_sol, ceiling_sol) {
if t.is_finite() && c.is_finite() && t > 0.0 && c > 0.0 {
// True shape is never costlier than the clamped
Expand Down Expand Up @@ -403,7 +417,7 @@ impl FpmForwardOp {
// and the error names the op.
let sol_failure: std::cell::RefCell<Option<AicError>> = std::cell::RefCell::new(None);
let sol = |sol_coords: &[f64]| -> f64 {
match sol_total(&self.sol_ops, self.phase, db, sol_coords) {
match self.sol_at(db, sol_coords) {
Ok(v) => v,
Err(err) => {
let mut slot = sol_failure.borrow_mut();
Expand Down Expand Up @@ -665,6 +679,17 @@ mod tests {
}
}

#[test]
fn recorded_dcp_disables_unmodeled_sol_transfer() {
let tmp = tempfile::tempdir().unwrap();
write_pair(tmp.path(), &default_rows());
let db = db_with_pair(tmp.path());
let mut measured = op(FpmPhase::Prefill);
measured.match_identity.push("4".into());
let error = measured.sol_at(&db, &[1.0, 1024.0, 0.0]).unwrap_err();
assert!(error.to_string().contains("not op-level SOL transfer"));
}

fn ctx(batch_size: u32, s: u32, prefix: u32) -> RuntimeContext {
RuntimeContext {
batch_size,
Expand Down
Loading
Loading