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; };