Skip to content

Commit 9b13d51

Browse files
steven-vargasteven-varga
committed
Merge remote-tracking branch 'origin/241-feature-wire-threaded-pipeline' into staging
2 parents afa3d68 + 5dcceed commit 9b13d51

4 files changed

Lines changed: 175 additions & 22 deletions

File tree

‎h5cpp/H5Dappend.hpp‎

Lines changed: 62 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,10 @@
77
#include "H5capi.hpp"
88
#include "H5Tmeta.hpp"
99
#include "H5cout.hpp"
10-
#include <cstring>
10+
#include "H5Zpipeline_threaded.hpp"
11+
#include <memory>
1112
#include <string>
13+
#include <variant>
1214
#include <vector>
1315
#include <stdexcept>
1416
#include <type_traits>
@@ -19,18 +21,35 @@ namespace h5 {
1921
}
2022
std::ostream& operator<<(std::ostream& os, const h5::pt_t& pt);
2123

24+
namespace h5::impl {
25+
// pt_t::pipeline selects between the synchronous basic pipeline (default,
26+
// bytewise-identical to pre-241 behavior) and the parallel threaded pipeline.
27+
// Both alternatives are indirect-owned through unique_ptr so that the
28+
// variant remains move-assignable regardless of the underlying pipeline's
29+
// move semantics (threaded_pipeline_t deletes moves because it owns
30+
// std::jthread workers and atomics; basic_pipeline_t inherits a manually-
31+
// written move-assign from pipeline_t<Derived> that suppresses the
32+
// implicit move-ctor needed by variant assignment).
33+
using pt_pipeline_t = std::variant<
34+
std::unique_ptr<impl::basic_pipeline_t>,
35+
std::unique_ptr<impl::threaded_pipeline_t>
36+
>;
37+
}
38+
2239
// packet table template specialization with inheritance
2340
namespace h5 {
2441
struct pt_t {
2542
pt_t();
26-
pt_t( const h5::ds_t& handle ); // conversion ctor
27-
// deep copy with own cache memory
43+
pt_t( const h5::ds_t& handle ); // conversion ctor — synchronous pipeline
44+
pt_t( const h5::ds_t& handle, h5::filter::threads workers ); // threaded pipeline
45+
// deep copy with own cache memory — always uses the synchronous pipeline,
46+
// since the threaded pipeline owns workers that cannot be duplicated.
2847
pt_t( const h5::pt_t& pt ) : h5::pt_t(pt.ds) {
2948
};
3049
~pt_t();
3150

3251
pt_t& operator=( h5::pt_t&& pt ){
33-
// prevent self assign
52+
// prevent self assign
3453
if (this == &pt) return *this;
3554
if(H5Iis_valid(this->ds)){ // flush and close dataset
3655
this->flush();
@@ -75,7 +94,19 @@ namespace h5 {
7594
void> append( const T& ref );
7695
void append( const std::string& ref );
7796

78-
impl::pipeline_t<impl::basic_pipeline_t> pipeline;
97+
// Resolve the variant to a reference to the live pipeline_t<Derived> CRTP
98+
// instance, regardless of which alternative is active. Both alternatives
99+
// expose the same set of fields (chunk0, block_size, ds, dxpl, ...) inherited
100+
// from pipeline_t<Derived>, so the visiting lambda can use the result
101+
// duck-typed across alternatives.
102+
template <typename F>
103+
decltype(auto) visit_pipeline(F&& f) {
104+
return std::visit([&](auto& p) -> decltype(auto) {
105+
return std::forward<F>(f)(*p); // both alternatives are unique_ptr
106+
}, pipeline);
107+
}
108+
109+
impl::pt_pipeline_t pipeline;
79110
h5::dxpl_t dxpl;
80111
h5::ds_t ds;
81112
h5::dt_t<void> dt;
@@ -91,19 +122,30 @@ namespace h5 {
91122
/* initialized to invalid state
92123
* */
93124
inline h5::pt_t::pt_t() :
125+
pipeline{std::make_unique<impl::basic_pipeline_t>()},
94126
dxpl{H5Pcreate(H5P_DATASET_XFER)},ds{H5I_UNINIT},n{0},fill_value{nullptr}{
95127
for(hsize_t i=0; i<H5CPP_MAX_RANK; i++ )
96128
count[i] = 1, offset[i] = 0;
97129
}
98130

99-
// conversion ctor
131+
// conversion ctor — synchronous pipeline (default)
100132
inline
101133
h5::pt_t::pt_t( const h5::ds_t& handle ) : pt_t() {
102134
/*default ctor has an invalid state -- skip initialization */
103135
if( !is_valid(handle) ) return;
104136
init(handle);
105137
}
106138

139+
// conversion ctor — threaded pipeline with N compression workers
140+
inline
141+
h5::pt_t::pt_t( const h5::ds_t& handle, h5::filter::threads workers ) : pt_t() {
142+
if( !is_valid(handle) ) return;
143+
auto threaded = std::make_unique<impl::threaded_pipeline_t>();
144+
threaded->set_worker_count(workers.n);
145+
pipeline.emplace<std::unique_ptr<impl::threaded_pipeline_t>>(std::move(threaded));
146+
init(handle);
147+
}
148+
107149
inline
108150
h5::pt_t::~pt_t(){
109151
/*default ctor has an invalid state -- skip flushing cache */
@@ -137,12 +179,14 @@ void h5::pt_t::init( const h5::ds_t& handle ){
137179
h5::dt_t<void*> type = h5::get_type<void*>( ds );
138180
hsize_t size = h5::get_size( type );
139181
this->fill_value = h5::get_fill_value(dcpl, type, size);
140-
pipeline.set_cache(dcpl, size);
141-
this->ptr = pipeline.chunk0;
142-
this->block_size = pipeline.block_size;
143-
this->element_size = pipeline.element_size;
144-
this->N = pipeline.n;
145-
pipeline.ds = ds; pipeline.dxpl = dxpl;
182+
visit_pipeline([&](auto& p) {
183+
p.set_cache(dcpl, size);
184+
this->ptr = p.chunk0;
185+
this->block_size = p.block_size;
186+
this->element_size = p.element_size;
187+
this->N = p.n;
188+
p.ds = ds; p.dxpl = dxpl;
189+
});
146190
h5::get_chunk_dims( dcpl, chunk_dims );
147191
for(hsize_t i=1; i<rank; i++)
148192
current_dims[i] = chunk_dims[i];
@@ -156,7 +200,7 @@ void> h5::pt_t::append( const T* ptr ) try {
156200
*offset = *current_dims;
157201
*current_dims += *chunk_dims;
158202
h5::set_extent(ds, current_dims);
159-
pipeline.write_chunk(offset,block_size,ptr);
203+
visit_pipeline([&](auto& p){ p.write_chunk(offset, block_size, ptr); });
160204
} catch( const std::runtime_error& err ){
161205
throw h5::error::io::dataset::append( err.what() );
162206
}
@@ -209,7 +253,7 @@ void> h5::pt_t::append( const T& ref ) try {
209253
*offset = *current_dims;
210254
*current_dims += *chunk_dims;
211255
h5::set_extent(ds, current_dims);
212-
pipeline.write_chunk(offset,block_size,ptr);
256+
visit_pipeline([&](auto& p){ p.write_chunk(offset, block_size, ptr); });
213257
} catch( const std::runtime_error& err ){
214258
throw h5::error::io::dataset::append( err.what() );
215259
}
@@ -235,21 +279,21 @@ void> h5::pt_t::append( const T& ref ) try {
235279
switch( dims_.size() ){
236280
case 1: // vector
237281
if( dims[0] * element_size == block_size )
238-
pipeline.write_chunk(offset, block_size, (void*) ptr_ );
282+
visit_pipeline([&](auto& p){ p.write_chunk(offset, block_size, (void*) ptr_ ); });
239283
else throw h5::error::io::packet_table::write(
240284
H5CPP_ERROR_MSG("dimension mismatch: "
241285
+ std::to_string( dims[0] * element_size) + " != " + std::to_string(block_size) ));
242286
break;
243287
case 2: //matrix
244288
if( dims[0] * dims[1] * element_size == block_size )
245-
pipeline.write_chunk(offset, block_size, (void*) ptr_ );
289+
visit_pipeline([&](auto& p){ p.write_chunk(offset, block_size, (void*) ptr_ ); });
246290
else throw h5::error::io::packet_table::write(
247291
H5CPP_ERROR_MSG("dimension mismatch: "
248292
+ std::to_string( dims[0] * dims[1] * element_size) + " != " + std::to_string(block_size) ));
249293
break;
250294
case 3: // cube
251295
if( dims[0] * dims[1] * dims[2] * element_size == block_size )
252-
pipeline.write_chunk(offset, block_size, (void*) ptr_ );
296+
visit_pipeline([&](auto& p){ p.write_chunk(offset, block_size, (void*) ptr_ ); });
253297
else throw h5::error::io::packet_table::write(
254298
H5CPP_ERROR_MSG("dimension mismatch: "
255299
+ std::to_string( dims[0] * dims[1] * dims[2] * element_size) + " != " + std::to_string(block_size) ));
@@ -283,7 +327,7 @@ void h5::pt_t::flush(){
283327
for(hsize_t i=0; i<(N-n); i++)
284328
for(size_t j=0; j < element_size; j++)
285329
static_cast<char*>( ptr )[(n + i) * element_size + j] = static_cast<char*>( fill_value )[ j ];
286-
pipeline.write_chunk( offset, block_size, ptr );
330+
visit_pipeline([&](auto& p){ p.write_chunk(offset, block_size, ptr); });
287331
}
288332
n = 0;
289333
}

‎h5cpp/H5Zpipeline_threaded.hpp‎

Lines changed: 33 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,19 @@
2222
// so no HDF5 API calls occur on worker threads, preserving compatibility with
2323
// non-thread-safe HDF5 builds.
2424

25+
namespace h5::filter {
26+
// Opt-in tag selecting the threaded filter pipeline.
27+
// h5::filter::threads{} -> std::thread::hardware_concurrency() workers
28+
// h5::filter::threads{N} -> N compression workers
29+
// Passed to pt_t constructors (and h5::write) to switch from the default
30+
// synchronous basic_pipeline_t to the parallel threaded_pipeline_t.
31+
struct threads {
32+
unsigned n;
33+
constexpr threads() noexcept : n(0) {}
34+
constexpr explicit threads(unsigned count) noexcept : n(count) {}
35+
};
36+
}
37+
2538
namespace h5::impl {
2639

2740
namespace detail {
@@ -79,6 +92,11 @@ struct threaded_pipeline_t : public pipeline_t<threaded_pipeline_t> {
7992
threaded_pipeline_t(threaded_pipeline_t&&) = delete;
8093
threaded_pipeline_t& operator=(threaded_pipeline_t&&) = delete;
8194

95+
// Runtime override of worker pool size. If non-zero, takes precedence over
96+
// the H5CPP_PIPELINE_WORKERS compile-time default. Must be called before
97+
// the first write_chunk_impl, since workers spawn lazily via std::call_once.
98+
void set_worker_count(unsigned n) noexcept { worker_count_override_ = n; }
99+
82100
// Drains all in-flight work and calls H5Dwrite_chunk from the calling
83101
// (main) thread. Must be called before reading back written data.
84102
void flush() {
@@ -152,8 +170,10 @@ struct threaded_pipeline_t : public pipeline_t<threaded_pipeline_t> {
152170
void ensure_workers() {
153171
std::call_once(init_flag_, [this] {
154172
constexpr unsigned cfg = static_cast<unsigned>(H5CPP_PIPELINE_WORKERS);
155-
const unsigned n = (cfg > 0) ? cfg
156-
: std::max(1u, std::thread::hardware_concurrency());
173+
const unsigned n =
174+
(worker_count_override_ > 0) ? worker_count_override_
175+
: (cfg > 0) ? cfg
176+
: std::max(1u, std::thread::hardware_concurrency());
157177
workers_.reserve(n);
158178
for (unsigned i = 0; i < n; ++i)
159179
workers_.emplace_back([this](std::stop_token st) { worker_loop(st); });
@@ -223,6 +243,7 @@ struct threaded_pipeline_t : public pipeline_t<threaded_pipeline_t> {
223243
std::atomic<int> in_flight_{0};
224244
std::vector<std::jthread> workers_;
225245
std::once_flag init_flag_;
246+
unsigned worker_count_override_{0};
226247
};
227248

228249
#else
@@ -316,6 +337,11 @@ struct threaded_pipeline_t : public pipeline_t<threaded_pipeline_t> {
316337
threaded_pipeline_t(threaded_pipeline_t&&) = delete;
317338
threaded_pipeline_t& operator=(threaded_pipeline_t&&) = delete;
318339

340+
// Runtime override of worker pool size. If non-zero, takes precedence over
341+
// the H5CPP_PIPELINE_WORKERS compile-time default. Must be called before
342+
// the first write_chunk_impl, since workers spawn lazily via std::call_once.
343+
void set_worker_count(unsigned n) noexcept { worker_count_override_ = n; }
344+
319345
// Drains all in-flight work and calls H5Dwrite_chunk from the calling
320346
// (main) thread. Must be called before reading back written data.
321347
void flush() {
@@ -389,8 +415,10 @@ struct threaded_pipeline_t : public pipeline_t<threaded_pipeline_t> {
389415
void ensure_workers() {
390416
std::call_once(init_flag_, [this] {
391417
constexpr unsigned cfg = static_cast<unsigned>(H5CPP_PIPELINE_WORKERS);
392-
const unsigned n = (cfg > 0) ? cfg
393-
: std::max(1u, std::thread::hardware_concurrency());
418+
const unsigned n =
419+
(worker_count_override_ > 0) ? worker_count_override_
420+
: (cfg > 0) ? cfg
421+
: std::max(1u, std::thread::hardware_concurrency());
394422
workers_.reserve(n);
395423
for (unsigned i = 0; i < n; ++i)
396424
workers_.emplace_back(
@@ -461,6 +489,7 @@ struct threaded_pipeline_t : public pipeline_t<threaded_pipeline_t> {
461489
std::atomic<int> in_flight_{0};
462490
std::vector<h5::detail::stoppable_thread_t> workers_;
463491
std::once_flag init_flag_;
492+
unsigned worker_count_override_{0};
464493
};
465494

466495
#endif // __cplusplus >= 202002L

‎h5cpp/core‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@
6060
#include "H5Pall.hpp"
6161
#include "H5Zpipeline.hpp"
6262
#include "H5Zpipeline_basic.hpp"
63+
#include "H5Zpipeline_threaded.hpp"
6364
#include "H5Pdapl.hpp"
6465

6566
#include "H5Ialgorithm.hpp"

‎test/H5Dappend.cpp‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -162,6 +162,85 @@ TEST_CASE("packet table output stream for invalid handle") {
162162
CHECK(oss.str().find("H5I_UNINIT") != std::string::npos);
163163
}
164164

165+
// =====================================================================
166+
// [#241] Threaded filter pipeline opt-in via h5::filter::threads{N}
167+
// =====================================================================
168+
169+
TEST_CASE("[#241] h5::filter::threads tag construction") {
170+
// Default-constructed tag means "use hardware_concurrency() workers".
171+
constexpr h5::filter::threads default_t{};
172+
CHECK(default_t.n == 0);
173+
174+
// Explicit count.
175+
constexpr h5::filter::threads explicit_t{4};
176+
CHECK(explicit_t.n == 4);
177+
}
178+
179+
TEST_CASE("[#241] pt_t with threaded pipeline — basic round-trip (no filter)") {
180+
h5::test::file_fixture_t f("test-pt-threaded-nofilter.h5");
181+
h5::ds_t ds = h5::create<int>(f.fd, "ds", h5::current_dims_t{0},
182+
h5::max_dims_t{H5S_UNLIMITED}, h5::chunk{16});
183+
{
184+
h5::pt_t pt(ds, h5::filter::threads{2});
185+
for (int i = 0; i < 64; ++i)
186+
h5::append(pt, i);
187+
h5::flush(pt); // ensure all workers drain before close
188+
}
189+
auto readback = h5::read<std::vector<int>>(f.fd, "ds");
190+
REQUIRE(readback.size() == 64);
191+
for (int i = 0; i < 64; ++i) CHECK(readback[i] == i);
192+
}
193+
194+
TEST_CASE("[#241] pt_t with threaded pipeline — gzip-compressed bytewise equivalence") {
195+
// Write the same data with basic and threaded pipelines into two files,
196+
// read back, assert content matches. Different filter chain ordering can
197+
// produce different on-disk bytes; we assert decompressed content equivalence.
198+
constexpr int N = 256;
199+
std::vector<int> expected(N);
200+
for (int i = 0; i < N; ++i) expected[i] = i * 7 + 3;
201+
202+
auto write_file = [&](const char* path, auto pt_factory) {
203+
h5::test::file_fixture_t f(path);
204+
h5::ds_t ds = h5::create<int>(f.fd, "ds", h5::current_dims_t{0},
205+
h5::max_dims_t{H5S_UNLIMITED}, h5::chunk{32} | h5::gzip{6});
206+
{
207+
auto pt = pt_factory(ds);
208+
for (int v : expected) h5::append(pt, v);
209+
h5::flush(pt);
210+
}
211+
};
212+
213+
write_file("test-pt-basic-gzip.h5",
214+
[](const h5::ds_t& ds) { return h5::pt_t(ds); });
215+
write_file("test-pt-threaded-gzip.h5",
216+
[](const h5::ds_t& ds) { return h5::pt_t(ds, h5::filter::threads{4}); });
217+
218+
h5::fd_t basic = h5::open("test-pt-basic-gzip.h5", H5F_ACC_RDONLY);
219+
h5::fd_t threaded = h5::open("test-pt-threaded-gzip.h5", H5F_ACC_RDONLY);
220+
auto basic_data = h5::read<std::vector<int>>(basic, "ds");
221+
auto threaded_data = h5::read<std::vector<int>>(threaded, "ds");
222+
REQUIRE(basic_data.size() == expected.size());
223+
REQUIRE(threaded_data.size() == expected.size());
224+
CHECK(basic_data == expected);
225+
CHECK(threaded_data == expected);
226+
CHECK(basic_data == threaded_data);
227+
}
228+
229+
TEST_CASE("[#241] pt_t with threaded pipeline — default worker count (hw_concurrency)") {
230+
h5::test::file_fixture_t f("test-pt-threaded-default.h5");
231+
h5::ds_t ds = h5::create<int>(f.fd, "ds", h5::current_dims_t{0},
232+
h5::max_dims_t{H5S_UNLIMITED}, h5::chunk{16} | h5::gzip{1});
233+
{
234+
// h5::filter::threads{} with no number → hw_concurrency() workers
235+
h5::pt_t pt(ds, h5::filter::threads{});
236+
for (int i = 0; i < 96; ++i) h5::append(pt, i);
237+
h5::flush(pt);
238+
}
239+
auto readback = h5::read<std::vector<int>>(f.fd, "ds");
240+
REQUIRE(readback.size() == 96);
241+
for (int i = 0; i < 96; ++i) CHECK(readback[i] == i);
242+
}
243+
165244
TEST_CASE("[#232] std::forward_list<int> append streams elements into chunked dataset") {
166245
h5::test::file_fixture_t f("test-pt-fwdlist.h5");
167246
// forward_list is append/view only — h5::write/read intentionally unsupported.

0 commit comments

Comments
 (0)