Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 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
4 changes: 4 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -17,5 +17,9 @@ htmlcov/
*.so
*.dylib

# Whole-forward FPM datasets are external runtime inputs, not source assets.
**/fpm_forward_perf.parquet
**/fpm_forward_perf.metadata.json

# Rust build output
target/
6 changes: 6 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,12 @@

This file adds explicit repository-wide development guards.

## Pull request titles

- Use `<type>: <short description>` for every AISimulate PR title.
- Allowed types: `feat|fix|docs|style|refactor|perf|test|chore|ci|build|revert`.
- Check the title before creating or updating a PR.

## Required First Step

Before making any change under:
Expand Down
5 changes: 5 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,11 @@ configurations offline, without bringing up a GPU serving cluster.
[FPE Support Matrix](https://ai-dynamo.org/aisimulate/fpe-support-matrix/) ·
[Legacy AIC Support Matrix](https://ai-dynamo.org/aisimulate/support-matrix/)

Whole-forward FPM data is supplied at runtime rather than shipped in this
repository. Set `forward_model="fpm"` and pass `fpm_parquet_path` through the
prediction/recommendation YAML, Python API, or Rust API; see the
Comment thread
simone-chen marked this conversation as resolved.
Resolved
[core API guide](docs/core-api.md#choosing-a-forward-pass-api).

AISimulate is the successor to the
[AIConfigurator (AIC)](https://github.com/ai-dynamo/aiconfigurator)
repository. It brings the complete AIC application and estimator into one
Expand Down
30 changes: 29 additions & 1 deletion crates/core/src/perfmodel/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,32 @@
//! root, so `crate::EngineConfig`, `crate::BackendKind`, ... resolve unchanged.

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};

/// Validate an explicit FPM input before loading data or entering Python.
pub(crate) fn validate_fpm_parquet_path(
path: Option<&Path>,
is_fpm: bool,
) -> Result<Option<&str>, crate::AicError> {
let Some(path) = path else { return Ok(None) };
let path = path.to_str().ok_or_else(|| {
crate::AicError::InvalidEngineConfig("fpm_parquet_path must be valid UTF-8".into())
})?;
if path.is_empty() {
return Err(crate::AicError::InvalidEngineConfig(
"fpm_parquet_path cannot be empty".into(),
));
}
if !is_fpm {
return Err(crate::AicError::InvalidEngineConfig(
"fpm_parquet_path requires forward_model='fpm' with exactly one FpmForward op per phase".into(),
));
}
Ok(Some(path))
}

pub const ENGINE_CONFIG_SCHEMA_VERSION: u32 = 1;
// bincode op payloads are positional, so a producer/consumer skew is only
// distinguishable by this version — `EngineSpec::from_bincode` reads and
Expand Down Expand Up @@ -115,6 +137,12 @@ pub struct EngineConfig {
#[serde(default)]
pub forward_model: Option<String>,

/// Optional external FPM parquet used when `forward_model == "fpm"`.
/// The required metadata sidecar is resolved by replacing the parquet
/// extension with `.metadata.json`.
#[serde(default)]
pub fpm_parquet_path: Option<PathBuf>,

// KV
pub kv_block_size: Option<u32>,

Expand Down
85 changes: 67 additions & 18 deletions crates/core/src/perfmodel/engine/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -335,6 +335,22 @@ impl Engine {
spec.engine.database_mode, db.database_mode
)));
}
let nextn = spec
.engine
.speculative
.as_ref()
.and_then(|s| s.nextn)
.unwrap_or(0);
Self::validate_fpm_spec(&spec)?;
Ok(Engine {
context_ops: spec.context_ops,
generation_ops: spec.generation_ops,
db,
nextn,
})
}

fn validate_fpm_spec(spec: &EngineSpec) -> Result<(), AicError> {
let nextn = spec
.engine
.speculative
Expand All @@ -361,7 +377,10 @@ impl Engine {
})
}
let any_fpm = contains_fpm(&spec.context_ops) || contains_fpm(&spec.generation_ops);
if any_fpm {
if any_fpm
|| spec.engine.fpm_parquet_path.is_some()
|| spec.engine.forward_model.as_deref() == Some("fpm")
{
// Hybrid speculative shape: the FIRST op of each phase is the
// whole-model FpmForward (target), optionally followed by
// op-level DRAFT ops (the Python rewrite keeps a scheme's
Expand All @@ -376,6 +395,10 @@ impl Engine {
Some(Op::FpmForward(d)) if d.phase == FpmPhase::Decode
) && !contains_fpm(&spec.context_ops[1..])
&& !contains_fpm(&spec.generation_ops[1..]);
crate::config::validate_fpm_parquet_path(
spec.engine.fpm_parquet_path.as_deref(),
shape_ok,
)?;
if !shape_ok {
return Err(AicError::InvalidEngineConfig(
"forward_model='fpm' spec must contain exactly one FpmForward op per phase \
Expand Down Expand Up @@ -410,12 +433,7 @@ impl Engine {
));
}
}
Ok(Engine {
context_ops: spec.context_ops,
generation_ops: spec.generation_ops,
db,
nextn,
})
Ok(())
}

/// FPM whole-model engine: each phase list LEADS with one `FpmForward`
Expand Down Expand Up @@ -451,6 +469,7 @@ impl Engine {
systems_root: &std::path::Path,
) -> Result<Engine, AicError> {
let spec = EngineSpec::from_bincode(bytes)?;
Self::validate_fpm_spec(&spec)?;
Self::validate_engine_database_mode(spec.engine.database_mode)?;
let version = spec.engine.backend_version.as_deref().ok_or_else(|| {
AicError::InvalidEngineConfig(
Expand All @@ -468,7 +487,7 @@ impl Engine {
// engines would lazily re-parse the same parquet files on its first
// query (~0.5s per engine on data-rich systems). Mode/policy, memo
// caches, and the provenance accumulator stay per-engine.
let db = PerfDatabase::load_resolved_shared(
let db = PerfDatabase::load_resolved_shared_with_fpm(
systems_root,
&spec.engine.system_name,
spec.engine.backend.as_str(),
Expand All @@ -493,6 +512,7 @@ impl Engine {
spec.engine.database_mode,
DatabaseMode::Empirical | DatabaseMode::Sol
) || spec.engine.tolerate_dirless_version,
spec.engine.fpm_parquet_path.as_deref(),
)?
.with_mode(spec.engine.database_mode, transfer_policy);
Engine::build(spec, Arc::new(db))
Expand Down Expand Up @@ -1915,6 +1935,7 @@ mod tests {
backend: BackendKind::Vllm,
backend_version: Some("0.24.0".to_string()),
forward_model: None,
fpm_parquet_path: None,
kv_block_size: None,
parallel: ParallelMapping {
tp_size: 8,
Expand Down Expand Up @@ -2526,14 +2547,17 @@ mod tests {
use crate::perf_database::fpm_forward::tests::{
default_identity, default_rows, write_pair,
};
write_pair(tmp, &default_rows());
let mut db = PerfDatabase::load(&systems_root(), "b200_sxm", "vllm", "0.24.0").unwrap();
db.set_fpm_forward_for_test(crate::perf_database::FpmForwardTable::new(
tmp.to_path_buf(),
"b200_sxm",
"vllm",
"0.25.1",
));
let parquet = write_pair(tmp, &default_rows());
std::fs::copy(
systems_root().join("b200_sxm.yaml"),
tmp.join("b200_sxm.yaml"),
)
.unwrap();
let mut config = fixture_engine_config(nextn);
config.backend_version = Some("0.25.1".into());
config.forward_model = Some("fpm".into());
config.fpm_parquet_path = Some(parquet);
config.systems_path = Some(tmp.to_path_buf());
let fpm_op = |phase: FpmPhase| {
Op::FpmForward(FpmForwardOp {
name: format!("fpm_forward_{}", phase.as_str()),
Expand All @@ -2546,11 +2570,36 @@ mod tests {
})
};
let spec = EngineSpec::new(
fixture_engine_config(nextn),
config,
vec![fpm_op(FpmPhase::Prefill)],
vec![fpm_op(FpmPhase::Decode)],
);
Engine::build(spec, Arc::new(db))
Engine::from_spec_bytes(&spec.to_bincode()?, tmp)
}

#[test]
fn external_fpm_engines_share_tables_without_backend_data() {
let tmp = tempfile::tempdir().unwrap();
let first = build_fpm_engine(tmp.path(), None).unwrap();
let second = build_fpm_engine(tmp.path(), None).unwrap();
assert!(!tmp.path().join("data").exists());
assert!(Arc::ptr_eq(
first.database().tables_arc(),
second.database().tables_arc()
));
}

#[test]
fn external_fpm_rejects_invalid_specs_before_database_loading() {
let tmp = tempfile::tempdir().unwrap();
for path in ["", "/missing/reviewed-fpm.parquet"] {
let mut config = fixture_engine_config(None);
config.fpm_parquet_path = Some(path.into());
let spec = EngineSpec::new(config, context_ops(), generation_ops());
let err = Engine::from_spec_bytes(&spec.to_bincode().unwrap(), tmp.path()).unwrap_err();
assert!(matches!(err, AicError::InvalidEngineConfig(_)), "{err}");
assert!(err.to_string().contains("fpm_parquet_path"), "{err}");
}
Comment thread
simone-chen marked this conversation as resolved.
Resolved
}

#[test]
Expand Down
1 change: 1 addition & 0 deletions crates/core/src/perfmodel/engine/spec.rs
Original file line number Diff line number Diff line change
Expand Up @@ -767,6 +767,7 @@ mod tests {
backend: crate::BackendKind::Trtllm,
backend_version: Some("1.0.0rc3".into()),
forward_model: None,
fpm_parquet_path: None,
kv_block_size: Some(64),
parallel: ParallelMapping {
tp_size: 8,
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 @@ -106,6 +106,7 @@ fn fixture_engine_config() -> EngineConfig {
backend: BackendKind::Vllm,
backend_version: Some("0.24.0".to_string()),
forward_model: None,
fpm_parquet_path: None,
kv_block_size: None,
parallel: ParallelMapping {
tp_size: 8,
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 @@ -507,6 +507,7 @@ mod tests {
backend: BackendKind::Vllm,
backend_version: Some("test-version".to_string()),
forward_model: None,
fpm_parquet_path: None,
kv_block_size: None,
parallel: ParallelMapping {
tp_size: 1,
Expand Down
67 changes: 62 additions & 5 deletions crates/core/src/perfmodel/perf_database/fpm_forward.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,12 @@
//! collector pair
//!
//! ```text
//! <data_root>/fpm_forward_perf.parquet
//! <data_root>/fpm_forward_perf.metadata.json
//! /data/reviewed-fpm.parquet
//! /data/reviewed-fpm.metadata.json
//! ```
//!
//! is validated (sidecar schema/sha256/row_count, per-row workload checks,
//! The parquet may be supplied outside the AISimulate repository and is
//! validated (sidecar schema/sha256/row_count, per-row workload checks,
//! duplicate physical row keys), healed of
//! `kv_seed_regime == "fake_fallback"` values (fabricated-KV measurements
//! replaced in memory by in-station extrapolation; see
Expand Down Expand Up @@ -188,8 +189,25 @@ impl FpmForwardTable {
/// identity, enforced against every row (a misplaced pair — e.g. an h200
/// parquet copied into a b200 tree — must fail loudly, not merge).
pub fn new(data_root: PathBuf, system: &str, backend: &str, version: &str) -> Self {
Self::from_parquet_path(
data_root.join(FPM_FORWARD_BASENAME),
system,
backend,
version,
)
}

/// Construct a table backed by an explicitly supplied parquet outside the
/// bundled systems-data tree. The adjacent `.metadata.json` sidecar keeps
/// the same atomic identity, digest, and schema validation as bundled data.
pub fn from_parquet_path(
parquet_path: PathBuf,
system: &str,
backend: &str,
version: &str,
) -> Self {
let raw = std::env::var(FPM_FAKE_FALLBACK_RAW_ENV).is_ok_and(|v| v == "1");
Self::new_with_replacement(data_root, system, backend, version, !raw)
Self::from_parquet_path_with_replacement(parquet_path, system, backend, version, !raw)
}

/// Explicit-replacement constructor: lets tests pin both semantics
Expand All @@ -200,9 +218,25 @@ impl FpmForwardTable {
backend: &str,
version: &str,
replace_fake_fallback: bool,
) -> Self {
Self::from_parquet_path_with_replacement(
data_root.join(FPM_FORWARD_BASENAME),
system,
backend,
version,
replace_fake_fallback,
)
}

fn from_parquet_path_with_replacement(
parquet_path: PathBuf,
system: &str,
backend: &str,
version: &str,
replace_fake_fallback: bool,
) -> Self {
Self {
parquet_path: data_root.join(FPM_FORWARD_BASENAME),
parquet_path,
system: system.to_string(),
backend: backend.to_string(),
version: version.to_string(),
Expand Down Expand Up @@ -1401,6 +1435,29 @@ pub(crate) mod tests {
FpmForwardTable::new_with_replacement(dir.to_path_buf(), "b200_sxm", "vllm", "0.25.1", true)
}

#[test]
fn external_parquet_path_supports_an_arbitrary_stem() {
let tmp = tempfile::tempdir().expect("tmpdir");
let canonical = write_pair(tmp.path(), &default_rows());
let external = tmp.path().join("reviewed-fpm.parquet");
std::fs::rename(&canonical, &external).expect("rename parquet");
std::fs::rename(
canonical.with_extension("metadata.json"),
external.with_extension("metadata.json"),
)
.expect("rename sidecar");

let table =
FpmForwardTable::from_parquet_path(external.clone(), "b200_sxm", "vllm", "0.25.1");

assert_eq!(table.parquet_path(), external);
assert!(
table
.select_cell(&default_identity(4), "org/model-a")
.is_ok()
);
Comment thread
simone-chen marked this conversation as resolved.
Resolved
}

fn loaded_table_raw(dir: &Path) -> FpmForwardTable {
FpmForwardTable::new_with_replacement(
dir.to_path_buf(),
Expand Down
Loading