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) <noreply@anthropic.com>

reader/JFJochCBFReader.cpp

@@ -338,10 +338,9 @@ bool JFJochCBFReader::LoadImage_i(std::shared_ptr<JFJochReaderDataset> &dataset,
     return true;
 }
 
-std::shared_ptr<JFJochReaderRawImage> JFJochCBFReader::GetRawImage(int64_t image_number) {
-    auto ret = std::make_shared<JFJochReaderRawImage>();
-    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<SpotToSave> JFJochCBFReader::ReadSpots(int64_t) const {

reader/JFJochCBFReader.h

@@ -49,6 +49,6 @@ public:
     [[nodiscard]] uint64_t GetNumberOfImages() const override;
     void Close() override;
 
-    std::shared_ptr<JFJochReaderRawImage> GetRawImage(int64_t image_number) override;
+    bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) override;
     [[nodiscard]] std::vector<SpotToSave> ReadSpots(int64_t image) const override;
 };

reader/JFJochHDF5Reader.cpp

@@ -56,9 +56,7 @@ HDF5ImageLocator::Location JFJochHDF5Reader::GetImageLocation(int64_t image_numb
     return image_source_.Resolve(image_number);
 }
 
-std::shared_ptr<JFJochReaderRawImage> JFJochHDF5Reader::GetRawImage(int64_t image_number) {
-    auto ret = std::make_shared<JFJochReaderRawImage>();
-
+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<JFJochReaderRawImage> 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<JFJochReaderDataset> &dataset,

reader/JFJochHDF5Reader.h

@@ -66,7 +66,7 @@ public:
 
     CompressedImage ReadCalibration(std::vector<uint8_t> &tmp, const std::string &name) const;
 
-    std::shared_ptr<JFJochReaderRawImage> 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.

reader/JFJochReader.cpp

@@ -57,6 +57,13 @@ std::shared_ptr<JFJochReaderImage> JFJochReader::LoadImage(int64_t image_number,
     return {};
 }
 
+std::shared_ptr<JFJochReaderRawImage> JFJochReader::GetRawImage(int64_t image_number) {
+    auto ret = std::make_shared<JFJochReaderRawImage>();
+    if (!ReadRawImage(image_number, *ret))
+        return {};
+    return ret;
+}
+
 void JFJochReader::SetStartMessage(const std::shared_ptr<JFJochReaderDataset> &val) {
     std::unique_lock ul(m);
     dataset = val;

reader/JFJochReader.h

@@ -49,7 +49,14 @@ public:
     void UpdateGeomMetadata(const DiffractionExperiment& experiment);
     void UpdateUserMask(const std::vector<uint32_t>& mask);
 
-    virtual std::shared_ptr<JFJochReaderRawImage> 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<JFJochReaderRawImage> GetRawImage(int64_t image_number);
 
     virtual std::vector<SpotToSave> ReadSpots(int64_t image) const = 0;
     std::shared_ptr<JFJochReaderSpots> ReadAllSpots(int64_t start_image, int64_t end_image, int64_t stride = 1) const;

rugnux/Rugnux.cpp

@@ -697,6 +697,7 @@ std::optional<CalibrationResult> Rugnux::RebinAndRefit(const CalibrationResult &
         for (int t = 0; t < nthreads; ++t) {
             workers.emplace_back(std::async(std::launch::async, [&]() {
                 std::vector<uint8_t> decompression_buffer;
+                JFJochReaderRawImage img;
                 ImagePreprocessorCPU preprocessor(experiment_, pixel_mask_);
                 ImagePreprocessorBuffer buffer(experiment_.GetPixelsNum());
                 AzIntEngineCPU azint(rebinned_mapping);
@@ -707,10 +708,9 @@ std::optional<CalibrationResult> 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<uint8_t> shadow_buffer;
         std::vector<uint8_t> decompression_buffer;
+        JFJochReaderRawImage raw_image;
         std::unique_ptr<ImagePreprocessorCPU> preprocessor;
         std::unique_ptr<ImagePreprocessorBuffer> preprocessed;
         std::unique_ptr<ImageSpotFinder> 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<JFJochReaderRawImage> 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<JFJochReaderRawImage> 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<JFJochReaderRawImage> 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<SpotToSave> 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<MXAnalysisWithoutFPGA> analysis;
             std::unique_ptr<AzimuthalIntegrationProfile> profile;
+            JFJochReaderRawImage raw_image;
         };
         std::vector<SpotEngine> 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<AzimuthalIntegrationProfile>(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<uint8_t> decompression_buffer;
+        std::shared_ptr<JFJochReaderRawImage> 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<JFJochReaderRawImage> 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<JFJochReaderRawImage>();
+            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<JFJochReaderRawImage> 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<JFJochReaderRawImage> 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<JFJochReaderRawImage>();
+            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;

viewer/JFJochHttpReader.cpp

@@ -398,39 +398,38 @@ bool JFJochHttpReader::LoadImage_i(std::shared_ptr<JFJochReaderDataset> &dataset
     }
 }
 
-std::shared_ptr<JFJochReaderRawImage> 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<uint8_t *>(res.body.data()), res.body.size());
 
         if (msg->msg_type != CBORImageType::IMAGE)
-            return {};
+            return false;
 
-        std::shared_ptr<JFJochReaderRawImage> image = std::make_shared<JFJochReaderRawImage>();
-        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;
     }
 }
 

viewer/JFJochHttpReader.h

@@ -79,7 +79,7 @@ public:
 
     BrokerStatus GetBrokerStatus() const;
 
-    std::shared_ptr<JFJochReaderRawImage> GetRawImage(int64_t image_number) override;
+    bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) override;
     std::vector<SpotToSave> ReadSpots(int64_t image) const override;
 };