From 4bac6d3f21b90a8b6b690e495e8a1ebe522fcd9d Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Wed, 9 Sep 2026 16:26:52 +0200 Subject: [PATCH] reader: a worker reads every frame into the same raw image GetRawImage() allocated a fresh JFJochReaderRawImage per frame, and on a miniCBF sweep its buffer holds the whole decoded image: 72.6 MB on a 4150 x 4371 detector. That is over glibc's 32 MB mmap ceiling, so the block is mmapped and munmapped every image and the decoder faults in 17 700 untouched pages as it writes them. Measured on one 18.1 Mpx frame, decoding into a fresh buffer takes 61 ms against 16 ms into one that has been written before - the allocation costs nearly three times what the decode does, and the per-image loop pays it twice per sweep. ReadRawImage() fills a raw image the caller owns, so a worker declares one outside its loop and keeps the pages. GetRawImage() stays as the allocating wrapper for the single-image callers. The two loops that hand the image to the process-file writer keep a shared_ptr and recycle it only while the writer thread is not still reading the previous frame's pixels. Co-Authored-By: Claude Opus 5 (1M context) --- reader/JFJochCBFReader.cpp | 7 ++-- reader/JFJochCBFReader.h | 2 +- reader/JFJochHDF5Reader.cpp | 12 +++---- reader/JFJochHDF5Reader.h | 2 +- reader/JFJochReader.cpp | 7 ++++ reader/JFJochReader.h | 9 ++++- rugnux/Rugnux.cpp | 67 +++++++++++++++++++++++-------------- viewer/JFJochHttpReader.cpp | 23 ++++++------- viewer/JFJochHttpReader.h | 2 +- 9 files changed, 78 insertions(+), 53 deletions(-) diff --git a/reader/JFJochCBFReader.cpp b/reader/JFJochCBFReader.cpp index 6cc40182f..3b35f4718 100644 --- a/reader/JFJochCBFReader.cpp +++ b/reader/JFJochCBFReader.cpp @@ -338,10 +338,9 @@ bool JFJochCBFReader::LoadImage_i(std::shared_ptr &dataset, return true; } -std::shared_ptr JFJochCBFReader::GetRawImage(int64_t image_number) { - auto ret = std::make_shared(); - ret->image = DecodeInto(image_number, ret->image_buffer); - return ret; +bool JFJochCBFReader::ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) { + image.image = DecodeInto(image_number, image.image_buffer); + return true; } std::vector JFJochCBFReader::ReadSpots(int64_t) const { diff --git a/reader/JFJochCBFReader.h b/reader/JFJochCBFReader.h index b153493da..b57c73454 100644 --- a/reader/JFJochCBFReader.h +++ b/reader/JFJochCBFReader.h @@ -49,6 +49,6 @@ public: [[nodiscard]] uint64_t GetNumberOfImages() const override; void Close() override; - std::shared_ptr GetRawImage(int64_t image_number) override; + bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) override; [[nodiscard]] std::vector ReadSpots(int64_t image) const override; }; diff --git a/reader/JFJochHDF5Reader.cpp b/reader/JFJochHDF5Reader.cpp index 4f41d9be0..4009a8cdf 100644 --- a/reader/JFJochHDF5Reader.cpp +++ b/reader/JFJochHDF5Reader.cpp @@ -56,9 +56,7 @@ HDF5ImageLocator::Location JFJochHDF5Reader::GetImageLocation(int64_t image_numb return image_source_.Resolve(image_number); } -std::shared_ptr JFJochHDF5Reader::GetRawImage(int64_t image_number) { - auto ret = std::make_shared(); - +bool JFJochHDF5Reader::ReadRawImage(int64_t image_number, JFJochReaderRawImage &ret) { // 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. @@ -73,13 +71,13 @@ std::shared_ptr JFJochHDF5Reader::GetRawImage(int64_t imag 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 = image_source_.ReadImageAt(ret.image_buffer, loc); + return true; } } - ret->image = HDF5ImageSource::ReadDirect(ret->image_buffer, *chunk); - return ret; + ret.image = HDF5ImageSource::ReadDirect(ret.image_buffer, *chunk); + return true; } bool JFJochHDF5Reader::LoadImage_i(std::shared_ptr &dataset, diff --git a/reader/JFJochHDF5Reader.h b/reader/JFJochHDF5Reader.h index b18a45d07..d281c0b10 100644 --- a/reader/JFJochHDF5Reader.h +++ b/reader/JFJochHDF5Reader.h @@ -66,7 +66,7 @@ public: CompressedImage ReadCalibration(std::vector &tmp, const std::string &name) const; - std::shared_ptr GetRawImage(int64_t image_number) override; + bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) override; // Metadata snapshots over the same images. The original file is registered as "Original" on // ReadFile and is the initial active snapshot; reprocessing results can be added later. diff --git a/reader/JFJochReader.cpp b/reader/JFJochReader.cpp index 11da4bd96..e8912ce0a 100644 --- a/reader/JFJochReader.cpp +++ b/reader/JFJochReader.cpp @@ -57,6 +57,13 @@ std::shared_ptr JFJochReader::LoadImage(int64_t image_number, return {}; } +std::shared_ptr JFJochReader::GetRawImage(int64_t image_number) { + auto ret = std::make_shared(); + if (!ReadRawImage(image_number, *ret)) + return {}; + return ret; +} + void JFJochReader::SetStartMessage(const std::shared_ptr &val) { std::unique_lock ul(m); dataset = val; diff --git a/reader/JFJochReader.h b/reader/JFJochReader.h index 39de784d6..31a5b59d8 100644 --- a/reader/JFJochReader.h +++ b/reader/JFJochReader.h @@ -49,7 +49,14 @@ public: void UpdateGeomMetadata(const DiffractionExperiment& experiment); void UpdateUserMask(const std::vector& mask); - virtual std::shared_ptr GetRawImage(int64_t image_number) = 0; + // Read one image into a raw image the caller owns, false where there is nothing to read. A + // worker that walks a whole sweep hands the same object back on every frame and so keeps its + // buffer: GetRawImage() allocates a fresh one per call, which on a 4-byte 16 Mpx frame is 72 MB + // of untouched pages the decoder then faults in one by one, measurably more than the decode. + virtual bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) = 0; + + // The same image in a freshly allocated raw image. + std::shared_ptr GetRawImage(int64_t image_number); virtual std::vector ReadSpots(int64_t image) const = 0; std::shared_ptr ReadAllSpots(int64_t start_image, int64_t end_image, int64_t stride = 1) const; diff --git a/rugnux/Rugnux.cpp b/rugnux/Rugnux.cpp index e0b91a704..b53322e9f 100644 --- a/rugnux/Rugnux.cpp +++ b/rugnux/Rugnux.cpp @@ -697,6 +697,7 @@ std::optional Rugnux::RebinAndRefit(const CalibrationResult & for (int t = 0; t < nthreads; ++t) { workers.emplace_back(std::async(std::launch::async, [&]() { std::vector decompression_buffer; + JFJochReaderRawImage img; ImagePreprocessorCPU preprocessor(experiment_, pixel_mask_); ImagePreprocessorBuffer buffer(experiment_.GetPixelsNum()); AzIntEngineCPU azint(rebinned_mapping); @@ -707,10 +708,9 @@ std::optional Rugnux::RebinAndRefit(const CalibrationResult & const int image_idx = start_image + ordinal * config_.stride; if (image_idx >= end_image) break; try { - auto img = reader_.GetRawImage(image_idx); - if (!img) continue; - const uint8_t *image_ptr = img->image.GetUncompressedPtr(decompression_buffer); - preprocessor.Analyze(buffer, image_ptr, img->image.GetMode()); + if (!reader_.ReadRawImage(image_idx, img)) continue; + const uint8_t *image_ptr = img.image.GetUncompressedPtr(decompression_buffer); + preprocessor.Analyze(buffer, image_ptr, img.image.GetMode()); azint.Run(buffer, local); } catch (const std::exception &e) { if (IsFatalResourceError(e)) throw; @@ -858,6 +858,7 @@ void Rugnux::PreScan(int start_image, int images_to_process, int frame_count, Ru struct PreScanWorker { std::vector shadow_buffer; std::vector decompression_buffer; + JFJochReaderRawImage raw_image; std::unique_ptr preprocessor; std::unique_ptr preprocessed; std::unique_ptr spot_finder; @@ -991,19 +992,19 @@ void Rugnux::PreScan(int start_image, int images_to_process, int frame_count, Ru const size_t i = visit[v]; const int ordinal = ordinals[i]; const int image_idx = start_image + ordinal * config_.stride; - std::shared_ptr img; + bool read = false; try { - img = reader_.GetRawImage(image_idx); + read = reader_.ReadRawImage(image_idx, w.raw_image); } catch (const std::exception &e) { if (IsFatalResourceError(e)) throw; logger.Warning("Pre-scan: failed to load image {}: {}", image_idx, e.what()); continue; } - if (!img) continue; + if (!read) continue; DataMessage msg{}; - msg.image = img->image; + msg.image = w.raw_image.image; msg.number = ordinal; msg.original_number = image_idx; frames_read.fetch_add(1, std::memory_order_relaxed); @@ -1254,19 +1255,19 @@ void Rugnux::PreScan(int start_image, int images_to_process, int frame_count, Ru PreScanWorker w = make_worker(); for (size_t i = next.fetch_add(1); i < extra.size(); i = next.fetch_add(1)) { const int image_idx = start_image + extra[i] * config_.stride; - std::shared_ptr img; + bool read = false; try { - img = reader_.GetRawImage(image_idx); + read = reader_.ReadRawImage(image_idx, w.raw_image); } catch (const std::exception &e) { if (IsFatalResourceError(e)) throw; logger.Warning("Pre-scan: failed to load image {}: {}", image_idx, e.what()); continue; } - if (img) { + if (read) { extra_read[i] = 1; // The extra frames go to the beam centre alone; the width has already // been measured on the projection's own sample. - find_spots(w, img->image, image_idx, true, extra_spots[i], + find_spots(w, w.raw_image.image, image_idx, true, extra_spots[i], false, width_curves); } } @@ -1414,6 +1415,7 @@ void Rugnux::RefineStillsGeometry(int start_image, int end_image, int images_to_ MXAnalysisWithoutFPGA analysis(experiment_, mapping, pixel_mask_, indexer, /*enable_fused_adaptive_gpu=*/true); AzimuthalIntegrationProfile profile(mapping); + JFJochReaderRawImage img; for (int stripe = next_stripe.fetch_add(1); stripe < STRIPES && !cancelled_; stripe = next_stripe.fetch_add(1)) { @@ -1424,18 +1426,18 @@ void Rugnux::RefineStillsGeometry(int start_image, int end_image, int images_to_ const int ordinal = sample[idx]; const int image_idx = start_image + ordinal * config_.stride; - std::shared_ptr img; + bool read = false; try { - img = reader_.GetRawImage(image_idx); + read = reader_.ReadRawImage(image_idx, img); } catch (const std::exception &e) { if (IsFatalResourceError(e)) throw; logger.Warning("Geometry refinement: failed to load image {}: {}", image_idx, e.what()); continue; } - if (!img) continue; + if (!read) continue; DataMessage msg{}; - msg.image = img->image; + msg.image = img.image; msg.number = ordinal; msg.original_number = image_idx; if (dataset->efficiency.size() > image_idx) @@ -2242,18 +2244,18 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b // the next, which is why the main image loop can give one to each of its workers - so it does // not matter which engine, or which thread, finds them. const auto find_spots = [&](MXAnalysisWithoutFPGA &analysis, AzimuthalIntegrationProfile &profile, - int ordinal) { + JFJochReaderRawImage &img, int ordinal) { const int image_idx = start_image + ordinal * config_.stride; std::vector spots; try { - if (auto img = reader_.GetRawImage(image_idx)) { + if (reader_.ReadRawImage(image_idx, img)) { DataMessage m{}; m.number = ordinal; m.original_number = image_idx; auto first_pass = config_.spot_finding; first_pass.indexing = false; first_pass.quick_integration = false; - m.image = img->image; + m.image = img.image; if (dataset->efficiency.size() > image_idx) m.image_collection_efficiency = dataset->efficiency[image_idx]; analysis.Analyze(m, profile, first_pass); @@ -2286,6 +2288,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b struct SpotEngine { std::unique_ptr analysis; std::unique_ptr profile; + JFJochReaderRawImage raw_image; }; std::vector engines(std::max(spot_workers, 1)); const auto use_gpu_of_worker = [](size_t t) { @@ -2320,7 +2323,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b e.profile = std::make_unique(mapping); } for (size_t i = next.fetch_add(1); i < wanted.size(); i = next.fetch_add(1)) - found[i] = find_spots(*e.analysis, *e.profile, wanted[i]); + found[i] = find_spots(*e.analysis, *e.profile, e.raw_image, wanted[i]); })); for (auto &f : futures) f.get(); @@ -3403,6 +3406,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b auto azint_worker = [&]() { std::vector decompression_buffer; + std::shared_ptr img; ImagePreprocessorCPU preprocessor(experiment_, pixel_mask_); ImagePreprocessorBuffer buffer(experiment_.GetPixelsNum()); AzIntEngineCPU azint(mapping); @@ -3413,15 +3417,20 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b const int image_idx = start_image + ordinal * config_.stride; if (image_idx >= end_image) break; - std::shared_ptr img; + // Reuse the raw image frame to frame - a fresh 4-byte 16 Mpx frame is 72 MB of + // untouched pages the decoder faults in one by one - except while the writer thread + // still holds the previous one's pixels. + if (!img || img.use_count() > 1) + img = std::make_shared(); + bool read = false; try { - img = reader_.GetRawImage(image_idx); + read = reader_.ReadRawImage(image_idx, *img); } catch (const std::exception &e) { if (IsFatalResourceError(e)) throw; logger.Error("Failed to load image {}: {}", image_idx, e.what()); continue; } - if (!img) continue; + if (!read) continue; DataMessage msg{}; msg.image = img->image; @@ -3467,21 +3476,27 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b /*enable_fused_adaptive_gpu=*/true); AzimuthalIntegrationProfile profile(mapping); HarmonicEvidence worker_harmonic; + std::shared_ptr img; while (!cancelled_) { const int ordinal = next_ordinal.fetch_add(1); const int image_idx = start_image + ordinal * config_.stride; if (image_idx >= end_image) break; - std::shared_ptr img; + // Reuse the raw image frame to frame - a fresh 4-byte 16 Mpx frame is 72 MB of + // untouched pages the decoder faults in one by one - except while the writer thread + // still holds the previous one's pixels. + if (!img || img.use_count() > 1) + img = std::make_shared(); + bool read = false; try { - img = reader_.GetRawImage(image_idx); + read = reader_.ReadRawImage(image_idx, *img); } catch (const std::exception &e) { if (IsFatalResourceError(e)) throw; logger.Error("Failed to load image {}: {}", image_idx, e.what()); continue; } - if (!img) continue; + if (!read) continue; DataMessage msg{}; msg.image = img->image; diff --git a/viewer/JFJochHttpReader.cpp b/viewer/JFJochHttpReader.cpp index b4c6e0f98..31419cbf1 100644 --- a/viewer/JFJochHttpReader.cpp +++ b/viewer/JFJochHttpReader.cpp @@ -398,39 +398,38 @@ bool JFJochHttpReader::LoadImage_i(std::shared_ptr &dataset } } -std::shared_ptr JFJochHttpReader::GetRawImage(int64_t image_number) { +bool JFJochHttpReader::ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) { if (addr.empty()) - return {}; + return false; auto res = Request("GET", "/image_buffer/image.cbor?id=" + std::to_string(image_number)); if (!res.ok || res.status != 200 || res.body.empty()) - return {}; + return false; try { auto msg = CBORStream2Deserialize(reinterpret_cast(res.body.data()), res.body.size()); if (msg->msg_type != CBORImageType::IMAGE) - return {}; + return false; - std::shared_ptr image = std::make_shared(); - image->image_buffer.resize(msg->data_message->image.GetCompressedSize()); - memcpy(image->image_buffer.data(), + image.image_buffer.resize(msg->data_message->image.GetCompressedSize()); + memcpy(image.image_buffer.data(), msg->data_message->image.GetCompressed(), msg->data_message->image.GetCompressedSize()); - image->image = CompressedImage( - image->image_buffer.data(), - image->image_buffer.size(), + image.image = CompressedImage( + image.image_buffer.data(), + image.image_buffer.size(), msg->data_message->image.GetWidth(), msg->data_message->image.GetHeight(), msg->data_message->image.GetMode(), msg->data_message->image.GetCompressionAlgorithm() ); - return image; + return true; } catch (std::exception &e) { - return {}; + return false; } } diff --git a/viewer/JFJochHttpReader.h b/viewer/JFJochHttpReader.h index d973dc4c0..a50dadcd1 100644 --- a/viewer/JFJochHttpReader.h +++ b/viewer/JFJochHttpReader.h @@ -79,7 +79,7 @@ public: BrokerStatus GetBrokerStatus() const; - std::shared_ptr GetRawImage(int64_t image_number) override; + bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) override; std::vector ReadSpots(int64_t image) const override; };