use std::fmt;
use std::io::{self, Write};
use std::ops::ControlFlow;
#[cfg(not(target_arch = "wasm32"))]
use henad_core::action::Fire;
use henad_core::action::{RefusedActions, Schedule, Scheduled};
use henad_core::explore::replay::Replay;
use henad_core::explore::value::{ValueError, check_value, parse_value};
use henad_core::export::state as state_export;
use henad_core::model::SimState;
use henad_core::params::ParamValue;
use henad_core::send_sync::WasmNotSend;
use henad_core::view::{EdgeView, GridView, PointView, StatEntry, StatValue};
use crate::entry::{ModelEntry, ModelState};
use crate::fault::{BUILDING, Fault, STEPPING, catching};
use crate::gpu::GpuContext;
#[cfg(not(target_arch = "wasm32"))]
use crate::gpu::{GpuSimState, fault::catching_on, stepping};
#[derive(Debug, Clone)]
pub struct RunSetup {
entry: ModelEntry,
values: Vec<ParamValue>,
seed: Option<u64>,
schedule: Schedule,
}
impl RunSetup {
pub(crate) fn new(entry: ModelEntry) -> Self {
let values = entry
.param_descriptors()
.iter()
.map(|descriptor| descriptor.kind.default_value())
.collect();
Self {
entry,
values,
seed: None,
schedule: Schedule::default(),
}
}
pub fn set(mut self, id: &str, value: impl Into<ParamValue>) -> Result<Self, SetupError> {
let value = value.into();
let index = param_index(&self.entry, id)?;
check_value(&self.entry.param_descriptors()[index].kind, &value).map_err(|error| param_error(id, error))?;
self.values[index] = value;
Ok(self)
}
pub fn set_text(mut self, id: &str, text: &str) -> Result<Self, SetupError> {
let index = param_index(&self.entry, id)?;
let value =
parse_value(&self.entry.param_descriptors()[index].kind, text).map_err(|error| param_error(id, error))?;
self.values[index] = value;
Ok(self)
}
pub fn with_seed(mut self, seed: u64) -> Self {
self.seed = Some(seed);
self
}
pub fn act_at(mut self, action_id: &str, tick: u64) -> Result<Self, SetupError> {
let index = action_index(&self.entry, action_id)?;
let mut entries = self.schedule.entries().to_vec();
entries.push(Scheduled {
index,
id: action_id.to_owned(),
tick,
});
self.schedule = Schedule::from_entries(entries);
Ok(self)
}
pub fn from_replay(entry: &ModelEntry, replay: &Replay) -> Result<Self, SetupError> {
if replay.model != entry.id() {
return Err(SetupError::WrongModel {
expected: entry.id().to_owned(),
found: replay.model.clone(),
});
}
Self::from_parts(entry, &replay.params, Some(replay.seed), replay.schedule.clone())
}
pub fn from_parts(
entry: &ModelEntry,
values: &[ParamValue],
seed: Option<u64>,
schedule: Schedule,
) -> Result<Self, SetupError> {
let descriptors = entry.param_descriptors();
if values.len() != descriptors.len() {
return Err(SetupError::ParamCount {
expected: descriptors.len(),
found: values.len(),
});
}
for (descriptor, value) in descriptors.iter().zip(values) {
check_value(&descriptor.kind, value).map_err(|error| param_error(descriptor.id, error))?;
}
let actions = entry.action_descriptors();
for scheduled in schedule.entries() {
if actions
.get(scheduled.index)
.is_none_or(|action| action.id != scheduled.id)
{
return Err(SetupError::UnknownAction {
id: scheduled.id.clone(),
});
}
}
Ok(Self {
entry: entry.clone(),
values: values.to_vec(),
seed,
schedule,
})
}
pub fn entry(&self) -> &ModelEntry {
&self.entry
}
pub fn values(&self) -> &[ParamValue] {
&self.values
}
pub fn seed(&self) -> Option<u64> {
self.seed
}
pub fn schedule(&self) -> &Schedule {
&self.schedule
}
pub fn build(&self, gpu: Option<&GpuContext>) -> Result<Simulation, Fault> {
#[cfg(target_arch = "wasm32")]
if self.entry.gpu_needs().is_some() {
return Err(gpu_in_browser(self.entry.id()));
}
let schedule = &self.schedule;
let engine = match self.entry.build(&self.values, self.seed, gpu)? {
ModelState::Cpu(mut state) => {
if !schedule.is_empty() {
cpu_call(BUILDING, || refusal(&schedule.run_due(&mut *state), BUILDING))?;
}
Engine::Cpu(state)
}
#[cfg(not(target_arch = "wasm32"))]
ModelState::Gpu(mut state) => {
let ctx = gpu
.ok_or_else(|| Fault::refused(BUILDING, "a GPU model was built with no device"))?
.clone();
if !schedule.is_empty() {
gpu_call(&ctx, BUILDING, || {
refusal(&stepping::run_due(&mut *state, &ctx, schedule), BUILDING)?;
stepping::wait(&ctx)
})?;
}
Engine::Gpu { state, ctx }
}
#[cfg(target_arch = "wasm32")]
ModelState::Gpu(_) => return Err(gpu_in_browser(self.entry.id())),
};
Ok(Simulation {
setup: self.clone(),
engine,
earlier_fault: None,
})
}
}
fn param_index(entry: &ModelEntry, id: &str) -> Result<usize, SetupError> {
entry.param_index(id).ok_or_else(|| {
SetupError::Param(ValueError::UnknownParam {
id: id.to_owned(),
known: entry
.param_descriptors()
.iter()
.map(|descriptor| descriptor.id)
.collect(),
})
})
}
fn action_index(entry: &ModelEntry, id: &str) -> Result<usize, SetupError> {
entry
.action_index(id)
.ok_or_else(|| SetupError::UnknownAction { id: id.to_owned() })
}
fn param_error(id: &str, error: ValueError) -> SetupError {
SetupError::Param(ValueError::Param {
id: id.to_owned(),
source: Box::new(error),
})
}
fn refusal(refused: &RefusedActions<'_>, during: &'static str) -> Result<(), Fault> {
match refused.first() {
None => Ok(()),
Some(action) => Err(Fault::refused(
during,
format!(
"the model refused its own action '{}' at tick {}",
action.id, action.tick
),
)),
}
}
#[cfg(target_arch = "wasm32")]
fn gpu_in_browser(id: &str) -> Fault {
Fault::refused(
BUILDING,
format!("model '{id}' runs on the GPU, and a browser steps a GPU model through GpuSimThread alone"),
)
}
fn cpu_call<T: WasmNotSend>(
during: &'static str,
task: impl FnOnce() -> Result<T, Fault> + WasmNotSend,
) -> Result<T, Fault> {
in_pool(|| catching(during, task)?)
}
#[cfg(not(target_arch = "wasm32"))]
fn in_pool<T: Send>(task: impl FnOnce() -> T + Send) -> T {
rayon::scope(|_| task())
}
#[cfg(target_arch = "wasm32")]
fn in_pool<T>(task: impl FnOnce() -> T) -> T {
task()
}
#[derive(Debug)]
#[non_exhaustive]
pub enum SetupError {
Param(ValueError),
ParamCount {
expected: usize,
found: usize,
},
UnknownAction {
id: String,
},
ReloadOnly {
id: String,
},
WrongModel {
expected: String,
found: String,
},
Fault(Fault),
}
impl fmt::Display for SetupError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Param(_) => f.write_str("cannot set the parameter"),
Self::ParamCount { expected, found } => {
write!(f, "expected {expected} parameter values, found {found}")
}
Self::UnknownAction { id } => write!(f, "model has no action '{id}'"),
Self::ReloadOnly { id } => write!(f, "parameter '{id}' applies only when the model is built"),
Self::WrongModel { expected, found } => {
write!(f, "the run is of model '{found}', not '{expected}'")
}
Self::Fault(_) => f.write_str("the model faulted"),
}
}
}
impl std::error::Error for SetupError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Param(error) => Some(error),
Self::Fault(fault) => Some(fault),
Self::ParamCount { .. }
| Self::UnknownAction { .. }
| Self::ReloadOnly { .. }
| Self::WrongModel { .. } => None,
}
}
}
impl From<Fault> for SetupError {
fn from(fault: Fault) -> Self {
Self::Fault(fault)
}
}
enum Engine {
Cpu(Box<dyn SimState>),
#[cfg(not(target_arch = "wasm32"))]
Gpu {
state: Box<dyn GpuSimState>,
ctx: GpuContext,
},
}
impl Engine {
fn state(&self) -> &dyn SimState {
match self {
Self::Cpu(state) => &**state,
#[cfg(not(target_arch = "wasm32"))]
Self::Gpu { state, .. } => &**state,
}
}
}
pub struct Simulation {
setup: RunSetup,
engine: Engine,
earlier_fault: Option<String>,
}
#[cfg(not(target_arch = "wasm32"))]
const _: fn() = || {
fn send<T: Send>() {}
send::<Simulation>();
};
impl fmt::Debug for Simulation {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Simulation")
.field("setup", &self.setup)
.field("tick", &self.tick())
.finish_non_exhaustive()
}
}
impl Simulation {
pub fn setup(&self) -> &RunSetup {
&self.setup
}
pub fn tick(&self) -> u64 {
self.engine.state().tick()
}
pub fn population(&self) -> u64 {
self.engine.state().population()
}
pub fn heap_bytes(&self) -> usize {
self.engine.state().heap_bytes()
}
pub fn parallel_jobs(&self) -> Option<usize> {
self.engine.state().parallel_jobs()
}
pub fn step(&mut self) -> Result<(), Fault> {
self.run_for(1)
}
pub fn run_for(&mut self, ticks: u64) -> Result<(), Fault> {
let end = self.tick().saturating_add(ticks);
self.advance(end)
}
pub fn run_to(&mut self, tick: u64) -> Result<(), Fault> {
self.advance(tick)
}
pub fn run_sampled<B: WasmNotSend>(
&mut self,
end_tick: u64,
interval: u64,
mut on_sample: impl FnMut(&StatSample) -> ControlFlow<B> + WasmNotSend,
) -> Result<ControlFlow<B>, Fault> {
assert!(interval > 0, "a sampling interval is at least one tick");
self.check_earlier_fault()?;
let Self { setup, engine, .. } = self;
let schedule = &setup.schedule;
let sampled = match engine {
Engine::Cpu(state) => in_pool(|| {
let first = catching(STEPPING, || sample_cpu(&mut **state))?;
if let ControlFlow::Break(stop) = on_sample(&first) {
return Ok(ControlFlow::Break(stop));
}
while state.tick() < end_tick {
let next = next_sample_tick(state.tick(), interval, end_tick);
let sample = catching(STEPPING, || {
step_cpu(&mut **state, schedule, next)?;
Ok(sample_cpu(&mut **state))
})??;
if let ControlFlow::Break(stop) = on_sample(&sample) {
return Ok(ControlFlow::Break(stop));
}
}
Ok(ControlFlow::Continue(()))
}),
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { state, ctx } => run_sampled_gpu(&mut **state, ctx, schedule, end_tick, interval, on_sample),
};
self.record(sampled)
}
pub fn stats(&mut self) -> Result<StatSample, Fault> {
self.check_earlier_fault()?;
let sample = match &mut self.engine {
Engine::Cpu(state) => cpu_call(STEPPING, || Ok(sample_cpu(&mut **state))),
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { state, ctx } => sample_gpu(&mut **state, ctx),
};
self.record(sample)
}
pub fn set_param(&mut self, id: &str, value: impl Into<ParamValue>) -> Result<(), SetupError> {
let value = value.into();
let index = param_index(&self.setup.entry, id)?;
let descriptor = &self.setup.entry.param_descriptors()[index];
check_value(&descriptor.kind, &value).map_err(|error| param_error(id, error))?;
self.check_earlier_fault()?;
let accepted = match &mut self.engine {
Engine::Cpu(state) => catching(STEPPING, || state.set_param(index, &value)),
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { state, ctx } => gpu_call(ctx, STEPPING, || Ok(state.set_param(index, &value))),
};
if self.record(accepted)? {
Ok(())
} else {
Err(SetupError::ReloadOnly { id: id.to_owned() })
}
}
pub fn act(&mut self, action_id: &str) -> Result<(), SetupError> {
let index = action_index(&self.setup.entry, action_id)?;
self.check_earlier_fault()?;
let refused = || Fault::refused(STEPPING, format!("the model refused its own action '{action_id}'"));
let acted = match &mut self.engine {
Engine::Cpu(state) => cpu_call(STEPPING, || if state.act(index) { Ok(()) } else { Err(refused()) }),
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { state, ctx } => gpu_call(ctx, STEPPING, || {
let mut encoder = ctx.device.create_command_encoder(&wgpu::CommandEncoderDescriptor {
label: Some("henad_simulation_action"),
});
if !state.encode_action(&mut encoder, index) {
return Err(refused());
}
ctx.queue.submit(Some(encoder.finish()));
stepping::wait(ctx)
}),
};
Ok(self.record(acted)?)
}
pub fn views(&mut self) -> Result<SimulationViews<'_>, Fault> {
self.check_earlier_fault()?;
let prepared = match &mut self.engine {
Engine::Cpu(state) => cpu_call(STEPPING, || {
state.prepare_view();
Ok(())
}),
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { .. } => Ok(()),
};
self.record(prepared)?;
Ok(match &self.engine {
Engine::Cpu(state) => SimulationViews {
grid: state.grid_view(),
points: state.point_view(),
edges: state.edge_view(),
},
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { .. } => SimulationViews {
grid: None,
points: None,
edges: None,
},
})
}
pub fn relax_layout(&mut self, budget_ms: f32) -> Result<(), Fault> {
self.check_earlier_fault()?;
let relaxed = match &mut self.engine {
Engine::Cpu(state) => cpu_call(STEPPING, || {
if state.set_layout(true, budget_ms) {
state.relax_layout();
state.set_layout(false, budget_ms);
}
Ok(())
}),
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { .. } => Ok(()),
};
self.record(relaxed)
}
pub fn write_state(&mut self, writer: &mut dyn Write) -> Result<(), ExportError> {
self.check_earlier_fault()?;
match &mut self.engine {
Engine::Cpu(state) => {
let prepared = cpu_call(STEPPING, || {
state.prepare_view();
Ok(())
});
self.record(prepared)?;
write_views(self.engine.state(), writer)?;
Ok(())
}
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { .. } => Err(ExportError::GpuState),
}
}
fn advance(&mut self, end: u64) -> Result<(), Fault> {
self.check_earlier_fault()?;
let Self { setup, engine, .. } = self;
let schedule = &setup.schedule;
let advanced = match engine {
Engine::Cpu(state) => {
if state.tick() >= end {
return Ok(());
}
cpu_call(STEPPING, || step_cpu(&mut **state, schedule, end))
}
#[cfg(not(target_arch = "wasm32"))]
Engine::Gpu { state, ctx } => advance_gpu(&mut **state, ctx, schedule, end),
};
self.record(advanced)
}
fn check_earlier_fault(&self) -> Result<(), Fault> {
let Some(earlier) = &self.earlier_fault else {
return Ok(());
};
#[cfg(not(target_arch = "wasm32"))]
if let Engine::Gpu { ctx, .. } = &self.engine
&& ctx.is_lost()
{
return Err(Fault::device_lost(STEPPING));
}
Err(Fault::refused(
STEPPING,
format!("an earlier call faulted and the simulation has to be rebuilt ({earlier})"),
))
}
fn record<T>(&mut self, result: Result<T, Fault>) -> Result<T, Fault> {
if let Err(fault) = &result
&& self.earlier_fault.is_none()
{
self.earlier_fault = Some(fault.to_string());
}
result
}
}
#[cfg(not(target_arch = "wasm32"))]
fn run_sampled_gpu<B>(
state: &mut dyn GpuSimState,
ctx: &GpuContext,
schedule: &Schedule,
end_tick: u64,
interval: u64,
mut on_sample: impl FnMut(&StatSample) -> ControlFlow<B>,
) -> Result<ControlFlow<B>, Fault> {
let first = sample_gpu(state, ctx)?;
if let ControlFlow::Break(stop) = on_sample(&first) {
return Ok(ControlFlow::Break(stop));
}
while state.tick() < end_tick {
let next = next_sample_tick(state.tick(), interval, end_tick);
advance_gpu(state, ctx, schedule, next)?;
let sample = sample_gpu(state, ctx)?;
if let ControlFlow::Break(stop) = on_sample(&sample) {
return Ok(ControlFlow::Break(stop));
}
}
Ok(ControlFlow::Continue(()))
}
#[cfg(not(target_arch = "wasm32"))]
fn gpu_call<T>(ctx: &GpuContext, during: &'static str, f: impl FnOnce() -> Result<T, Fault>) -> Result<T, Fault> {
match catching_on(ctx, during, f) {
Ok(Ok(value)) => Ok(value),
Ok(Err(fault)) | Err(fault) => Err(if ctx.is_lost() {
Fault::device_lost(during)
} else {
fault
}),
}
}
#[cfg(not(target_arch = "wasm32"))]
fn advance_gpu(state: &mut dyn GpuSimState, ctx: &GpuContext, schedule: &Schedule, end: u64) -> Result<(), Fault> {
let count = end.saturating_sub(state.tick());
if count == 0 {
return Ok(());
}
gpu_call(ctx, STEPPING, || {
let refused = stepping::run_steps_acting(state, ctx, count, schedule, Fire::AfterStep)?;
refusal(&refused, STEPPING)
})
}
fn next_sample_tick(tick: u64, interval: u64, end_tick: u64) -> u64 {
(tick / interval)
.saturating_add(1)
.saturating_mul(interval)
.min(end_tick)
}
fn step_cpu(state: &mut dyn SimState, schedule: &Schedule, end: u64) -> Result<(), Fault> {
if schedule.is_empty() {
for _ in state.tick()..end {
state.step();
}
return Ok(());
}
while state.tick() < end {
let stop = schedule.next_due_after(state.tick()).map_or(end, |due| due.min(end));
for _ in state.tick()..stop {
state.step();
}
refusal(&schedule.run_due(state), STEPPING)?;
}
Ok(())
}
fn sample_cpu(state: &mut dyn SimState) -> StatSample {
state.prepare_view();
StatSample {
tick: state.tick(),
entries: state.stats(),
}
}
#[cfg(not(target_arch = "wasm32"))]
fn sample_gpu(state: &mut dyn GpuSimState, ctx: &GpuContext) -> Result<StatSample, Fault> {
let entries = gpu_call(ctx, STEPPING, || stepping::sample_stats(state, ctx))?;
stepping::wait(ctx)?;
Ok(StatSample {
tick: state.tick(),
entries,
})
}
fn write_views(state: &dyn SimState, mut writer: &mut dyn Write) -> io::Result<()> {
if let Some(grid) = state.grid_view() {
state_export::write_grid(&mut writer, grid.width, grid.height, grid.cells)?;
}
let points = state.point_view();
if let Some(points) = &points {
state_export::write_points(&mut writer, points.pos_x, points.pos_y, points.color)?;
}
if let (Some(points), Some(edges)) = (&points, state.edge_view()) {
let rows = state_export::point_rows(points.pos_x, points.pos_y);
state_export::write_edges(&mut writer, edges.src, edges.dst, edges.color.unwrap_or(&[]), &rows)?;
}
Ok(())
}
#[derive(Debug)]
#[non_exhaustive]
pub struct SimulationViews<'a> {
pub grid: Option<GridView<'a>>,
pub points: Option<PointView<'a>>,
pub edges: Option<EdgeView<'a>>,
}
#[derive(Debug)]
#[non_exhaustive]
pub enum ExportError {
GpuState,
Io(io::Error),
Fault(Fault),
}
impl fmt::Display for ExportError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::GpuState => f.write_str("a GPU model's state stays on the device and cannot be exported"),
Self::Io(_) => f.write_str("cannot write the state"),
Self::Fault(_) => f.write_str("the model faulted while preparing its views"),
}
}
}
impl std::error::Error for ExportError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::GpuState => None,
Self::Io(error) => Some(error),
Self::Fault(fault) => Some(fault),
}
}
}
impl From<io::Error> for ExportError {
fn from(error: io::Error) -> Self {
Self::Io(error)
}
}
impl From<Fault> for ExportError {
fn from(fault: Fault) -> Self {
Self::Fault(fault)
}
}
#[derive(Debug, Clone)]
pub struct StatSample {
tick: u64,
entries: Vec<StatEntry>,
}
impl StatSample {
pub fn tick(&self) -> u64 {
self.tick
}
pub fn entries(&self) -> &[StatEntry] {
&self.entries
}
pub fn get(&self, label: &str) -> Option<&StatValue> {
self.entries
.iter()
.find(|entry| entry.label == label)
.map(|entry| &entry.value)
}
pub fn scalar(&self, label: &str) -> Option<f64> {
self.get(label).map(StatValue::scalar)
}
}