Give the process file its own thread

Writing an image to the process file takes the global HDF5 mutex, which is the
same one every worker needs to find its next image. The write is short - the
file holds the per-image analysis, not the pixels - but with a worker per
hardware thread they were still taking turns at it. The workers now post to a
bounded queue and one thread owns the file.

A DataMessage does not own its pixels, it points into the reader's buffer, so
the raw image is parked in the queue beside its message; without that the worker
frees the pixels on its next iteration and the writer reads whatever landed
there. The queue is bounded at four per worker so a run whose analysis outpaces
its writer cannot accumulate every image it has ever processed, and a write that
throws - out of space, above all - is held and rethrown when the loop drains it,
before the end message is written and the file finalized.

Worth 6.8 s -> 6.5 s on a 16 Mpx rotation dataset at 48 workers, on top of the
much larger gain from taking the read out of the same lock. Both process files,
written with and without the writer thread, re-scale to the same 101215 unique
reflections at the same ISa.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

rugnux/Rugnux.cpp

@@ -8,12 +8,15 @@
 #include <atomic>
 #include <chrono>
 #include <cmath>
+#include <condition_variable>
+#include <deque>
 #include <functional>
 #include <future>
 #include <mutex>
 #include <numeric>
 #include <set>
 #include <sstream>
+#include <thread>
 
 #include "../reader/JFJochHDF5Reader.h"
 #include "../common/JFJochMath.h"
@@ -162,6 +165,101 @@ namespace {
         return scaled;
     }
 
+    // Writing an image to the process file takes the global HDF5 mutex, which is the same mutex every
+    // worker thread needs to fetch its next image. The write itself is short, but with a worker per
+    // hardware thread they spend longer queueing for the lock than the writing takes: on a 16 Mpx
+    // rotation dataset the per-image loop runs 7.7 s with 8 workers and 13.1 s with 48. Give the file
+    // to one thread and let the workers post to it, and the lock has one taker again.
+    //
+    // A DataMessage does not own its pixels - it points into the reader's buffer - so the raw image is
+    // parked in the queue beside it. Without that the worker frees the pixels on its next iteration and
+    // the writer reads whatever landed there.
+    class ProcessFileWriter {
+    public:
+        ProcessFileWriter(FileWriter &writer, size_t capacity)
+            : writer_(writer), capacity_(capacity),
+              thread_([this] { Run(); }) {}
+
+        // Post one image. Blocks while the queue is full, which is what stops a run whose analysis
+        // outpaces its writer from holding every image it has ever processed in memory.
+        void Post(const DataMessage &msg, std::shared_ptr<JFJochReaderRawImage> img) {
+            std::unique_lock lock(m_);
+            space_.wait(lock, [this] { return queue_.size() < capacity_ || failed_; });
+            if (failed_)
+                return;   // the error is rethrown by Finish(); dropping the rest is deliberate
+            queue_.push_back({msg, std::move(img)});
+            work_.notify_one();
+        }
+
+        // Drain, join, and rethrow whatever the writer thread hit. Must be called before the
+        // FileWriter is used for anything else - the end message, Finalize().
+        void Finish() {
+            {
+                std::lock_guard lock(m_);
+                done_ = true;
+            }
+            work_.notify_one();
+            if (thread_.joinable())
+                thread_.join();
+            if (error_)
+                std::rethrow_exception(error_);
+        }
+
+        ~ProcessFileWriter() {
+            {
+                std::lock_guard lock(m_);
+                done_ = true;
+                failed_ = true;   // an unwinding run should not wait for the backlog to be written
+            }
+            work_.notify_one();
+            space_.notify_all();
+            if (thread_.joinable())
+                thread_.join();
+        }
+
+    private:
+        struct Job {
+            DataMessage msg;
+            std::shared_ptr<JFJochReaderRawImage> img;
+        };
+
+        void Run() {
+            while (true) {
+                Job job;
+                {
+                    std::unique_lock lock(m_);
+                    work_.wait(lock, [this] { return !queue_.empty() || done_; });
+                    if (queue_.empty())
+                        return;
+                    job = std::move(queue_.front());
+                    queue_.pop_front();
+                }
+                space_.notify_one();
+                try {
+                    writer_.Write(job.msg);
+                } catch (...) {
+                    std::lock_guard lock(m_);
+                    if (!error_)
+                        error_ = std::current_exception();
+                    failed_ = true;
+                    space_.notify_all();
+                    return;
+                }
+            }
+        }
+
+        FileWriter &writer_;
+        const size_t capacity_;
+        std::mutex m_;
+        std::condition_variable work_;
+        std::condition_variable space_;
+        std::deque<Job> queue_;
+        bool done_ = false;
+        bool failed_ = false;
+        std::exception_ptr error_;
+        std::thread thread_;
+    };
+
 }
 
 Rugnux::Rugnux(JFJochHDF5Reader &reader, DiffractionExperiment experiment,
@@ -1011,9 +1109,14 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b
         reader_.GetHDF5DataSource(start_image, images_to_process, config_.stride);
 
     std::unique_ptr<FileWriter> writer;
-    if (write_files && config_.write_process_h5)
+    std::unique_ptr<ProcessFileWriter> writer_queue;
+    if (write_files && config_.write_process_h5) {
         writer = std::make_unique<FileWriter>(start_message, /*check_overwrite_at_start=*/true,
                                               /*trusted_path=*/true);
+        // Deep enough that a worker never waits for the writer in the normal case, shallow enough that
+        // the backlog is bounded by the worker count rather than by the length of the run.
+        writer_queue = std::make_unique<ProcessFileWriter>(*writer, 4 * std::max(1, config_.nthreads));
+    }
 
     const char *mode_name = full ? "full analysis"
                                  : calibration ? "powder calibration" : "azimuthal integration";
@@ -1477,7 +1580,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b
             msg.run_name = experiment_.GetRunName();
 
             plots.Add(msg, profile);
-            if (writer) writer->Write(msg);
+            if (writer_queue) writer_queue->Post(msg, img);
             note_written(ordinal);
             if (observer) observer->OnImageProcessed(msg);
             const int done = finished_count.fetch_add(1) + 1;
@@ -1531,7 +1634,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b
             }
 
             plots.Add(msg, profile);
-            if (writer) writer->Write(msg);
+            if (writer_queue) writer_queue->Post(msg, img);
             note_written(ordinal);
             if (observer) observer->OnImageProcessed(msg);
             const int done = finished_count.fetch_add(1) + 1;
@@ -1551,6 +1654,11 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b
         futures.push_back(std::async(std::launch::async, worker));
     for (auto &f: futures)
         f.get();
+    // Everything the workers posted has to be on disk before the end message is written and the file
+    // finalized, and a write that failed - out of space, above all - has to surface here rather than
+    // leave a truncated file behind.
+    if (writer_queue)
+        writer_queue->Finish();
     // Wall time of the per-image loop alone. The per-stage means below cover only what runs inside it,
     // so without this there is nothing to compare them against and no way to see how much of a run is
     // spent outside it - on the first-pass indexing and on scaling/merging.