//! The device backend (native targets). use std::collections::HashMap; use fieldcraft_gpu::select::{self, Policy}; use fieldcraft_lbm::backend::{Backend, Fields}; use fieldcraft_lbm::lattice::{C, Q, W}; use fieldcraft_lbm::setup::MAX_LATTICE_SPEED; use fieldcraft_lbm::{Error, Flow, Group, Result, Rule, Target}; use fieldcraft_model::ThermalCondition; use wgpu::util::DeviceExt; const SHADER: &str = include_str!("lbm.wgsl"); const WG: u32 = 256; const MAX_GROUPS_X: u32 = 32768; const KIND_PULL: u32 = 0; const KIND_BOUNCE: u32 = 1; const KIND_REFLECT: u32 = 2; const KIND_OPEN: u32 = 3; /// The scalar lattice's directions (the first seven of D3Q19) and weights. const QT: usize = 7; const WT: [f64; QT] = [0.25, 0.125, 0.125, 0.125, 0.125, 0.125, 0.125]; fn gpu_error(e: impl std::fmt::Display) -> Error { Error::Invalid(format!("GPU: {e}")) } /// Steps per submission: the stability check and a cancel take effect between submissions. const MAX_CHUNK: u64 = 64; /// How long one wait for the device may take before the solve gives up on it. const DEVICE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120); /// A bounded chunk's dispatches, shared by the normal and timestamped paths. type Dispatch = (usize, &'static str, (u32, u32)); /// A vector of `len` copies of `value`, or [`Error::Limit`] when memory runs out (the host /// arrays are sized by the grid: 16 M cells is over a gigabyte of populations). fn filled(len: usize, value: T, what: &str) -> Result> { let mut v = Vec::new(); v.try_reserve_exact(len).map_err(|_| Error::Limit(format!("not enough memory for the {what} ({len} entries)")))?; v.resize(len, value); Ok(v) } /// An empty vector with room for `len`, or [`Error::Limit`]. fn reserved(len: usize, what: &str) -> Result> { let mut v = Vec::new(); v.try_reserve_exact(len).map_err(|_| Error::Limit(format!("not enough memory for the {what} ({len} entries)")))?; Ok(v) } /// Workgroup counts (x, y) covering `count` items. fn groups(count: usize) -> Result<(u32, u32)> { let total = u32::try_from(count.div_ceil(WG as usize).max(1)).map_err(|_| Error::Limit("too many cells for one dispatch".into()))?; let gx = total.min(MAX_GROUPS_X); let gy = total.div_ceil(gx); if gy > 65535 { return Err(Error::Limit("too many cells for one dispatch".into())); } Ok((gx, gy)) } fn bytes_u32(v: &[u32]) -> Vec { v.iter().flat_map(|x| x.to_le_bytes()).collect() } fn bytes_f32(v: &[f32]) -> Vec { v.iter().flat_map(|x| x.to_le_bytes()).collect() } /// Register this backend with `fieldcraft_lbm` (an application does this once at start), so /// `solve.fluid` with `backend: "gpu"` steps on the device `FIELDCRAFT_GPU` picks. pub fn register() { fieldcraft_lbm::register_backend(Box::new(|flow, _setup| GpuFlow::from_env(flow).map(|g| Box::new(g) as Box))); } /// How the momentum populations stream (see `lbm.wgsl`). #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub enum Streaming { /// In place: measured faster on every device tried (RTX 4090, GB10, M4 Max; #563). #[default] Auto, /// One population buffer, collided and streamed in place (the AA pattern). InPlace, /// Two population buffers: collide in place, then pull into the other (A–B). TwoBuffers, } /// How a [`GpuFlow`] runs; the defaults are what a solve uses. #[derive(Clone, Copy, Debug, Default)] pub struct Options { pub streaming: Streaming, /// Time every kernel with timestamp queries ([`GpuFlow::kernel_times`]). Each dispatch then /// gets its own pass and every chunk waits for its timestamps: for measurements, not solves. pub profile: bool, } /// GPU time per kernel, summed over the dispatches since the flow was created (with /// [`Options::profile`]). #[derive(Clone, Debug, Default, PartialEq)] pub struct KernelTime { pub dispatches: u64, pub seconds: f64, } /// A prepared flow on the device. pub struct GpuFlow { device: wgpu::Device, queue: wgpu::Queue, /// With [`Options::profile`]: the GPU time of each kernel so far. kernel_times: Option>>, /// How long creating the compute pipelines (compiling the shader) took. pipeline_seconds: f64, aa: bool, adapter: String, pipelines: HashMap<&'static str, wgpu::ComputePipeline>, /// AB swaps populations; AA keeps one population buffer and changes its orientation. /// Both modes swap the separate scalar lattice buffers. bind: [wgpu::BindGroup; 2], /// Which bind group's `f` holds the state. current: usize, n: usize, cells: (u32, u32), link_count: usize, links_groups: (u32, u32), open_count: usize, open_groups: (u32, u32), group_count: usize, force_groups: (u32, u32), stats: wgpu::Buffer, moments: wgpu::Buffer, forces: wgpu::Buffer, partials: wgpu::Buffer, partial_count: usize, /// The scalar populations remain A–B with heat; and T₀ they are shifted by. thermal: Option<[wgpu::Buffer; 2]>, initial_temperature: f64, fluid: Vec, fluid_cells: usize, groups: Vec, empty_obstacles: Vec, steps_done: u64, /// The momentum (xyz) and heat (w) each group received in the last step, once read back /// (zero before the first step). exchange_cache: Option>, /// Each group's share of the momentum from the populations' weights, which the kernel leaves /// out (`link_forces` sums the deviations): added to what it reads back. weight_forces: Vec<[f64; 3]>, } impl GpuFlow { /// `flow` on the device `FIELDCRAFT_GPU` picks. pub fn from_env(flow: &Flow) -> Result { let policy = Policy::from_env().map_err(gpu_error)?; GpuFlow::new(flow, policy) } /// `flow` (its grid, rules and initial state: everything the step needs) on the device /// `policy` picks. pub fn new(flow: &Flow, policy: Policy) -> Result { GpuFlow::with_options(flow, policy, Options::default()) } /// [`GpuFlow::new`] with `options`. pub fn with_options(flow: &Flow, policy: Policy, options: Options) -> Result { let Some((adapter, tier)) = select::adapter(policy).map_err(gpu_error)? else { return Err(Error::Invalid(format!("{}=cpu: no GPU device; solve with `backend: \"cpu\"`", select::ENV))); }; let info = adapter.get_info(); let profile = options.profile; let features = if profile { wgpu::Features::TIMESTAMP_QUERY } else { wgpu::Features::empty() }; if !adapter.features().contains(features) { return Err(gpu_error("profiling needs timestamp queries, which this adapter does not offer")); } let have = adapter.limits(); let mut limits = wgpu::Limits::default().using_resolution(have.clone()); limits.max_buffer_size = have.max_buffer_size; limits.max_storage_buffer_binding_size = have.max_storage_buffer_binding_size; limits.max_storage_buffers_per_shader_stage = have.max_storage_buffers_per_shader_stage.max(limits.max_storage_buffers_per_shader_stage); let (device, queue) = pollster::block_on(adapter.request_device(&wgpu::DeviceDescriptor { label: Some("fieldcraft lbm"), required_features: features, required_limits: limits, ..Default::default() })) .map_err(gpu_error)?; let adapter_name = format!("{}{}", info.name, if tier == select::Tier::Software { " (software)" } else { "" }); let aa = match options.streaming { Streaming::Auto | Streaming::InPlace => true, Streaming::TwoBuffers => false, }; let n = flow.grid().len(); let elem = 4u64; let pop_bytes = elem * (Q * n) as u64; let cap = device.limits().max_storage_buffer_binding_size.min(device.limits().max_buffer_size); if pop_bytes > cap { return Err(Error::Limit(format!( "{n} cells need {} MB of populations per buffer; this GPU allows {} MB per buffer: coarsen the spacing", pop_bytes >> 20, cap >> 20 ))); } // The state and the flags. // Populations as deviations from their weights (see lbm.wgsl); solid cells stay zero. let mut f = filled(Q * n, 0.0f32, "populations")?; for q in 0..Q { let w = fieldcraft_lbm::lattice::W.get(q).copied().unwrap_or(0.0); for cell in 0..n { if let (Some(slot), true) = (f.get_mut(q * n + cell), flow.is_fluid(cell)) { *slot = (flow.population(q, cell) - w) as f32; } } } // The scalar populations as deviations from w_q T₀ (see lbm.wgsl). let thermal = flow.thermal(); let t0 = thermal.map_or(0.0, |t| t.initial()); let mut tpop = filled(if thermal.is_some() { QT * n } else { 0 }, 0.0f32, "temperature populations")?; if thermal.is_some() { for q in 0..QT { let w = WT.get(q).copied().unwrap_or(0.0) * t0; for cell in 0..n { if let (Some(slot), true) = (tpop.get_mut(q * n + cell), flow.is_fluid(cell)) { *slot = (flow.thermal_population(q, cell) - w) as f32; } } } } let mut fluid = filled(n, false, "cell flags")?; let mut fluid_u32 = filled(n, 0u32, "cell flags")?; for cell in 0..n { if let (Some(b), Some(u)) = (fluid.get_mut(cell), fluid_u32.get_mut(cell)) { *b = flow.is_fluid(cell); *u = u32::from(*b); } } // The links, by cell, and the ones with a force by group. let links = flow.links(); let mut link_u = reserved(4 * links.len(), "links")?; let mut link_f = reserved(links.len(), "links")?; let mut by_group: Vec<(u32, u32)> = reserved(links.len(), "links by group")?; let group_count = flow.groups().len(); let mut weight_forces = filled(group_count, [0.0f64; 3], "force groups")?; let c = |q: u32| C.get(q as usize).copied().unwrap_or([0; 3]).map(f64::from); let w = |q: u32| W.get(q as usize).copied().unwrap_or(0.0); for (l, link) in links.iter().enumerate() { let (kind, from, group, q2, du) = match link.rule { Rule::Pull { from } => (KIND_PULL, from, 0, 0, 0.0), Rule::Bounce { group, du } => (KIND_BOUNCE, 0, group, 0, du), Rule::Reflect { from, q, group } => (KIND_REFLECT, from, group, u32::from(q), 0.0), Rule::Open => (KIND_OPEN, 0, 0, 0, 0.0), }; // What the weights add to the link's exchange (see `link_forces`). let weight = match link.rule { Rule::Bounce { .. } => c(u32::from(link.q)).map(|x| -2.0 * w(u32::from(link.q)) * x), Rule::Reflect { .. } => std::array::from_fn(|a| (c(q2)[a] - c(u32::from(link.q))[a]) * w(q2)), _ => [0.0; 3], }; if matches!(link.rule, Rule::Bounce { .. } | Rule::Reflect { .. }) { by_group.push((group, l as u32)); if let Some(g) = weight_forces.get_mut(group as usize) { for (s, x) in g.iter_mut().zip(weight) { *s += x; } } } link_u.extend_from_slice(&[u32::from(link.q) | (kind << 8) | (q2 << 16), from, group, link.cell]); link_f.push(du as f32); } by_group.sort(); let mut group_start = filled(group_count + 1, 0u32, "force groups")?; for (group, _) in &by_group { if let Some(next) = group_start.get_mut(*group as usize + 1) { *next += 1; } } for g in 0..group_count { let prev = group_start.get(g).copied().unwrap_or(0); if let Some(next) = group_start.get_mut(g + 1) { *next += prev; } } let by_group: Vec = by_group.iter().map(|(_, l)| *l).collect(); // Each group's thermal condition (held temperatures shifted by T₀). let mut group_t = reserved(4 * group_count, "thermal conditions")?; for g in 0..group_count { let (kind, value) = match thermal.and_then(|t| t.groups().get(g)).copied().unwrap_or_default() { ThermalCondition::Adiabatic {} => (0.0, 0.0), ThermalCondition::Temperature { value } => (1.0, (value - t0) as f32), ThermalCondition::Flux { value } => (2.0, value as f32), }; group_t.extend_from_slice(&[kind, value, 0.0, 0.0]); } // The open cells. let open = flow.open_cells(); let mut open_u = reserved(4 * open.len(), "open cells")?; let mut open_f = reserved(4 * open.len(), "open cells")?; for oc in open { let (kind, target) = match oc.target { Target::Velocity(u) => (0u32, [u[0] as f32, u[1] as f32, u[2] as f32]), Target::Density(rho) => (1u32, [rho as f32, 0.0, 0.0]), }; open_u.extend_from_slice(&[oc.cell, oc.axis as u32, kind, 0]); open_f.extend_from_slice(&[target[0], target[1], target[2], oc.inward as f32]); } let mut open_t = reserved(4 * open.len(), "open cell temperatures")?; for k in 0..open.len() { match thermal.and_then(|t| t.open().get(k)).copied().flatten() { Some(value) => open_t.extend_from_slice(&[1.0, (value - t0) as f32, 0.0, 0.0]), None => open_t.extend_from_slice(&[0.0; 4]), } } let (omega_plus, omega_minus) = flow.omegas(); let acc = flow.acceleration(); let forced = acc != [0.0; 3]; let cells = groups(n)?; let links_groups = groups(links.len())?; let open_groups = groups(open.len())?; let force_groups = groups((group_count * WG as usize).max(1))?; let partial_count = (cells.0 as usize) * (cells.1 as usize); let mut params = Vec::with_capacity(64); let u = |v: usize| u32::try_from(v).map_err(|_| Error::Limit("too many cells for the GPU solver".into())); for v in [ u(n)?, u(flow.grid().cells[0])?, u(flow.grid().cells[1])?, u(flow.grid().cells[2])?, cells.0, u(links.len())?, u(open.len())?, u(group_count)?, ] { params.extend_from_slice(&v.to_le_bytes()); } for v in [ omega_plus as f32, omega_minus as f32, (1.0 - 0.5 * omega_plus) as f32, (1.0 - 0.5 * omega_minus) as f32, acc[0] as f32, acc[1] as f32, acc[2] as f32, if forced { 1.0 } else { 0.0 }, ] { params.extend_from_slice(&v.to_le_bytes()); } let buoyancy = thermal.map_or([0.0; 3], |t| t.buoyancy()); for v in [ buoyancy[0] as f32, buoyancy[1] as f32, buoyancy[2] as f32, if buoyancy != [0.0; 3] { 1.0 } else { 0.0 }, thermal.map_or(0.0, |t| t.omega()) as f32, t0 as f32, if thermal.is_some() { 1.0 } else { 0.0 }, 0.0, ] { params.extend_from_slice(&v.to_le_bytes()); } // Buffers, with the device's out-of-memory errors caught. let scopes = [wgpu::ErrorFilter::OutOfMemory, wgpu::ErrorFilter::Validation].map(|f| device.push_error_scope(f)); let storage = |label: &str, bytes: &[u8]| { device.create_buffer_init(&wgpu::util::BufferInitDescriptor { label: Some(label), contents: if bytes.is_empty() { &[0u8; 16][..] } else { bytes }, usage: wgpu::BufferUsages::STORAGE | wgpu::BufferUsages::COPY_SRC | wgpu::BufferUsages::COPY_DST, }) }; let empty = |label: &str, size: u64| { device.create_buffer(&wgpu::BufferDescriptor { label: Some(label), size: size.max(16), usage: wgpu::BufferUsages::STORAGE | wgpu::BufferUsages::COPY_SRC | wgpu::BufferUsages::COPY_DST, mapped_at_creation: false, }) }; let params = [0u32, 1].map(|phase| { let mut bytes = params.clone(); bytes.extend_from_slice(&bytes_u32(&[phase, 0, 0, 0])); device.create_buffer_init(&wgpu::util::BufferInitDescriptor { label: Some("lbm params"), contents: &bytes, usage: wgpu::BufferUsages::UNIFORM, }) }); let fluid_buf = storage("lbm fluid", &bytes_u32(&fluid_u32)); // AA reuses binding g for boundary scratch only, never another full population field. let pops = [storage("lbm f", &bytes_f32(&f)), empty("lbm g / AA boundary scratch", if aa { 4 * links.len() as u64 } else { pop_bytes })]; let link_u = storage("lbm links", &bytes_u32(&link_u)); let link_f = storage("lbm link du", &bytes_f32(&link_f)); let open_u = storage("lbm open cells", &bytes_u32(&open_u)); let open_f = storage("lbm open targets", &bytes_f32(&open_f)); let moments = empty("lbm moments", 16 * n as u64); let stats = storage("lbm stats", &bytes_u32(&[0, 0, 0, 0])); let contrib = empty("lbm link forces", 16 * links.len() as u64); let by_group = storage("lbm links by group", &bytes_u32(&by_group)); let group_start = storage("lbm group start", &bytes_u32(&group_start)); let forces = empty("lbm forces", 16 * group_count as u64); let partials = empty("lbm partials", 4 * partial_count as u64); let therm_bytes = elem * (QT * n) as u64; let therm = if thermal.is_some() { [storage("lbm t", &bytes_f32(&tpop)), empty("lbm t2", therm_bytes)] } else { [empty("lbm t", 16), empty("lbm t2", 16)] }; let group_t = storage("lbm thermal conditions", &bytes_f32(&group_t)); let open_t = storage("lbm open temperatures", &bytes_f32(&open_t)); let module = device .create_shader_module(wgpu::ShaderModuleDescriptor { label: Some("fieldcraft lbm"), source: wgpu::ShaderSource::Wgsl(SHADER.into()) }); let entry = |binding: u32, ty: wgpu::BufferBindingType| wgpu::BindGroupLayoutEntry { binding, visibility: wgpu::ShaderStages::COMPUTE, ty: wgpu::BindingType::Buffer { ty, has_dynamic_offset: false, min_binding_size: None }, count: None, }; let read = wgpu::BufferBindingType::Storage { read_only: true }; let write = wgpu::BufferBindingType::Storage { read_only: false }; let layout = device.create_bind_group_layout(&wgpu::BindGroupLayoutDescriptor { label: Some("fieldcraft lbm"), entries: &[ entry(0, wgpu::BufferBindingType::Uniform), entry(1, read), entry(2, write), entry(3, write), entry(5, read), entry(6, read), entry(7, read), entry(8, read), entry(9, write), entry(10, write), entry(11, write), entry(12, read), entry(13, read), entry(14, write), entry(15, write), entry(16, write), entry(17, write), entry(18, read), entry(19, read), ], }); let bind = [0usize, 1].map(|k| { device.create_bind_group(&wgpu::BindGroupDescriptor { label: Some("fieldcraft lbm"), layout: &layout, entries: &[ wgpu::BindGroupEntry { binding: 0, resource: params[k].as_entire_binding() }, wgpu::BindGroupEntry { binding: 1, resource: fluid_buf.as_entire_binding() }, wgpu::BindGroupEntry { binding: 2, resource: pops[if aa { 0 } else { k }].as_entire_binding() }, wgpu::BindGroupEntry { binding: 3, resource: pops[if aa { 1 } else { 1 - k }].as_entire_binding() }, wgpu::BindGroupEntry { binding: 5, resource: link_u.as_entire_binding() }, wgpu::BindGroupEntry { binding: 6, resource: link_f.as_entire_binding() }, wgpu::BindGroupEntry { binding: 7, resource: open_u.as_entire_binding() }, wgpu::BindGroupEntry { binding: 8, resource: open_f.as_entire_binding() }, wgpu::BindGroupEntry { binding: 9, resource: moments.as_entire_binding() }, wgpu::BindGroupEntry { binding: 10, resource: stats.as_entire_binding() }, wgpu::BindGroupEntry { binding: 11, resource: contrib.as_entire_binding() }, wgpu::BindGroupEntry { binding: 12, resource: by_group.as_entire_binding() }, wgpu::BindGroupEntry { binding: 13, resource: group_start.as_entire_binding() }, wgpu::BindGroupEntry { binding: 14, resource: forces.as_entire_binding() }, wgpu::BindGroupEntry { binding: 15, resource: partials.as_entire_binding() }, wgpu::BindGroupEntry { binding: 16, resource: therm[k].as_entire_binding() }, wgpu::BindGroupEntry { binding: 17, resource: therm[1 - k].as_entire_binding() }, wgpu::BindGroupEntry { binding: 18, resource: group_t.as_entire_binding() }, wgpu::BindGroupEntry { binding: 19, resource: open_t.as_entire_binding() }, ], }) }); let pl = device.create_pipeline_layout(&wgpu::PipelineLayoutDescriptor { label: Some("fieldcraft lbm"), bind_group_layouts: &[Some(&layout)], immediate_size: 0, }); let mut pipelines = HashMap::new(); let pipeline_start = std::time::Instant::now(); for name in [ "collide", "stream", "stream_links", "stream_links_exchange", "scatter_links", "open_boundaries", "link_forces", "group_forces", "cell_moments", "mass", "collide_thermal", "stream_thermal", "stream_links_thermal", "open_thermal", "thermal_check", ] { let p = device.create_compute_pipeline(&wgpu::ComputePipelineDescriptor { label: Some(name), layout: Some(&pl), module: &module, entry_point: Some(name), compilation_options: wgpu::PipelineCompilationOptions { constants: &[("AA_MODE", if aa { 1.0 } else { 0.0 })], ..Default::default() }, cache: None, }); pipelines.insert(name, p); } for scope in scopes.into_iter().rev() { if let Some(e) = pollster::block_on(scope.pop()) { return Err(gpu_error(e)); } } let pipeline_seconds = pipeline_start.elapsed().as_secs_f64(); Ok(GpuFlow { device, queue, kernel_times: profile.then(Default::default), pipeline_seconds, aa, adapter: adapter_name, pipelines, bind, current: 0, n, cells, link_count: links.len(), links_groups, open_count: open.len(), open_groups, group_count, force_groups, stats, moments, forces, partials, partial_count, thermal: thermal.map(|_| therm), initial_temperature: t0, fluid_cells: flow.fluid_cells(), fluid, groups: flow.groups().to_vec(), empty_obstacles: flow.empty_obstacles().to_vec(), steps_done: 0, exchange_cache: Some(vec![[0.0; 4]; group_count]), weight_forces, }) } /// The adapter's name. pub fn adapter_name(&self) -> &str { &self.adapter } /// Whether the populations stream in place (see [`Streaming`]). pub fn streams_in_place(&self) -> bool { self.aa } /// How long creating the compute pipelines took, in seconds. pub fn pipeline_seconds(&self) -> f64 { self.pipeline_seconds } /// With [`Options::profile`], the GPU time of each kernel so far, by kernel name. pub fn kernel_times(&self) -> Option> { self.kernel_times.as_ref().map(|t| t.borrow().clone()) } fn pipeline(&self, name: &str) -> Result<&wgpu::ComputePipeline> { self.pipelines.get(name).ok_or_else(|| Error::Invalid(format!("GPU: no kernel {name}"))) } fn dispatch(&self, pass: &mut wgpu::ComputePass<'_>, bind: usize, name: &str, (gx, gy): (u32, u32)) -> Result<()> { pass.set_bind_group(0, self.bind.get(bind).ok_or_else(|| Error::Invalid("GPU: no bind group".into()))?, &[]); pass.set_pipeline(self.pipeline(name)?); pass.dispatch_workgroups(gx, gy, 1); Ok(()) } /// Timestamped passes are diagnostic: their extra pass boundaries and readback are not /// used for throughput measurements. Normally every chunk remains one compute pass. fn submit(&self, label: &str, commands: &[Dispatch]) -> Result<()> { let mut enc = self.device.create_command_encoder(&wgpu::CommandEncoderDescriptor { label: Some(label) }); let Some(times) = &self.kernel_times else { { let mut pass = enc.begin_compute_pass(&wgpu::ComputePassDescriptor { label: Some(label), timestamp_writes: None }); for &(bind, name, groups) in commands { self.dispatch(&mut pass, bind, name, groups)?; } } self.queue.submit([enc.finish()]); return Ok(()); }; let count = u32::try_from(commands.len() * 2).map_err(|_| gpu_error("too many timestamp queries"))?; if count == 0 || count > wgpu::QUERY_SET_MAX_QUERIES { return Err(gpu_error("timestamp query count is outside the device limit")); } let query = self.device.create_query_set(&wgpu::QuerySetDescriptor { label: Some(label), ty: wgpu::QueryType::Timestamp, count }); let result = self.device.create_buffer(&wgpu::BufferDescriptor { label: Some("lbm timestamps"), size: u64::from(count) * 8, usage: wgpu::BufferUsages::QUERY_RESOLVE | wgpu::BufferUsages::COPY_SRC, mapped_at_creation: false, }); for (i, &(bind, name, groups)) in commands.iter().enumerate() { let mut pass = enc.begin_compute_pass(&wgpu::ComputePassDescriptor { label: Some(name), timestamp_writes: Some(wgpu::ComputePassTimestampWrites { query_set: &query, beginning_of_pass_write_index: Some(2 * i as u32), end_of_pass_write_index: Some(2 * i as u32 + 1), }), }); self.dispatch(&mut pass, bind, name, groups)?; } enc.resolve_query_set(&query, 0..count, &result, 0); self.queue.submit([enc.finish()]); let bytes = self.read(&result, u64::from(count) * 8)?; let period = f64::from(self.queue.get_timestamp_period()); let mut totals = times.borrow_mut(); for ((_, name, _), pair) in commands.iter().zip(bytes.as_chunks::<16>().0) { let [begin, end] = pair.as_chunks::<8>().0 else { return Err(gpu_error("invalid timestamp pair")) }; let ticks = u64::from_le_bytes(*end).checked_sub(u64::from_le_bytes(*begin)).ok_or_else(|| gpu_error("invalid GPU timestamps"))?; let total = totals.entry(name).or_default(); total.dispatches += 1; total.seconds += ticks as f64 * period * 1e-9; } Ok(()) } /// Copy `size` bytes of `src` back to the host. fn read(&self, src: &wgpu::Buffer, size: u64) -> Result> { let size = size.max(16); let staging = self.device.create_buffer(&wgpu::BufferDescriptor { label: Some("lbm readback"), size, usage: wgpu::BufferUsages::MAP_READ | wgpu::BufferUsages::COPY_DST, mapped_at_creation: false, }); let mut enc = self.device.create_command_encoder(&wgpu::CommandEncoderDescriptor { label: Some("lbm readback") }); enc.copy_buffer_to_buffer(src, 0, &staging, 0, size); self.queue.submit([enc.finish()]); let slice = staging.slice(..); let (tx, rx) = std::sync::mpsc::channel(); slice.map_async(wgpu::MapMode::Read, move |r| { let _ = tx.send(r); }); // A bounded wait: a hung or lost device is an error, never a hang. match self.device.poll(wgpu::PollType::Wait { submission_index: None, timeout: Some(DEVICE_TIMEOUT) }) { Ok(_) => {} Err(e) => return Err(gpu_error(format!("the device did not answer within {} s: {e}", DEVICE_TIMEOUT.as_secs()))), } rx.try_recv().map_err(|_| gpu_error("the readback did not complete"))?.map_err(gpu_error)?; let bytes = slice.get_mapped_range().map_err(gpu_error)?.to_vec(); staging.unmap(); Ok(bytes) } fn read_f32(&self, src: &wgpu::Buffer, count: usize) -> Result> { let bytes = self.read(src, 4 * count as u64)?; Ok(bytes.as_chunks::<4>().0.iter().take(count).map(|b| f32::from_le_bytes(*b)).collect()) } /// The momentum (xyz) and heat (w) each group received in the last step. fn exchange(&mut self) -> Result<&[[f64; 4]]> { if self.exchange_cache.is_none() { let v = self.read_f32(&self.forces, 4 * self.group_count)?; self.exchange_cache = Some( v.as_chunks::<4>() .0 .iter() .zip(&self.weight_forces) .map(|(g, w)| [f64::from(g[0]) + w[0], f64::from(g[1]) + w[1], f64::from(g[2]) + w[2], f64::from(g[3])]) .collect(), ); } Ok(self.exchange_cache.as_deref().unwrap_or_default()) } /// The moments of the current state into the moments buffer. fn compute_moments(&self) -> Result<()> { self.submit("lbm moments", &[(self.current, "cell_moments", self.cells)]) } } impl GpuFlow { /// One submission of `steps` (≤ MAX_CHUNK) time steps, then the stability check. fn advance_chunk(&mut self, steps: u64) -> Result<()> { self.queue.write_buffer(&self.stats, 0, &bytes_u32(&[0, 0, 0, 0])); let mut commands = Vec::with_capacity(8 * MAX_CHUNK as usize + 3); for step in 0..steps { commands.push((self.current, "collide", self.cells)); if !self.aa { commands.push((self.current, "stream", self.cells)); } if self.link_count > 0 { // The last step's momentum exchange is the one read back. let exchange = self.aa && step + 1 == steps; commands.push((self.current, if exchange { "stream_links_exchange" } else { "stream_links" }, self.links_groups)); if self.aa { commands.push((self.current, "scatter_links", self.links_groups)); } } if self.open_count > 0 { commands.push((self.current, "open_boundaries", self.open_groups)); } if self.thermal.is_some() { commands.push((self.current, "collide_thermal", self.cells)); commands.push((self.current, "stream_thermal", self.cells)); if self.link_count > 0 { commands.push((self.current, "stream_links_thermal", self.links_groups)); } if self.open_count > 0 { commands.push((self.current, "open_thermal", self.open_groups)); } } self.current ^= 1; } // The momentum (and heat) exchange of the last step, from its post-collision // populations. if self.link_count > 0 && self.group_count > 0 { commands.push((self.current ^ 1, "link_forces", self.links_groups)); commands.push((self.current ^ 1, "group_forces", self.force_groups)); } if self.thermal.is_some() { commands.push((self.current, "thermal_check", self.cells)); } self.submit("lbm steps", &commands)?; self.steps_done += steps; self.exchange_cache = None; let stats = self.read(&self.stats, 16)?; let word = |k: usize| stats.get(4 * k..4 * k + 4).and_then(|b| b.try_into().ok()).map(u32::from_le_bytes).unwrap_or(0); let max_u2 = f32::from_bits(word(0)); if word(1) != 0 || !(f64::from(max_u2) <= MAX_LATTICE_SPEED * MAX_LATTICE_SPEED) { return Err(Error::Unstable(format!( "the flow went unstable by time step {}: the lattice speed reached {:.3} (limit {MAX_LATTICE_SPEED}) or a density stopped being a positive number; lower the time step, refine the spacing or check the boundary conditions", self.steps_done, max_u2.sqrt() ))); } if word(2) != 0 { return Err(Error::Unstable(format!( "the heat went unstable by time step {}: a temperature stopped being a number; lower the time step, refine the spacing or check the heat fluxes", self.steps_done ))); } Ok(()) } } impl Backend for GpuFlow { fn name(&self) -> String { format!("gpu: {}, {} streaming", self.adapter, if self.aa { "in-place" } else { "two-buffer" }) } fn advance(&mut self, steps: u64) -> Result<()> { // Submissions of at most MAX_CHUNK steps: the stability check runs after each, so a // run that blows up is caught within that many steps whatever the caller asks for. let mut left = steps; while left > 0 { let chunk = left.min(MAX_CHUNK); self.advance_chunk(chunk)?; left -= chunk; } Ok(()) } fn steps_done(&self) -> u64 { self.steps_done } fn fields(&mut self) -> Result { self.compute_moments()?; let m = self.read_f32(&self.moments, 4 * self.n)?; let density = m.as_chunks::<4>().0.iter().map(|v| f64::from(v[3])).collect(); let velocity = m.as_chunks::<4>().0.iter().map(|v| [f64::from(v[0]), f64::from(v[1]), f64::from(v[2])]).collect(); // The temperature from the scalar populations, summed on the host: T₀ + Σ t (T₀ in // solid cells, whose populations stay zero). let temperature = match &self.thermal { None => Vec::new(), Some(therm) => { let t = self.read_f32(therm.get(self.current).ok_or_else(|| Error::Invalid("GPU: no thermal buffer".into()))?, QT * self.n)?; (0..self.n) .map(|cell| self.initial_temperature + (0..QT).map(|q| t.get(q * self.n + cell).copied().map_or(0.0, f64::from)).sum::()) .collect() } }; Ok(Fields { density, velocity, temperature }) } fn heat(&mut self) -> Result> { if self.thermal.is_none() { return Ok(Vec::new()); } Ok(self.exchange()?.iter().map(|g| g[3]).collect()) } fn mass(&mut self) -> Result { self.compute_moments()?; self.submit("lbm mass", &[(self.current, "mass", self.cells)])?; let partials = self.read_f32(&self.partials, self.partial_count)?; // Summed in a fixed order on the host, so the result is deterministic on this device. Ok(partials.iter().map(|v| f64::from(*v)).sum()) } fn forces(&mut self) -> Result> { Ok(self.exchange()?.iter().map(|g| [g[0], g[1], g[2]]).collect()) } fn groups(&self) -> &[Group] { &self.groups } fn fluid_cells(&self) -> usize { self.fluid_cells } fn is_fluid(&self, cell: usize) -> bool { self.fluid.get(cell).copied().unwrap_or(false) } fn empty_obstacles(&self) -> &[String] { &self.empty_obstacles } }