docs: add superpowers implementation plans (MPI-IO VOL backend, format write extensions, filter codecs)
This commit is contained in:
Omar Sobh
@@ -0,0 +1,757 @@
|
||||
# MPI-IO VOL Backend Implementation Plan
|
||||
|
||||
> **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<MpiComm>) -> Self` — wraps an MPI communicator
|
||||
- `MpiVol::new_world() -> Self` — convenience for `MPI_COMM_WORLD`
|
||||
|
||||
- [ ] **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<String>,
|
||||
#[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<Self, VolError> {
|
||||
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<Self, VolError> {
|
||||
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<VolCapability> {
|
||||
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<Vec<u8>, 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<Vec<u8>, 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<u8>;
|
||||
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<f64> = 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<f32> = 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"));
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **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.
|
||||
|
||||
- [ ] **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"]
|
||||
```
|
||||
|
||||
- [ ] **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;
|
||||
```
|
||||
|
||||
- [ ] **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.
|
||||
|
||||
- [ ] **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).
|
||||
|
||||
- [ ] **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)
|
||||
|
||||
- [ ] **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<f64> = 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"
|
||||
```
|
||||
|
||||
- [ ] **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.
|
||||
|
||||
- [ ] **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.
|
||||
|
||||
- [ ] **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`
|
||||
|
||||
- [ ] **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<f64> = 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();
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **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
|
||||
```
|
||||
|
||||
- [ ] **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.
|
||||
|
||||
- [ ] **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 <N>
|
||||
//!
|
||||
//! 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<String> = 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<f64> = (0..per_rank).map(|i| (rank * per_rank + i) as f64).collect();
|
||||
let shard_bytes: Vec<u8> = 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);
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **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"
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Verify it compiles**
|
||||
|
||||
```bash
|
||||
cargo check -p clawhdf5-bench --features mpi-io 2>&1 | tail -10
|
||||
```
|
||||
|
||||
Expected: no errors.
|
||||
|
||||
- [ ] **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.
|
||||
|
||||
- [ ] **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
|
||||
```
|
||||
Reference in New Issue
Block a user