Merge remote-tracking branch 'origin/263-parallel-decompress-read' into staging

steven-varga · steven-varga · commit d8e2fd6a0104 · 2026-05-25T09:41:36.000-04:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -329,7 +329,9 @@ jobs:
         run: cmake --build build --parallel --config Release
 
       - name: Test
+        shell: bash
         run: |
+          set -euxo pipefail
           if [ "$CDASH_SUBMIT" = "true" ]; then
             ctest -D Experimental --test-dir build -C Release --output-on-failure
           else
diff --git a/h5cpp/H5Tmeta.hpp b/h5cpp/H5Tmeta.hpp
@@ -386,29 +386,13 @@ namespace h5::meta {
         : is_transport_contiguous_t<typename meta::decay<T>::type> {};
 
     // Trivially copyable aggregates are safe for bulk memcpy: no padding surprises,
-    // no non-trivial copy semantics.  Excludes arrays and text already handled above,
-    // and plugin-registered types which may have non-trivial HDF5 field ordering.
+    // no non-trivial copy semantics.  Excludes arrays and text already handled above.
     template <class T> struct is_transport_contiguous_impl_t<T, std::enable_if_t<
         std::is_aggregate_v<T> &&
         !is_array_like<T>::value &&
         !is_text_like<T>::value &&
-        !is_reflected_compound_t<T>::value &&
         std::is_trivially_copyable_v<T>>> : std::true_type {};
 
-    template <class T> struct is_transport_contiguous_impl_t<T, std::enable_if_t<is_reflected_compound_t<T>::value>> {
-    private:
-        static_assert(compiler_meta_t<T>::version == metadata_version,
-            "H5CPP compiler metadata version mismatch");
-        using fields_t = typename compiler_meta_t<T>::fields_t;
-        template <std::size_t... Is>
-        static constexpr bool check_contiguous(std::index_sequence<Is...>) noexcept {
-            return (... && is_transport_contiguous_v<
-                typename std::tuple_element_t<Is, fields_t>::field_type>);
-        }
-    public:
-        static constexpr bool value = check_contiguous(std::make_index_sequence<std::tuple_size_v<fields_t>>{});
-    };
-
     // Gap 1: contiguous STL sequence containers (vector<T>, span<T>, linalg types, etc.)
     // Triggers when T exposes a data() pointer and size(), but is not a C/std::array,
     // not text, and not arithmetic.
@@ -591,30 +575,16 @@ namespace h5::meta {
         static constexpr std::size_t bytes(const T&) noexcept { return sizeof(T); }
     };
 
-    // Reflected compound structs (plugin-registered)
-    template <class T>
-    struct access_traits_t<T, std::enable_if_t<is_reflected_compound_t<T>::value>> {
-        using element_t  = T;
-        using pointer_t  = const T*;
-        static constexpr access_t kind = access_t::object;
-        static constexpr bool is_trivially_packable = is_transport_contiguous_v<T>;
-        static const T*  data(const T& v)  noexcept { return &v; }
-        static T*        data(T& v)        noexcept { return &v; }
-        static constexpr std::array<std::size_t,0> size(const T&) noexcept { return {}; }
-        static constexpr std::size_t bytes(const T&) noexcept { return sizeof(T); }
-    };
-
     // Plain aggregates: any struct/class that is an aggregate but not arithmetic,
-    // array-like, text-like, or already plugin-registered via is_reflected_compound_t.
+    // array-like, or text-like.
     // Provides the memory-access contract so h5::write(ds, pod_value) works once
     // storage_traits_impl_t<T> is populated (old dt_t path or future C++26 reflection).
     template <class T>
     struct access_traits_t<T, std::enable_if_t<
         std::is_aggregate_v<T> &&
         !std::is_arithmetic_v<T> &&
         !is_array_like<T>::value &&
-        !is_text_like<T>::value &&
-        !is_reflected_compound_t<T>::value>> {
+        !is_text_like<T>::value>> {
         using element_t  = T;
         using pointer_t  = const T*;
         static constexpr access_t kind = access_t::object;
diff --git a/h5cpp/H5Zpipeline.hpp b/h5cpp/H5Zpipeline.hpp
@@ -63,16 +63,28 @@ namespace h5{ namespace impl {
 		static constexpr size_t alignment = H5CPP_MEM_ALIGNMENT;
 		static constexpr size_t default_capacity = 256 * 1024 * 1024;
 
-		struct aligned_deleter {
-			void operator()(char* ptr) const { std::free(ptr); }
+		struct arena_deleter {
+			void operator()(char* ptr) const {
+#ifdef _WIN32
+				_aligned_free(ptr);
+#else
+				std::free(ptr);
+#endif
+			}
 		};
-		std::unique_ptr<char, aligned_deleter> base;
+		std::unique_ptr<char, arena_deleter> base;
 		char* bump = nullptr;
 		char* end = nullptr;
 
 		explicit chunk_arena_t(size_t capacity = default_capacity) {
 			void* ptr = nullptr;
+#ifdef _WIN32
+			ptr = _aligned_malloc(capacity, alignment);
+#else
 			if (posix_memalign(&ptr, alignment, capacity) != 0)
+				ptr = nullptr;
+#endif
+			if (!ptr)
 				throw std::bad_alloc();
 			base.reset(static_cast<char*>(ptr));
 			bump = base.get();
@@ -103,7 +115,13 @@ namespace h5{ namespace impl {
 	private:
 		[[nodiscard]] char* allocate_fallback(size_t size) {
 			void* ptr = nullptr;
+#ifdef _WIN32
+			ptr = _aligned_malloc(size, alignment);
+#else
 			if (posix_memalign(&ptr, alignment, size) != 0)
+				ptr = nullptr;
+#endif
+			if (!ptr)
 				throw std::bad_alloc();
 			return static_cast<char*>(ptr);
 		}
@@ -374,8 +392,10 @@ template< class Derived>
 	if (rank == 1 && (O[0] % B[0]) == 0) [[likely]] {
 		constexpr hsize_t prefetch_distance = 4;
 		for (hsize_t j = 0; j < N[0]; j += B[0]) {
+#ifndef _WIN32
 			if (j + (prefetch_distance + 1) * B[0] < N[0])
 				__builtin_prefetch(ptr + (j + prefetch_distance * B[0]) * element_size, 0, 3);
+#endif
 			hsize_t bytes_in_chunk = (j + B[0] <= N[0]) ? B[0] : (N[0] - j);
 			hsize_t bytes_to_copy = bytes_in_chunk * element_size;
 			if (bytes_to_copy < block_size) [[unlikely]]
diff --git a/h5cpp/H5Zpipeline_pool.hpp b/h5cpp/H5Zpipeline_pool.hpp
@@ -40,6 +40,7 @@ struct pool_pipeline_t : public pipeline_t<pool_pipeline_t> {
         // pipeline (and its shared_ptr to the pool) goes away.  Blocks the
         // destructor on each front future.
         while (!in_flight_.empty()) drain_in_flight(/*blocking=*/true);
+        while (!read_in_flight_.empty()) drain_in_flight_read(nullptr, /*blocking=*/true);
     }
 
     pool_pipeline_t(const pool_pipeline_t&) = delete;
@@ -52,11 +53,19 @@ struct pool_pipeline_t : public pipeline_t<pool_pipeline_t> {
     void write_chunk_impl(const hsize_t* offset, std::size_t nbytes, const void* src);
     void read_chunk_impl (const hsize_t* offset, std::size_t nbytes, void*       dst);
 
+    // Parallel read override — shadows pipeline_t<pool_pipeline_t>::read.
+    // Rank-1 chunked datasets are decompressed across the worker pool;
+    // rank>1 falls back to the synchronous base-class path.
+    void read(const h5::ds_t& ds, const h5::offset_t& offset,
+              const h5::stride_t& stride, const h5::block_t& block,
+              const h5::count_t& count, const h5::dxpl_t& dxpl, void* ptr);
+
     // Public quiesce — block until every in-flight future has been
     // drained.  Called by pt_t::flush so users get the expected
     // "data on disk after flush returns" semantics for the pool path.
     void drain() {
         while (!in_flight_.empty()) drain_in_flight(/*blocking=*/true);
+        while (!read_in_flight_.empty()) drain_in_flight_read(nullptr, /*blocking=*/true);
     }
 
 private:
@@ -72,9 +81,20 @@ struct pool_pipeline_t : public pipeline_t<pool_pipeline_t> {
 
     void drain_in_flight(bool blocking);
 
+    // Result of a worker decompression task for the parallel read path.
+    struct read_result_t {
+        std::unique_ptr<std::byte[]> data;
+        std::size_t                   copy_size{0};
+        std::size_t                   dst_offset{0};
+    };
+
+    void drain_in_flight_read(char* dst, bool blocking);
+
     std::shared_ptr<worker_pool_t>       pool_;
     unsigned                              cap_{0};
     std::deque<std::future<result_t>>     in_flight_;
+    std::deque<std::future<read_result_t>> read_in_flight_;
+    char*                                 read_dst_ = nullptr;
 };
 
 // ─── write path ──────────────────────────────────────────────────────────────
@@ -202,13 +222,144 @@ inline void pool_pipeline_t::write_chunk_impl(const hsize_t* offset_in,
         drain_in_flight(/*blocking=*/true);
 }
 
-// ─── read path (synchronous for v1) ──────────────────────────────────────────
+// ─── read path ───────────────────────────────────────────────────────────────
 //
-// Phase 1.3.3 keeps read synchronous, identical to basic_pipeline_t::read_chunk_impl.
-// Parallel decompression is a deliberate follow-up: it requires read-ahead
-// (the read path consumes chunks in order, but the user's buffer slots are
-// known up-front, so prefetching can fan out across the pool).  Tracked in
-// Phase 1.5+ work.
+// Phase 1.3.3 kept read synchronous.  Phase 1.5 adds parallel decompression
+// for rank-1 chunked datasets by running the reverse filter chain on the
+// worker pool while the calling thread performs sequential H5Dread_chunk I/O.
+
+inline void pool_pipeline_t::drain_in_flight_read(char* dst, bool blocking) {
+    using namespace std::chrono_literals;
+    char* base = dst ? dst : read_dst_;
+    if (!base) return;  // no destination available (shouldn't happen outside read())
+
+    if (blocking) {
+        if (read_in_flight_.empty()) return;
+        auto r = read_in_flight_.front().get();
+        std::memcpy(base + r.dst_offset, r.data.get(), r.copy_size);
+        read_in_flight_.pop_front();
+        return;
+    }
+    while (!read_in_flight_.empty() &&
+           read_in_flight_.front().wait_for(0s) == std::future_status::ready) {
+        auto r = read_in_flight_.front().get();
+        std::memcpy(base + r.dst_offset, r.data.get(), r.copy_size);
+        read_in_flight_.pop_front();
+    }
+}
+
+inline void pool_pipeline_t::read(const h5::ds_t& ds, const h5::offset_t& offset,
+                                  const h5::stride_t& stride, const h5::block_t& block,
+                                  const h5::count_t& count, const h5::dxpl_t& dxpl, void* ptr) {
+    // Fallback to synchronous base-class path when no pool is present or when
+    // the dataset is not rank-1.  Rank>1 parallel decompression is future work.
+    if (!pool_ || rank != 1) {
+        pipeline_t<pool_pipeline_t>::read(ds, offset, stride, block, count, dxpl, ptr);
+        return;
+    }
+
+    this->dxpl = dxpl;
+    this->ds   = ds;
+
+    h5::offset_t offset_;
+    h5::count_t  count_;
+    for (hsize_t i = 0; i < rank; ++i) {
+        offset_[i] = offset[i];
+        count_[i]  = count[i] * block[i];
+    }
+
+    // Rank-1 chunk decomposition (mirrors split_to_chunk_read).
+    hsize_t n  = count_[0];          // total elements to read
+    hsize_t b  = B[0];               // elements per chunk
+    hsize_t ry = n % b;              // trailing partial chunk
+
+    // Snapshot filter chain into a POD closure capture (same pattern as write).
+    struct filter_chain_t {
+        filter::call_t filter[H5CPP_MAX_FILTER];
+        unsigned       flags[H5CPP_MAX_FILTER];
+        std::size_t    cd_size[H5CPP_MAX_FILTER];
+        unsigned       cd_values[H5CPP_MAX_FILTER][H5CPP_MAX_FILTER_PARAM];
+        hsize_t        tail;
+    };
+    filter_chain_t fc{};
+    fc.tail = this->tail;
+    std::memcpy(fc.filter,    this->filter,    sizeof(fc.filter));
+    std::memcpy(fc.flags,     this->flags,     sizeof(fc.flags));
+    std::memcpy(fc.cd_size,   this->cd_size,   sizeof(fc.cd_size));
+    std::memcpy(fc.cd_values, this->cd_values, sizeof(fc.cd_values));
+
+    const std::size_t scratch = filter::filter_scratch_bound(block_size);
+    char* dst = static_cast<char*>(ptr);
+    read_dst_ = dst;
+    const std::size_t block_sz = this->block_size;
+
+    for (hsize_t j = 0; j < n; j += b) {
+        hsize_t chunk_offset = j + offset_[0];
+        hsize_t copy_size = (ry != 0 && j == n - ry) ? ry * element_size : b * element_size;
+        hsize_t dst_offset = j * element_size;
+
+        // Main thread performs HDF5 I/O (H5Dread_chunk is not thread-safe).
+        auto compressed = std::make_unique<std::byte[]>(scratch);
+        std::uint32_t filter_mask = 0;
+        hsize_t C[1] = {chunk_offset};
+#if H5_VERSION_GE(2,0,0)
+        std::size_t buf_size = scratch;
+        H5Dread_chunk2(static_cast<::hid_t>(ds), static_cast<::hid_t>(dxpl),
+                       C, &filter_mask, compressed.get(), &buf_size);
+#else
+        H5Dread_chunk(static_cast<::hid_t>(ds), static_cast<::hid_t>(dxpl),
+                      C, &filter_mask, compressed.get());
+#endif
+
+        // Submit decompression to the worker pool.
+        auto fut = pool_->submit([
+            compressed = std::move(compressed), block_sz, copy_size,
+            dst_offset, scratch, fc
+        ]() mutable -> read_result_t {
+            read_result_t result;
+            result.copy_size  = copy_size;
+            result.dst_offset = dst_offset;
+
+            if (fc.tail == 0) {
+                result.data = std::make_unique<std::byte[]>(copy_size);
+                std::memcpy(result.data.get(), compressed.get(), copy_size);
+                return result;
+            }
+
+            auto work_buf = std::make_unique<std::byte[]>(scratch);
+            void* src = compressed.get();
+            void* dst_buf = work_buf.get();
+            std::size_t length = block_sz;
+
+            for (hsize_t fi = fc.tail; fi > 0; --fi) {
+                const hsize_t idx = fi - 1;
+                length = fc.filter[idx](dst_buf, src, length,
+                                        fc.flags[idx] | H5Z_FLAG_REVERSE,
+                                        fc.cd_size[idx], fc.cd_values[idx]);
+                std::swap(src, dst_buf);
+            }
+
+            result.data = std::make_unique<std::byte[]>(copy_size);
+            std::memcpy(result.data.get(), src, copy_size);
+            return result;
+        });
+
+        read_in_flight_.push_back(std::move(fut));
+
+        // Opportunistic drain — if front future is ready, copy and free buffer.
+        drain_in_flight_read(dst, /*blocking=*/false);
+
+        // Bounded back-pressure: block when the deque reaches the cap.
+        while (read_in_flight_.size() >= cap_)
+            drain_in_flight_read(dst, /*blocking=*/true);
+    }
+
+    // Drain any remaining decompression tasks.
+    while (!read_in_flight_.empty())
+        drain_in_flight_read(dst, /*blocking=*/true);
+
+    read_dst_ = nullptr;
+}
 
 inline void pool_pipeline_t::read_chunk_impl(const hsize_t* offset_in,
                                               std::size_t nbytes,
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
@@ -98,6 +98,7 @@ add_test_case(H5Zfletcher32)
 add_test_case(H5Zlz4)
 add_test_case(H5Zzstd)
 add_test_case(H5Zszip)
+add_test_case(H5Zpipeline_parallel_read)
 add_test_case(H5Zgorilla)
 # add_test_case(H5Iall)
 add_test_case(H5async)        # Phase II scaffold lifecycle + submit_and_wait tests
diff --git a/test/H5Zpipeline_parallel_read.cpp b/test/H5Zpipeline_parallel_read.cpp
