Receivers and pullers: comments to one line, names to the conventions of the 2025 code

Comments say what the next line does or why; names follow the rest of the receiver: <thing>_mutex, thread
bodies <What>Thread (FPGAImagePuller: AcquireThread, ModuleThread, CompressThread, FinalizeMeasurement),
no single letters for objects. Code does not change. JFJochReceiverFPGA and what the style review's "leave
alone" list names (deleters, exception_ptr, the condition variables, LiveModel) stay as they were.

Moved here from the comments:

Pre-pass size: of 100 frames of a collection indexed at 18%, 7 indexed at a distance 2.6% off and the
geometry refinement did not run; of 500, 34 did and refined it to 0.4%. With the broker's default fixed
threshold 3 points fewer frames index than with the adaptive one the analysis takes.

Preflight sends FillMessage alone, measured on a JUNGFRAU 9M at 1.4 kB against 543 kB for the start
message, and the whole of it rather than the fields the check reads today, so the writer sees the description
it will be given at start.

FFT indexers and cuFFT plans are kept for the life of the broker because every pass of every collection
builds the same pools, and making a plan takes a process-wide lock and took over a second on a card busy
with the online analysis.

The host memory budget subtracts the image buffer, which has to keep the whole collection and so is most of
the machine; at worst what could run side by side runs one part after the other.

The receiver is released before the indexer pool is rebuilt because a fresh GPU indexer next to a stale
receiver (after a failed start the receiver is stopped but not destroyed until the next Start) can fail
to initialise.

ReceiveRingBuffer: one image of a large detector is tens of megabytes, so a queue of a thousand of them is a
quantity of memory nobody chose. Regions are handed out and reclaimed in arrival order; a region released
early is only remembered, and republish holds at most its queue's worth, so that is as far as the reclaim
can lag. A message larger than the whole ring goes on the heap, so an oversized START or END cannot wedge
the receiver. Republish is capped at a share of the ring because it keeps the oldest unsent images and
would otherwise pin everything received after them, running the writer at the republish rate.

ZMQImagePuller: with a full ring the broker's pusher waits one second, then drops the image and sends without
waiting for the rest of the run, so a ring too small for the writer's stalls costs images. TCPImagePuller: a
full ring blocks the pusher's send; its BUSY heartbeats tell it apart from a dead writer, and it gives up
only after a minute in which no image was written. Keep-alive detection is about 60 s, both sides.

Stills model step: the step on a provisional merge still running when the next merge is made is cut, one not
started is replaced; a molecular replacement or a refinement not started is left out at the deadline, a
refinement running is cut short; each step starts from the last placement. Without a space group stated the
frames are merged in the model's where its cell is the run's consensus, otherwise the model is left out.

Rotation, stills and stills with a model end to end against the build before this work: MTZ data columns,
both maps, placed and refined models and blobs are byte for byte the same; only the MTZ history's build
stamp differs. Scan results are the same with times stripped.

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

image_puller/ImagePuller.h

@@ -33,14 +33,11 @@ struct RawFrame {
 };
 
 struct ImagePullerOutput {
-    // The message as it came off the socket, in the puller's ring buffer. It stays there for as long as
-    // any copy of this struct is alive - the deserialised image in cbor points into it, and republish
-    // sends it as is.
+    // The message as it came off the socket, in the ring buffer; held there as long as any copy of this struct lives
     std::shared_ptr<ReceiveBufferRegion> raw;
     std::shared_ptr<RawFrame> tcp_msg;
     std::shared_ptr<CBORStream2DeserializerOutput> cbor;
-    // When the message came off the socket - before it was queued or deserialised, so a consumer that
-    // falls behind does not shift it.
+    // When the message came off the socket, before it was queued or deserialised
     std::chrono::steady_clock::time_point received{};
 };
 
@@ -51,16 +48,10 @@ struct ImagePullerStatistics {
 class ImagePuller {
 protected:
     ThreadSafeFIFO<ImagePullerOutput> outside_fifo; // assuming 1 file is <1000 files, this should be enough to buffer
-    // Every received message is held here until consumed; the queues count messages, this bounds bytes.
-    // Size 0 means no ring: messages are held on the heap, bounded by the queues alone, as they always
-    // were - the broker's own DECTRIS puller runs that way, jfjoch_writer sets a size.
+    // Bounds the bytes of received messages until consumed; size 0: on the heap, bounded by the queues alone
     ReceiveRingBuffer ring;
 
-    // Bytes republish holds right now. Republish keeps the OLDEST not yet sent images, and the ring is
-    // reclaimed in arrival order, so an image waiting for a slow republish consumer pins everything
-    // received after it - the writer would then run at the republish rate. What republish may hold is
-    // therefore capped at a share of the ring; an image over the cap is not republished, exactly as one
-    // that finds the republish queue full. Without a ring there is no cap.
+    // Bytes republish holds now: capped at a share of the ring, as a slow republish consumer pins all received after
     std::atomic<size_t> repub_bytes{0};
     bool RepubCanHold(size_t bytes) const {
         return ring.GetCapacity() == 0 || repub_bytes.load() + bytes <= ring.GetCapacity() / 4;

image_puller/ReceiveRingBuffer.cpp

@@ -7,24 +7,21 @@
 #include <optional>
 
 struct ReceiveRingBuffer::State {
-    // Uninitialised on purpose: a ring that is never filled should not cost its capacity in
-    // resident memory.
+    // Uninitialised: a ring that is never filled should not cost its capacity in resident memory
     std::unique_ptr<uint8_t[]> memory;
     const size_t capacity;
 
-    std::mutex m;
+    std::mutex state_mutex;
     std::condition_variable cv;
 
-    // Live regions in arrival order. Only the oldest can give back space, so a region released out of
-    // turn is marked and waits.
+    // Live regions in arrival order; only the oldest can give back space
     struct Slot {
         size_t offset;
         size_t size;
         bool released = false;
     };
     std::deque<Slot> slots;
-    // Next byte to hand out. The ring is empty when slots is empty; the oldest live byte is then
-    // slots.front().offset.
+    // Next byte to hand out; the oldest live byte is slots.front().offset
     size_t tail = 0;
 
     size_t utilization = 0;
@@ -35,14 +32,12 @@ struct ReceiveRingBuffer::State {
             : memory(in_capacity > 0 ? std::make_unique_for_overwrite<uint8_t[]>(in_capacity) : nullptr),
               capacity(in_capacity) {}
 
-    // Where a region of `size` bytes would start, or nothing if it does not fit right now. Called
-    // under the mutex.
+    // Where a region of `size` bytes would start, nothing if it does not fit now; under the mutex
     std::optional<size_t> FindSpace(size_t size) const {
         if (slots.empty())
             return 0;
         const size_t head = slots.front().offset;
-        // tail has come round to head: treated as full. (After a wrap the piece between the old tail
-        // and the end is free, but it is behind the live data and not worth tracking.)
+        // tail has come round to head: treated as full
         if (tail == head)
             return std::nullopt;
         if (tail > head) {
@@ -60,7 +55,7 @@ struct ReceiveRingBuffer::State {
     }
 
     void Release(size_t offset) {
-        std::unique_lock ul(m);
+        std::unique_lock ul(state_mutex);
         for (auto &s: slots) {
             if (s.offset == offset) {
                 s.released = true;
@@ -77,7 +72,7 @@ struct ReceiveRingBuffer::State {
     }
 
     void ReleaseHeap(size_t size) {
-        std::unique_lock ul(m);
+        std::unique_lock ul(state_mutex);
         utilization -= size;
     }
 };
@@ -89,7 +84,7 @@ std::shared_ptr<ReceiveBufferRegion> ReceiveRingBuffer::Allocate(size_t size) {
     if (size == 0)
         return std::make_shared<ReceiveBufferRegion>();
 
-    std::unique_lock ul(state->m);
+    std::unique_lock ul(state->state_mutex);
 
     if (size > state->capacity) {
         // Heap fallback: counted, not bounded.
@@ -126,7 +121,7 @@ std::shared_ptr<ReceiveBufferRegion> ReceiveRingBuffer::Allocate(size_t size) {
 }
 
 void ReceiveRingBuffer::Stop() {
-    std::unique_lock ul(state->m);
+    std::unique_lock ul(state->state_mutex);
     state->stopped = true;
     state->cv.notify_all();
 }
@@ -136,11 +131,11 @@ size_t ReceiveRingBuffer::GetCapacity() const {
 }
 
 size_t ReceiveRingBuffer::GetUtilization() const {
-    std::unique_lock ul(state->m);
+    std::unique_lock ul(state->state_mutex);
     return state->utilization;
 }
 
 size_t ReceiveRingBuffer::GetMaxUtilization() const {
-    std::unique_lock ul(state->m);
+    std::unique_lock ul(state->state_mutex);
     return state->max_utilization;
 }

image_puller/ReceiveRingBuffer.h

@@ -9,37 +9,20 @@
 #include <memory>
 #include <mutex>
 
-// A received message lives here from the moment it comes off the socket until the last consumer is
-// done with it - the writer has stored the image and the republish thread has sent it. Both consumers
-// hold a shared_ptr to the region, and the region's destructor is what gives the bytes back.
+// A received message, from the socket until the last consumer lets go of the shared_ptr to it
 struct ReceiveBufferRegion {
     uint8_t *data = nullptr;
     size_t size = 0;
 };
 
-// Bounds the memory a puller holds in received, not yet consumed, messages - by bytes, which is the
-// quantity that matters, where the queues in front of and behind it count messages. One image of a
-// large detector is tens of megabytes, so a queue of a thousand of them is a quantity of memory nobody
-// chose.
-//
-// Regions are handed out in arrival order and the space is taken back in arrival order too, which is
-// what makes it a ring: the writer and the republish thread may finish with messages out of order, so
-// a region released early is only remembered, and the space is reclaimed once everything received
-// before it is released as well. Republish holds at most its queue's worth of messages, so that is as
-// far as the reclaim can lag.
-//
-// A message larger than the whole ring goes on the heap instead, so an oversized START or END message
-// cannot wedge the receiver; everything else waits for space, which is the back-pressure that keeps the
-// sender from running ahead.
+// Bounds the bytes a puller holds in received messages, handed out and taken back in arrival order
 class ReceiveRingBuffer {
     struct State;
     std::shared_ptr<State> state;
 public:
     explicit ReceiveRingBuffer(size_t capacity);
 
-    // Blocks until the bytes are available; returns nullptr once Stop() has been called. A request
-    // larger than the capacity (so every request of a ring of capacity 0) is served from the heap,
-    // at once and whether stopped or not.
+    // Blocks for the bytes; nullptr once stopped. A request larger than the capacity is served from the heap, at once
     std::shared_ptr<ReceiveBufferRegion> Allocate(size_t size);
 
     // Releases every waiter in Allocate; the regions already handed out stay valid.

image_puller/TCPImagePuller.cpp

@@ -134,8 +134,7 @@ void TCPImagePuller::CBORThread() {
             } else {
                 ret.cbor = CBORStream2Deserialize(ret.raw ? ret.raw->data : nullptr, ret.raw ? ret.raw->size : 0);
                 outside_fifo.PutBlocking(ret);
-                // A PREFLIGHT carries a StartMessage payload but starts nothing, so republishing it
-                // would announce a run to downstream consumers that is not going to happen.
+                // A PREFLIGHT starts nothing, so it is not republished
                 if (repub_socket && (type != TCPFrameType::PREFLIGHT)) {
                     if ((ret.cbor->msg_type == CBORImageType::START)
                         || (ret.cbor->msg_type == CBORImageType::END))
@@ -145,8 +144,7 @@ void TCPImagePuller::CBORThread() {
                 }
             }
         } catch (const std::exception &e) {
-            // std::exception: the deserialiser allocates a vector per per-image array of an END message,
-            // and a bad_alloc leaving this thread would end the process.
+            // std::exception: a bad_alloc in the deserialiser leaving this thread would end the process
             logger.ErrorException(e);
         }
         ret = cbor_fifo.GetBlocking();
@@ -236,10 +234,7 @@ bool TCPImagePuller::EnsureConnected() {
         if (receive_buffer_size)
             setsockopt(new_fd, SOL_SOCKET, SO_RCVBUF, &receive_buffer_size.value(), sizeof(int32_t));
 
-        // OS-level TCP keep-alive: detect a silently-dead pusher (no RST, e.g. crash or
-        // network partition) even while recv() is blocked in a long inter-image gap. Mirror
-        // the pusher's settings so detection is symmetric (~60s) and stays well clear of the
-        // 250 ms app-level BUSY heartbeat.
+        // Detects a silently dead pusher while recv() blocks; the pusher's settings, well clear of the 250 ms heartbeat
         int one = 1;
         setsockopt(new_fd, SOL_SOCKET, SO_KEEPALIVE, &one, sizeof(one));
         int idle = 30;
@@ -380,10 +375,7 @@ void TCPImagePuller::ReceiverThread() {
             out.tcp_msg->header = frame.header;
 
             if (frame.header.payload_size > 0) {
-                // Read straight into the ring (or, with no ring, onto the heap). A full ring holds the
-                // read here, the socket buffer fills, and the pusher's send blocks - its BUSY heartbeats
-                // tell it apart from a dead writer, and it gives up only after a minute in which not one
-                // image was written.
+                // A full ring holds the read here: the socket buffer fills and the pusher's send blocks
                 out.raw = ring.Allocate(frame.header.payload_size);
                 if (!out.raw)
                     break; // Stop() - the puller is being disconnected.
@@ -411,10 +403,7 @@ void TCPImagePuller::ReceiverThread() {
 }
 
 void TCPImagePuller::HeartbeatThread() {
-    // While connected, periodically tell the pusher we are alive even if the
-    // consuming pipeline is stalled (e.g. blocked on a slow filesystem). This lets
-    // the pusher distinguish a busy-but-healthy writer from a dead one and keep
-    // waiting through arbitrarily long backpressure instead of dropping the run.
+    // Tells the pusher we are alive even if the consuming pipeline is stalled
     while (!disconnect) {
         // Sleep in small slices so shutdown stays prompt.
         for (int i = 0; i < 5 && !disconnect; i++)
@@ -434,8 +423,7 @@ void TCPImagePuller::HeartbeatThread() {
         h.ack_fifo_occupancy = cbor_fifo.GetCurrentUtilization();
         h.ack_fifo_max_occupancy = cbor_fifo.Size();
 
-        // Best effort: a failure here just means the socket is gone, which
-        // ReceiverThread will detect and reconnect on its own.
+        // Best effort: ReceiverThread notices a socket that is gone
         std::lock_guard lg(send_mutex);
         SendAll(&h, sizeof(h));
     }
@@ -447,8 +435,7 @@ void TCPImagePuller::Disconnect() {
 
     CloseSocket();
 
-    // Same reason as in ZMQImagePuller: once the puller is going away nobody drains outside_fifo,
-    // so a full queue would leave CBORThread blocked in PutBlocking and the joins below hanging.
+    // As in ZMQImagePuller: nobody drains outside_fifo any more, and the joins below would hang
     cbor_fifo.Stop();
     repub_fifo.Stop();
     outside_fifo.Stop();

image_puller/TCPImagePuller.h

@@ -31,8 +31,7 @@ class TCPImagePuller : public ImagePuller {
     std::thread repub_thread;
     std::thread heartbeat_thread;
 
-    // Serializes all writer->pusher sends (ACKs, keepalive pongs, busy heartbeats)
-    // so that concurrently-sent frames cannot interleave on the socket.
+    // Frames sent to the pusher (ACKs, pongs, busy heartbeats) must not interleave on the socket
     std::mutex send_mutex;
 
     Logger logger{"TCPImagePuller"};

image_puller/ZMQImagePuller.cpp

@@ -40,10 +40,7 @@ ZMQImagePuller::~ZMQImagePuller() {
 void ZMQImagePuller::Disconnect() {
     disconnect = 1;
 
-    // The threads below hand messages on through bounded blocking queues, and by the time a puller
-    // is disconnected nothing drains outside_fifo any more - the receiver that did is gone. A full
-    // queue would then park CBORThread in PutBlocking for ever, back up cbor_fifo and park
-    // PullerThread too, and the joins below would never return. Stop() releases them.
+    // Nothing drains outside_fifo any more: without Stop() a full queue would leave the joins below hanging
     cbor_fifo.Stop();
     repub_fifo.Stop();
     outside_fifo.Stop();
@@ -87,11 +84,7 @@ void ZMQImagePuller::PullerThread() {
             ret.raw = std::shared_ptr<ReceiveBufferRegion>(new ReceiveBufferRegion{const_cast<uint8_t *>(msg->data()), msg->size()},
                                                            [msg](ReceiveBufferRegion *r) { delete r; });
         } else {
-            // Copied into the ring, and the ZeroMQ message is released with the next receive. When the
-            // ring is full this waits and the socket is not read: its queue fills to the watermark and
-            // the sender's send blocks - for as long as the sender is prepared to wait. The broker's
-            // ZeroMQ pusher waits one second, then drops the image and sends without waiting for the
-            // rest of the run, so a ring that is too small for the writer's stalls costs images.
+            // When the ring is full this waits, and the socket is not read: the sender's send blocks for as long as it waits
             ret.raw = ring.Allocate(msg->size());
             if (!ret.raw) {
                 // Stop() - the puller is being disconnected.
@@ -111,9 +104,7 @@ void ZMQImagePuller::CBORThread() {
     while (ret.raw) {
         try {
             ret.cbor = CBORStream2Deserialize(ret.raw->data, ret.raw->size);
-            // Even if we suspend consuming of the messages by the receiver,
-            // it is still reasonable to have them republished
-            // so republish functionality is not affected by Suspend()
+            // Suspend() of the receiver does not affect republish
             if (!suspend)
                 outside_fifo.PutBlocking(ret);
             if (repub_socket) {
@@ -124,8 +115,7 @@ void ZMQImagePuller::CBORThread() {
                     repub_bytes += ret.raw->size;
             }
         } catch (const std::exception &e) {
-            // std::exception: the deserialiser allocates a vector per per-image array of an END message,
-            // and a bad_alloc leaving this thread would end the process.
+            // std::exception: a bad_alloc in the deserialiser leaving this thread would end the process
             logger.ErrorException(e);
         }
         ret = cbor_fifo.GetBlocking();

image_puller/ZMQImagePuller.h

@@ -15,8 +15,7 @@
 
 
 class ZMQImagePuller : public ImagePuller {
-    // Messages, not bytes: what the socket holds beyond the ring buffer. Kept small for that reason -
-    // the ring is where the memory is accounted for.
+    // Messages, not bytes: small, as the ring is where the memory is accounted for
     const uint32_t default_receive_watermark = 10;
     const uint32_t default_repub_watermark = 220;
 

receiver/FPGAImagePuller.cpp

@@ -32,8 +32,7 @@ FPGAImagePuller::FPGAImagePuller(const DiffractionExperiment &in_experiment, con
     if (experiment.GetDetectorType() != DetectorType::JUNGFRAU)
         calibration = nullptr;
 
-    // Making a slot takes tens of milliseconds on a large detector, under the lock all modules go through:
-    // the first ones are ready when the frames start. The rest only if the receiver falls behind.
+    // Making a slot is slow, under the lock all modules go through: the first ones are ready when the frames start
     for (size_t i = 0; i < std::min<size_t>(max_slots, InitialSlots); i++) {
         free_slots.push_back(std::make_shared<Slot>(experiment));
         slots_created++;
@@ -46,20 +45,20 @@ FPGAImagePuller::FPGAImagePuller(const DiffractionExperiment &in_experiment, con
                       .received = std::chrono::steady_clock::now()});
 
     for (int i = 0; i < compression_threads; i++)
-        compress_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::CompressImages, this));
+        compress_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::CompressThread, this));
 
     for (int d = 0; d < ndatastreams; d++)
         for (int m = 0; m < experiment.GetModulesNum(d); m++)
-            module_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::ReadModule, this, d, m));
+            module_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::ModuleThread, this, d, m));
 
     std::latch ready(ndatastreams);
     for (int d = 0; d < ndatastreams; d++)
-        acquire_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::Acquire, this,
+        acquire_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::AcquireThread, this,
                                                 static_cast<uint16_t>(d), std::cref(mask), calibration,
                                                 std::ref(ready)));
     ready.wait();
 
-    finish_future = std::async(std::launch::async, &FPGAImagePuller::Finish, this);
+    finish_future = std::async(std::launch::async, &FPGAImagePuller::FinalizeMeasurement, this);
 
     if (failed) {
         try {
@@ -101,7 +100,7 @@ void FPGAImagePuller::Stop() {
         std::rethrow_exception(std::exchange(error, nullptr));
 }
 
-void FPGAImagePuller::Acquire(uint16_t data_stream, const PixelMask &mask, const JFCalibration *calibration,
+void FPGAImagePuller::AcquireThread(uint16_t data_stream, const PixelMask &mask, const JFCalibration *calibration,
                               std::latch &ready) {
     auto &device = devices[data_stream];
     try {
@@ -133,8 +132,7 @@ int64_t FPGAImagePuller::FastestFrame() const {
     return ret;
 }
 
-// Polled rather than waited on: a dead card has to be noticed from the other cards' frames, and the
-// counters wake every waiter on every completion.
+// Polled: a dead card has to be noticed from the other cards' frames
 void FPGAImagePuller::WaitForFrame(uint16_t data_stream, uint16_t module_number, int64_t frame) const {
     const auto &counters = devices[data_stream].Counters();
     // A frame is complete once the module is two further on, as in JFJochReceiverFPGA
@@ -144,8 +142,7 @@ void FPGAImagePuller::WaitForFrame(uint16_t data_stream, uint16_t module_number,
         std::this_thread::sleep_for(std::chrono::milliseconds(1));
 }
 
-// The last frame any card got, for a module whose own card has finished. Waits for the other cards to
-// finish too, so every module of the detector stops at the same image.
+// The last frame any card got; waits for all cards, so every module stops at the same image
 int64_t FPGAImagePuller::LastFrame() const {
     for (int d = 0; d < experiment.GetDataStreamsNum(); d++)
         while (!devices[d].Counters().IsAcquisitionFinished() && !failed)
@@ -185,7 +182,7 @@ void FPGAImagePuller::ReturnSlot(const std::shared_ptr<Slot> &slot) {
     free_slots.push_back(slot);
 }
 
-void FPGAImagePuller::ReadModule(uint16_t data_stream, uint16_t module_number) {
+void FPGAImagePuller::ModuleThread(uint16_t data_stream, uint16_t module_number) {
     try {
         auto &device = devices[data_stream];
         const auto &counters = device.Counters();
@@ -235,7 +232,7 @@ void FPGAImagePuller::ReadModule(uint16_t data_stream, uint16_t module_number) {
     }
 }
 
-void FPGAImagePuller::CompressImages() {
+void FPGAImagePuller::CompressThread() {
     while (const auto slot = images_to_compress.GetBlocking()) {
         const int64_t image_number = slot->image_number;
         try {
@@ -288,7 +285,7 @@ void FPGAImagePuller::Emit(int64_t image_number, std::optional<ImagePullerOutput
     }
 }
 
-void FPGAImagePuller::Finish() {
+void FPGAImagePuller::FinalizeMeasurement() {
     for (auto &f: module_futures)
         f.get();
     for (size_t i = 0; i < compress_futures.size(); i++)

receiver/FPGAImagePuller.h

@@ -19,20 +19,12 @@
 #include "../image_puller/ImagePuller.h"
 #include "../jungfrau/JFCalibration.h"
 
-// The images of a collection from the FPGA cards, as a detector stream for JFJochReceiverLite: start
-// message, images assembled and compressed here in image-number order, end message. The card converts
-// the pixels; its analysis results are not used, the receiver analyses the assembled image.
-//
-// One thread per module takes that module's frames off the card and writes them into the image they
-// belong to, so no module ever waits for another and none waits for the consumer. Images are held in a
-// bounded pool of slots, from the first module to reach them until the receiver is done with them. An
-// image finding no free slot is dropped whole, and its frames go back to the card unread.
+// The images of a collection from the FPGA cards, as a detector stream: assembled and compressed here, in image order
 class FPGAImagePuller : public ImagePuller {
 public:
     static constexpr size_t DefaultSlots = 128;
     static constexpr size_t InitialSlots = 32;
-    // A module this far behind the fastest one is taken as missing from that frame. It is what the
-    // device itself allows before it starts handing late frames straight back to the card.
+    // A module this far behind the fastest one is taken as missing from that frame, as the device itself does
     static constexpr int64_t MissingAfterFrames = AcquisitionCounters::ThresholdFramesLost + 2;
 private:
     struct Slot {
@@ -75,10 +67,10 @@ private:
     std::future<void> finish_future;
     std::once_flag finish_once;
 
-    void Acquire(uint16_t data_stream, const PixelMask &mask, const JFCalibration *calibration, std::latch &ready);
-    void ReadModule(uint16_t data_stream, uint16_t module_number);
-    void CompressImages();
-    void Finish();
+    void AcquireThread(uint16_t data_stream, const PixelMask &mask, const JFCalibration *calibration, std::latch &ready);
+    void ModuleThread(uint16_t data_stream, uint16_t module_number);
+    void CompressThread();
+    void FinalizeMeasurement();
 
     void WaitForFrame(uint16_t data_stream, uint16_t module_number, int64_t frame) const;
     int64_t FastestFrame() const;

receiver/JFJochReceiver.cpp

@@ -30,8 +30,7 @@ JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
           scan_result(in_experiment),
           serialmx_filter(in_experiment),
           pixel_mask(in_pixel_mask),
-          // retain_outcomes=false for now: online scaling/merge at the end is not enabled yet, so the
-          // whole-run integration_outcome is unused here. Flip back to true when online scaling lands.
+          // retain_outcomes=false: no online scaling/merge at the end, so the integration outcomes are not used
           indexer(experiment, indexing_thread_pool, false, /*real_time=*/true) {
     logger.Debug("Initializing receiver");
     // Ensure there is nothing running for now
@@ -115,10 +114,10 @@ JFJochReceiverStatus JFJochReceiver::GetStatus() const {
 }
 
 
-StartMessage JFJochReceiver::MakeStartMessage(const DiffractionExperiment &x, const PixelMask &mask,
+StartMessage JFJochReceiver::MakeStartMessage(const DiffractionExperiment &in_experiment, const PixelMask &mask,
                                               const AzimuthalIntegrationMapping &mapping) {
     StartMessage message{};
-    x.FillMessage(message);
+    in_experiment.FillMessage(message);
     message.arm_date = time_UTC(std::chrono::system_clock::now());
     message.az_int_q_bin_count = mapping.GetQBinCount();
     message.az_int_bin_to_q = mapping.GetBinToQ();
@@ -129,11 +128,11 @@ StartMessage JFJochReceiver::MakeStartMessage(const DiffractionExperiment &x, co
         message.az_int_map = mapping.GetPixelToBin();
     }
     message.writer_notification_zmq_addr = image_pusher.GetWriterNotificationSocketAddress();
-    message.rois = x.ROI().ExportMetadata();
-    if (!x.ROI().empty())
-        message.roi_map = x.ExportROIMap();
-    message.max_spot_count = x.GetMaxSpotCount();
-    message.pixel_mask["default"] = mask.GetMask(x);
+    message.rois = in_experiment.ROI().ExportMetadata();
+    if (!in_experiment.ROI().empty())
+        message.roi_map = in_experiment.ExportROIMap();
+    message.max_spot_count = in_experiment.GetMaxSpotCount();
+    message.pixel_mask["default"] = mask.GetMask(in_experiment);
     return message;
 }
 
@@ -162,22 +161,14 @@ void JFJochReceiver::SaveStartMessageToImageBuffer(const StartMessage &msg) {
     image_buffer.SaveStartMessage(buffer);
 }
 
-// Serialize a processed image into its buffer slot. The slot holds the compressed
-// image plus the per-image CBOR metadata (spot list, reflection list, ...); if an
-// unusually rich frame makes that metadata too large to fit alongside the image,
-// drop just this frame (log + recycle the slot) instead of letting the serializer
-// throw, which would abort the whole data collection. Returns true when the image
-// was serialized and the slot is ready for the caller to commit.
+// A frame whose metadata does not fit its slot is dropped, without aborting the collection
 bool JFJochReceiver::SerializeImageOrDrop(CBORStream2Serializer &serializer, DataMessage &msg,
                                           ZeroCopyReturnValue &loc, size_t slot_size, Logger &logger) {
     try {
         serializer.SerializeImage(msg);
         return true;
     } catch (const JFJochException &) {
-        // Confirm this is a slot-overflow (not some other CBOR error) by re-serializing
-        // the metadata alone with an empty image placeholder: that always fits when the
-        // image is what overflowed, and gives the exact metadata size for the log. If it
-        // throws too, it is not a simple overflow and the exception propagates.
+        // The metadata alone fits if it is the image that overflowed; if not, the exception propagates
         const auto image = msg.image;
         msg.image = CompressedImage(nullptr, 0, image.GetWidth(), image.GetHeight(),
                                     image.GetMode(), image.GetCompressionAlgorithm(),
@@ -289,15 +280,14 @@ void JFJochReceiver::SendEndMessage() {
         rotation_indexing_lattice_type = message.rotation_lattice_type;
     }
     message.unit_cell = indexer.GetConsensusUnitCell();
-    // The collection does not stop for an indexer that failed - the images are written regardless - but
-    // its indexing rate is then not a statement about the crystal, and the log has to say so.
-    if (const int64_t n = indexer.GetIndexerErrorCount(); n > 0) {
+    // The images are written regardless, but the indexing rate is then not a statement about the crystal
+    if (const int64_t error_count = indexer.GetIndexerErrorCount(); error_count > 0) {
         try {
             indexer.RethrowIndexerError();
         } catch (const std::exception &e) {
             logger.Error("Indexing failed {} times in this collection, first with: {}. Those attempts are "
                          "left out of the indexing rate, and a rotation lattice may be missing for the same "
-                         "reason", n, e.what());
+                         "reason", error_count, e.what());
         }
     }
 

receiver/JFJochReceiver.h

@@ -94,7 +94,7 @@ protected:
     void UpdateMaxDelay(uint64_t delay);
     JFJochReceiverStatus GetStatus() const;
     // The start message of this collection, stating that experiment, mask and azimuthal mapping.
-    StartMessage MakeStartMessage(const DiffractionExperiment &x, const PixelMask &mask,
+    StartMessage MakeStartMessage(const DiffractionExperiment &experiment, const PixelMask &mask,
                                   const AzimuthalIntegrationMapping &mapping);
     void SendStartMessage();
     void SendStartMessage(StartMessage message);
@@ -110,8 +110,7 @@ protected:
     // Serializes into the slot and sends it on; false if there is no slot or the image does not fit
     bool ForwardImage(DataMessage &message, ZeroCopyReturnValue *loc);
 
-    // The last word on the end message before it is sent: a receiver whose analysis is not the
-    // per-image one fills in what it found.
+    // The last word on the end message: a receiver with an analysis of its own fills in what it found
     virtual void AddToEndMessage(EndMessage &message) {}
     virtual uint64_t GetImagesDropped() const { return 0; }
     virtual void Cancel(const JFJochException &e);

receiver/JFJochReceiverOutput.h

@@ -26,8 +26,7 @@ struct JFJochReceiverOutput {
     std::string writer_err;
     std::optional<int64_t> images_written;
     MeanProcessingTime processing_time;
-    // A problem with the analysis alone (the images are all there), for the broker's status: why a live
-    // rugnux run was skipped. Empty otherwise.
+    // Why a live rugnux run was skipped, for the broker's status
     std::string analysis_warning;
 };
 

receiver/JFJochReceiverPedestal.h

@@ -12,8 +12,7 @@
 #include "../acquisition_device/AcquisitionDeviceGroup.h"
 #include "../jungfrau/JFPedestalCalc.h"
 
-// A pedestal run: raw frames from the acquisition devices go directly into JFPedestalCalc. It forwards no
-// images, so it uses no image buffer slots and sends nothing to the writer.
+// A pedestal run: raw frames from the devices go into JFPedestalCalc; no image buffer slots, nothing to the writer
 class JFJochReceiverPedestal : public JFJochReceiver {
     std::vector<std::unique_ptr<JFPedestalCalc>> pedestal;
 

receiver/JFJochReceiverRugnuxRotation.cpp

@@ -17,42 +17,35 @@
 #include "RugnuxStructure.h"
 
 namespace {
-    // Kept for the rest of the broker - the receiver, the pushers, the preview - when the host memory
-    // rugnux plans against is decided.
+    // Kept for the rest of the broker when the host memory rugnux plans against is decided
     constexpr size_t RUGNUX_HOST_MEMORY_MARGIN = 16ULL * 1024 * 1024 * 1024;
 
-    // The host memory rugnux plans against (HostMemory.h) is the machine's budget less what the broker
-    // already holds: the image buffer, which has to keep the whole collection and so is most of the
-    // machine, and a margin. Never 0, which would take the whole machine's budget: at worst what could
-    // run side by side runs one part after the other.
+    // The machine's budget (HostMemory.h) less the image buffer and the margin; never 0, the whole machine's budget
     size_t RugnuxHostBudget(const ImageBuffer &buffer) {
         const size_t budget = DefaultHostMemoryBudget();
         const size_t taken = buffer.GetBufferSize() + RUGNUX_HOST_MEMORY_MARGIN;
         return budget > taken ? budget - taken : 1;
     }
 
-    // A run reachable for Cut while it exists: registered in `slot` (under `m`, cancelled at once where the
-    // cut came first) and taken off again before it is destroyed, an exception included.
+    // A run reachable for Cut while it exists: registered in `slot`, cancelled at once where the cut came first
     class Reachable {
-        std::mutex &m;
+        std::mutex &slot_mutex;
         Rugnux *&slot;
     public:
-        Reachable(std::mutex &m, Rugnux *&slot, Rugnux &run, const bool &cut, const std::atomic<bool> &cancelled)
-            : m(m), slot(slot) {
-            std::unique_lock ul(m);
+        Reachable(std::mutex &in_mutex, Rugnux *&slot, Rugnux &run, const bool &cut, const std::atomic<bool> &cancelled)
+            : slot_mutex(in_mutex), slot(slot) {
+            std::unique_lock ul(slot_mutex);
             slot = &run;
             if (cut || cancelled)
                 run.Cancel();
         }
         ~Reachable() {
-            std::unique_lock ul(m);
+            std::unique_lock ul(slot_mutex);
             slot = nullptr;
         }
     };
 
-    // What rugnux reports as it goes: the per-image results of its canonical pass into the receiver's
-    // plots and scan result - the plots accumulate, so an image reported by the pre-pass as well would be
-    // counted twice - and its steps into the live status (not the log: they are processing results).
+    // Per-image results of rugnux's canonical pass to the plots and scan result, its steps to the live status
     class LiveObserver : public RugnuxObserver {
         JFJochReceiverPlots &plots;
         ScanResultGenerator &scan_result;
@@ -64,7 +57,7 @@ namespace {
                      const AzimuthalIntegrationMapping &mapping, RugnuxLiveStatus &status)
             : plots(plots), scan_result(scan_result), mapping(mapping), status(status) {}
 
-        void OnImagePass(bool c) override { canonical = c; }
+        void OnImagePass(bool input) override { canonical = input; }
         void OnImageProcessed(const DataMessage &msg) override {
             if (!canonical)
                 return;
@@ -114,10 +107,7 @@ JFJochReceiverRugnuxRotation::JFJochReceiverRugnuxRotation(const DiffractionExpe
       model(std::move(in_model)),
       deadline_s(in_deadline_s),
       live_status(in_live_status) {
-    // GPU FFT indexers and cuFFT plans are kept for reuse for the life of the broker, not of one run as
-    // in `rugnux` (FFTIndexerReuse, FFTPlanReuse): every pass of every collection builds the same pools,
-    // and making a plan takes a process-wide lock and took over a second on a card busy with the online
-    // analysis. Never destroyed - at exit the CUDA runtime may already be gone.
+    // Indexers and cuFFT plans kept for the life of the broker; never destroyed, the CUDA runtime may be gone at exit
     [[maybe_unused]] static const FFTIndexerReuse *fft_indexers = new FFTIndexerReuse;
     [[maybe_unused]] static const FFTPlanReuse *fft_plans = new FFTPlanReuse;
     logger.Info("rugnux live: starting {} forward threads; rugnux runs on {} threads while the images arrive, "
@@ -130,7 +120,7 @@ JFJochReceiverRugnuxRotation::~JFJochReceiverRugnuxRotation() {
     cancelled = true;
     reader.Ended();
     {
-        std::unique_lock ul(rugnux_m);
+        std::unique_lock ul(rugnux_mutex);
         if (rugnux != nullptr)
             rugnux->Cancel();
         if (probe != nullptr)
@@ -142,8 +132,7 @@ JFJochReceiverRugnuxRotation::~JFJochReceiverRugnuxRotation() {
 void JFJochReceiverRugnuxRotation::OnStart(std::chrono::steady_clock::time_point received) {
     SendStartMessage();
 
-    // Rugnux reads the images back from the buffer, so it needs one slot per image: a slot taken
-    // again would hand it a later image. Without that the collection is only forwarded.
+    // Rugnux reads the images back from the buffer: without one slot per image the collection is only forwarded
     const auto slots = image_buffer.GetStatus().total_slots;
     // The image time as the detector produces images, summation included
     const double image_time_s = std::chrono::duration<double>(experiment.GetImageTime()).count();
@@ -169,8 +158,7 @@ void JFJochReceiverRugnuxRotation::OnImage(uint32_t id, DataMessage &message, st
 
 void JFJochReceiverRugnuxRotation::FinishCollection() {
     reader.Ended();
-    // The latency of the demonstration runs from the last image in (as the puller took it off the socket)
-    // to every result in place.
+    // The latency runs from the last image in to every result in place
     const auto last_image = live_status.LastImage();
     logger.Info("Receiving finished, {} images", images_collected.load());
 
@@ -247,9 +235,7 @@ void JFJochReceiverRugnuxRotation::RugnuxThread() {
         const auto dataset = reader.GetDataset();
         RugnuxLiveSetup setup = MakeRugnuxLiveSetup(*dataset, rugnux_nthreads, model);
         setup.config.host_memory_budget_bytes = host_budget;
-        // The cards are idle until the first images arrive: the indexers and plans the passes will
-        // need are built then, beside the run, which meanwhile waits for its first frames. Waited for
-        // when the run ends.
+        // Built beside the run, which waits for its first frames meanwhile; waited for when the run ends
         auto gpu_warm_up = std::async(std::launch::async, [this, setup] {
             try {
                 const auto t0 = std::chrono::steady_clock::now();
@@ -262,25 +248,24 @@ void JFJochReceiverRugnuxRotation::RugnuxThread() {
             }
         });
         LiveObserver observer(plots, scan_result, *az_int_mapping, live_status);
-        // The run goes - with the pinned buffers its engines kept - once its results are out (below), beside
-        // the end message, which reads only the result: a value of its own, no part of the run.
+        // The run goes with its pinned buffers once its results are out (below), beside the end message
         std::optional<Rugnux> process;
         process.emplace(reader, setup.experiment, *dataset->pixel_mask, setup.config);
         std::optional<Reachable> reachable;
-        reachable.emplace(rugnux_m, rugnux, *process, cut, cancelled);
+        reachable.emplace(rugnux_mutex, rugnux, *process, cut, cancelled);
         ProcessResult result = process->Run(&observer);
         // The verdict and summary of the report a `rugnux` run writes, without writing it.
         const ReportDocument doc = BuildReportDocument(experiment.GetFilePrefix(), experiment.GetFilePrefix(),
                                                        setup.experiment, result);
         std::vector<std::string> warnings;
-        for (const auto &w: doc.warnings)
-            if (!w.developer_only)
-                warnings.push_back(w.text);
+        for (const auto &warning: doc.warnings)
+            if (!warning.developer_only)
+                warnings.push_back(warning.text);
         const double cc_anom = result.merge_statistics.overall.cc_anom;
         live_status.Result(doc.verdict, doc.verdict_text, doc.summary_rows, warnings,
                            std::isfinite(cc_anom) && cc_anom > CC_ANOM_SIGNAL);
-        if (const auto m = LiveModel(result))
-            live_status.Model(*m);
+        if (const auto live_model = LiveModel(result))
+            live_status.Model(*live_model);
         // The numbers the summary's Space group, Resolution and Signal rows give
         std::optional<double> d_min, isa;
         if (result.has_merge_statistics)
@@ -289,17 +274,16 @@ void JFJochReceiverRugnuxRotation::RugnuxThread() {
         if (!result.merge_statistics.stills && result.error_model_isa_resolved && result.error_model_isa > 0.0)
             isa = result.error_model_isa;
         live_status.Headline(result.space_group ? result.space_group->xhm() : "", d_min, isa);
-        // The placed model and a copy of its map coefficients (the end message takes the originals), and the
-        // refined model with its own, for the broker's viewer routes.
+        // For the broker's viewer routes; a copy of the map coefficients, as the end message takes the originals
         if (result.outputs && result.outputs->placed_model) {
-            const RugnuxOutputs &o = *result.outputs;
-            PublishStructures(live_status, *o.placed_model, o.maps.get(),
-                              o.refined_model ? &*o.refined_model : nullptr, o.refined_maps.get());
+            const RugnuxOutputs &outputs = *result.outputs;
+            PublishStructures(live_status, *outputs.placed_model, outputs.maps.get(),
+                              outputs.refined_model ? &*outputs.refined_model : nullptr, outputs.refined_maps.get());
         }
         live_status.Finish(result.cancelled ? RugnuxLiveState::Cancelled : RugnuxLiveState::Done);
         reachable.reset();
         {
-            std::unique_lock ul(rugnux_m);
+            std::unique_lock ul(rugnux_mutex);
             rugnux_result = std::move(result);
         }
         published = true;
@@ -315,10 +299,7 @@ void JFJochReceiverRugnuxRotation::RugnuxThread() {
 }
 
 void JFJochReceiverRugnuxRotation::ProbeThread() {
-    // Rugnux waits for half the sweep before it decides on a lattice. In the meantime the machine is
-    // idle, so a small run of its own indexes the first 60 degrees and says at once whether there is
-    // a crystal - quick feedback for the person at the beamline, above all where there is none. It
-    // decides nothing for the run, which goes on as it would without it.
+    // A small run of its own on the first degrees says early whether there is a crystal; it decides nothing
     constexpr double PROBE_DEG = 60.0;
     try {
         const auto dataset = reader.GetDataset();
@@ -334,23 +315,22 @@ void JFJochReceiverRugnuxRotation::ProbeThread() {
         setup.config.end_image = static_cast<int>(frames);
         setup.config.rotation_postrefine_geometry = false;
         setup.config.run_scaling = false;
-        // The file's centre only: the check captures the centre on the GPUs and runs a second first pass,
-        // both beside the main run's pre-scan, for a centre the probe does not report.
+        // The file's centre only: the probe does not report one
         setup.config.beam_center_check = false;
-        ProcessResult r;
+        ProcessResult result;
         {
             Rugnux probe_run(reader, setup.experiment, *dataset->pixel_mask, setup.config);
-            Reachable reachable(rugnux_m, probe, probe_run, cut, cancelled);
-            r = probe_run.Run(nullptr);
+            Reachable reachable(rugnux_mutex, probe, probe_run, cut, cancelled);
+            result = probe_run.Run(nullptr);
         }
-        if (r.consensus_cell && r.rotation_lattice_type) {
-            const auto &c = *r.consensus_cell;
+        if (result.consensus_cell && result.rotation_lattice_type) {
+            const auto &cell = *result.consensus_cell;
             live_status.Probe(fmt::format("lattice on the first {:.0f} deg: {}-centred {}, {:.1f} {:.1f} {:.1f} "
                                           "{:.1f} {:.1f} {:.1f}, {:.0f}% of the images indexed", PROBE_DEG,
-                                          r.rotation_lattice_type->centering,
-                                          gemmi::crystal_system_str(r.rotation_lattice_type->crystal_system),
-                                          c.a, c.b, c.c, c.alpha, c.beta, c.gamma,
-                                          100.0 * r.indexing_rate.value_or(0.0f)));
+                                          result.rotation_lattice_type->centering,
+                                          gemmi::crystal_system_str(result.rotation_lattice_type->crystal_system),
+                                          cell.a, cell.b, cell.c, cell.alpha, cell.beta, cell.gamma,
+                                          100.0 * result.indexing_rate.value_or(0.0f)));
         } else
             live_status.Probe(fmt::format("no lattice on the first {:.0f} deg", PROBE_DEG));
     } catch (const NoLatticeFound &e) {
@@ -365,49 +345,48 @@ void JFJochReceiverRugnuxRotation::AddToEndMessage(EndMessage &message) {
     JFJochReceiverStream::AddToEndMessage(message);
     // A failure here costs the end message the analysis only: it goes out as the receiver made it.
     try {
-        std::unique_lock ul(rugnux_m);
+        std::unique_lock ul(rugnux_mutex);
         if (!rugnux_result)
             return;
-        const EndMessage &r = rugnux_result->end_message;
-        message.indexing_rate = r.indexing_rate;
-        message.bkg_estimate = r.bkg_estimate;
-        message.spindle_blind_fraction = r.spindle_blind_fraction;
-        message.ice_ring_score_mean = r.ice_ring_score_mean;
-        message.az_int_result = r.az_int_result;
-        message.rotation_lattice = r.rotation_lattice;
-        message.rotation_lattice_type = r.rotation_lattice_type;
-        message.rotation_extra_lattices = r.rotation_extra_lattices;
-        message.unit_cell = r.unit_cell;
-        message.unit_cell_esd = r.unit_cell_esd;
-        message.space_group_name = r.space_group_name;
-        message.space_group_number = r.space_group_number;
-        message.reindex_matrix = r.reindex_matrix;
-        message.refined_beam_center_x = r.refined_beam_center_x;
-        message.refined_beam_center_y = r.refined_beam_center_y;
-        message.refined_poni_rot1 = r.refined_poni_rot1;
-        message.refined_poni_rot2 = r.refined_poni_rot2;
-        message.refined_poni_rot3 = r.refined_poni_rot3;
-        message.refined_rotation_axis = r.refined_rotation_axis;
-        // Merged data, map coefficients and placed model, for the writer to write beside the master. Moved:
-        // the result is not read again.
+        const EndMessage &rugnux_end = rugnux_result->end_message;
+        message.indexing_rate = rugnux_end.indexing_rate;
+        message.bkg_estimate = rugnux_end.bkg_estimate;
+        message.spindle_blind_fraction = rugnux_end.spindle_blind_fraction;
+        message.ice_ring_score_mean = rugnux_end.ice_ring_score_mean;
+        message.az_int_result = rugnux_end.az_int_result;
+        message.rotation_lattice = rugnux_end.rotation_lattice;
+        message.rotation_lattice_type = rugnux_end.rotation_lattice_type;
+        message.rotation_extra_lattices = rugnux_end.rotation_extra_lattices;
+        message.unit_cell = rugnux_end.unit_cell;
+        message.unit_cell_esd = rugnux_end.unit_cell_esd;
+        message.space_group_name = rugnux_end.space_group_name;
+        message.space_group_number = rugnux_end.space_group_number;
+        message.reindex_matrix = rugnux_end.reindex_matrix;
+        message.refined_beam_center_x = rugnux_end.refined_beam_center_x;
+        message.refined_beam_center_y = rugnux_end.refined_beam_center_y;
+        message.refined_poni_rot1 = rugnux_end.refined_poni_rot1;
+        message.refined_poni_rot2 = rugnux_end.refined_poni_rot2;
+        message.refined_poni_rot3 = rugnux_end.refined_poni_rot3;
+        message.refined_rotation_axis = rugnux_end.refined_rotation_axis;
+        // For the writer to write beside the master; moved, the result is not read again
         if (!rugnux_result->outputs)
             return;
-        RugnuxOutputs &o = *rugnux_result->outputs;
-        if (o.merged)
-            message.merged = std::move(o.merged->mtz);
-        if (o.maps)
-            message.map_coefficients = std::move(o.maps->coefficients);
-        if (o.placed_model)
-            message.placed_model = PlacedModelFromStructure(*o.placed_model->model, o.placed_model->cell,
-                                                            o.placed_model->space_group.xhm());
+        RugnuxOutputs &outputs = *rugnux_result->outputs;
+        if (outputs.merged)
+            message.merged = std::move(outputs.merged->mtz);
+        if (outputs.maps)
+            message.map_coefficients = std::move(outputs.maps->coefficients);
+        if (outputs.placed_model)
+            message.placed_model = PlacedModelFromStructure(*outputs.placed_model->model, outputs.placed_model->cell,
+                                                            outputs.placed_model->space_group.xhm());
         // And the refinement's, as the CLI writes them (_refined_model.cif, _refined_maps.mtz, _blobs.json).
-        if (o.refined_model)
-            message.refined_model = PlacedModelFromStructure(*o.refined_model->model, o.refined_model->cell,
-                                                             o.refined_model->space_group.xhm());
-        if (o.refined_maps)
-            message.refined_map_coefficients = std::move(o.refined_maps->coefficients);
-        if (o.blobs)
-            message.blobs = DensityBlobsToMessage(*o.blobs);
+        if (outputs.refined_model)
+            message.refined_model = PlacedModelFromStructure(*outputs.refined_model->model, outputs.refined_model->cell,
+                                                             outputs.refined_model->space_group.xhm());
+        if (outputs.refined_maps)
+            message.refined_map_coefficients = std::move(outputs.refined_maps->coefficients);
+        if (outputs.blobs)
+            message.blobs = DensityBlobsToMessage(*outputs.blobs);
 
         DropResultBlocksThatDoNotFit(message, &live_status, logger);
     } catch (const std::exception &e) {
@@ -421,16 +400,14 @@ void JFJochReceiverRugnuxRotation::Cancel(bool silent) {
     // The silent cancel is the detector reading idle at the end of every collection: rugnux goes on.
     if (silent)
         return;
-    // /cancel is the one way in for the beamline, which cannot tell the collection from its analysis: it
-    // ends the collection while the images still arrive, as JFJochReceiverLite does, and in every case
-    // cuts the analysis, so the end message goes out at once with what rugnux has established.
+    // Ends the collection while images still arrive; in every case the analysis is cut and the end message goes out
     if (!end_message_received)
         JFJochReceiver::Cancel(silent);
     Cut("cancel");
 }
 
 void JFJochReceiverRugnuxRotation::Cut(const std::string &reason) {
-    std::unique_lock ul(rugnux_m);
+    std::unique_lock ul(rugnux_mutex);
     if (cut || rugnux_result)
         return;
     cut = true;
@@ -451,13 +428,13 @@ JFJochReceiverOutput JFJochReceiverRugnuxRotation::GetFinalStatistics() const {
     JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics();
     ret.analysis_warning = skipped_warning;
     // The receiver's own rotation indexer sees no image here: the lattice is rugnux's.
-    std::unique_lock ul(rugnux_m);
+    std::unique_lock ul(rugnux_mutex);
     if (rugnux_result) {
-        const EndMessage &r = rugnux_result->end_message;
-        ret.scan_result.rotation_lattice = r.rotation_lattice;
-        if (r.rotation_lattice_type) {
-            ret.scan_result.rotation_crystal_system = r.rotation_lattice_type->crystal_system;
-            ret.scan_result.rotation_centering = r.rotation_lattice_type->centering;
+        const EndMessage &rugnux_end = rugnux_result->end_message;
+        ret.scan_result.rotation_lattice = rugnux_end.rotation_lattice;
+        if (rugnux_end.rotation_lattice_type) {
+            ret.scan_result.rotation_crystal_system = rugnux_end.rotation_lattice_type->crystal_system;
+            ret.scan_result.rotation_centering = rugnux_end.rotation_lattice_type->centering;
         }
     }
     return ret;

receiver/JFJochReceiverRugnuxRotation.h

@@ -14,9 +14,7 @@
 #include "../reader/JFJochImageBufferReader.h"
 #include "../rugnux/Rugnux.h"
 
-// A rotation collection forwarded to the writer as JFJochReceiverLite does, but analysed by rugnux while it
-// arrives instead of image by image: one rugnux thread reads the collection back from the image buffer.
-// The end message waits for rugnux, and carries what it found (see AddToEndMessage).
+// A rotation collection forwarded to the writer and analysed by rugnux, which reads it back from the image buffer
 class JFJochReceiverRugnuxRotation : public JFJochReceiverStream {
     using JFJochReceiverStream::Cancel;
 
@@ -29,7 +27,7 @@ class JFJochReceiverRugnuxRotation : public JFJochReceiverStream {
     RugnuxLiveStatus &live_status;
     std::string skipped_warning;   // why the collection was not analysed, for the broker's status
 
-    mutable std::mutex rugnux_m;
+    mutable std::mutex rugnux_mutex;
     Rugnux *rugnux = nullptr;   // while it runs, for Cut
     Rugnux *probe = nullptr;    // the same for the lattice probe
     std::optional<ProcessResult> rugnux_result;
@@ -44,13 +42,11 @@ class JFJochReceiverRugnuxRotation : public JFJochReceiverStream {
     void OnStart(std::chrono::steady_clock::time_point received) override;
     void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override;
     void FinishCollection() override;
-    // Opens the reader on the collection and starts the rugnux and probe threads; a failure is the
-    // analysis's alone (live status Failed), never the collection's.
+    // A failure here is the analysis's alone (live status Failed), never the collection's
     void StartRugnux();
     void RugnuxThread();
     void ProbeThread();
-    // Stops the analysis where it is: rugnux returns what it has established (see Rugnux::Cancel), and
-    // the end message goes out with that. Reason: "deadline" or "cancel".
+    // Stops rugnux where it is, the end message goes out with what it established; reason "deadline" or "cancel"
     void Cut(const std::string &reason);
     void AddToEndMessage(EndMessage &message) override;
 public:

receiver/JFJochReceiverRugnuxStills.cpp

@@ -64,9 +64,7 @@ void JFJochReceiverRugnuxStills::OnImage(uint32_t id, DataMessage &message, std:
     CountImage(message);
     const size_t size = message.image.GetCompressedSize();
 
-    // Copied into a slot of the image buffer, where it waits for the analysis; SendAnalysedImage writes it, with
-    // its analysis, to a second slot, which is sent. A frame that finds the buffer full, but for a slot kept
-    // free for each analysis thread to send with, is dropped, as any frame that finds the image buffer full.
+    // Waits for the analysis in a slot of the image buffer; a slot is kept free for each analysis thread to send with
     auto held = std::make_shared<HeldImage>();
     held->slot = image_buffer.GetImageSlot(stills_keep_free);
     if (held->slot == nullptr) {
@@ -90,8 +88,7 @@ void JFJochReceiverRugnuxStills::OnImage(uint32_t id, DataMessage &message, std:
 void JFJochReceiverRugnuxStills::FinishCollection() {
     logger.Info("Receiving finished, {} images", images_collected.load());
 
-    // Where the start message never went out (or the analysis could not be set up), there is no end message
-    // either, and the error is the collection's result.
+    // With no start message there is no end message either, and the error is the collection's result
     std::exception_ptr error;
     try {
         stills->Finish();
@@ -129,8 +126,7 @@ void JFJochReceiverRugnuxStills::StillsReady(const StillsSetup &setup) {
 }
 
 SpotFindingSettings JFJochReceiverRugnuxStills::AnalysisSettings() {
-    // The broker's settings, with the spot finder rugnux uses on stills (the pre-pass decided the centre and mask
-    // with it): adaptive threshold, single-pixel spots kept.
+    // The broker's settings, with the spot finder of the pre-pass: adaptive threshold, single-pixel spots kept
     SpotFindingSettings settings = GetSpotFindingSettings();
     settings.adaptive_threshold = true;
     settings.min_pix_per_spot = std::nullopt;
@@ -143,9 +139,7 @@ void JFJochReceiverRugnuxStills::StillsSetupFailed(const JFJochException &error)
 }
 
 ZeroCopyReturnValue *JFJochReceiverRugnuxStills::AnalysedImageSlot() {
-    // An analysed frame is not dropped because the writer lags behind the catch-up from a full buffer: the analysis
-    // waits for the slots being sent to come back, and the frames arriving meanwhile find the buffer full. A
-    // writer taking nothing for this long loses the frame.
+    // An analysed frame waits for the writer, which lags behind the catch-up from a full buffer; this long at most
     constexpr auto WRITER_WAIT = std::chrono::seconds(2);
     const auto t0 = std::chrono::steady_clock::now();
     auto loc = image_buffer.GetImageSlot();
@@ -195,9 +189,7 @@ void JFJochReceiverRugnuxStills::Cancel(bool silent) {
     // The silent cancel is the detector reading idle at the end of every collection: rugnux goes on.
     if (silent)
         return;
-    // /cancel is the one way in for the beamline, which cannot tell the collection from its analysis: it
-    // ends the collection while the images still arrive, as JFJochReceiverLite does, and in every case
-    // cuts the analysis, so the end message goes out at once with what rugnux has established.
+    // Ends the collection while images still arrive; in every case the model step is cut and the end message goes out
     if (!end_message_received)
         JFJochReceiver::Cancel(silent);
     Cut("cancel");

receiver/JFJochReceiverRugnuxStills.h

@@ -11,9 +11,7 @@
 #include "RugnuxLiveStatus.h"
 #include "StillsGPUAnalysis.h"
 
-// A serial collection analysed frame by frame (StillsGPUAnalysis): each image the puller delivers waits in the
-// image buffer for the pre-pass, is analysed at the geometry it decides, and goes to the writer with its analysis.
-// The start message waits for the pre-pass, the end message carries the last merge and model step.
+// A serial collection analysed frame by frame at the geometry of a pre-pass, which the start message waits for
 class JFJochReceiverRugnuxStills : public JFJochReceiverStream, public StillsSink {
     using JFJochReceiverStream::Cancel;
 

receiver/JFJochReceiverService.cpp

@@ -80,18 +80,11 @@ std::optional<JFJochReceiverStatus> JFJochReceiverService::GetStatus() {
 }
 
 void JFJochReceiverService::Preflight(const DiffractionExperiment &experiment) {
-    // Same condition JFJochReceiver uses to decide whether it pushes to a writer at all: with no
-    // images or no file prefix nothing is written, so there is nothing to check.
+    // As JFJochReceiver decides whether it pushes to a writer at all
     if ((experiment.GetImageNum() <= 0) || experiment.GetFilePrefix().empty())
         return;
 
-    // FillMessage alone, which is what makes this frame small: the per-pixel arrays that dominate a
-    // real start message - the pixel mask, the azimuthal-integration map, the ROI map - are added
-    // by JFJochReceiver::SendStartMessage and are not here. Measured on a JUNGFRAU 9M: 1.4 kB
-    // against 543 kB for the start message that follows. It is deliberately the whole of
-    // FillMessage rather than the handful of fields the check reads today, so that the writer sees
-    // the same description of the run it will be given at start - a hand-picked subset would
-    // answer for a different set of file names the day the check learns to read one more field.
+    // FillMessage alone, so this frame is small: the per-pixel arrays are added by JFJochReceiver::SendStartMessage
     StartMessage message{};
     experiment.FillMessage(message);
     image_pusher.Preflight(message);
@@ -207,9 +200,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
         state = ReceiverState::Running;
     } catch (const std::exception &e) {
         receiver.reset(); // its threads may be polling the puller reset below
-        // The receiver never started, so drop the status its base constructor had already reset -
-        // otherwise /status and /statistics keep reporting a zero-progress run that never happened,
-        // until the next start overwrites it.
+        // The receiver never started: /status must not keep reporting a run that never happened
         receiver_status.Clear();
         receiver_status.SetProgress({});
         if (fpga_puller) {
@@ -425,8 +416,7 @@ void JFJochReceiverService::ClearResults() {
 
 void JFJochReceiverService::ClearImageBuffer() {
     std::unique_lock ul(state_mutex);
-    // Clearing image buffer during data collection could be catastrophic, so better protect here, even if redundant
-    // with JFJochStateMachine
+    // Clearing image buffer during data collection could be catastrophic, so better protect here
     if (state == ReceiverState::Idle)
         image_buffer.Finalize(std::chrono::milliseconds(2500));
     else
@@ -436,15 +426,9 @@ void JFJochReceiverService::ClearImageBuffer() {
 
 JFJochReceiverService &JFJochReceiverService::Indexing(const IndexingSettings &input) {
     std::unique_lock ul(state_mutex);
-    // Clearing image buffer during data collection could be catastrophic, so better protect here, even if redundant
-    // with JFJochStateMachine
+    // Clearing image buffer during data collection could be catastrophic, so better protect here
     if (state == ReceiverState::Idle) {
-        // Release the previous run's receiver first. It holds a raw pointer to the indexer pool and
-        // its own GPU resources; keeping it alive while we rebuild the pool means a fresh GPU indexer
-        // has to coexist with a stale receiver (e.g. after a failed acquisition start, where the
-        // receiver is stopped but not destroyed until the next Start), which can make the GPU indexer
-        // initialisation fail. Destroying the receiver before the pool also avoids the dangling
-        // pointer. Safe here: state is Idle, so no measurement is using it.
+        // The receiver goes first: it holds a raw pointer to the pool, and a stale one's GPU resources can fail the new indexers
         receiver.reset();
 
         logger.Info("Resetting indexing thread pool");

receiver/JFJochReceiverService.h

@@ -26,16 +26,14 @@ class JFJochReceiverService {
     std::shared_ptr<ImagePuller> image_puller;
     int64_t nthreads = 8;
 
-    // Rotation collections from a DECTRIS detector analysed by rugnux while they arrive
-    // (JFJochReceiverRugnuxRotation), instead of image by image. Off unless asked for.
+    // Collections analysed by rugnux while they arrive, instead of image by image. Off unless asked for.
     bool rugnux_enable = false;
     std::shared_ptr<const RugnuxModel> rugnux_model;
     int rugnux_reserved_cpus = 4;   // left to the forward threads and the pusher
     double rugnux_deadline_s = 60;   // the analysis is cut this long after the last image (0 = never)
     RugnuxLiveStatus rugnux_status;   // of the last collection, kept after it ends
 
-    // A PSI detector's frames go through FPGAImagePuller and the stream receivers, instead of
-    // JFJochReceiverFPGA. Off unless asked for.
+    // A PSI detector's frames go through FPGAImagePuller and the stream receivers. Off unless asked for.
     bool fpga_image_puller = false;
     size_t fpga_image_puller_slots = FPGAImagePuller::DefaultSlots;
     std::shared_ptr<FPGAImagePuller> fpga_puller;
@@ -82,8 +80,7 @@ public:
     void LoadInternalGeneratorImage(const DiffractionExperiment& experiment,
                                     const std::vector<uint16_t> &raw_expected_image,
                                     uint64_t image_number);
-    // Ask the writer whether this run's output could be written, before anything is started.
-    // Throws if it could not.
+    // Ask the writer whether this run's output could be written, throws if not
     void Preflight(const DiffractionExperiment &experiment);
     void Start(const DiffractionExperiment &experiment,
                const PixelMask &pixel_mask,
@@ -109,8 +106,7 @@ public:
     std::string GetJPEGFromBuffer(const PreviewImageSettings &settings, int64_t image_number = ImageBuffer::MaxImage);
     ImageBufferStatus GetImageBufferStatus() const;
     void ClearImageBuffer();
-    // Drop everything the previous run left readable - its status, plots, live rugnux status and
-    // buffered images - so nothing of it is served under the next run's tokens. Idle only.
+    // Drop everything the previous run left readable, so nothing of it is served under the next run's tokens. Idle only.
     void ClearResults();
 
     ImagePusherStatus GetImagePusherStatus() const;

receiver/JFJochReceiverStream.cpp

@@ -45,19 +45,14 @@ void JFJochReceiverStream::ApplyStartMessage(DiffractionExperiment &experiment,
     experiment.Detector().SensorMaterial(msg.sensor_material);
     experiment.Detector().SensorThickness_um(msg.sensor_thickness * 1e6);
     experiment.Detector().SaturationLimit(SaturationLimitFromValue(msg.saturation_value));
-    // The detector decides the corrections applied to the pixels (DECTRIS enables them by default);
-    // the outgoing start message is rebuilt from the experiment, so without copying them NXmx would
-    // record them as not applied.
+    // The outgoing start message is rebuilt from the experiment, NXmx would record these as not applied
     experiment.Detector().CountRateCorrectionApplied(msg.countrate_correction_enabled);
     experiment.Detector().CountRateCorrectionLookupTable(msg.countrate_correction_lookup_table);
     experiment.Detector().FlatfieldApplied(msg.flatfield_enabled);
     experiment.Detector().VirtualPixelInterpolationApplied(msg.virtual_pixel_interpolation_enabled);
-    // The mask written with the data is ours (the detector's plus user masking); it is never
-    // uploaded to the detector, so the pixels arrive without it applied.
+    // The mask written with the data is ours, never uploaded to the detector
     experiment.ApplyPixelMask(false);
-    // Images are forwarded byte-for-byte, so the stream's own image_dtype - not anything configured
-    // locally - decides both the width and the sign the outgoing metadata must declare. Taking only
-    // the width used to leave a signed stream declared unsigned in NXmx.
+    // Images are forwarded byte-for-byte, so the stream decides both width and sign
     experiment.Detector().BitDepthImage(msg.bit_depth_image);
     experiment.PixelSigned(msg.pixel_signed);
 }
@@ -121,8 +116,7 @@ void JFJochReceiverStream::JoinThreads() {
 
 void JFJochReceiverStream::MeasurementThread() {
     try {
-        // Wait for start message to arrive. Once the detector has reported it is done, nothing
-        // more is coming: allow a start message that is still in flight to land, then give up.
+        // Once the detector has reported it is done, a start message in flight may still land, then give up
         std::optional<std::chrono::steady_clock::time_point> give_up_at;
         auto msg = image_puller.PollImage();
 

receiver/JFJochReceiverStream.h

@@ -12,10 +12,7 @@
 #include "JFJochReceiver.h"
 #include "../image_puller/ImagePuller.h"
 
-// A collection coming from an ImagePuller: waits for the start message, takes the images on a number of
-// threads until the end message, then finishes the collection. What is done with an image is the child's.
-// The most-derived class starts the threads as the last statement of its constructor and joins them as the
-// first of its destructor, nothing here does.
+// A collection from an ImagePuller; the most-derived class starts the threads last and joins them first thing
 class JFJochReceiverStream : public JFJochReceiver {
 protected:
     using JFJochReceiver::Cancel;
@@ -23,8 +20,7 @@ protected:
     ImagePuller &image_puller;
     std::atomic<bool> end_message_received{false};
 
-    // Set when the caller reports the run is over - JFJochServices::Stop cancels the receiver silently once
-    // the detector reads back idle. A silent cancel does not interrupt images that are still being processed.
+    // A silent cancel: the detector reads back idle. Images still being processed go on.
     std::atomic<bool> detector_finished{false};
 
     // How long a start message may still be in flight after the detector reported it had finished

receiver/RugnuxLiveStatus.cpp

@@ -17,140 +17,140 @@ double RugnuxLiveStatus::Now() const {
 }
 
 void RugnuxLiveStatus::CloseStep() {
-    if (!s.steps.empty() && !s.steps.back().duration_s)
-        s.steps.back().duration_s = Now() - s.steps.back().started_s;
+    if (!snapshot.steps.empty() && !snapshot.steps.back().duration_s)
+        snapshot.steps.back().duration_s = Now() - snapshot.steps.back().started_s;
 }
 
 void RugnuxLiveStatus::Start(int64_t images_expected, std::optional<double> expected_collection_s,
                              std::chrono::steady_clock::time_point start_received) {
-    std::unique_lock ul(m);
-    s = RugnuxLiveSnapshot{};
-    s.state = RugnuxLiveState::Running;
-    s.images_expected = images_expected;
-    s.expected_collection_s = expected_collection_s;
+    std::unique_lock ul(status_mutex);
+    snapshot = RugnuxLiveSnapshot{};
+    snapshot.state = RugnuxLiveState::Running;
+    snapshot.images_expected = images_expected;
+    snapshot.expected_collection_s = expected_collection_s;
     start = start_received;
     last_image.reset();
 }
 
 void RugnuxLiveStatus::ImageReceived(std::chrono::steady_clock::time_point received) {
-    std::unique_lock ul(m);
-    ++s.images_received;
+    std::unique_lock ul(status_mutex);
+    ++snapshot.images_received;
     // Several forward threads report, so the images need not come here in the order they arrived.
     const double t = Since(received);
-    if (!s.start_to_first_image_s || t < s.start_to_first_image_s.value())
-        s.start_to_first_image_s = t;
+    if (!snapshot.start_to_first_image_s || t < snapshot.start_to_first_image_s.value())
+        snapshot.start_to_first_image_s = t;
     if (!last_image || received > last_image.value())
         last_image = received;
 }
 
 void RugnuxLiveStatus::Step(const std::string &title) {
-    std::unique_lock ul(m);
+    std::unique_lock ul(status_mutex);
     CloseStep();
-    s.steps.push_back(RugnuxLiveStep{.title = title, .started_s = Now()});
-    s.step_progress.reset();
+    snapshot.steps.push_back(RugnuxLiveStep{.title = title, .started_s = Now()});
+    snapshot.step_progress.reset();
 }
 
 void RugnuxLiveStatus::Note(const std::string &text) {
-    std::unique_lock ul(m);
-    if (s.steps.empty())
-        s.steps.push_back(RugnuxLiveStep{.title = "", .started_s = Now()});
-    s.steps.back().notes.push_back(text);
+    std::unique_lock ul(status_mutex);
+    if (snapshot.steps.empty())
+        snapshot.steps.push_back(RugnuxLiveStep{.title = "", .started_s = Now()});
+    snapshot.steps.back().notes.push_back(text);
 }
 
 void RugnuxLiveStatus::Progress(uint64_t done, uint64_t total) {
-    std::unique_lock ul(m);
+    std::unique_lock ul(status_mutex);
     if (total > 0)
-        s.step_progress = static_cast<float>(done) / static_cast<float>(total);
+        snapshot.step_progress = static_cast<float>(done) / static_cast<float>(total);
 }
 
 void RugnuxLiveStatus::Finish(RugnuxLiveState state, const std::string &message) {
-    std::unique_lock ul(m);
+    std::unique_lock ul(status_mutex);
     CloseStep();
-    s.elapsed_s = Now();
-    s.state = state;
-    s.message = message;
-    s.step_progress.reset();
+    snapshot.elapsed_s = Now();
+    snapshot.state = state;
+    snapshot.message = message;
+    snapshot.step_progress.reset();
 }
 
 void RugnuxLiveStatus::Result(const std::string &verdict, const std::string &verdict_text,
                               const std::vector<std::pair<std::string, std::string>> &summary,
                               const std::vector<std::string> &warnings, bool anomalous_signal) {
-    std::unique_lock ul(m);
-    s.verdict = verdict;
-    s.verdict_text = verdict_text;
-    s.summary = summary;
-    s.warnings = warnings;
-    s.anomalous_signal = anomalous_signal;
+    std::unique_lock ul(status_mutex);
+    snapshot.verdict = verdict;
+    snapshot.verdict_text = verdict_text;
+    snapshot.summary = summary;
+    snapshot.warnings = warnings;
+    snapshot.anomalous_signal = anomalous_signal;
 }
 
 void RugnuxLiveStatus::Model(const RugnuxLiveModel &model) {
-    std::unique_lock ul(m);
-    s.model = model;
+    std::unique_lock ul(status_mutex);
+    snapshot.model = model;
 }
 
 void RugnuxLiveStatus::Headline(const std::string &space_group, std::optional<double> d_min,
                                 std::optional<double> isa) {
-    std::unique_lock ul(m);
-    s.space_group = space_group;
-    s.d_min = d_min;
-    s.isa = isa;
+    std::unique_lock ul(status_mutex);
+    snapshot.space_group = space_group;
+    snapshot.d_min = d_min;
+    snapshot.isa = isa;
 }
 
 void RugnuxLiveStatus::Stills(const RugnuxLiveStills &stills) {
-    std::unique_lock ul(m);
-    s.stills = stills;
+    std::unique_lock ul(status_mutex);
+    snapshot.stills = stills;
 }
 
 void RugnuxLiveStatus::Latency(double last_image_to_written_s) {
-    std::unique_lock ul(m);
-    s.last_image_to_written_s = last_image_to_written_s;
+    std::unique_lock ul(status_mutex);
+    snapshot.last_image_to_written_s = last_image_to_written_s;
 }
 
 std::optional<std::chrono::steady_clock::time_point> RugnuxLiveStatus::LastImage() {
-    std::unique_lock ul(m);
+    std::unique_lock ul(status_mutex);
     if (last_image)
-        s.last_image_s = Since(last_image.value());
+        snapshot.last_image_s = Since(last_image.value());
     return last_image;
 }
 
 void RugnuxLiveStatus::Cut(const std::string &reason) {
-    std::unique_lock ul(m);
-    s.cut = reason;
-    s.cut_s = Now();
+    std::unique_lock ul(status_mutex);
+    snapshot.cut = reason;
+    snapshot.cut_s = Now();
 }
 
 void RugnuxLiveStatus::Probe(const std::string &text) {
-    std::unique_lock ul(m);
-    s.probe = text;
-    s.probe_s = Now();
+    std::unique_lock ul(status_mutex);
+    snapshot.probe = text;
+    snapshot.probe_s = Now();
 }
 
 void RugnuxLiveStatus::Warning(const std::string &text) {
-    std::unique_lock ul(m);
-    s.warnings.push_back(text);
+    std::unique_lock ul(status_mutex);
+    snapshot.warnings.push_back(text);
 }
 
 void RugnuxLiveStatus::Structure(std::shared_ptr<const RugnuxStructure> structure,
                                  std::shared_ptr<const RugnuxStructure> refined_structure) {
-    std::unique_lock ul(m);
-    s.structure = std::move(structure);
-    s.refined_structure = std::move(refined_structure);
+    std::unique_lock ul(status_mutex);
+    snapshot.structure = std::move(structure);
+    snapshot.refined_structure = std::move(refined_structure);
 }
 
 void RugnuxLiveStatus::Clear() {
-    std::unique_lock ul(m);
-    s = RugnuxLiveSnapshot{};
+    std::unique_lock ul(status_mutex);
+    snapshot = RugnuxLiveSnapshot{};
     last_image.reset();
 }
 
 RugnuxLiveSnapshot RugnuxLiveStatus::Get() const {
-    std::unique_lock ul(m);
-    RugnuxLiveSnapshot ret = s;
-    if (s.state == RugnuxLiveState::Running)
+    std::unique_lock ul(status_mutex);
+    RugnuxLiveSnapshot ret = snapshot;
+    if (snapshot.state == RugnuxLiveState::Running)
         ret.elapsed_s = Now();
 
     // Held since START, reported since the first image - or, before it has come, since now.
-    const double zero = s.start_to_first_image_s.value_or(s.state == RugnuxLiveState::Running ? ret.elapsed_s : 0.0);
+    const double zero = snapshot.start_to_first_image_s.value_or(snapshot.state == RugnuxLiveState::Running ? ret.elapsed_s : 0.0);
     ret.elapsed_s -= zero;
     for (auto &step: ret.steps)
         step.started_s -= zero;

receiver/RugnuxLiveStatus.h

@@ -14,14 +14,7 @@
 class RugnuxStructure;   // RugnuxStructure.h
 struct ProcessResult;     // rugnux/Rugnux.h
 
-// Where the live rugnux run of the last collection is, for the broker's status route: the steps it has
-// taken with what each found ("lattice found", "space group P 21 21 21"), the step it is on and how far
-// into its images, and - once it is over - its verdict and summary as the report states them, and the
-// latency from the last image to every result written.
-//
-// Times are those of the image puller - when a message came off the socket - and every time reported is
-// in seconds since the first image arrived; until it has, since now. The START message comes earlier, by
-// start_to_first_image_s. The collection's frame rate is (images_received - 1) / last_image_s.
+// Where the live rugnux run of the last collection is; times in seconds since the first image arrived
 enum class RugnuxLiveState { Idle, Running, Done, Failed, Cancelled, Skipped };
 
 struct RugnuxLiveStep {
@@ -57,9 +50,7 @@ struct RugnuxLiveRefinement {
     std::vector<RugnuxLiveBlob> blobs;     // strongest first
 };
 
-// What the run made of the model it was given (ProcessResult::model_validation and ::model_prior), in the
-// numbers the report's MODEL VALIDATION section gives. Unset values were not computed: the model was not
-// placed, or the placement was not by molecular replacement.
+// What the run made of the model it was given, as the report's MODEL VALIDATION; unset: not computed
 struct RugnuxLiveModel {
     std::string placement;                 // RIGID_BODY | MR | NOT_PLACED
     std::string model_name;
@@ -83,8 +74,7 @@ struct RugnuxLiveModel {
     std::optional<RugnuxLiveRefinement> refinement;
 };
 
-// Serial frames analysed on the GPU after a pre-pass (StillsGPUAnalysis): the pre-pass and what it decided,
-// and the analysis since.
+// Serial frames analysed on the GPU after a pre-pass (StillsGPUAnalysis)
 enum class RugnuxLivePrePass { Holding, Running, Done, Failed };
 
 struct RugnuxLiveStills {
@@ -110,14 +100,12 @@ struct RugnuxLiveSnapshot {
     double elapsed_s = 0.0;
     std::optional<float> step_progress;   // fraction of the current pass's images
     std::vector<RugnuxLiveStep> steps;
-    // The lattice probe on the first 60 degrees (JFJochReceiverRugnuxRotation::ProbeThread): an early answer
-    // to "is there a crystal", seconds into the collection; the run's own answer comes later.
+    // The lattice probe on the first 60 degrees (JFJochReceiverRugnuxRotation::ProbeThread)
     std::string probe;
     std::optional<double> probe_s;        // when it answered
     std::optional<double> start_to_first_image_s; // from the START message to the first image
     std::optional<double> last_image_s;   // when the last image arrived; set once the collection is over
-    // The analysis was cut - "deadline" (the broker's deadline_s after the last image) or "cancel" - at
-    // cut_s; what it had established went out, the steps after it were left out.
+    // The analysis was cut, "deadline" or "cancel", at cut_s
     std::string cut;
     std::optional<double> cut_s;
     std::optional<double> last_image_to_written_s;
@@ -131,16 +119,15 @@ struct RugnuxLiveSnapshot {
     std::string space_group;
     std::optional<double> d_min;          // of the merge
     std::optional<double> isa;            // where the error model determined it
-    // The placed model and its maps, once the run has placed one, and the refined model and its maps, once the
-    // run has refined it; for the viewer routes.
+    // The placed and refined models and their maps, for the viewer routes
     std::shared_ptr<const RugnuxStructure> structure;
     std::shared_ptr<const RugnuxStructure> refined_structure;
     std::optional<RugnuxLiveStills> stills;
 };
 
 class RugnuxLiveStatus {
-    mutable std::mutex m;
-    RugnuxLiveSnapshot s;
+    mutable std::mutex status_mutex;
+    RugnuxLiveSnapshot snapshot;
     // Kept since the START message, shifted to the first image by Get().
     std::chrono::steady_clock::time_point start;
     std::optional<std::chrono::steady_clock::time_point> last_image;
@@ -148,8 +135,7 @@ class RugnuxLiveStatus {
     double Now() const;
     void CloseStep();
 public:
-    // start_received, received: when the START message and the image came off the socket (ImagePullerOutput);
-    // expected_collection_s: how long the collection is set up to take, none if not known
+    // start_received, received: when the message came off the socket; expected_collection_s: none if not known
     void Start(int64_t images_expected, std::optional<double> expected_collection_s,
                std::chrono::steady_clock::time_point start_received);
     void ImageReceived(std::chrono::steady_clock::time_point received);

receiver/RugnuxStructure.cpp

@@ -51,7 +51,7 @@ bool RugnuxStructure::HasAnomalousMap() const {
 }
 
 std::string RugnuxStructure::ModelCif() const {
-    std::unique_lock ul(m);
+    std::unique_lock ul(structure_mutex);
     if (!model_cif)
         model_cif = PlacedModelMmcif(*model, cell, space_group);
     return *model_cif;
@@ -60,7 +60,7 @@ std::string RugnuxStructure::ModelCif() const {
 std::string RugnuxStructure::Map(RugnuxMapType type) const {
     if (!coefficients || (type == RugnuxMapType::Anomalous && !HasAnomalousMap()))
         return {};
-    std::unique_lock ul(m);
+    std::unique_lock ul(structure_mutex);
     std::optional<std::string> &cached = type == RugnuxMapType::TwoFoFc ? map_2fofc
                                          : type == RugnuxMapType::FoFc ? map_fofc : map_anom;
     if (cached)

receiver/RugnuxStructure.h

@@ -22,16 +22,14 @@ struct PlacedModelOutput;
 
 enum class RugnuxMapType { TwoFoFc, FoFc, Anomalous };
 
-// The placed model and the map coefficients of a live rugnux run, for the broker to serve to a viewer: the
-// model as mmCIF, the same text as <prefix>_model.cif (PlacedModelMmcif), and the maps as CCP4 map files - made in memory when
-// first asked for and kept, never written to disk.
+// The placed model and map coefficients of a live run for a viewer: the maps made in memory when first asked for
 class RugnuxStructure {
     std::shared_ptr<const gemmi::Structure> model;
     UnitCell cell;
     gemmi::SpaceGroup space_group;
     std::optional<MtzTable> coefficients;   // FWT/PHWT and DELFWT/PHDELWT, as <prefix>_maps.mtz holds them
 
-    mutable std::mutex m;
+    mutable std::mutex structure_mutex;
     mutable std::optional<std::string> model_cif;
     mutable std::optional<std::string> map_2fofc, map_fofc, map_anom;
 public:
@@ -44,20 +42,14 @@ public:
     // The coefficients include the anomalous difference map (ANOM/PHANOM): the merge kept the Bijvoet pairs.
     bool HasAnomalousMap() const;
     std::string ModelCif() const;
-    // A CCP4 map (mode 2), sampled every d_min/3, on one unit cell centred on the model - along an axis the model
-    // reaches further than the cell, on the model plus MAP_MARGIN_A instead, so that all of it has density. The
-    // header's mean and RMS are those of the cell. Empty where there are no map coefficients, or for the anomalous
-    // map none of its coefficients.
+    // A CCP4 map sampled every d_min/3, on one unit cell around the model; empty where there are no coefficients
     std::string Map(RugnuxMapType type) const;
 };
 
-// The model step's structures, as the placed model with the maps of its placement and the refined one with its own,
-// to the live status for the broker's viewer routes.
+// The model step's structures, to the live status for the broker's viewer routes
 void PublishStructures(RugnuxLiveStatus &status, const PlacedModelOutput &placed, const ModelMaps *maps,
                        const PlacedModelOutput *refined, const ModelMaps *refined_maps);
 
-// The end message's merged data, maps and models, for the writer. Every pusher serialises END into a buffer of
-// MESSAGE_SIZE_FOR_START_END, and an END that does not fit there fails the collection; so it is tried here, and one
-// that does not fit goes without them, with a warning in the status.
+// An END that does not fit MESSAGE_SIZE_FOR_START_END fails the collection: it goes without them, with a warning
 void DropResultBlocks(EndMessage &message);
 void DropResultBlocksThatDoNotFit(EndMessage &message, RugnuxLiveStatus *status, Logger &logger);

receiver/StillsGPUAnalysis.cpp

@@ -111,8 +111,7 @@ void StillsGPUAnalysis::Finish() {
         }
     }
     if (model_future.valid()) {
-        // The model step on the merge of the collection, until the deadline after the last image: the step on a
-        // provisional merge still running is cut, and this one starts from its placement.
+        // The step on the merge of the collection, until the deadline; the provisional one running is cut, this starts from it
         if (merged && model_left_out.empty()) {
             const auto deadline = last_image.value_or(finish_start) + std::chrono::duration_cast<
                     std::chrono::steady_clock::duration>(std::chrono::duration<double>(deadline_s));
@@ -155,17 +154,15 @@ void StillsGPUAnalysis::ReportOutcome() {
 }
 
 void StillsGPUAnalysis::PrePassThread() {
-    const auto masked = [](const PixelMask &m) {
-        return std::count_if(m.GetMask().begin(), m.GetMask().end(), [](uint32_t v) { return v != 0; });
+    const auto masked = [](const PixelMask &pixel_mask) {
+        return std::count_if(pixel_mask.GetMask().begin(), pixel_mask.GetMask().end(), [](uint32_t v) { return v != 0; });
     };
     setup.experiment = experiment;
     setup.mask = mask;
     const int64_t all_images = experiment.GetImageNum();
     std::optional<StillsPrePassResult> result;
     std::chrono::steady_clock::time_point prepass_start = std::chrono::steady_clock::now();
-    // At a low hit rate the first frames may not be enough to fit the geometry on, or may not hold a crystal: with
-    // a cell to fit, the pre-pass runs again on all that is held once that has doubled, until the fit has its
-    // frames, or it holds hold_limit frames (the last try), or the collection ended.
+    // With a cell, the pre-pass runs again on all that is held once it has doubled, until the fit has its frames or the end
     const int64_t most_held = std::min(all_images, hold_limit);
     for (int64_t needed = std::min(PREPASS_IMAGES, most_held); ;) {
         std::map<int64_t, std::shared_ptr<HeldImage>> sample;
@@ -187,14 +184,14 @@ void StillsGPUAnalysis::PrePassThread() {
         Publish();
         try {
             const auto t0 = std::chrono::steady_clock::now();
-            auto r = RunStillsPrePass(start, sample, static_cast<int>(std::thread::hardware_concurrency()));
+            auto ret = RunStillsPrePass(start, sample, static_cast<int>(std::thread::hardware_concurrency()));
             logger.Info("Pre-pass on {} images ({:.1f} s): beam centre ({:.2f}, {:.2f}) -> ({:.2f}, {:.2f}) px, "
-                        "distance {:.2f} -> {:.2f} mm on {} frames, masked pixels {} -> {}", r.images,
+                        "distance {:.2f} -> {:.2f} mm on {} frames, masked pixels {} -> {}", ret.images,
                         std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count(),
-                        experiment.GetBeamX_pxl(), experiment.GetBeamY_pxl(), r.beam_x_pxl, r.beam_y_pxl,
-                        experiment.GetDetectorDistance_mm(), r.distance_mm, r.geometry_frames,
-                        masked(mask), masked(r.mask));
-            result = std::move(r);
+                        experiment.GetBeamX_pxl(), experiment.GetBeamY_pxl(), ret.beam_x_pxl, ret.beam_y_pxl,
+                        experiment.GetDetectorDistance_mm(), ret.distance_mm, ret.geometry_frames,
+                        masked(mask), masked(ret.mask));
+            result = std::move(ret);
         } catch (const std::exception &e) {
             logger.Warning("Pre-pass on {} images failed - {}", sample.size(), e.what());
         }
@@ -215,19 +212,18 @@ void StillsGPUAnalysis::PrePassThread() {
     const bool failed = !result;
     int geometry_frames = 0;
     if (result) {
-        const auto &r = *result;
-        setup.experiment.BeamX_pxl(r.beam_x_pxl).BeamY_pxl(r.beam_y_pxl).DetectorDistance_mm(r.distance_mm);
-        setup.mask = r.mask;
-        geometry_frames = r.geometry_frames;
-        const auto &m = r.mask.GetMask();
+        setup.experiment.BeamX_pxl(result->beam_x_pxl).BeamY_pxl(result->beam_y_pxl).DetectorDistance_mm(result->distance_mm);
+        setup.mask = result->mask;
+        geometry_frames = result->geometry_frames;
+        const auto &result_mask = result->mask.GetMask();
         std::unique_lock ul(live_mutex);
-        live.beam_x_pxl = r.beam_x_pxl;
-        live.beam_y_pxl = r.beam_y_pxl;
-        live.distance_mm = r.distance_mm;
-        live.shadow_pixels = std::count_if(m.begin(), m.end(), [](uint32_t v) {
+        live.beam_x_pxl = result->beam_x_pxl;
+        live.beam_y_pxl = result->beam_y_pxl;
+        live.distance_mm = result->distance_mm;
+        live.shadow_pixels = std::count_if(result_mask.begin(), result_mask.end(), [](uint32_t v) {
             return (v & (1u << PixelMask::BeamStopPixelBit)) != 0;
         });
-        live.bad_pixels = std::count_if(m.begin(), m.end(), [](uint32_t v) {
+        live.bad_pixels = std::count_if(result_mask.begin(), result_mask.end(), [](uint32_t v) {
             return (v & (1u << PixelMask::HotPixelBit)) != 0;
         });
     } else
@@ -259,13 +255,12 @@ void StillsGPUAnalysis::PrePassThread() {
         }
         sink.StillsReady(setup);
     } catch (const std::exception &e) {
-        // Nothing can be written without the start message: the frames are dropped (AnalysisThread), and
-        // the collection is stopped.
-        const auto *j = dynamic_cast<const JFJochException *>(&e);
-        const JFJochException error = j ? *j : JFJochException(JFJochExceptionCategory::WrongDAQState,
+        // Nothing can be written without the start message: the frames are dropped, the collection is stopped
+        const auto *jfjoch_error = dynamic_cast<const JFJochException *>(&e);
+        const JFJochException error = jfjoch_error ? *jfjoch_error : JFJochException(JFJochExceptionCategory::WrongDAQState,
                                                                std::string("GPU analysis setup failed: ") + e.what());
         logger.ErrorException(error);
-        setup_error = j ? std::current_exception() : std::make_exception_ptr(error);
+        setup_error = jfjoch_error ? std::current_exception() : std::make_exception_ptr(error);
         sink.StillsSetupFailed(error);
         prepass_done.count_down();
         return;
@@ -372,25 +367,24 @@ const std::optional<ModelStepResult> &StillsGPUAnalysis::GetModel() const {
 void StillsGPUAnalysis::MergeThread() {
     constexpr int64_t PREPARE_EVERY = 200;   // frames between two Prepare calls after a checkpoint
     const int64_t images = experiment.GetImageNum();
-    // The cell merged in, as rugnux merges it: the consensus of the frames indexed by the first merge (the
-    // collection's cell only picks the frames that are merged).
+    // The cell merged in, as rugnux merges it: the consensus of the frames indexed by the first merge
     bool cell_set = false;
     const auto have_cell = [&] {
         if (!cell_set)
-            if (const auto c = indexer->GetConsensusUnitCell()) {
-                merge_experiment.SetUnitCell(*c);
-                merge->Cell(*c);
+            if (const auto cell = indexer->GetConsensusUnitCell()) {
+                merge_experiment.SetUnitCell(*cell);
+                merge->Cell(*cell);
                 cell_set = true;
                 if (model) {
-                    const ModelStepGroup g = ModelStepSpaceGroup(merge_experiment.GetGemmiSpaceGroup(), *c, *model);
-                    if (!g.space_group) {
-                        model_left_out = "the model is left out: " + g.reason;
+                    const ModelStepGroup group = ModelStepSpaceGroup(merge_experiment.GetGemmiSpaceGroup(), *cell, *model);
+                    if (!group.space_group) {
+                        model_left_out = "the model is left out: " + group.reason;
                         logger.Warning("{}", model_left_out);
                         status.Warning(model_left_out);
                     } else if (!merge_experiment.GetGemmiSpaceGroup()) {
-                        merge_experiment.SetSpaceGroup(*g.space_group);
-                        merge->SpaceGroup(*g.space_group);
-                        logger.Info("Merged in the model's space group {}", g.space_group->xhm());
+                        merge_experiment.SetSpaceGroup(*group.space_group);
+                        merge->SpaceGroup(*group.space_group);
+                        logger.Info("Merged in the model's space group {}", group.space_group->xhm());
                     }
                 }
             }
@@ -398,17 +392,17 @@ void StillsGPUAnalysis::MergeThread() {
     };
     size_t next = 0;
     try {
-        const auto add = [&](int64_t n, IntegrationOutcome &&o) {
-            merge->Add(n, std::move(o));
+        const auto add = [&](int64_t n, IntegrationOutcome &&outcome) {
+            merge->Add(n, std::move(outcome));
             if (next < CHECKPOINTS.size()
                 && merge->Added() >= std::llround(CHECKPOINTS[next] * static_cast<double>(images))) {
                 if (have_cell()) {
-                    StillsMergeOutput c = merge->Checkpoint(MERGE_THREADS);
+                    StillsMergeOutput checkpoint = merge->Checkpoint(MERGE_THREADS);
                     const std::string title = fmt::format("{:.0f}% of the images (provisional)",
                                                           100.0 * CHECKPOINTS[next]);
-                    PublishMerge(c, title);
+                    PublishMerge(checkpoint, title);
                     if (model_future.valid() && model_left_out.empty())
-                        SubmitModel(c, title, false);
+                        SubmitModel(checkpoint, title, false);
                 }
                 ++next;
             } else if (next > 0 && merge->Added() % PREPARE_EVERY == 0)
@@ -429,13 +423,13 @@ void StillsGPUAnalysis::MergeThread() {
 }
 
 void StillsGPUAnalysis::PublishMerge(const StillsMergeOutput &out, const std::string &title) {
-    const auto &o = out.merge.statistics.overall;
+    const auto &overall = out.merge.statistics.overall;
     const std::string text = fmt::format("Merge of {}: {} crystals, {:.2f} A, CC1/2 {:.3f}, R_split {:.3f}, "
                                          "completeness {:.1f}%, <I/sigma> {:.1f} ({:.1f} s)", title, out.images,
-                                         o.d_min, o.cc_half, o.r_split,
-                                         o.possible_unique_reflections > 0
-                                         ? 100.0 * o.unique_reflections / o.possible_unique_reflections : 0.0,
-                                         o.mean_i_over_sigma, out.seconds);
+                                         overall.d_min, overall.cc_half, overall.r_split,
+                                         overall.possible_unique_reflections > 0
+                                         ? 100.0 * overall.unique_reflections / overall.possible_unique_reflections : 0.0,
+                                         overall.mean_i_over_sigma, out.seconds);
     logger.Info("{}", text);
     status.Note(text);
 }
@@ -515,49 +509,49 @@ void StillsGPUAnalysis::ModelThread() {
             settings.cut = &step_cut;
             if (job.final)
                 settings.step = [this](const std::string &title) { status.Step(title); };
-            const DiffractionExperiment &x = merge_experiment;
-            ModelStepResult r = RunModelStep(job.merged, x.GetUnitCell().value(), x.GetSpaceGroupOrP1(),
-                                                            *model, x.GetWavelength_A(),
-                                                            x.GetScalingSettings().GetMergeFriedel(), job.shells,
-                                                            settings, model_placed ? &*model_placed : nullptr, logger);
-            PublishModel(r, job.title);
-            if (r.placed_model)
-                model_placed = r;
+            ModelStepResult result = RunModelStep(job.merged, merge_experiment.GetUnitCell().value(),
+                                                  merge_experiment.GetSpaceGroupOrP1(), *model,
+                                                  merge_experiment.GetWavelength_A(),
+                                                  merge_experiment.GetScalingSettings().GetMergeFriedel(), job.shells,
+                                                  settings, model_placed ? &*model_placed : nullptr, logger);
+            PublishModel(result, job.title);
+            if (result.placed_model)
+                model_placed = result;
             if (job.final)
-                model_final = std::move(r);
+                model_final = std::move(result);
         } catch (const std::exception &e) {
             logger.Warning("Model step on the merge of {} failed: {}", job.title, e.what());
         }
     }
 }
 
-void StillsGPUAnalysis::PublishModel(const ModelStepResult &r, const std::string &title) {
+void StillsGPUAnalysis::PublishModel(const ModelStepResult &result, const std::string &title) {
     std::string text = fmt::format("Model on the merge of {}: ", title);
-    if (r.placement == "NOT_PLACED")
+    if (result.placement == "NOT_PLACED")
         text += "not placed";
     else
-        text += fmt::format("placed by {}{}", r.placement == "MR" ? "molecular replacement" : "its rigid body",
-                            r.warm ? " (from the earlier placement)" : "");
-    if (r.validation && r.validation->ok)
-        text += fmt::format(", R-work {:.3f}, R-free {:.3f}", r.validation->r_work, r.validation->r_free);
-    if (r.refinement && r.refinement->ok)
-        text += fmt::format("; refined{} R-work {:.3f}, R-free {:.3f}, {} blobs", r.refinement->cancelled
-                            ? " (cut short)" : "", r.refinement->r_work, r.refinement->r_free,
-                            r.refinement->blobs.size());
-    for (const auto &l : r.left_out)
-        text += "; " + l + " left out";
-    text += fmt::format(" ({:.1f} s placement, {:.1f} s refinement)", r.placement_s, r.refinement_s);
+        text += fmt::format("placed by {}{}", result.placement == "MR" ? "molecular replacement" : "its rigid body",
+                            result.warm ? " (from the earlier placement)" : "");
+    if (result.validation && result.validation->ok)
+        text += fmt::format(", R-work {:.3f}, R-free {:.3f}", result.validation->r_work, result.validation->r_free);
+    if (result.refinement && result.refinement->ok)
+        text += fmt::format("; refined{} R-work {:.3f}, R-free {:.3f}, {} blobs", result.refinement->cancelled
+                            ? " (cut short)" : "", result.refinement->r_work, result.refinement->r_free,
+                            result.refinement->blobs.size());
+    for (const auto &left_out : result.left_out)
+        text += "; " + left_out + " left out";
+    text += fmt::format(" ({:.1f} s placement, {:.1f} s refinement)", result.placement_s, result.refinement_s);
     logger.Info("{}", text);
     status.Note(text);
-    if (const auto m = LiveModel(ModelStepReport(r)))
-        status.Model(*m);
-    if (!r.placed_model)
+    if (const auto live_model = LiveModel(ModelStepReport(result)))
+        status.Model(*live_model);
+    if (!result.placed_model)
         return;
     std::optional<PlacedModelOutput> refined;
-    if (r.refinement && r.refinement->ok && r.refinement->model)
-        refined = PlacedModelOutput{r.refinement->model, r.placed_model->cell, r.placed_model->space_group};
-    PublishStructures(status, *r.placed_model, r.maps.get(), refined ? &*refined : nullptr,
-                      refined && r.refinement->maps ? r.refinement->maps.get() : nullptr);
+    if (result.refinement && result.refinement->ok && result.refinement->model)
+        refined = PlacedModelOutput{result.refinement->model, result.placed_model->cell, result.placed_model->space_group};
+    PublishStructures(status, *result.placed_model, result.maps.get(), refined ? &*refined : nullptr,
+                      refined && result.refinement->maps ? result.refinement->maps.get() : nullptr);
 }
 
 void StillsGPUAnalysis::FillEndMessage(EndMessage &message) const {
@@ -565,25 +559,24 @@ void StillsGPUAnalysis::FillEndMessage(EndMessage &message) const {
     message.refined_beam_center_x = GetBeamX_pxl();
     message.refined_beam_center_y = GetBeamY_pxl();
     message.refined_detector_distance_mm = GetDetectorDistance_mm();
-    // The merged data, for the writer to put beside the master file: in the model's frame where a model was
-    // placed against them, with its maps, the model as placed and as refined, and the refined model's blobs.
-    if (const auto &mx = GetModel()) {
-        DiffractionExperiment x(GetMergeExperiment());
-        x.SetSpaceGroup(mx->space_group);
-        message.merged = MergedMtzTable(mx->merged, mx->cell, x, OutputProvenance{});
-        if (mx->maps)
-            message.map_coefficients = mx->maps->coefficients;
-        if (mx->placed_model)
-            message.placed_model = PlacedModelFromStructure(*mx->placed_model->model, mx->placed_model->cell,
-                                                            mx->placed_model->space_group.xhm());
-        if (const auto &rf = mx->refinement; rf && rf->ok && rf->model) {
-            message.refined_model = PlacedModelFromStructure(*rf->model, mx->cell, mx->space_group.xhm());
-            if (rf->maps)
-                message.refined_map_coefficients = rf->maps->coefficients;
-            message.blobs = DensityBlobsToMessage(rf->blobs);
+    // In the model's frame where a model was placed against them
+    if (const auto &model_step = GetModel()) {
+        DiffractionExperiment model_experiment(GetMergeExperiment());
+        model_experiment.SetSpaceGroup(model_step->space_group);
+        message.merged = MergedMtzTable(model_step->merged, model_step->cell, model_experiment, OutputProvenance{});
+        if (model_step->maps)
+            message.map_coefficients = model_step->maps->coefficients;
+        if (model_step->placed_model)
+            message.placed_model = PlacedModelFromStructure(*model_step->placed_model->model, model_step->placed_model->cell,
+                                                            model_step->placed_model->space_group.xhm());
+        if (const auto &refinement = model_step->refinement; refinement && refinement->ok && refinement->model) {
+            message.refined_model = PlacedModelFromStructure(*refinement->model, model_step->cell, model_step->space_group.xhm());
+            if (refinement->maps)
+                message.refined_map_coefficients = refinement->maps->coefficients;
+            message.blobs = DensityBlobsToMessage(refinement->blobs);
         }
-    } else if (const auto &m = GetMerged(); m && GetMergeExperiment().GetUnitCell())
-        message.merged = MergedMtzTable(m->merge.merged, *GetMergeExperiment().GetUnitCell(),
+    } else if (const auto &merge_output = GetMerged(); merge_output && GetMergeExperiment().GetUnitCell())
+        message.merged = MergedMtzTable(merge_output->merge.merged, *GetMergeExperiment().GetUnitCell(),
                                         GetMergeExperiment(), OutputProvenance{});
     if (message.merged)
         DropResultBlocksThatDoNotFit(message, &status, logger);

receiver/StillsGPUAnalysis.h

@@ -44,33 +44,10 @@ public:
     virtual void StillsSetupFailed(const JFJochException &error) = 0;
 };
 
-// Serial frames analysed on the GPU (MXAnalysisWithoutFPGA), after a pre-pass on
-// the first PREPASS_IMAGES (RunStillsPrePass) has decided the beam centre and the mask they are analysed
-// with. Every frame given before the pre-pass is done is held; the sink is told once it is, with the
-// geometry and mask the frames are then analysed at, and gets each frame once analysed. The run is reported in
-// `status` (pre-pass, its decisions, the analysis).
-// Every frame given is taken and analysed: the slots of the image buffer are what bounds them. The pre-pass holds
-// and fits on at most `max_held` of them; those arriving beyond that, or while it runs, wait for the analysis.
-// Where the setup after the pre-pass fails, the sink is told at once, the frames are
-// dropped unanalysed and Finish throws the error. An analysis thread that fails passes its frames on
-// unanalysed, and the run is reported as failed.
-// The indexed frames are merged as `rugnux --mode scale` merges them (StillsLiveMerge), beside the analysis:
-// a provisional merge at each of CHECKPOINTS of the images, on MERGE_THREADS threads, reported as a step
-// of the status, and the merge of the whole collection once the last frame is analysed, on every thread,
-// before Finish returns (GetMerged). The cell is the run's consensus at the first merge, as rugnux merges in
-// it (the collection's cell picks the frames merged); the space group the collection's, else P1.
-// merge = false leaves it out.
-// With a cell given, the pre-pass runs again on all the frames held each time they have doubled, until the geometry
-// fit has its frames, `max_held` frames are held or the collection ends; the analysis starts after that and
-// catches up from the frames held. The geometry is fixed when the pre-pass ends: every frame is analysed and merged at it. Where the pre-pass could
-// not refine it (too few of the frames it had indexed: a small image buffer, a serial hit rate, a chip whose first
-// frames are empty; or no cell given), the header's distance stays for the whole run, with a warning in the log
-// and the status.
+// Serial frames analysed on the GPU at the geometry of a pre-pass (fixed from there on), and merged beside the analysis
 class StillsGPUAnalysis {
 public:
-    // Enough frames for the distance to be refined at a serial hit rate: of 100 frames of a collection indexed
-    // at 18%, 7 indexed at a distance 2.6% off and the geometry refinement did not run; of 500, 34 did and
-    // refined it to 0.4%. They wait in the image buffer.
+    // Enough frames for the distance to be refined at a serial hit rate; they wait in the image buffer
     constexpr static int64_t PREPASS_IMAGES = 500;
     constexpr static std::array<double, 2> CHECKPOINTS = {0.5, 0.75};
     constexpr static int MERGE_THREADS = 8;
@@ -83,11 +60,9 @@ public:
     ~StillsGPUAnalysis();
 
     void Hold(const std::shared_ptr<HeldImage> &image);
-    // No frame comes after this: the pre-pass runs on what came, if it has not, and every frame is
-    // analysed before it returns. Throws where the setup after the pre-pass failed.
+    // No frame comes after this: every frame is analysed before it returns. Throws if the setup after the pre-pass failed.
     void Finish();
-    // After Finish, once the end message is out and the writer is done: the status gets the run's outcome.
-    // Without it the outcome is reported when the analysis goes.
+    // After Finish, once everything is written; without it the outcome is reported when the analysis goes
     void ReportOutcome();
     // POST /cancel: the model step stops where it is, as at the deadline, and the frames are still analysed and merged.
     void Cut(const std::string &reason);
@@ -102,8 +77,7 @@ public:
     [[nodiscard]] const DiffractionExperiment &GetMergeExperiment() const;
     // The model step on the merge of the collection, where one was made.
     [[nodiscard]] const std::optional<ModelStepResult> &GetModel() const;
-    // What the analysis found, for the collection's end message: the cell, geometry, and the merged data for the
-    // writer to put beside the master file, in the model's frame where a model was placed against them.
+    // The cell, geometry and merged data, in the model's frame where a model was placed
     void FillEndMessage(EndMessage &message) const;
 
 private:
@@ -149,13 +123,7 @@ private:
     std::optional<StillsMergeOutput> merged;
     std::future<void> merge_future;
 
-    // Where a model is given, it is placed against each merge, mapped and refined (RunModelStep), on a
-    // thread of its own: the provisional merges' beside the analysis, published to the status as they are made, and the
-    // collection's in Finish, which waits for it until deadline_s after the last image (0: no deadline) and then cuts it -
-    // a refinement running is cut short, a molecular replacement or a refinement not started is left out. A step still
-    // running when the next merge is made is cut the same way, and one not started is replaced by it. Each step starts
-    // from the placement of the last one that placed the model. Without a space group stated, the frames are merged in
-    // the model's where its cell is the run's consensus (ModelStepSpaceGroup); otherwise the model is left out.
+    // Placed against each merge, mapped and refined on a thread of its own; Finish waits until deadline_s, then cuts it
     const std::shared_ptr<const RugnuxModel> model;
     const double deadline_s;
     std::string model_left_out;   // why the model is left out; set by the merge thread before any step
@@ -180,7 +148,7 @@ private:
     void SubmitModel(StillsMergeOutput &out, const std::string &title, bool final);
     void StopModel(bool cut);
     void ModelThread();
-    void PublishModel(const ModelStepResult &r, const std::string &title);
+    void PublishModel(const ModelStepResult &result, const std::string &title);
 
     void StartPrePass();
     void PrePassThread();

receiver/StillsPrePass.cpp

@@ -54,8 +54,7 @@ StillsPrePassResult RunStillsPrePass(const StartMessage &start, const std::map<i
     Rugnux::ReleasePinnedPool();
 
     return StillsPrePassResult{
-        // Where the beam lands: the receiver analyses with the detector square to it, as stated, and a tilt
-        // the run refined moves its PONI, not this.
+        // Where the beam lands: a tilt the run refined moves its PONI, not this
         .beam_x_pxl = static_cast<float>(result.used_direct_beam_x_pxl),
         .beam_y_pxl = static_cast<float>(result.used_direct_beam_y_pxl),
         .distance_mm = result.used_distance_mm,

receiver/StillsPrePass.h

@@ -11,9 +11,7 @@
 #include "../common/PixelMask.h"
 #include "../common/ZeroCopyReturnValue.h"
 
-// An image the receiver holds until it is analysed: its message, with the image pointing into `slot`, the
-// image buffer slot the compressed frame was written to. The slot goes back to the buffer, unsent, with the
-// last copy of the image held.
+// An image held until analysed, pointing into `slot`, which goes back unsent with the last copy
 struct HeldImage {
     DataMessage message;
     ZeroCopyReturnValue *slot = nullptr;
@@ -40,9 +38,6 @@ struct StillsPrePassResult {
     int geometry_frames = 0;   // the indexed frames the distance and centre were refined on; 0: kept as stated
 };
 
-// The first frames of a serial collection through rugnux, as `rugnux --no-merge` on them: the beam centre
-// it decides (from the scattered background, and from the frames that index where there are any), the
-// distance (refined where a cell is given) and the mask it ends with (the collection's, plus the beam-stop
-// shadow and the defective pixels). `images` by number; the start message is the collection's.
+// The first frames of a serial collection through rugnux: beam centre, distance (with a cell) and mask it decides
 StillsPrePassResult RunStillsPrePass(const StartMessage &start, const std::map<int64_t, std::shared_ptr<HeldImage>> &images,
                                      int nthreads);