diff --git a/reader/HDF5ImageLocator.cpp b/reader/HDF5ImageLocator.cpp index df62d7e2..1f912869 100644 --- a/reader/HDF5ImageLocator.cpp +++ b/reader/HDF5ImageLocator.cpp @@ -63,7 +63,8 @@ HDF5ImageLocator::Location HDF5ImageLocator::Resolve(int64_t global_image) const if (layout_.format == FileWriterFormat::NXmxLegacy) { const uint32_t file_id = global_image / layout_.images_per_file; const uint32_t local_index = global_image % layout_.images_per_file; - return {OpenCached(layout_.legacy_files.at(file_id)), local_index}; + const auto &path = layout_.legacy_files.at(file_id); + return {OpenCached(path), local_index, path}; } if (layout_.format == FileWriterFormat::NXmxVDS @@ -72,7 +73,8 @@ HDF5ImageLocator::Location HDF5ImageLocator::Resolve(int64_t global_image) const for (const auto &mapping: layout_.vds_mappings) { if (!mapping.ContainsVirtualImage(image)) continue; - return {OpenCached(mapping.filename), static_cast(mapping.SourceImage(image))}; + return {OpenCached(mapping.filename), static_cast(mapping.SourceImage(image)), + mapping.filename}; } throw JFJochException(JFJochExceptionCategory::HDF5, "Image not covered by /entry/data/data VDS mappings"); @@ -81,7 +83,7 @@ HDF5ImageLocator::Location HDF5ImageLocator::Resolve(int64_t global_image) const // Contiguous / integrated: pixels live in the master file at the global index. if (!layout_.master_file) throw JFJochException(JFJochExceptionCategory::HDF5, "Master file not loaded"); - return {layout_.master_file, static_cast(global_image)}; + return {layout_.master_file, static_cast(global_image), layout_.master_filename}; } std::vector HDF5ImageLocator::GetSourceMapping(uint64_t first_image, diff --git a/reader/HDF5ImageLocator.h b/reader/HDF5ImageLocator.h index c91e5a69..7b13cc06 100644 --- a/reader/HDF5ImageLocator.h +++ b/reader/HDF5ImageLocator.h @@ -25,6 +25,9 @@ public: struct Location { std::shared_ptr file; uint32_t local_index = 0; + // Path the file was opened from. Needed to open it a second time as a plain file, for the + // positional reads HDF5ImageSource does outside the mutex. + std::string path; }; // Layout description, filled by the reader once the master file has been parsed. All paths diff --git a/reader/HDF5ImageSource.cpp b/reader/HDF5ImageSource.cpp index fe624134..ce360209 100644 --- a/reader/HDF5ImageSource.cpp +++ b/reader/HDF5ImageSource.cpp @@ -4,11 +4,64 @@ #include "HDF5ImageSource.h" #include "../common/JFJochException.h" +#ifdef _WIN32 +#include +#else +#include +#include +#endif + +// Positional reads: pread() on POSIX, ReadFile() with an OVERLAPPED offset on Windows. Both take the +// offset as an argument instead of moving a shared file position, which is what lets every worker +// thread read through one handle at the same time. +HDF5ImageSource::RawFile::RawFile(const std::string &path) { +#ifdef _WIN32 + HANDLE h = CreateFileA(path.c_str(), GENERIC_READ, FILE_SHARE_READ | FILE_SHARE_WRITE, nullptr, + OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, nullptr); + handle_ = (h == INVALID_HANDLE_VALUE) ? -1 : reinterpret_cast(h); +#else + handle_ = ::open(path.c_str(), O_RDONLY); +#endif +} + +HDF5ImageSource::RawFile::~RawFile() { + if (handle_ == -1) + return; +#ifdef _WIN32 + CloseHandle(reinterpret_cast(handle_)); +#else + ::close(static_cast(handle_)); +#endif +} + +void HDF5ImageSource::RawFile::ReadAt(void *dst, size_t size, uint64_t address) const { + auto *out = static_cast(dst); + size_t done = 0; + while (done < size) { +#ifdef _WIN32 + OVERLAPPED ov{}; + ov.Offset = static_cast((address + done) & 0xFFFFFFFFULL); + ov.OffsetHigh = static_cast((address + done) >> 32); + DWORD got = 0; + const bool ok = ReadFile(reinterpret_cast(handle_), out + done, + static_cast(size - done), &got, &ov); + const long long n = ok ? static_cast(got) : -1; +#else + const long long n = ::pread(static_cast(handle_), out + done, size - done, address + done); +#endif + if (n <= 0) + throw JFJochException(JFJochExceptionCategory::HDF5, "Error reading image chunk from file"); + done += static_cast(n); + } +} + void HDF5ImageSource::Configure(HDF5ImageLocator::Layout layout) { + dataset_cache_.clear(); locator_.Configure(std::move(layout)); } void HDF5ImageSource::Clear() { + dataset_cache_.clear(); locator_.Clear(); } @@ -23,45 +76,90 @@ std::vector HDF5ImageSource::GetSourceMapping(uint64_t fi return locator_.GetSourceMapping(first_image, image_count, total_images, stride); } -CompressedImage HDF5ImageSource::ReadImageAt(std::vector &buffer, - const HDF5ImageLocator::Location &loc) const { - return LoadImageDataset(buffer, *loc.file, loc.local_index); -} +const HDF5ImageSource::OpenDataset & +HDF5ImageSource::GetDataset(const HDF5ImageLocator::Location &loc) const { + if (auto it = dataset_cache_.find(loc.file.get()); it != dataset_cache_.end()) + return it->second; -CompressedImage HDF5ImageSource::LoadImageDataset(std::vector &tmp, HDF5Object &file, hsize_t number) { - std::vector start = {static_cast(number), 0, 0}; + OpenDataset entry; + entry.file = loc.file; + entry.dataset = std::make_unique(*loc.file, "/entry/data/data"); - const std::string dataset_name = "/entry/data/data"; - - HDF5DataSet dataset(file, dataset_name); - HDF5DataSpace dataspace(dataset); - HDF5DataType datatype(dataset); - HDF5Dcpl dcpl(dataset); + HDF5DataSpace dataspace(*entry.dataset); + HDF5DataType datatype(*entry.dataset); + HDF5Dcpl dcpl(*entry.dataset); if (dataspace.GetNumOfDimensions() != 3) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "/entry/data/data dataset must be 3D"); - auto dim = dataspace.GetDimensions(); - - CompressionAlgorithm algorithm = CompressionAlgorithm::NO_COMPRESSION; - auto chunk_size = dcpl.GetChunking(); - - if ((chunk_size.size() == 3) && (chunk_size[0] == 1) && (chunk_size[1] == dim[1]) && (chunk_size[2] == dim[2])) { - dataset.ReadDirectChunk(tmp, start); - algorithm = dcpl.GetCompression(); - } else { - dataset.ReadVectorToU8(tmp, start, {1, dim[1], dim[2]}); - algorithm = CompressionAlgorithm::NO_COMPRESSION; - } - if (datatype.IsFloat()) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Float datasets not supported at this time"); - return { - tmp, dim[2], dim[1], - CalcImageMode(datatype.GetElemSize(), datatype.IsFloat(), datatype.IsSigned()), - algorithm - }; + auto dim = dataspace.GetDimensions(); + entry.height = dim[1]; + entry.width = dim[2]; + entry.mode = CalcImageMode(datatype.GetElemSize(), datatype.IsFloat(), datatype.IsSigned()); + + auto chunk_size = dcpl.GetChunking(); + entry.direct_chunk = (chunk_size.size() == 3) && (chunk_size[0] == 1) + && (chunk_size[1] == dim[1]) && (chunk_size[2] == dim[2]); + if (entry.direct_chunk) + entry.algorithm = dcpl.GetCompression(); + + if (entry.direct_chunk && !loc.path.empty()) { + entry.raw = std::make_unique(loc.path); + if (!entry.raw->IsOpen()) + entry.raw.reset(); + hid_t fcpl = H5Fget_create_plist(loc.file->GetID()); + if (fcpl >= 0) { + hsize_t user_block = 0; + if (H5Pget_userblock(fcpl, &user_block) >= 0) + entry.user_block = user_block; + H5Pclose(fcpl); + } + } + + return dataset_cache_.emplace(loc.file.get(), std::move(entry)).first->second; +} + +std::optional +HDF5ImageSource::PrepareDirectRead(const HDF5ImageLocator::Location &loc) const { + const auto &ds = GetDataset(loc); + if (!ds.raw) + return {}; + + const hsize_t coord[3] = {static_cast(loc.local_index), 0, 0}; + unsigned filter_mask = 0; + haddr_t address = HADDR_UNDEF; + hsize_t size = 0; + if (H5Dget_chunk_info_by_coord(ds.dataset->GetID(), coord, &filter_mask, &address, &size) < 0) + return {}; + // A chunk nobody ever wrote has no address and no bytes; only HDF5 knows it reads as the fill + // value, so hand those back to it. + if (address == HADDR_UNDEF || size == 0) + return {}; + + return DirectChunk{ds.raw.get(), ds.user_block + address, static_cast(size), + ds.width, ds.height, ds.mode, ds.algorithm}; +} + +CompressedImage HDF5ImageSource::ReadDirect(std::vector &buffer, const DirectChunk &chunk) { + buffer.resize(chunk.size); + chunk.file->ReadAt(buffer.data(), chunk.size, chunk.address); + return {buffer, chunk.width, chunk.height, chunk.mode, chunk.algorithm}; +} + +CompressedImage HDF5ImageSource::ReadImageAt(std::vector &buffer, + const HDF5ImageLocator::Location &loc) const { + const auto &ds = GetDataset(loc); + const std::vector start = {static_cast(loc.local_index), 0, 0}; + + if (ds.direct_chunk) + ds.dataset->ReadDirectChunk(buffer, start); + else + ds.dataset->ReadVectorToU8(buffer, start, {1, ds.height, ds.width}); + + return {buffer, ds.width, ds.height, ds.mode, ds.algorithm}; } diff --git a/reader/HDF5ImageSource.h b/reader/HDF5ImageSource.h index 8b1a9176..2443a898 100644 --- a/reader/HDF5ImageSource.h +++ b/reader/HDF5ImageSource.h @@ -4,6 +4,8 @@ #pragma once #include +#include +#include #include #include @@ -16,6 +18,35 @@ // touches it. Caller must hold the global hdf5_mutex (HDF5 is not thread-safe). class HDF5ImageSource { public: + // Plain positional-read handle on a data file, opened alongside the HDF5 one. Owns the handle. + class RawFile { + public: + explicit RawFile(const std::string &path); + ~RawFile(); + RawFile(const RawFile &) = delete; + RawFile &operator=(const RawFile &) = delete; + + bool IsOpen() const { return handle_ != -1; } + // Read `size` bytes from byte `address`. Positional and stateless, so any number of threads + // may call it on the same handle at once. Throws on a short read. + void ReadAt(void *dst, size_t size, uint64_t address) const; + + private: + intptr_t handle_ = -1; // a file descriptor on POSIX, a HANDLE on Windows + }; + + // Where the bytes of one image are, and what they decode to. Everything needed to read an image + // without calling HDF5 again. + struct DirectChunk { + const RawFile *file = nullptr; + uint64_t address = 0; + uint32_t size = 0; + hsize_t width = 0; + hsize_t height = 0; + CompressedImageMode mode{}; + CompressionAlgorithm algorithm = CompressionAlgorithm::NO_COMPRESSION; + }; + void Configure(HDF5ImageLocator::Layout layout); void Clear(); @@ -26,6 +57,20 @@ public: // Read the pixels at a resolved location into a CompressedImage backed by `buffer`. CompressedImage ReadImageAt(std::vector &buffer, const HDF5ImageLocator::Location &loc) const; + // Ask HDF5 where image `loc` is in the file rather than asking it for the image. This is a + // lookup in the chunk index and nothing else - no read - so the mutex is held for a fraction of + // what an actual read costs, and the read itself then happens on any number of threads at once + // through ReadDirect(). Caller must hold hdf5_mutex. + // + // Empty when this file cannot be served that way: one chunk per image is what makes an image a + // single contiguous run of bytes, and a chunk that has never been written has no address at all. + // The caller falls back to ReadImageAt() then. + std::optional PrepareDirectRead(const HDF5ImageLocator::Location &loc) const; + + // Read what PrepareDirectRead() found. Touches no HDF5 and no shared state, so it needs no + // mutex; this is the whole point of the two-step split. + static CompressedImage ReadDirect(std::vector &buffer, const DirectChunk &chunk); + std::vector GetSourceMapping(uint64_t first_image, std::optional image_count, uint64_t total_images, @@ -33,5 +78,27 @@ public: private: HDF5ImageLocator locator_; - static CompressedImage LoadImageDataset(std::vector &tmp, HDF5Object &file, hsize_t number); + + // /entry/data/data and everything asked of it here - its rank and dimensions, its element type, + // its chunking, its compression - are properties of the file, identical for every image in it. + // They used to be looked up again for each image: four HDF5 object opens per frame, inside the + // global hdf5_mutex that every worker thread queues on. Resolve them once per file instead. + // + // The entry keeps the file alive, so the pointer it is keyed by cannot be recycled underneath it + // and the dataset handle cannot outlive the file it belongs to. + struct OpenDataset { + std::shared_ptr file; + std::unique_ptr dataset; + std::unique_ptr raw; + // HDF5 addresses count from the end of the user block, so they are file offsets only once + // its size is added. Zero for everything this project writes, but not for every file. + uint64_t user_block = 0; + hsize_t width = 0; + hsize_t height = 0; + CompressedImageMode mode{}; + CompressionAlgorithm algorithm = CompressionAlgorithm::NO_COMPRESSION; + bool direct_chunk = false; + }; + mutable std::map dataset_cache_; + const OpenDataset &GetDataset(const HDF5ImageLocator::Location &loc) const; }; diff --git a/reader/JFJochHDF5Reader.cpp b/reader/JFJochHDF5Reader.cpp index 953d130b..5750ccd4 100644 --- a/reader/JFJochHDF5Reader.cpp +++ b/reader/JFJochHDF5Reader.cpp @@ -57,15 +57,28 @@ HDF5ImageLocator::Location JFJochHDF5Reader::GetImageLocation(int64_t image_numb } std::shared_ptr JFJochHDF5Reader::GetRawImage(int64_t image_number) { - std::unique_lock ul(hdf5_mutex); - - if (!active_metadata_) - throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, - "Cannot load image if file not loaded"); - - auto loc = GetImageLocation(image_number); auto ret = std::make_shared(); - ret->image = image_source_.ReadImageAt(ret->image_buffer, loc); + + // Every worker thread of an offline run comes through here, and HDF5 lets only one of them in at + // a time. So ask HDF5 only where the image is - a chunk-index lookup - and read the bytes after + // dropping the lock, which is the part that takes any time and the part that parallelises. + std::optional chunk; + { + std::unique_lock ul(hdf5_mutex); + + if (!active_metadata_) + throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, + "Cannot load image if file not loaded"); + + auto loc = GetImageLocation(image_number); + chunk = image_source_.PrepareDirectRead(loc); + if (!chunk) { + ret->image = image_source_.ReadImageAt(ret->image_buffer, loc); + return ret; + } + } + + ret->image = HDF5ImageSource::ReadDirect(ret->image_buffer, *chunk); return ret; }