Jungfraujoch/docs/review/03-data-path.md

Review: data path (receiver / pushers / devices)

Scope

Streaming data path of Jungfraujoch:

• receiver/ — JFJochReceiver (base), JFJochReceiverFPGA, JFJochReceiverLite, FrameTransformation, LossyFilter, JFJochReceiverPlots, JFJochReceiverService, JFJochReceiverCurrentStatus, ImageMetadata.
• image_pusher/ — TCPStreamPusher (priority: zero-copy buffer-reuse fix), ImagePusher base.
• Supporting plumbing in common/: ImageBuffer, ZeroCopyReturnValue, ThreadSafeFIFO.
• image_puller/, acquisition_device/, jungfrau/, preview/, detector_control/ skimmed for the data-flow and ownership picture (not exhaustively audited).

Vendored libs and broker/gen/ treated as black boxes. ffbidx reached only transitively (via the indexer); no direct touchpoints in this subsystem — see final section.

Architecture overview (threading & data flow)

There are two receiver implementations sharing the abstract JFJochReceiver. The base owns all the cross-cutting state — counters (images_collected/sent/skipped, compressed sizes), MovingAverage stat windows, the ImageBuffer and ImagePusher references, the ScanResultGenerator, LossyFilter, indexer, and the SendStartMessage/ SendEndMessage/GetStatus machinery (JFJochReceiver.cpp:108-210). Each subclass adds its own ingestion threads.

FPGA path (JFJochReceiverFPGA): one AcquireThread per data stream drives the FPGA, plus a pool of FrameTransformationThreads (count = forward_and_sum_nthreads). Each transformation thread pulls an image number from a ThreadSafeFIFO<int64_t> work queue (images_to_go), waits on AcquisitionDevice::Counters for the frame, pulls per-module DeviceOutput straight from the device buffer, runs metadata/azimuthal/spot-finding analysis, geometry-transforms + compresses into an ImageBuffer slot, then hands the slot to the pusher (JFJochReceiverFPGA.cpp:272-503). Frame ownership in the device buffer is released explicitly via FrameBufferRelease. A separate FinalizeMeasurement future joins all worker futures, sends the END message, and finalizes the writer (:550-588). CPU-summation mode fans each image out to nested SummationThreads.

Lite path (JFJochReceiverLite): a MeasurementThread waits for the upstream START message off an ImagePuller, configures the experiment from it, then a pool of DataAnalysisThreads each poll the puller's FIFO directly for DATA/END/CALIBRATION messages, analyze, and forward into ImageBuffer slots (JFJochReceiverLite.cpp:227-354). There is no explicit work-distribution queue — every analysis thread competes on image_puller.PollImage(), and the END message is observed via the shared end_message_received flag.

Buffer handoff & zero-copy. ImageBuffer is a fixed ring of fixed-size slots. GetImageSlot() returns a ZeroCopyReturnValue* (pointer into a stable std::vector<ZeroCopyReturnValue>), the worker serializes/compresses the CBOR image directly into the slot, marks ReadyToSend(), then calls image_pusher.SendImage(*loc). For TCP the slot is not copied: TCPStreamPusher::SendImage(ZeroCopyReturnValue&) enqueues the slot pointer into a per-connection ThreadSafeFIFO; the per-connection WriterThread sends it and calls z->release() (returning the slot to the ring) only after the bytes are handed to the kernel — and for MSG_ZEROCOPY sends, only after the kernel errqueue completion notification (TCPStreamPusher.cpp:351-388,544-568). The synchronous SendImage(ptr,size,number) overload takes a transient caller buffer (z == nullptr) and must never zero-copy it — this is the recently-fixed hazard.

Plot production (JFJochReceiverPlots): both receivers call plots.Add(message, az_int_profile) per image from the worker threads. Internally each metric is a StatusVector (documented as internally thread-safe), so the per-metric appends need no outer lock; only the std::map<string,ROIStatus> (roi_m, a shared_mutex) and the az_int_profile / XFEL maps (m) are mutex-guarded (JFJochReceiverPlots.h:34-88). The broker pulls snapshots via GetPlots/GetPlotRaw/GetAzIntProfile, and GetStatus reads aggregate means (indexing rate, bkg estimate). So plot data is produced incrementally on the hot path and read concurrently by broker/UI threads — characterized here; frontend↔broker line is out of scope.

Findings

[High] LossyFilter::ApplyFilter mutates shared RNG/counter but the body is unsynchronized — receiver/LossyFilter.cpp:19-29

Category: Bug ApplyFilter is called concurrently from every FPGA FrameTransformationThread and every Lite DataAnalysisThread (JFJochReceiverFPGA.cpp:438, JFJochReceiverLite.cpp:308). RollDice() does lock random_m, but the rest of ApplyFilter does not: message.number = image_number++ reads-modifies-writes the std::atomic<int64_t> image_number via ++ (atomic, OK on its own) but the decision to renumber and the renumber itself are not atomic together. Two threads can interleave between RollDice() returning true and the image_number++, so the renumbered sequence is fine numerically (atomic post-increment) but is assigned in nondeterministic order relative to original image order — filtered serial-MX output image numbers are not guaranteed monotonic/contiguous per source order. More importantly the p == 1.0 fast path returns without touching image_number, while the p < 1.0 path renumbers from a separate counter, so for 0 < p < 1 the emitted message.number is decoupled from the original frame number across threads with no ordering guarantee. Worth confirming this matches the intended serial-MX renumbering contract; if monotonic per-arrival numbering is expected, the dice-roll + assignment needs to be one critical section.

[High] images_sent is incremented at enqueue time, not send time (TCP) — receiver/JFJochReceiverFPGA.cpp:484-485, JFJochReceiverLite.cpp:331-332

Category: Bug / Inconsistency Both receivers do image_pusher.SendImage(*loc); ++images_sent; with the comment "Handle case when image not sent properly". For the TCP pusher SendImage(ZeroCopyReturnValue&) is asynchronous — it only enqueues onto the per-connection FIFO and returns void; the actual send can later fail (broken socket → z->release() with no send, TCPStreamPusher.cpp:552-555,1071-1094). So images_sent over-counts whenever a connection breaks mid-run or the 2-second enqueue deadline expires and the slot is dropped. The authoritative "written" count comes from GetImagesWritten() = total_data_acked_ok (ACK-based), so the receiver-side images_sent and the pusher-side acked count can disagree; the "wrong number of images" symptom the zero-copy commit chased can also surface here. The file-based pushers return a bool from SendImage(z) (via the base overload) but that bool is discarded for the ZeroCopyReturnValue overload, which is void — there is no way for the caller to know the async send failed.

[Medium] ZeroCopyCompletionThread reads c->zc_pending.empty() without holding zc_mutex — image_pusher/TCPStreamPusher.cpp:392

Category: Bug (data race) The loop condition while (c->active || !c->zc_pending.empty()) inspects std::deque::empty() without zc_mutex, while EnqueueZeroCopyPending/ReleaseCompletedZeroCopy/ForceReleasePendingZeroCopy mutate the deque under the lock (:362-381,351-360). Concurrent push_back/pop_front against an unsynchronized empty() is a data race (UB), even if in practice it only affects the loop's exit timing. Since StopDataCollectionThreads already drains via WaitForZeroCopyDrain + ForceReleasePendingZeroCopy before joining zc_future, the unlocked check is also redundant for correctness — gating purely on c->active (atomic) and letting the drain/force path handle the tail would remove the race. The same loop drives c->fd.load() and c->broken (both atomic, fine).

[Medium] FPGA SendStartMessage runs before acquisition threads, but END/finalize ordering differs from Lite — receiver/JFJochReceiverFPGA.cpp:111-137 vs JFJochReceiverLite.cpp:94-157

Category: Architecture / Inconsistency The two receivers implement the same lifecycle (start msg → images → end msg → finalize writer → check buffer returned) but in structurally different places and orders. FPGA sends START in the constructor before spawning acquire threads and runs END/finalize in a dedicated FinalizeMeasurement future joined by StopReceiver. Lite does START and END inside its single MeasurementThread, and crucially Lite calls CheckIfBufferReturned before image_pusher.Finalize() (:141-149) while FPGA also does buffer-return-then-Finalize (:575-582) — those match, but the surrounding error handling diverges: Lite's MeasurementThread throws out of the future on buffer-return timeout, FPGA's FinalizeMeasurement also throws, yet FPGA additionally cancels devices on END. The duplicated "send end / check buffer / finalize / set status" tail (≈15 lines) is copy-pasted with subtle drift; a shared protected helper on the base (e.g. FinalizeWriterAndBuffer()) would remove the divergence risk. This is the clearest instance of the FPGA-vs-Lite copy-paste-with-drift the review asked to look for.

[Medium] WriterThread does not break the loop on a broken connection — image_pusher/TCPStreamPusher.cpp:544-568

Category: Smell / timeout correctness Once c->broken is set, the writer keeps looping, draining the queue element-by-element and calling e.z->release() per element (:552-555) until it finally sees e.end. Each GetBlocking() blocks until something is enqueued, so during the 2s producer enqueue window (SendImage PutTimeout loop, :1076-1089) the thread spins through release-and-continue. Functionally it drains correctly, but on a broken socket it would be cheaper and clearer to flush the queue once and exit; as written, shutdown relies on the producer eventually pushing {.end=true} via StopDataCollectionThreads. Confirm there is no path where active is cleared and the {.end} sentinel fails to enqueue (the PutTimeout 200ms fallback at :851-856 clears + re-Puts, which looks correct, but the interaction is subtle).

[Low] JFJochReceiverFPGA::AcquireThread logs the same exception twice — receiver/JFJochReceiverFPGA.cpp:192-196

Category: Inconsistency In the second catch block, logger.ErrorException(e); Cancel(e); logger.ErrorException(e); logs e twice (and Cancel(e) itself logs again at JFJochReceiver.cpp:221-222), so a single device error produces three log entries. The first catch block (:179-184) logs it once. Minor, but inconsistent and noisy.

[Low] JFJochReceiver::GetStatus() reads many non-atomic / mutex-guarded members without their locks — receiver/JFJochReceiver.cpp:80-105

Category: Bug (benign race) / Inconsistency GetStatus() is const and is called both from worker threads (after every image) and indirectly from the broker via current_status. It reads max_image_number_sent (guarded elsewhere by max_image_number_sent_mutex) and max_delay (guarded by max_delay_mutex) without taking those mutexes, and reads plain int64_t/std::optional members non-atomically. The atomic counters are fine; the mutex-guarded scalars are technically racy reads. In practice these are status/telemetry values so a torn read is cosmetic, but it is inconsistent with the locking the same fields get on the write side. Either make them atomic or take the locks.

[Low] IsConnectionAlive is racy as a pre-send gate — image_pusher/TCPStreamPusher.cpp:169-191, used at :1047,1199

Category: Smell SendImage(ptr,size,number) checks IsConnectionAlive(c) then sends; the connection can break between the check and the send, which is unavoidable and handled by SendAll's error paths — so the check is an optimization, not a guarantee. SendCalibration re-checks broken/connected/active four times before and after taking send_mutex (:1165-1203); this is defensive but verbose (≈40 lines of repeated guards). A single re-check under the lock would be equivalent given broken/connected/active are atomics. Simplification opportunity, not a bug.

[Low] ImageBuffer::GetImageSlot waits on a preview reader while holding the global mutex — common/ImageBuffer.cpp:111-135

Category: Architecture When a recycled slot still has readers > 0 (a preview/JPEG copy in flight), GetImageSlot does cv_preview_done.wait(ul, …) while holding m. The wait releases m, so it is not a self-deadlock, but it stalls every other producer/consumer that needs the buffer mutex (status reads, ReadyToSend, ReleaseSlot) until the preview memcpy of one image completes. GetImage does drop the lock for its memcpy (:225-230), so the window is just the copy duration, but on a large image at high frame rate this serializes the hot path behind a UI read. Worth noting as a latency coupling between preview and acquisition.

[Nit] nexus_mask declared and never used — receiver/JFJochReceiver.cpp:125

Category: Simplification std::vector<uint32_t> nexus_mask; in SendStartMessage is dead. Remove.

[Nit] SummationThread returns int64_t but always returns 0 — receiver/JFJochReceiverFPGA.cpp:241-270

Category: Simplification The return value is never read (futures … f.get() discards it, :345-346). Should be void. The maintainer's "no superfluous code" principle applies.

[Nit] FillNotCollectedModule / err_value const-correctness and naming — receiver/FrameTransformation.cpp:138-140

Category: Nit Minor: precompression_buffer/compressed_buffer/err_value are std::vector<char> and repeatedly C-cast to typed pointers; consistent reinterpret_cast or a typed view would read cleaner. Also image_mode/pixel_depth are initialized in the member-init list in an order that does not match declaration order (pixel_depth declared after image_mode’s use); harmless here but compilers with -Wreorder will warn.

FPGA vs Lite receiver: shared vs duplicated

Shared (good): All telemetry/counters, SendStartMessage/SendEndMessage, GetStatus, GetFinalStatistics skeleton, the ImageBuffer slot protocol, LossyFilter, ScanResultGenerator, JFJochReceiverPlots.Add, and the Cancel base behavior all live on JFJochReceiver. The two subclasses correctly funnel through these.

Duplicated with drift (risk): The per-image forwarding tail is near-identical but copy-pasted in both subclasses: get slot → nullptr ⇒ writer_queue_full=true → else serialize into slot → SetImageNumber/SetImageSize/ SetIndexed → ReadyToSend() (FPGA) / release() (Lite dark-mask) → optional preview/metadata sockets → SendImage

• ++images_sent else release() → UpdateMaxImageSent. Compare JFJochReceiverFPGA.cpp:440-489 against JFJochReceiverLite.cpp:310-336. They differ in small ways (FPGA does compression into the slot and computes compression_ratio inline; Lite serializes an already-compressed DataMessage), but the slot lifecycle and the preview/metadata/pusher dispatch are the same logic written twice. The MaskThread (Lite:179-225) duplicates a third, slightly different copy. The finalize tail (send end → CheckIfBufferReturned → Finalize writer → set status) is also duplicated (see Medium finding above). A shared ForwardImageToWriter(ZeroCopyReturnValue&, DataMessage&) helper on the base would collapse three near-copies into one and remove the drift surface (e.g. only FPGA guards the image_collection_efficiency == 0 empty-image case, only Lite forwards calibration messages).

Efficiency / progress semantics differ legitimately: FPGA derives efficiency from packet counts and progress from device frame counters; Lite derives both from images_collected / max_image_number_received vs GetFrameNum(). That divergence is inherent (Lite has no packet accounting) and is fine.

Inconsistencies with rest of repo

• Logging style mismatch. TCPStreamPusher builds log messages with string + concatenation throughout ("... " + std::to_string(x) + ..., e.g. :127,565,694) while the rest of the receiver subsystem uses the fmt-style logger.Info("... {}", x) API (e.g. JFJochReceiver.cpp:48, JFJochReceiverService.cpp:223). SendCalibration in the same file uses the fmt style (:1166), so the file is internally inconsistent too.
• SendImage(z) return type asymmetry. Base ImagePusher::SendImage(const uint8_t*, size_t, int64_t) returns bool; the ZeroCopyReturnValue overload returns void (ImagePusher.h:43-44). Callers therefore cannot detect async TCP send failures (ties into the images_sent over-count finding). Other pushers (HDF5/ZMQ) also override the void overload, so the asymmetry is repo-wide, but it is the root of the silent-failure path here.
• Mutex-guarded-vs-atomic inconsistency in the base receiver. max_image_number_sent/max_image_number_received/ max_delay use a dedicated mutex on write but are read lock-free in GetStatus/GetProgress. Elsewhere in the repo similar telemetry is plain atomic. Pick one idiom.
• ThreadSafeFIFO::Get decrements utilization but not via the blocking path's accounting is consistent within the class; no issue, just noting the FIFO is the shared primitive both paths rely on and it is sound (all ops mutex-guarded).

ffbidx touchpoints (if any)

None directly in the data path. ffbidx is reached only transitively through IndexAndRefine/IndexerThreadPool (JFJochReceiver.h:23-24, member indexer), which the receivers feed via MXAnalysisAfterFPGA/MXAnalysisWithoutFPGA and finalize via indexer.FinalizeRotationIndexing() / GetConsensusUnitCell() in SendEndMessage (JFJochReceiver.cpp:165-176). The indexer object is shared across all worker threads by reference; its internal thread-safety is owned by the image-analysis/indexing subsystem (separate review). No ffbidx API is called from receiver/pusher/device code in scope, so no ffbidx-specific misuse to flag here.