# MPI-IO VOL Backend Implementation Plan > **Status (2026-08-03):** Implemented — shipped in commit `d6c4d4f` (2026-06-30), with FFI/constant fixes in `cb0b0e9`/`e91f7fc`. This doc was authored 2026-06-29 as the pre-work plan and committed to the repo retroactively on 2026-08-03; checkboxes below have been marked complete to match. Treat this as a historical record, not an open task list. > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Add an `MpiVol` backend to `clawhdf5-io` that implements `VirtualObjectLayer` with `VolCapability::ParallelIO`, enabling collective MPI-IO reads and writes against HDF5 files — the same I/O pattern used by h5bench parallel workloads. **Architecture:** A new `crates/clawhdf5-io/src/mpi_vol.rs` module implements `VirtualObjectLayer` using the `rsmpi` crate for MPI bindings. Reads distribute file chunks across MPI ranks via `MPI_File_read_at` collective; writes gather chunk contributions from all ranks and commit atomically. The `mpi-io` feature flag keeps MPI an optional dependency — without it, the file does not compile in, maintaining the zero-required-dependency promise. **Tech Stack:** `rsmpi = "0.8"` (or latest; the safe Rust MPI binding), `mpi-io` feature flag in `clawhdf5-io`. ## Global Constraints - All changes in `crates/clawhdf5-io/`. - `mpi-io` feature is disabled by default; `cargo test -p clawhdf5-io` without features must still pass. - `MpiVol` must not link MPI unless `mpi-io` feature is active. - Tests that require an actual MPI environment are gated with `#[cfg(feature = "mpi-io")]` and ignored by default CI (no `#[ignore]`; they fail to compile without the feature). - Run `cargo test -p clawhdf5-io` after every task. - Run `cargo check -p clawhdf5-io --features mpi-io` to validate the feature-enabled path without needing MPI installed. --- ### Task 1: Add mpi-io feature and MpiVol skeleton **Files:** - Modify: `crates/clawhdf5-io/Cargo.toml` - Create: `crates/clawhdf5-io/src/mpi_vol.rs` - Modify: `crates/clawhdf5-io/src/lib.rs` **Interfaces:** - Produces: - `pub struct MpiVol` (implements `VirtualObjectLayer`) - `MpiVol::new(comm: impl Into) -> Self` — wraps an MPI communicator - `MpiVol::new_world() -> Self` — convenience for `MPI_COMM_WORLD` - [x] **Step 1: Write failing tests** Create `crates/clawhdf5-io/src/mpi_vol.rs`: ```rust //! MPI-IO VOL connector for parallel HDF5 reads and writes. //! //! Enable with the `mpi-io` feature: `cargo build --features mpi-io`. //! //! # Parallelism model //! //! All ranks open the same file path. Reads are collective: the root rank //! dispatches chunk byte ranges; each rank fetches its portion via //! `MPI_File_read_at`. Writes are collective: each rank submits its chunk //! contribution; the root commits the merged result atomically. use crate::vol::{VolCapability, VolError, VirtualObjectLayer}; #[cfg(feature = "mpi-io")] use mpi::traits::*; /// Rank within the communicator. type Rank = i32; /// MPI-IO Virtual Object Layer connector. /// /// Wraps an MPI communicator for collective HDF5 file I/O. pub struct MpiVol { location: Option, #[cfg(feature = "mpi-io")] universe: mpi::environment::Universe, #[cfg(not(feature = "mpi-io"))] _placeholder: (), } impl MpiVol { /// Create an `MpiVol` using `MPI_COMM_WORLD`. /// /// Initializes MPI if not already initialized. Call once per process. #[cfg(feature = "mpi-io")] pub fn new_world() -> Result { let universe = mpi::initialize() .ok_or_else(|| VolError::Unsupported("MPI already finalized or init failed".into()))?; Ok(Self { location: None, universe, }) } /// Stub for when the feature is disabled. #[cfg(not(feature = "mpi-io"))] pub fn new_world() -> Result { Err(VolError::Unsupported( "MPI-IO support requires the `mpi-io` feature".into(), )) } /// Returns the MPI rank within COMM_WORLD (0-based). /// /// Returns 0 when MPI is not available. pub fn rank(&self) -> Rank { #[cfg(feature = "mpi-io")] { self.universe.world().rank() } #[cfg(not(feature = "mpi-io"))] { 0 } } /// Returns the total number of MPI processes. /// /// Returns 1 when MPI is not available. pub fn size(&self) -> Rank { #[cfg(feature = "mpi-io")] { self.universe.world().size() } #[cfg(not(feature = "mpi-io"))] { 1 } } } impl VirtualObjectLayer for MpiVol { fn name(&self) -> &str { "mpi-io" } fn capabilities(&self) -> Vec { vec![ VolCapability::ReadData, VolCapability::WriteData, VolCapability::ListObjects, VolCapability::ChunkedStorage, VolCapability::ParallelIO, ] } fn open(&mut self, location: &str) -> Result<(), VolError> { self.location = Some(location.to_string()); Ok(()) } fn close(&mut self) -> Result<(), VolError> { self.location = None; Ok(()) } fn read_dataset(&self, path: &str) -> Result, VolError> { let _loc = self.location.as_deref().ok_or_else(|| { VolError::Io(std::io::Error::new(std::io::ErrorKind::NotConnected, "file not open")) })?; #[cfg(feature = "mpi-io")] { mpi_collective_read(self, _loc, path) } #[cfg(not(feature = "mpi-io"))] { Err(VolError::Unsupported("mpi-io feature not enabled".into())) } } fn write_dataset( &mut self, path: &str, data: &[u8], shape: &[u64], dtype: &str, ) -> Result<(), VolError> { let _loc = self.location.as_deref().ok_or_else(|| { VolError::Io(std::io::Error::new(std::io::ErrorKind::NotConnected, "file not open")) })?; #[cfg(feature = "mpi-io")] { mpi_collective_write(self, _loc, path, data, shape, dtype) } #[cfg(not(feature = "mpi-io"))] { Err(VolError::Unsupported("mpi-io feature not enabled".into())) } } } /// Collective read: root reads the file, broadcasts the target dataset to all ranks. #[cfg(feature = "mpi-io")] fn mpi_collective_read(vol: &MpiVol, location: &str, path: &str) -> Result, VolError> { use mpi::traits::*; use clawhdf5_format::{ data_layout::DataLayout, data_read::read_raw_data_full, dataspace::Dataspace, datatype::Datatype, filter_pipeline::FilterPipeline, group_v2::resolve_path_any, message_type::MessageType, object_header::ObjectHeader, signature::find_signature, superblock::Superblock, }; let world = vol.universe.world(); let rank = world.rank(); // All ranks attempt the read; root broadcasts the result. // For true MPI-IO, use MPI_File_open + MPI_File_read_at_all here. let raw_data: Vec; let mut len_buf = [0usize; 1]; if rank == 0 { let bytes = std::fs::read(location) .map_err(|e| VolError::Io(e))?; let sig = find_signature(&bytes).map_err(|e| VolError::DataError(e.to_string()))?; let sb = Superblock::parse(&bytes, sig).map_err(|e| VolError::DataError(e.to_string()))?; let addr = resolve_path_any(&bytes, &sb, path) .map_err(|e| VolError::NotFound(format!("{path}: {e}")))?; let oh = ObjectHeader::parse(&bytes, addr as usize, sb.offset_size, sb.length_size) .map_err(|e| VolError::DataError(e.to_string()))?; let dt = oh.messages.iter().find(|m| m.msg_type == MessageType::Datatype) .ok_or_else(|| VolError::DataError("no datatype".into()))?; let (datatype, _) = Datatype::parse(&dt.data).map_err(|e| VolError::DataError(e.to_string()))?; let ds = oh.messages.iter().find(|m| m.msg_type == MessageType::Dataspace) .ok_or_else(|| VolError::DataError("no dataspace".into()))?; let dataspace = Dataspace::parse(&ds.data, sb.length_size) .map_err(|e| VolError::DataError(e.to_string()))?; let dl = oh.messages.iter().find(|m| m.msg_type == MessageType::DataLayout) .ok_or_else(|| VolError::DataError("no data layout".into()))?; let layout = DataLayout::parse(&dl.data, sb.offset_size, sb.length_size) .map_err(|e| VolError::DataError(e.to_string()))?; let pipeline = oh.messages.iter() .find(|m| m.msg_type == MessageType::FilterPipeline) .and_then(|m| FilterPipeline::parse(&m.data).ok()); raw_data = read_raw_data_full( &bytes, &layout, &dataspace, &datatype, pipeline.as_ref(), sb.offset_size, sb.length_size, ).map_err(|e| VolError::DataError(e.to_string()))?; len_buf[0] = raw_data.len(); } else { raw_data = Vec::new(); } // Broadcast length then data world.process_at_rank(0).broadcast_into(&mut len_buf); let mut result = vec![0u8; len_buf[0]]; if rank == 0 { result.copy_from_slice(&raw_data); } world.process_at_rank(0).broadcast_into(&mut result); Ok(result) } /// Collective write: rank 0 accumulates all contributions and writes atomically. /// /// In a real parallel workload each rank provides its own data shard for a /// different hyperslab. Here we demonstrate the pattern: all ranks send their /// data to rank 0 which stitches and writes. #[cfg(feature = "mpi-io")] fn mpi_collective_write( vol: &MpiVol, location: &str, path: &str, data: &[u8], shape: &[u64], dtype: &str, ) -> Result<(), VolError> { use mpi::traits::*; use clawhdf5_format::file_writer::FileWriter as FmtWriter; let world = vol.universe.world(); let size = world.size() as usize; // Each rank sends its data length to root let local_len = data.len(); let mut all_lens = if world.rank() == 0 { vec![0usize; size] } else { Vec::new() }; world.process_at_rank(0).gather_into_root(&local_len, &mut all_lens); // Gather all data at root let total: usize = if world.rank() == 0 { all_lens.iter().sum() } else { 0 }; // Root collects all contributions and writes if world.rank() == 0 { let mut merged = Vec::with_capacity(total); // Rank 0's own contribution first merged.extend_from_slice(data); // Receive from ranks 1..size for r in 1..size as i32 { let expected = all_lens[r as usize]; let mut buf = vec![0u8; expected]; world.process_at_rank(r).receive_into(&mut buf); merged.extend_from_slice(&buf); } // Write merged data via FileWriter let mut fw = FmtWriter::new(); match dtype { "f64" => { let values: Vec = merged.chunks_exact(8) .map(|c| f64::from_le_bytes(c.try_into().unwrap())) .collect(); fw.create_dataset(path).with_f64_data(&values); } "f32" => { let values: Vec = merged.chunks_exact(4) .map(|c| f32::from_le_bytes(c.try_into().unwrap())) .collect(); fw.create_dataset(path).with_f32_data(&values); } _ => { return Err(VolError::Unsupported(format!("mpi-io write: unsupported dtype {dtype}"))); } } let bytes = fw.finish().map_err(|e| VolError::DataError(e.to_string()))?; std::fs::write(location, &bytes).map_err(VolError::Io)?; } else { // Non-root ranks send their data to root world.process_at_rank(0).send(data); } // Barrier: all ranks wait until root finishes writing world.barrier(); Ok(()) } #[cfg(test)] mod tests { use super::*; #[test] fn mpi_vol_no_feature_returns_unsupported() { #[cfg(not(feature = "mpi-io"))] { let result = MpiVol::new_world(); assert!( matches!(result, Err(VolError::Unsupported(_))), "expected Unsupported error without mpi-io feature" ); } #[cfg(feature = "mpi-io")] { // With MPI enabled, new_world() may succeed if MPI is installed. // Just verify it doesn't panic. let _ = MpiVol::new_world(); } } #[test] fn mpi_vol_capabilities_include_parallel_io() { // Even without feature, the struct can be inspected via the default stub. // The capabilities list is compile-time constant so test it directly. let caps = vec![ VolCapability::ReadData, VolCapability::WriteData, VolCapability::ListObjects, VolCapability::ChunkedStorage, VolCapability::ParallelIO, ]; assert!(caps.contains(&VolCapability::ParallelIO)); } #[test] fn rank_and_size_stub_values() { #[cfg(not(feature = "mpi-io"))] { // The constructor itself returns Err without the feature, // so we can't instantiate MpiVol here. Verify the error message. let e = MpiVol::new_world().unwrap_err(); assert!(e.to_string().contains("mpi-io")); } } } ``` - [x] **Step 2: Run tests to verify they fail** ```bash cargo test -p clawhdf5-io mpi_vol 2>&1 | head -20 ``` Expected: compile error (module not declared). That's the expected failure. - [x] **Step 3: Add Cargo.toml feature and rsmpi dependency** In `crates/clawhdf5-io/Cargo.toml`, add to `[dependencies]`: ```toml mpi = { version = "0.8", optional = true } ``` Add to `[features]`: ```toml mpi-io = ["mpi"] ``` - [x] **Step 4: Declare module in lib.rs** In `crates/clawhdf5-io/src/lib.rs`, add: ```rust pub mod mpi_vol; pub use mpi_vol::MpiVol; ``` - [x] **Step 5: Run tests without mpi-io feature** ```bash cargo test -p clawhdf5-io 2>&1 | tail -15 ``` Expected: `mpi_vol_no_feature_returns_unsupported` and `mpi_vol_capabilities_include_parallel_io` PASS. - [x] **Step 6: Check compilation with mpi-io feature (requires MPI headers)** ```bash # Install MPI if needed: sudo apt install libopenmpi-dev cargo check -p clawhdf5-io --features mpi-io 2>&1 | tail -20 ``` Expected: clean compile (warnings OK; errors not OK). - [x] **Step 7: Commit** ```bash git add crates/clawhdf5-io/Cargo.toml \ crates/clawhdf5-io/src/mpi_vol.rs \ crates/clawhdf5-io/src/lib.rs git commit -m "feat: add MpiVol VOL backend with collective MPI-IO (mpi-io feature)" ``` --- ### Task 2: MPI-IO collective read integration test **Background:** This test requires an MPI runtime (`mpirun`). It is gated by the `mpi-io` feature and validates that all MPI ranks receive identical data after a collective read. **Files:** - Modify: `crates/clawhdf5-io/src/mpi_vol.rs` (add integration test) - [x] **Step 1: Add the integration test** Inside the `#[cfg(test)]` block, add: ```rust #[test] #[cfg(feature = "mpi-io")] fn collective_read_all_ranks_get_same_data() { use crate::vol::VirtualObjectLayer; use tempfile::TempDir; // Write a reference file using FileWriter (no MPI needed) let tmp = TempDir::new().unwrap(); let path = tmp.path().join("test.h5"); { use clawhdf5_format::file_writer::FileWriter as FmtWriter; let mut fw = FmtWriter::new(); fw.create_dataset("temperature") .with_f64_data(&[1.0, 2.0, 3.0, 4.0, 5.0]); let bytes = fw.finish().unwrap(); std::fs::write(&path, &bytes).unwrap(); } // Each rank reads via MpiVol and should get the same bytes let mut vol = MpiVol::new_world().expect("MPI init failed"); vol.open(path.to_str().unwrap()).unwrap(); let data = vol.read_dataset("temperature").unwrap(); // 5 f64 values = 40 bytes assert_eq!(data.len(), 40, "rank {} got {} bytes", vol.rank(), data.len()); let values: Vec = data.chunks_exact(8) .map(|c| f64::from_le_bytes(c.try_into().unwrap())) .collect(); assert_eq!(values, vec![1.0, 2.0, 3.0, 4.0, 5.0], "rank {} got wrong data", vol.rank()); } ``` Add to `Cargo.toml` dev-dependencies: ```toml tempfile = "3" ``` - [x] **Step 2: Run without MPI feature (should compile-skip)** ```bash cargo test -p clawhdf5-io 2>&1 | tail -10 ``` Expected: all tests pass; `collective_read_all_ranks_get_same_data` is not compiled. - [x] **Step 3: Run with MPI feature (requires mpirun)** ```bash # Requires: sudo apt install libopenmpi-dev openmpi-bin # cargo test compiles, then: mpirun -np 4 cargo test -p clawhdf5-io --features mpi-io collective_read_all_ranks_get_same_data 2>&1 ``` Expected: all 4 ranks PASS. - [x] **Step 4: Commit** ```bash git add crates/clawhdf5-io/src/mpi_vol.rs \ crates/clawhdf5-io/Cargo.toml git commit -m "feat: add MpiVol collective read integration test" ``` --- ### Task 3: MPI-IO collective write integration test **Background:** Validates that N ranks each contribute a shard of a dataset; rank 0 assembles and writes the complete file. **Files:** - Modify: `crates/clawhdf5-io/src/mpi_vol.rs` - [x] **Step 1: Add the integration test** ```rust #[test] #[cfg(feature = "mpi-io")] fn collective_write_assembles_all_shards() { use crate::vol::VirtualObjectLayer; use tempfile::TempDir; use mpi::traits::*; let tmp = TempDir::new().unwrap(); let path = tmp.path().join("parallel_out.h5"); let mut vol = MpiVol::new_world().expect("MPI init failed"); vol.open(path.to_str().unwrap()).unwrap(); let world = vol.universe.world(); let rank = world.rank() as usize; // Each rank contributes one f64 value: rank * 10.0 let shard = ((rank as f64) * 10.0f64).to_le_bytes().to_vec(); vol.write_dataset("values", &shard, &[world.size() as u64], "f64") .unwrap(); // All ranks verify the written file has 4 values (one per rank) let total_size = world.size() as usize; if rank == 0 { let bytes = std::fs::read(&path).unwrap(); use clawhdf5_format::{ data_layout::DataLayout, data_read::read_raw_data_full, dataspace::Dataspace, datatype::Datatype, group_v2::resolve_path_any, message_type::MessageType, object_header::ObjectHeader, signature::find_signature, superblock::Superblock, }; let sig = find_signature(&bytes).unwrap(); let sb = Superblock::parse(&bytes, sig).unwrap(); let addr = resolve_path_any(&bytes, &sb, "values").unwrap(); let oh = ObjectHeader::parse( &bytes, addr as usize, sb.offset_size, sb.length_size, ).unwrap(); let (dt, _) = Datatype::parse( &oh.messages.iter().find(|m| m.msg_type == MessageType::Datatype).unwrap().data, ).unwrap(); let ds = Dataspace::parse( &oh.messages.iter().find(|m| m.msg_type == MessageType::Dataspace).unwrap().data, sb.length_size, ).unwrap(); let dl = DataLayout::parse( &oh.messages.iter().find(|m| m.msg_type == MessageType::DataLayout).unwrap().data, sb.offset_size, sb.length_size, ).unwrap(); let raw = read_raw_data_full( &bytes, &dl, &ds, &dt, None, sb.offset_size, sb.length_size, ).unwrap(); assert_eq!(raw.len(), total_size * 8, "expected {} f64 values", total_size); let values: Vec = raw.chunks_exact(8) .map(|c| f64::from_le_bytes(c.try_into().unwrap())) .collect(); for (i, &v) in values.iter().enumerate() { assert!((v - (i as f64 * 10.0)).abs() < 1e-9, "rank {i} shard wrong: got {v}"); } } world.barrier(); } ``` - [x] **Step 2: Run** ```bash cargo test -p clawhdf5-io 2>&1 | tail -5 # no feature — should pass mpirun -np 4 cargo test -p clawhdf5-io --features mpi-io collective_write 2>&1 ``` - [x] **Step 3: Commit** ```bash git add crates/clawhdf5-io/src/mpi_vol.rs git commit -m "feat: add MpiVol collective write integration test (4 ranks)" ``` --- ### Task 4: MpiVol parallel benchmark binary **Background:** Adds a benchmark binary to `clawhdf5-bench` that runs h5bench-equivalent write/read workloads using `MpiVol`. This provides the throughput numbers needed to compare clawhdf5 against standard libhdf5 + h5bench. **Files:** - Create: `crates/clawhdf5-bench/src/bin/mpi_io_bench.rs` - Modify: `crates/clawhdf5-bench/Cargo.toml` (add `mpi-io` feature, `mpi_io_bench` binary) **Produces:** `cargo run -p clawhdf5-bench --features mpi-io --bin mpi_io_bench -- --size 100000` outputs MB/s throughput numbers comparable to h5bench output. - [x] **Step 1: Create the binary** Create `crates/clawhdf5-bench/src/bin/mpi_io_bench.rs`: ```rust //! h5bench-equivalent MPI-IO performance benchmark. //! //! Usage: mpirun -np N cargo run -p clawhdf5-bench --features mpi-io --bin mpi_io_bench -- --size //! //! Measures collective write and read throughput in MB/s for f64 arrays. #[cfg(feature = "mpi-io")] fn main() { use clawhdf5_io::mpi_vol::MpiVol; use clawhdf5_io::vol::VirtualObjectLayer; use std::time::Instant; use mpi::traits::*; let args: Vec = std::env::args().collect(); let n_elements: usize = args.iter() .position(|a| a == "--size") .and_then(|i| args.get(i + 1)) .and_then(|s| s.parse().ok()) .unwrap_or(100_000); let mut vol = MpiVol::new_world().expect("MPI init failed"); let world = vol.universe.world(); let rank = world.rank() as usize; let size = world.size() as usize; let path = format!("/tmp/clawhdf5_mpiio_bench_{n_elements}.h5"); vol.open(&path).unwrap(); // Each rank contributes n_elements/size f64 values let per_rank = n_elements / size; let shard: Vec = (0..per_rank).map(|i| (rank * per_rank + i) as f64).collect(); let shard_bytes: Vec = shard.iter().flat_map(|v| v.to_le_bytes()).collect(); // Collective write world.barrier(); let t0 = Instant::now(); vol.write_dataset("data", &shard_bytes, &[n_elements as u64], "f64").unwrap(); world.barrier(); let write_elapsed = t0.elapsed().as_secs_f64(); // Collective read let t1 = Instant::now(); let _data = vol.read_dataset("data").unwrap(); world.barrier(); let read_elapsed = t1.elapsed().as_secs_f64(); if rank == 0 { let total_mb = (n_elements * 8) as f64 / 1e6; println!("=== clawhdf5 MPI-IO Benchmark ==="); println!("Elements : {n_elements}"); println!("Ranks : {size}"); println!("Total : {total_mb:.1} MB"); println!("Write : {:.1} MB/s", total_mb / write_elapsed); println!("Read : {:.1} MB/s", total_mb / read_elapsed); } } #[cfg(not(feature = "mpi-io"))] fn main() { eprintln!("mpi_io_bench requires the `mpi-io` feature."); eprintln!("Run: mpirun -np N cargo run -p clawhdf5-bench --features mpi-io --bin mpi_io_bench"); std::process::exit(1); } ``` - [x] **Step 2: Add to Cargo.toml** In `crates/clawhdf5-bench/Cargo.toml`, add: ```toml [dependencies] clawhdf5-io = { path = "../clawhdf5-io", features = [] } [features] mpi-io = ["clawhdf5-io/mpi-io", "mpi"] [dependencies.mpi] version = "0.8" optional = true [[bin]] name = "mpi_io_bench" path = "src/bin/mpi_io_bench.rs" ``` - [x] **Step 3: Verify it compiles** ```bash cargo check -p clawhdf5-bench --features mpi-io 2>&1 | tail -10 ``` Expected: no errors. - [x] **Step 4: Run with 4 ranks** ```bash mpirun -np 4 cargo run --release -p clawhdf5-bench --features mpi-io --bin mpi_io_bench -- --size 1000000 2>&1 ``` Expected output (numbers will vary by hardware): ``` === clawhdf5 MPI-IO Benchmark === Elements : 1000000 Ranks : 4 Total : 8.0 MB Write : xxx.x MB/s Read : xxx.x MB/s ``` Record results in `BENCHMARKS.md` under a new `## MPI-IO Parallel I/O` section. - [x] **Step 5: Commit** ```bash git add crates/clawhdf5-bench/src/bin/mpi_io_bench.rs \ crates/clawhdf5-bench/Cargo.toml git commit -m "feat: add mpi_io_bench binary for h5bench-comparable parallel I/O throughput" ``` --- ## Verification ```bash # Without MPI feature — all existing tests still pass cargo test -p clawhdf5-io 2>&1 | tail -10 # With MPI feature — compile check (requires libopenmpi-dev) cargo check -p clawhdf5-io --features mpi-io 2>&1 | tail -5 # Integration tests (requires openmpi-bin) mpirun -np 4 cargo test -p clawhdf5-io --features mpi-io 2>&1 | tail -20 # Benchmark (requires openmpi-bin) mpirun -np 4 cargo run --release -p clawhdf5-bench --features mpi-io --bin mpi_io_bench -- --size 1000000 2>&1 ```