use std::cell::UnsafeCell;
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::ffi::c_void;
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, AtomicUsize};
use std::sync::{Arc, Condvar, Mutex};
use std::time::Instant;
use crate::config::{GraphConfig, GraphPlan, StatsLevel};
use crate::context::{Context, Options};
use crate::executor::{DelegatingExecutor, Executor, ThreadPool, DEFAULT_EXECUTOR_NAME};
use crate::kernel::{Contract, KernelInstance, PortTable};
use crate::runtime::{self, GraphShared};
use crate::status::{Error, Result};
use crate::timestamp::Timestamp;
use super::{
Activity, Edge, EdgeId, GraphInner, InputPolicy, InputQueueStats, Node, NodeSched, NodeStats,
OnError, State,
};
impl GraphInner {
pub(super) fn build(cfg: GraphConfig) -> Result<Self> {
let plan = GraphPlan::build(cfg)?;
let diagnostics = plan.diagnostics();
let cfg = plan.config;
let configured_stats = cfg.stats.unwrap_or_else(|| {
cfg.stats_timing.map_or(StatsLevel::Basic, |enabled| {
if enabled {
StatsLevel::Full
} else {
StatsLevel::Basic
}
})
});
let stats_level = cfg.effective_stats_level();
let edges: Vec<Edge> = plan
.edges
.iter()
.map(|planned| {
let mut edge = Edge::new(planned.name.clone());
edge.is_graph_input = planned.graph_input;
edge.is_graph_output = planned.graph_output;
edge.producer = planned.producer;
edge.consumers = planned.consumer_ports.clone();
edge
})
.collect();
let edge_by_name: BTreeMap<String, EdgeId> = plan.edge_by_name.clone();
let graph_inputs = plan.graph_inputs.clone();
let graph_outputs = plan.graph_outputs.clone();
let input_by_name = plan.input_by_name.clone();
let output_by_name = plan.output_by_name.clone();
let node_port_tables: Vec<(Arc<PortTable>, Arc<PortTable>)> = plan
.nodes
.iter()
.map(|node| (node.input_ports.clone(), node.output_ports.clone()))
.collect();
for diagnostic in diagnostics {
runtime::log_warn(&diagnostic.message);
}
let back_edge_mask = plan.back_edge_mask.clone();
let mut executors: Vec<Executor> = vec![Executor::Pool(default_thread_pool(stats_level))];
for (executor_index, e) in cfg.executors.iter().enumerate() {
executors.push(
build_executor(&e.name, e, stats_level)
.map_err(|error| error.context(format!("executors[{executor_index}]")))?,
);
}
let mut contracts = Vec::with_capacity(cfg.nodes.len());
let mut kernels = Vec::with_capacity(cfg.nodes.len());
for (idx, n) in cfg.nodes.iter().enumerate() {
let name = node_label(n, idx);
let (ins, outs) = node_port_tables[idx].clone();
let mut contract = Contract::new(ins, outs);
if n.r#type != "route" {
let contract_result = unsafe {
KernelInstance::fill_contract(
&n.kernel,
&mut contract as *mut Contract as *mut c_void,
)
};
contract_result.map_err(|error| error.context(format!("nodes[{idx}].kernel")))?;
}
if let Some(error) = contract.take_error() {
return Err(Error::InvalidArg(format!(
"nodes[{idx}].kernel (node `{name}`): GetContract failed: {error}"
)));
}
validate_contract(&format!("nodes[{idx}]"), &name, &contract)?;
contracts.push(contract);
kernels.push(if n.r#type == "route" {
KernelInstance::create_route(
n.route
.clone()
.expect("validated route node has route configuration"),
)
} else {
KernelInstance::create(&n.kernel)
.map_err(|error| error.context(format!("nodes[{idx}].kernel")))?
});
}
check_edge_type_compatibility(&cfg, &edges, &contracts)?;
let shared = Arc::new(GraphShared::new(cfg.clone()));
let mut nodes = Vec::new();
let mut required: Vec<(String, String)> = Vec::new();
for ((idx, n), (contract, kernel)) in cfg
.nodes
.iter()
.enumerate()
.zip(contracts.into_iter().zip(kernels))
{
let name = node_label(n, idx);
let (ins, outs) = node_port_tables[idx].clone();
let planned = &plan.nodes[idx];
let input_edges = planned.input_edges.clone();
let output_edges = planned.output_edges.clone();
let mif = n.max_in_flight.max(1);
let options = Arc::new(Options::new(n.options.clone()));
let kernel_name = if n.r#type == "route" {
"__lmflow.route".to_string()
} else {
n.kernel.clone()
};
let make_ctx = || {
Context::new(
name.clone(),
kernel_name.clone(),
ins.clone(),
outs.clone(),
options.clone(),
Arc::new(BTreeMap::new()), shared.clone(),
)
};
let ctxs: Vec<UnsafeCell<Context>> =
(0..mif).map(|_| UnsafeCell::new(make_ctx())).collect();
let executor = planned.executor_index;
nodes.push(Node {
name,
kernel_name,
inputs: input_edges,
outputs: output_edges,
in_ports: ins.clone(),
out_ports: outs,
executor,
policy: InputPolicy::from_config(&n.input_policy, &ins)?,
input_types: contract.input_types.clone(),
output_types: contract.output_types.clone(),
kernel,
ctxs,
max_in_flight: mif,
sched: Mutex::new(NodeSched::new(mif)),
stats: NodeStats::default(),
input_queues: (0..ins.len())
.map(|_| Mutex::new(VecDeque::new()))
.collect(),
input_queue_capacity: ins
.names()
.iter()
.map(|port| {
n.input_queues
.ports
.get(port)
.and_then(|limits| limits.packets)
.unwrap_or(n.input_queues.packets)
})
.map(|capacity| (capacity != 0).then_some(capacity))
.collect(),
input_queue_reserved: (0..ins.len()).map(|_| AtomicUsize::new(0)).collect(),
input_queue_bytes: (0..ins.len()).map(|_| AtomicU64::new(0)).collect(),
input_queue_stats: (0..ins.len()).map(|_| InputQueueStats::default()).collect(),
input_closed: (0..ins.len()).map(|_| AtomicBool::new(false)).collect(),
on_error: OnError::from_config(&n.on_error),
min_period: if n.rate > 0.0 {
Some(std::time::Duration::from_secs_f64(1.0 / n.rate))
} else {
None
},
last_fire: Mutex::new(None),
input_is_back_edge: back_edge_mask[idx].clone(),
source_done: AtomicBool::new(false),
source_waiting: AtomicBool::new(false),
source_wait_reason: AtomicUsize::new(0),
source_wake_deadline_us: AtomicI64::new(0),
source_yield_count: AtomicU64::new(0),
source_wake_generation: AtomicU64::new(0),
input_bounds: (0..ins.len())
.map(|_| Mutex::new(Timestamp::pre_stream()))
.collect(),
});
for name in &contract.required_side_packets {
required.push((
name.clone(),
nodes.last().expect("just inserted").name.clone(),
));
}
}
if configured_stats != StatsLevel::Full && shared.config.watchdog_ms > 0 {
runtime::log_info(
"stats is overridden to full because watchdog_ms > 0 (the watchdog needs per-call timing)",
);
} else if stats_level != StatsLevel::Full {
runtime::log_info(
"stats is not full: per-call timing, latency percentiles, CoW copies, and executor timing are disabled",
);
}
Ok(Self {
shared,
nodes,
edges,
graph_inputs,
graph_outputs,
input_by_name,
output_by_name,
edge_by_name,
state: Mutex::new(State::Initialized),
executors,
in_flight: AtomicUsize::new(0),
delegated_cursor: AtomicUsize::new(0),
delegated_running: AtomicBool::new(false),
activity: (Mutex::new(Activity::default()), Condvar::new()),
wakeup_callback: Mutex::new(None),
wakeup_pending: AtomicBool::new(false),
wakeup_generation: AtomicU64::new(0),
paused: AtomicBool::new(false),
blocked_flush_nodes: Mutex::new(BTreeSet::new()),
side_packets: Mutex::new(BTreeMap::new()),
required_side_packets: required,
epoch: Instant::now(),
run_started_us: AtomicI64::new(0),
dot_intervals: Mutex::new(super::dot::DotIntervalBaselines::default()),
stats_level,
})
}
}
fn default_thread_pool(stats_level: StatsLevel) -> ThreadPool {
let n = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
ThreadPool::new_with_stats(DEFAULT_EXECUTOR_NAME, n, Vec::new(), 0, stats_level)
}
fn build_executor(
name: &str,
e: &crate::config::ExecutorConfig,
stats_level: StatsLevel,
) -> Result<Executor> {
if e.r#type == "DelegatingExecutor" {
for (field, set) in [
("num_threads", e.num_threads != 0),
("affinity", !e.affinity.is_empty()),
("priority", e.priority != 0),
] {
if set {
return Err(Error::InvalidArg(format!(
"executor `{name}`: DelegatingExecutor owns no threads, so `{field}` is \
meaningless -- it hands ready nodes back to the host thread. Drop the field, \
or use type: \"ThreadPoolExecutor\" if you wanted engine-owned threads"
)));
}
}
return Ok(Executor::Delegating(DelegatingExecutor::new_with_stats(
name,
stats_level,
)));
}
Ok(Executor::Pool(ThreadPool::new_with_stats(
name,
e.num_threads,
e.affinity.clone(),
e.priority,
stats_level,
)))
}
fn validate_contract(path: &str, name: &str, contract: &Contract) -> Result<()> {
for (which, types) in [
("input", &contract.input_types),
("output", &contract.output_types),
] {
for (i, &type_id) in types.iter().enumerate() {
if type_id == crate::packet::type_id::HOST_OBJECT {
return Err(Error::InvalidArg(format!(
"{path}.kernel (node `{name}`): {which} port {i} declares LMFLOW_TYPE_HOST_OBJECT, \
which is reserved and not enabled (see ADR #26). Host-language native \
objects (e.g. PyObject) would create a second type system invisible to \
the YAML graph, and their refcount can drop on an engine worker thread \
where releasing them needs the GIL. Use LMFLOW_TYPE_BUFFER for numeric \
collections, or LMFLOW_TYPE_STR carrying JSON for arbitrary metadata"
)));
}
if (8..16).contains(&type_id) {
return Err(Error::InvalidArg(format!(
"{path}.kernel (node `{name}`): {which} port {i} declares reserved type id {type_id}; \
built-in ids currently end at LMFLOW_TYPE_HOST_OBJECT (7), and custom \
type ids must be >= 16"
)));
}
if type_id >= 16 && crate::packet::type_descriptor(type_id).is_none() {
return Err(Error::InvalidArg(format!(
"{path}.kernel (node `{name}`): {which} port {i} declares unregistered custom type id \
{type_id}; register its stable name, size, and alignment with \
lmflow_register_type_descriptor before building the graph"
)));
}
}
}
let mut required = std::collections::BTreeSet::new();
for side_packet in &contract.required_side_packets {
if side_packet.is_empty() {
return Err(Error::InvalidArg(format!(
"{path}.kernel (node `{name}`): GetContract declares an empty required side packet name"
)));
}
if !required.insert(side_packet) {
return Err(Error::InvalidArg(format!(
"{path}.kernel (node `{name}`): GetContract declares required side packet `{side_packet}` more than once"
)));
}
}
Ok(())
}
fn check_edge_type_compatibility(
cfg: &GraphConfig,
edges: &[Edge],
contracts: &[Contract],
) -> Result<()> {
for edge in edges {
let Some(producer) = edge.producer else {
continue; };
let producer_port = contracts[producer]
.outputs
.index_by_name(&edge.name)
.expect("edge producer output must exist");
let produced = contracts[producer].output_types[producer_port];
if produced == crate::packet::type_id::NONE {
continue; }
for &(consumer, consumer_port) in &edge.consumers {
let wanted = contracts[consumer].input_types[consumer_port];
if wanted == crate::packet::type_id::NONE || wanted == produced {
continue;
}
return Err(Error::InvalidArg(format!(
"nodes[{consumer}].input_ports[{consumer_port}]: type mismatch on edge `{}`: \
nodes[{producer}] `{}` output port `{}` declares {}, but node `{}` input port `{}` declares {}",
edge.name,
node_label(&cfg.nodes[producer], producer),
contracts[producer]
.outputs
.name(producer_port)
.unwrap_or("?"),
crate::packet::type_name(produced),
node_label(&cfg.nodes[consumer], consumer),
contracts[consumer]
.inputs
.name(consumer_port)
.unwrap_or("?"),
crate::packet::type_name(wanted),
)));
}
}
Ok(())
}
fn node_label(n: &crate::config::NodeConfig, idx: usize) -> String {
if n.name.is_empty() {
format!("{}#{}", n.kernel, idx)
} else {
n.name.clone()
}
}