1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
//! Concurrent create + append + metadata stress test for the `threadsafe`
//! writer (Stage 3d of `docs/threadsafe-fine-grained-locking.md`).
//!
//! Where `threadsafe_parallel_write.rs` exercises concurrent *writes* to
//! datasets created up front, this test drives the paths Stage 3c moved onto
//! the shared read guard: every thread, holding only `&H5File`, **creates**
//! its own unlimited chunked dataset, **appends** to it in several batches
//! (create → extend → append → write_chunk), and writes a per-dataset
//! **string attribute** (a global-heap metadata write). All of this runs
//! concurrently across threads on distinct datasets, so the only
//! serialization is the per-slot mutex (never contended here — distinct
//! datasets), the brief spine mutex (registry push/clone), the atomic
//! allocator, and positioned pwrite.
//!
//! Every iteration reopens the file and asserts each dataset's full contents
//! and its attribute; the run repeats so a data race surfaces as a wrong or
//! short read rather than passing silently. h5py reads the result back on the
//! first iteration for standard-tool correctness.
//!
//! Gated on `threadsafe` (the concurrency under test). Present under the
//! default feature set plus `--features threadsafe`.
#![cfg(feature = "threadsafe")]
use rust_hdf5::types::VarLenUnicode;
use rust_hdf5::H5File;
const TEST_PYTHON: &str = "/Users/stevek/mamba/envs/bs2026.1/bin/python";
fn python() -> Option<&'static str> {
static PY: std::sync::OnceLock<Option<String>> = std::sync::OnceLock::new();
PY.get_or_init(|| {
let candidate =
std::env::var("RUST_HDF5_TEST_PYTHON").unwrap_or_else(|_| TEST_PYTHON.to_string());
if std::path::Path::new(&candidate).exists() {
Some(candidate)
} else {
eprintln!("skipping h5py cross-check: {candidate} not present");
None
}
})
.as_deref()
}
fn tmp(name: &str) -> std::path::PathBuf {
use std::sync::atomic::{AtomicU64, Ordering};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let n = COUNTER.fetch_add(1, Ordering::Relaxed);
std::env::temp_dir().join(format!(
"rust_hdf5_cc_{}_{}_{}.h5",
name,
std::process::id(),
n
))
}
/// Value at element `i` of dataset `k`. Distinct per (k, i) so a cross-dataset
/// or cross-chunk mix-up shows up as a wrong value on read-back.
fn value(k: usize, i: usize) -> i32 {
(k * 1_000_000 + i) as i32
}
#[test]
fn concurrent_create_append_metadata_distinct_datasets() {
const N: usize = 8; // threads / datasets
const BATCHES: usize = 8; // appends per dataset
const BATCH: usize = 32; // elements per append
const LEN: usize = BATCHES * BATCH; // 256 elements per dataset
const CHUNK: usize = 16; // -> several chunks, partial-chunk buffering exercised
const ITERS: usize = 16; // repeat to shake out races
for iter in 0..ITERS {
let path = tmp(&format!("iter{iter}"));
{
let file = H5File::create(&path).unwrap();
// Each thread owns one dataset end to end: create, append in
// batches, then attach a string attribute. Threads share only
// `&file` (so creation itself runs under the read guard).
std::thread::scope(|s| {
for k in 0..N {
let file = &file;
s.spawn(move || {
let ds = file
.new_dataset::<i32>()
.shape([0])
.chunk(&[CHUNK])
.max_shape(&[None])
.create(&format!("ds{k}"))
.unwrap_or_else(|e| panic!("create ds{k}: {e}"));
for b in 0..BATCHES {
let batch: Vec<i32> =
(0..BATCH).map(|i| value(k, b * BATCH + i)).collect();
ds.append(&batch)
.unwrap_or_else(|e| panic!("append ds{k} batch {b}: {e}"));
}
let attr = ds
.new_attr::<VarLenUnicode>()
.shape(())
.create("origin")
.unwrap_or_else(|e| panic!("create attr ds{k}: {e}"));
attr.write_string(&format!("thread-{k}"))
.unwrap_or_else(|e| panic!("write attr ds{k}: {e}"));
});
}
});
file.close().unwrap();
}
// Reopen and assert every dataset's full contents and its attribute.
// A race in the allocator, the positioned writes, the per-slot index,
// or the registry push would surface here.
{
let file = H5File::open(&path).unwrap();
for k in 0..N {
let ds = file.dataset(&format!("ds{k}")).unwrap();
let got = ds.read_raw::<i32>().unwrap();
let want: Vec<i32> = (0..LEN).map(|i| value(k, i)).collect();
assert_eq!(
got.len(),
LEN,
"dataset ds{k} length mismatch on iter {iter}"
);
assert_eq!(got, want, "dataset ds{k} content mismatch on iter {iter}");
let origin = ds
.attr("origin")
.and_then(|a| a.read_string())
.unwrap_or_else(|e| panic!("read attr ds{k} iter {iter}: {e}"));
assert_eq!(
origin,
format!("thread-{k}"),
"dataset ds{k} attribute mismatch on iter {iter}"
);
}
}
// h5py reads the concurrently-created datasets + attributes back on
// the first iteration: confirms standard-tool readability, not merely
// round-trippability by our own reader.
if iter == 0 {
if let Some(py) = python() {
let script = format!(
"import h5py, numpy as np\n\
f = h5py.File(r'{}', 'r')\n\
for k in range({N}):\n\
\x20 d = f['ds%d' % k]\n\
\x20 v = d[...]\n\
\x20 exp = np.arange({LEN}, dtype='i8') + k * 1000000\n\
\x20 assert v.shape == ({LEN},), (k, v.shape)\n\
\x20 assert (v.astype('i8') == exp).all(), (k, v[:8], exp[:8])\n\
\x20 assert d.attrs['origin'] == ('thread-%d' % k), (k, d.attrs['origin'])\n\
f.close()\n",
path.display()
);
let status = std::process::Command::new(py)
.arg("-c")
.arg(&script)
.status()
.expect("failed to spawn python");
assert!(status.success(), "h5py cross-check failed for {path:?}");
}
}
std::fs::remove_file(&path).ok();
}
}
/// Concurrent creation of the *same* dataset name must produce exactly one
/// dataset, not a file with two same-named links.
///
/// Stage 3c lets every thread create on a shared `&H5File`. The name-uniqueness
/// check and the registry insert must be atomic (the `create_lock` gate);
/// otherwise two threads could both pass the duplicate-name check against a
/// snapshot and both push, writing an invalid HDF5 file. Here `N` threads race
/// to create the one name "dup": exactly one must win, the rest must get the
/// "already exists" error, and the reopened file must contain a single readable
/// "dup" dataset holding the winner's data. Repeated to hammer the race window.
#[test]
fn concurrent_create_same_name_yields_one_dataset() {
use std::sync::atomic::{AtomicUsize, Ordering};
const N: usize = 8; // threads racing for one name
const ITERS: usize = 64; // repeat to shake out the check/insert race
const PATTERN: [i32; 4] = [10, 20, 30, 40];
for iter in 0..ITERS {
let path = tmp(&format!("dup{iter}"));
let successes = AtomicUsize::new(0);
{
let file = H5File::create(&path).unwrap();
std::thread::scope(|s| {
for _ in 0..N {
let file = &file;
let successes = &successes;
s.spawn(move || {
// All threads race to create the single name "dup"
// through a gated create path (chunked unlimited). The
// winner appends a known pattern; losers are rejected
// with an error (the `Err` arm), never silently
// duplicated.
if let Ok(ds) = file
.new_dataset::<i32>()
.shape([0])
.chunk(&[4])
.max_shape(&[None])
.create("dup")
{
successes.fetch_add(1, Ordering::Relaxed);
ds.append(&PATTERN).expect("winner append");
}
});
}
});
file.close().unwrap();
}
let n_ok = successes.load(Ordering::Relaxed);
assert_eq!(
n_ok, 1,
"exactly one create must win on iter {iter}, got {n_ok}"
);
// Reopen: exactly one "dup" link, and it holds the winner's data.
let file = H5File::open(&path).unwrap();
let dup_links = file
.dataset_names()
.into_iter()
.filter(|n| n.trim_start_matches('/') == "dup")
.count();
assert_eq!(
dup_links, 1,
"file must contain exactly one 'dup' link on iter {iter}, got {dup_links}"
);
let got = file.dataset("dup").unwrap().read_raw::<i32>().unwrap();
assert_eq!(got, PATTERN.to_vec(), "winner's data on iter {iter}");
drop(file);
std::fs::remove_file(&path).ok();
}
}