#[cfg_attr(not(feature = "gpu"), allow(unused_imports))]
use std::sync::Arc;
use daedalus_planner::{Graph, PlannerConfig, PlannerInput, build_plan};
use daedalus_registry::store::Registry;
use daedalus_runtime::ExecutionTelemetry;
use daedalus_runtime::executor::{Executor, NodeHandler};
use daedalus_runtime::{HostBridgeManager, RuntimePlan, SchedulerConfig, build_runtime};
use crate::config::{EngineConfig, GpuBackend, RuntimeMode};
use crate::error::EngineError;
pub struct RunResult {
pub runtime_plan: RuntimePlan,
pub telemetry: ExecutionTelemetry,
}
pub struct Engine {
config: EngineConfig,
#[cfg(feature = "gpu")]
gpu_handle: std::sync::Mutex<Option<Arc<daedalus_gpu::GpuContextHandle>>>,
}
impl Engine {
pub fn new(config: EngineConfig) -> Result<Self, EngineError> {
config.validate().map_err(EngineError::Config)?;
if matches!(config.gpu, GpuBackend::Device | GpuBackend::Mock) && !cfg!(feature = "gpu") {
return Err(EngineError::FeatureDisabled("gpu"));
}
if config.runtime.lockfree_queues && !cfg!(feature = "lockfree-queues") {
return Err(EngineError::FeatureDisabled("lockfree-queues"));
}
Ok(Self {
config,
#[cfg(feature = "gpu")]
gpu_handle: std::sync::Mutex::new(None),
})
}
pub fn config(&self) -> &EngineConfig {
&self.config
}
pub fn plan(
&self,
registry: &Registry,
graph: Graph,
) -> Result<daedalus_planner::PlannerOutput, EngineError> {
let planner_cfg = self.planner_config()?;
let output = build_plan(PlannerInput { graph, registry }, planner_cfg);
let has_errors = output
.diagnostics
.iter()
.any(|d| !matches!(d.code, daedalus_planner::DiagnosticCode::LintWarning));
if has_errors {
return Err(EngineError::Planner(output.diagnostics));
}
Ok(output)
}
pub fn build_runtime_plan(
&self,
plan: &daedalus_planner::ExecutionPlan,
) -> Result<RuntimePlan, EngineError> {
let sched = SchedulerConfig {
default_policy: self.config.runtime.default_policy.clone(),
backpressure: self.config.runtime.backpressure.clone(),
lockfree_queues: self.config.runtime.lockfree_queues,
};
Ok(build_runtime(plan, &sched))
}
pub fn execute<H: NodeHandler + Send + Sync + 'static>(
&self,
runtime_plan: RuntimePlan,
handler: H,
) -> Result<ExecutionTelemetry, EngineError> {
let mut exec = Executor::new(&runtime_plan, handler)
.with_metrics_level(self.config.runtime.metrics_level);
#[cfg(feature = "gpu")]
{
if let Some(gpu) = self.get_gpu_handle()? {
exec = exec.with_gpu((*gpu).clone());
}
}
#[cfg(not(feature = "gpu"))]
{
if !matches!(self.config.gpu, GpuBackend::Cpu) {
return Err(EngineError::FeatureDisabled("gpu"));
}
let _ = exec;
}
if matches!(self.config.runtime.mode, RuntimeMode::Parallel) {
exec = exec.with_pool_size(self.config.runtime.pool_size);
}
let telemetry = match self.config.runtime.mode {
RuntimeMode::Serial => exec.run(),
RuntimeMode::Parallel => exec.run_parallel(),
}?;
Ok(telemetry)
}
pub fn execute_with_host<H: NodeHandler + Send + Sync + 'static>(
&self,
runtime_plan: RuntimePlan,
host: HostBridgeManager,
handler: H,
) -> Result<ExecutionTelemetry, EngineError> {
let mut exec = Executor::new(&runtime_plan, handler)
.with_host_bridges(host)
.with_metrics_level(self.config.runtime.metrics_level);
#[cfg(feature = "gpu")]
{
if let Some(gpu) = self.get_gpu_handle()? {
exec = exec.with_gpu((*gpu).clone());
}
}
#[cfg(not(feature = "gpu"))]
{
if !matches!(self.config.gpu, GpuBackend::Cpu) {
return Err(EngineError::FeatureDisabled("gpu"));
}
let _ = exec;
}
if matches!(self.config.runtime.mode, RuntimeMode::Parallel) {
exec = exec.with_pool_size(self.config.runtime.pool_size);
}
let telemetry = match self.config.runtime.mode {
RuntimeMode::Serial => exec.run(),
RuntimeMode::Parallel => exec.run_parallel(),
}?;
Ok(telemetry)
}
pub fn run<H: NodeHandler + Send + Sync + 'static>(
&self,
registry: &Registry,
graph: Graph,
handler: H,
) -> Result<RunResult, EngineError> {
let planner_output = self.plan(registry, graph)?;
let runtime_plan = self.build_runtime_plan(&planner_output.plan)?;
let telemetry = self.execute(runtime_plan.clone(), handler)?;
Ok(RunResult {
runtime_plan,
telemetry,
})
}
fn planner_config(&self) -> Result<PlannerConfig, EngineError> {
if self.config.planner.enable_gpu
&& !cfg!(feature = "gpu")
&& matches!(self.config.gpu, GpuBackend::Device | GpuBackend::Mock)
{
return Err(EngineError::FeatureDisabled("gpu"));
}
#[allow(unused_mut)]
let mut cfg = PlannerConfig {
enable_gpu: self.config.planner.enable_gpu,
enable_lints: self.config.planner.enable_lints,
active_features: self.config.planner.active_features.clone(),
#[cfg(feature = "gpu")]
gpu_caps: None,
};
#[cfg(feature = "gpu")]
{
if self.config.planner.enable_gpu {
let ctx = self.get_gpu_handle()?;
cfg.gpu_caps = ctx.as_ref().map(|c| c.capabilities());
}
}
#[cfg(not(feature = "gpu"))]
{
let _ = cfg.enable_gpu;
}
Ok(cfg)
}
#[cfg(feature = "gpu")]
fn get_gpu_handle(&self) -> Result<Option<Arc<daedalus_gpu::GpuContextHandle>>, EngineError> {
use daedalus_gpu::{GpuBackendKind, GpuOptions, select_backend};
if matches!(self.config.gpu, GpuBackend::Cpu) {
return Ok(None);
}
let mut guard = self.gpu_handle.lock().unwrap();
if let Some(handle) = guard.as_ref() {
return Ok(Some(handle.clone()));
}
let opts = match self.config.gpu {
GpuBackend::Cpu => return Ok(None),
GpuBackend::Mock => GpuOptions {
preferred_backend: Some(GpuBackendKind::Mock),
adapter_label: None,
allow_software: true,
},
GpuBackend::Device => GpuOptions {
preferred_backend: Some(GpuBackendKind::Wgpu),
adapter_label: None,
allow_software: false,
},
};
let handle = Arc::new(select_backend(&opts)?);
*guard = Some(handle.clone());
Ok(Some(handle))
}
}