XTC reader: frames handed up as stored, streams indexed in parallel

The main loop's read (ReadRawImage) now hands a camera frame up as the
file stores it - 16-bit pixels, UINT16_WIDENED, as the marCCD reader
does - wherever there is no pixel offset to take off and no gain to
divide by; the GPU preprocessor widens it on the device. The host no
longer widens 3.7 Mpx to int32 per frame (the loop that was ~79% of
the main pass's user cycles). xtc::ReadFrame becomes xtc::ReadPixels,
the stored pixels only; the int32 decode (LoadImage, or an offset or
gain) widens them in one plain loop.

The streams are indexed in parallel (one worker per stream, up to the
hardware threads), merged in the order given: each index is a walk of
small reads, waiting on the disk one at a time.

Measured on a Rayonix MX340 run, 1920x1920 (RTX 5080, 5950X, page
cache warm, other agents' builds running; output identical but for the
known run-to-run flip of one observation on one set, which the
baseline shows between its own runs):
  main pass, 2000 frames:  479 -> 789 frames/s (lcls222),
                           413 -> 477, 388 -> 452 (the two forms of
                           lcls242, there indexing-bound)
  Step 1, 48 streams/82k frames: 34-51 s -> 2.4 s with the headers
                           warm, 58.2 s -> 5.7 s evicted from cache

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015CeEGgpKdfkCTJb3YNMJcu

reader/JFJochXTCReader.cpp

@@ -4,9 +4,12 @@
 #include "JFJochXTCReader.h"
 
 #include <algorithm>
+#include <atomic>
 #include <cmath>
+#include <future>
 #include <map>
 #include <numeric>
+#include <thread>
 
 #include "../common/JFJochException.h"
 #include "../common/Logger.h"
@@ -42,8 +45,24 @@ void JFJochXTCReader::ReadFiles(const std::vector<std::string> &streams) {
     std::string model;
     uint32_t binning = 0;
     size_t damaged = 0;
-    for (const auto &stream: streams) {
-        auto index = xtc::IndexFile(stream);
+    // The streams are indexed in parallel: each is a walk of small reads from one place in the file to the
+    // next, which on a run not in the page cache waits on the disk one read at a time.
+    std::vector<xtc::Index> indices(streams.size());
+    {
+        std::atomic<size_t> next = 0;
+        const size_t nthreads = std::min<size_t>(streams.size(), std::max(1u, std::thread::hardware_concurrency()));
+        std::vector<std::future<void>> workers;
+        for (size_t t = 0; t < nthreads; t++)
+            workers.push_back(std::async(std::launch::async, [&] {
+                for (size_t i = next.fetch_add(1); i < streams.size(); i = next.fetch_add(1))
+                    indices[i] = xtc::IndexFile(streams[i]);
+            }));
+        for (auto &w: workers)
+            w.get();
+    }
+    for (size_t s = 0; s < streams.size(); s++) {
+        const auto &stream = streams[s];
+        const auto &index = indices[s];
         all.insert(all.end(), index.frames.begin(), index.frames.end());
         energies.insert(energies.end(), index.photon_energy_eV.begin(), index.photon_energy_eV.end());
         if (!index.rayonix_model.empty()) {
@@ -142,11 +161,17 @@ CompressedImage JFJochXTCReader::DecodeInto(int64_t image_number, Buffer &buffer
     if (image_number < 0 || static_cast<size_t>(image_number) >= frames_.size())
         throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Image number out of range");
     const auto &f = frames_[image_number];
-    buffer.resize(static_cast<size_t>(f.width) * f.height * sizeof(int32_t));
+    const size_t npixel = static_cast<size_t>(f.width) * f.height;
+    buffer.resize(npixel * sizeof(int32_t));
+    scratch.resize(npixel * sizeof(uint16_t));
+    xtc::ReadPixels(f, reinterpret_cast<uint16_t *>(scratch.data()));
+    const auto in = reinterpret_cast<const uint16_t *>(scratch.data());
     auto out = reinterpret_cast<int32_t *>(buffer.data());
-    xtc::ReadFrame(f, out, scratch);
+    const auto offset = static_cast<int32_t>(f.pixel_offset);
+    for (size_t i = 0; i < npixel; i++)
+        out[i] = static_cast<int32_t>(in[i]) - offset;
     if (adu_per_photon_ != 1.0f)
-        for (size_t i = 0; i < static_cast<size_t>(f.width) * f.height; i++)
+        for (size_t i = 0; i < npixel; i++)
             out[i] = static_cast<int32_t>(std::lround(static_cast<float>(out[i]) / adu_per_photon_));
     return CompressedImage(buffer.data(), buffer.size(), f.width, f.height,
                            CompressedImageMode::Int32, CompressionAlgorithm::NO_COMPRESSION);
@@ -169,6 +194,16 @@ bool JFJochXTCReader::LoadImage_i(std::shared_ptr<JFJochReaderDataset> &dataset,
 bool JFJochXTCReader::ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) {
     if (image_number < 0 || static_cast<size_t>(image_number) >= frames_.size())
         return false;
+    const auto &f = frames_[image_number];
+    if (f.pixel_offset == 0 && adu_per_photon_ == 1.0f) {
+        // The pixels as the file stores them, two bytes each, where there is nothing to take off or divide
+        // by: the consumer widens them, on the device where there is one.
+        image.image_buffer.resize(static_cast<size_t>(f.width) * f.height * sizeof(uint16_t));
+        xtc::ReadPixels(f, reinterpret_cast<uint16_t *>(image.image_buffer.data()));
+        image.image = CompressedImage(image.image_buffer.data(), image.image_buffer.size(), f.width, f.height,
+                                      CompressedImageMode::Int32, CompressionAlgorithm::UINT16_WIDENED);
+        return true;
+    }
     image.image = DecodeInto(image_number, image.image_buffer, image.read_buffer);
     return true;
 }