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
//! Concurrent same-dataset appends under the `threadsafe` writer.
//!
//! An append is several dataset-slot acquisitions: take the buffered tail,
//! write the chunk-aligned rows, re-buffer the new tail, extend the extent.
//! The per-slot mutex serializes each acquisition but not the operation, so
//! without the per-dataset op lock two appends interleave between
//! acquisitions — two calls take the same base row, rows are lost or
//! overwritten. These tests drive that exact schedule: N threads append to
//! *one* dataset, with a frame count that never divides the chunk so the
//! buffered tail is always in play.
//!
//! Every append call must land as a contiguous, in-order block of rows
//! (atomicity), every appended frame must appear exactly once (no lost or
//! doubled rows), and no frame may be torn. Order *between* calls is the
//! scheduler's.
#![cfg(feature = "threadsafe")]
use rust_hdf5::H5File;
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_same_ds_{}_{}_{}.h5",
name,
std::process::id(),
n
))
}
/// Tag of frame `f` of append call `j` by thread `k`. Distinct per
/// (k, j, f) so a lost, doubled, or split append shows up as a wrong
/// multiset or a broken run on read-back.
fn tag(k: usize, j: usize, f: usize) -> i32 {
(k * 1_000_000 + j * 100 + f) as i32
}
#[test]
fn concurrent_same_dataset_appends_serialize_wholly() {
const N: usize = 4; // threads
const M: usize = 40; // append calls per thread
const FRAMES: usize = 3; // frames per call — never a multiple of CHUNK0
const W: usize = 8; // row width
const CHUNK0: usize = 4; // chunk rows: every call leaves a buffered tail
const ITERS: usize = 10; // repeat to shake out schedules
for iter in 0..ITERS {
let path = tmp(&format!("iter{iter}"));
{
let file = H5File::create(&path).unwrap();
let ds = file
.new_dataset::<i32>()
.shape([0, W])
.chunk(&[CHUNK0, W])
.max_shape(&[None, Some(W)])
.create("d")
.unwrap();
std::thread::scope(|s| {
for k in 0..N {
let ds = &ds;
s.spawn(move || {
for j in 0..M {
let data: Vec<i32> = (0..FRAMES)
.flat_map(|f| std::iter::repeat_n(tag(k, j, f), W))
.collect();
ds.append(&data)
.unwrap_or_else(|e| panic!("append k={k} j={j}: {e}"));
}
});
}
});
file.close().unwrap();
}
let file = H5File::open(&path).unwrap();
let ds = file.dataset("d").unwrap();
assert_eq!(ds.shape(), vec![N * M * FRAMES, W], "iter {iter}");
let got = ds.read_raw::<i32>().unwrap();
// No torn frames: each row is uniform.
let rows: Vec<i32> = got
.as_chunks::<W>()
.0
.iter()
.enumerate()
.map(|(r, row)| {
assert!(
row.iter().all(|&v| v == row[0]),
"iter {iter}: torn frame at row {r}: {row:?}"
);
row[0]
})
.collect();
// Atomicity: each call's frames form one contiguous in-order run.
// Walking runs also proves every frame appears exactly once — the
// walk consumes exactly FRAMES rows per (k, j) and the total length
// already matched.
let mut seen = std::collections::HashSet::new();
let mut r = 0;
while r < rows.len() {
let first = rows[r];
let (k, rem) = ((first / 1_000_000) as usize, first % 1_000_000);
let (j, f) = ((rem / 100) as usize, (rem % 100) as usize);
assert_eq!(f, 0, "iter {iter}: run at row {r} starts mid-call: {first}");
for f in 0..FRAMES {
assert_eq!(
rows[r + f],
tag(k, j, f),
"iter {iter}: call (k={k}, j={j}) split at row {}",
r + f
);
}
assert!(
seen.insert((k, j)),
"iter {iter}: call (k={k}, j={j}) appended twice"
);
r += FRAMES;
}
assert_eq!(seen.len(), N * M, "iter {iter}");
std::fs::remove_file(&path).ok();
}
}
#[test]
fn concurrent_same_dataset_vlen_appends_serialize_wholly() {
const N: usize = 4; // threads
const M: usize = 30; // append calls per thread
const STRINGS: usize = 3; // strings per call
const CHUNK: usize = 4; // never a multiple of STRINGS: tail always buffered
const ITERS: usize = 10;
for iter in 0..ITERS {
let path = tmp(&format!("vlen{iter}"));
{
let file = H5File::create(&path).unwrap();
file.create_appendable_vlen_dataset("strs", CHUNK, None)
.unwrap();
std::thread::scope(|s| {
for k in 0..N {
let file = &file;
s.spawn(move || {
for j in 0..M {
let batch: Vec<String> =
(0..STRINGS).map(|f| format!("{k}:{j}:{f}")).collect();
let refs: Vec<&str> = batch.iter().map(String::as_str).collect();
file.append_vlen_strings("strs", &refs)
.unwrap_or_else(|e| panic!("append k={k} j={j}: {e}"));
}
});
}
});
file.close().unwrap();
}
let file = H5File::open(&path).unwrap();
let ds = file.dataset("strs").unwrap();
let got = ds.read_vlen_strings().unwrap();
assert_eq!(got.len(), N * M * STRINGS, "iter {iter}");
// Each call's strings must form one contiguous in-order run, and the
// runs must cover every (k, j) exactly once.
let mut seen = std::collections::HashSet::new();
let mut r = 0;
while r < got.len() {
let parts: Vec<usize> = got[r].split(':').map(|p| p.parse().unwrap()).collect();
let (k, j, f) = (parts[0], parts[1], parts[2]);
assert_eq!(
f, 0,
"iter {iter}: run at element {r} starts mid-call: {}",
got[r]
);
for f in 0..STRINGS {
assert_eq!(
got[r + f],
format!("{k}:{j}:{f}"),
"iter {iter}: call (k={k}, j={j}) split at element {}",
r + f
);
}
assert!(
seen.insert((k, j)),
"iter {iter}: call (k={k}, j={j}) appended twice"
);
r += STRINGS;
}
assert_eq!(seen.len(), N * M, "iter {iter}");
std::fs::remove_file(&path).ok();
}
}