mx/Jungfraujoch
3
2 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 93 Wiki Activity

JFJochReceiverRugnuxRotation and JFJochReceiverRugnuxStills replace JFJochReceiverRugnux

Browse Source 
The two workflows shared nothing beyond what JFJochReceiverStream now holds, so they become children of
it and stills_run goes with its seven branches. The service chooses between them with RugnuxRotationData.
Cut, AddToEndMessage and SetSpotFindingSettings are each one body; the stills child no longer builds an
image buffer reader it does not read, and has a mutex and a cut flag of its own.

JFJochReceiverService::Start no longer reassigns the detector type to steer a switch: it asks whether the
input is a stream.

A cancel after the images are in is told by end_message_received rather than a flag of its own, set
at the same moment give or take one poll.

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

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK
This commit is contained in:
leonarski_fandClaude Opus 5.5 committed 2026-11-07 18:56:13 +01:00
1 parent d87508c6b8
commit 1ea363876e
12 files changed
+451 -377
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

receiver/CMakeLists.txt
+2 -1
View File
@@ -14,7 +14,8 @@ ADD_LIBRARY(JFJochReceiver STATIC
JFJochReceiverStream.h JFJochReceiverStream.cpp
JFJochReceiverLite.h JFJochReceiverLite.cpp
FPGAImagePuller.h FPGAImagePuller.cpp
JFJochReceiverRugnux.h JFJochReceiverRugnux.cpp
JFJochReceiverRugnuxRotation.h JFJochReceiverRugnuxRotation.cpp
JFJochReceiverRugnuxStills.h JFJochReceiverRugnuxStills.cpp
StillsPrePass.h StillsPrePass.cpp
StillsGPUAnalysis.h StillsGPUAnalysis.cpp
RugnuxLiveStatus.h RugnuxLiveStatus.cpp
receiver/JFJochReceiverRugnux.h
-99
View File
@@ -1,99 +0,0 @@
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only
#pragma once
#include <future>
#include <latch>
#include "JFJochReceiverStream.h"
#include "../reader/JFJochImageBufferReader.h"
#include "../rugnux/Rugnux.h"
#include "RugnuxLiveStatus.h"
#include "StillsGPUAnalysis.h"
// A DECTRIS rotation collection forwarded to the writer as JFJochReceiverLite does, but analysed by
// rugnux while it arrives instead of image by image: the forward threads put each image into the
// image buffer and send it on, and one rugnux thread reads the collection back from that buffer. The
// end message waits for rugnux, and carries what it found - the lattice, refined geometry, per-image
// results and its merged data, map coefficients and placed model (see AddToEndMessage).
// A serial collection (no goniometer scan) is analysed frame by frame instead (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 merges and the model step
// run beside it, and the end message carries the last of them. Only what the puller delivers is used: a
// DataMessage with a compressed image, whatever produced it.
class JFJochReceiverRugnux : public JFJochReceiverStream {
using JFJochReceiverStream::Cancel;
JFJochImageBufferReader reader;
const int rugnux_nthreads;
const size_t host_budget; // what the main run and the probe plan their host memory against
const std::shared_ptr<const RugnuxModel> model;
const double deadline_s; // after the last image, the analysis is cut (0 = never)
RugnuxLiveStatus &live_status;
std::string skipped_warning; // why the collection was not analysed, for the broker's status
mutable std::mutex rugnux_m;
Rugnux *rugnux = nullptr; // while it runs, for Cut
Rugnux *probe = nullptr; // the same for the lattice probe
std::optional<ProcessResult> rugnux_result;
std::promise<void> rugnux_published; // rugnux_result is in place, or the run has failed (RugnuxThread)
bool cut = false;
std::atomic<bool> images_done{false}; // the forward threads have ended
const bool stills_run;
IndexerThreadPool *const indexing_thread_pool;
const int64_t stills_nthreads;
int64_t stills_keep_free = 0; // slots frames do not take, for the analysis threads to send with
std::unique_ptr<StillsGPUAnalysis> stills; // under rugnux_m until the frames are in, for Cut
// rugnux_out is ready once the run's result is in place, rugnux_done once the run is gone as well.
std::future<void> rugnux_out;
std::future<void> probe_done;
std::future<void> rugnux_done;
void StartStills();
void FinishStills();
void HoldStillsImage(const DataMessage &message);
ZeroCopyReturnValue *AnalysedImageSlot();
void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile);
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.
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".
void Cut(const std::string &reason);
void AddToEndMessage(EndMessage &message) override;
public:
JFJochReceiverRugnux(const DiffractionExperiment &experiment,
const PixelMask &pixel_mask,
ImagePuller &image_puller,
ImagePusher &image_pusher,
Logger &logger,
int64_t forward_nthreads,
int rugnux_nthreads,
std::shared_ptr<const RugnuxModel> model,
double deadline_s,
RugnuxLiveStatus &live_status,
const SpotFindingSettings &spot_finding_settings,
JFJochReceiverCurrentStatus &current_status,
JFJochReceiverPlots &plots,
ImageBuffer &image_buffer,
ZMQPreviewSocket *zmq_preview_socket = nullptr,
ZMQMetadataSocket *zmq_metadata_socket = nullptr,
IndexerThreadPool *indexing_thread_pool = nullptr);
~JFJochReceiverRugnux() override;
void Cancel(bool silent) override;
JFJochReceiverOutput GetFinalStatistics() const override;
void SetSpotFindingSettings(const SpotFindingSettings &spot_finding_settings) override;
};
receiver/JFJochReceiverRugnux.cpp → receiver/JFJochReceiverRugnuxRotation.cpp
+49 -224
View File
@@ -1,26 +1,21 @@
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only
#include "JFJochReceiverRugnux.h"
#include "JFJochReceiverRugnuxRotation.h"
#include <cmath>
#include <cstring>
#include <thread>
#include "spdlog/fmt/fmt.h"
#include "../common/HostMemory.h"
#include "../common/time_utc.h"
#include "../frame_serialize/CBORStream2Deserializer.h"
#include "../image_analysis/indexing/IndexerThreadPool.h"
#include "../rugnux/RugnuxDefaults.h"
#include "../rugnux/RugnuxLive.h"
#include "../rugnux/ResultReport.h"
#include "../writer/RugnuxResultBlocks.h"
#include "RugnuxStructure.h"
using namespace std::chrono_literals;
namespace {
// Kept for the rest of the broker - the receiver, the pushers, the preview - when the host memory
// rugnux plans against is decided.
@@ -94,23 +89,22 @@ namespace {
}
}
JFJochReceiverRugnux::JFJochReceiverRugnux(const DiffractionExperiment &in_experiment,
const PixelMask &in_pixel_mask,
ImagePuller &in_image_puller,
ImagePusher &in_image_pusher,
Logger &in_logger,
int64_t in_forward_nthreads,
int in_rugnux_nthreads,
std::shared_ptr<const RugnuxModel> in_model,
double in_deadline_s,
RugnuxLiveStatus &in_live_status,
const SpotFindingSettings &in_spot_finding_settings,
JFJochReceiverCurrentStatus &in_current_status,
JFJochReceiverPlots &in_plots,
ImageBuffer &in_image_buffer,
ZMQPreviewSocket *in_zmq_preview_socket,
ZMQMetadataSocket *in_zmq_metadata_socket,
IndexerThreadPool *in_indexing_thread_pool)
JFJochReceiverRugnuxRotation::JFJochReceiverRugnuxRotation(const DiffractionExperiment &in_experiment,
const PixelMask &in_pixel_mask,
ImagePuller &in_image_puller,
ImagePusher &in_image_pusher,
Logger &in_logger,
int64_t in_forward_nthreads,
int in_rugnux_nthreads,
std::shared_ptr<const RugnuxModel> in_model,
double in_deadline_s,
RugnuxLiveStatus &in_live_status,
const SpotFindingSettings &in_spot_finding_settings,
JFJochReceiverCurrentStatus &in_current_status,
JFJochReceiverPlots &in_plots,
ImageBuffer &in_image_buffer,
ZMQPreviewSocket *in_zmq_preview_socket,
ZMQMetadataSocket *in_zmq_metadata_socket)
: JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger,
in_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots,
in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr),
@@ -119,25 +113,20 @@ JFJochReceiverRugnux::JFJochReceiverRugnux(const DiffractionExperiment &in_exper
host_budget(RugnuxHostBudget(in_image_buffer)),
model(std::move(in_model)),
deadline_s(in_deadline_s),
live_status(in_live_status),
stills_run(!RugnuxRotationData(in_experiment)),
indexing_thread_pool(in_indexing_thread_pool),
stills_nthreads(std::min<int64_t>(16, in_forward_nthreads)) {
if (!stills_run) {
// 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.
[[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, "
"planning against {:.1f} GB of host memory", nthreads, rugnux_nthreads,
static_cast<double>(host_budget) / 1e9);
}
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.
[[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, "
"planning against {:.1f} GB of host memory", nthreads, rugnux_nthreads,
static_cast<double>(host_budget) / 1e9);
StartThreads();
}
JFJochReceiverRugnux::~JFJochReceiverRugnux() {
JFJochReceiverRugnuxRotation::~JFJochReceiverRugnuxRotation() {
cancelled = true;
reader.Ended();
{
@@ -150,11 +139,7 @@ JFJochReceiverRugnux::~JFJochReceiverRugnux() {
JoinThreads();
}
void JFJochReceiverRugnux::OnStart(std::chrono::steady_clock::time_point received) {
if (stills_run) {
StartStills();
return;
}
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
@@ -175,13 +160,15 @@ void JFJochReceiverRugnux::OnStart(std::chrono::steady_clock::time_point receive
StartRugnux();
}
void JFJochReceiverRugnux::FinishCollection() {
if (stills_run) {
FinishStills();
return;
}
void JFJochReceiverRugnuxRotation::OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) {
live_status.ImageReceived(received);
CountImage(message);
StampImage(message, image_puller.GetCurrentFifoUtilization());
ForwardImage(message, image_buffer.GetImageSlot());
}
void JFJochReceiverRugnuxRotation::FinishCollection() {
reader.Ended();
images_done = true;
// The latency of the demonstration runs from the last image in (as the puller took it off the socket)
// to every result in place.
const auto last_image = live_status.LastImage();
@@ -238,164 +225,23 @@ void JFJochReceiverRugnux::FinishCollection() {
current_status.SetStatus(GetStatus());
}
void JFJochReceiverRugnux::StartStills() {
const int64_t slots = image_buffer.GetStatus().total_slots;
stills_keep_free = std::min(stills_nthreads, slots / 2);
auto analysis = std::make_unique<StillsGPUAnalysis>(
experiment, pixel_mask, MakeStartMessage(experiment, pixel_mask, *az_int_mapping),
indexing_thread_pool, stills_nthreads,
// What the pre-pass holds and fits on; the frames beyond it wait in the image buffer as well.
slots / 2, logger,
[this](const StillsGPUAnalysis::Setup &s) {
// The profiles are binned on the pre-pass's geometry.
plots.Setup(s.experiment, *s.mapping);
SendStartMessage(MakeStartMessage(s.experiment, s.mask, *s.mapping));
},
[this](DataMessage &m, const AzimuthalIntegrationProfile &p) { SendAnalysedImage(m, p); },
[this] {
// 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. With the broker's
// default fixed threshold 3 points fewer frames index.
SpotFindingSettings settings = GetSpotFindingSettings();
settings.adaptive_threshold = true;
settings.min_pix_per_spot = std::nullopt;
return settings;
},
// The start message did not go out (or the analysis could not be set up): nothing can be
// written, so the collection stops.
[this](const JFJochException &e) { Cancel(e); },
&live_status, true, model, deadline_s);
std::unique_lock ul(rugnux_m);
stills = std::move(analysis);
if (cut)
stills->Cut("cancel");
}
void JFJochReceiverRugnux::FinishStills() {
images_done = true;
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.
std::exception_ptr error;
try {
stills->Finish();
} catch (const std::exception &) {
error = std::current_exception();
}
current_status.SetProgress(1.0);
current_status.SetStatus(GetStatus());
if (!error)
SendEndMessage();
end_time = std::chrono::system_clock::now();
CheckIfImagesSent();
if (error) {
current_status.SetProgress({});
current_status.SetStatus(GetStatus());
std::rethrow_exception(error);
}
WriterFinalize();
stills->Written();
current_status.SetProgress({});
current_status.SetStatus(GetStatus());
}
void JFJochReceiverRugnux::HoldStillsImage(const DataMessage &message) {
// The analysis keeps per-image results by number, for the images the collection declares; the stream is not ours.
if (message.number < 0 || message.number >= experiment.GetImageNum()) {
logger.Error("Dropping image {} - the collection has {} images", message.number, experiment.GetImageNum());
return;
}
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.
auto held = std::make_shared<HeldImage>();
held->slot = image_buffer.GetImageSlot(stills_keep_free);
if (held->slot == nullptr) {
writer_queue_full = true;
return;
}
if (size > static_cast<size_t>(experiment.GetImageBufferLocationSize())) {
logger.Error("Dropping image {} - {} B do not fit in buffer slot {} B", message.number, size,
experiment.GetImageBufferLocationSize());
return;
}
memcpy(held->slot->GetImage(), message.image.GetCompressed(), size);
held->message = message;
held->message.image = CompressedImage(held->slot->GetImage(), size, message.image.GetWidth(),
message.image.GetHeight(), message.image.GetMode(),
message.image.GetCompressionAlgorithm(), message.image.GetChannel());
StampImage(held->message, image_puller.GetCurrentFifoUtilization());
stills->Hold(held);
}
ZeroCopyReturnValue *JFJochReceiverRugnux::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.
constexpr auto WRITER_WAIT = std::chrono::seconds(2);
const auto t0 = std::chrono::steady_clock::now();
auto loc = image_buffer.GetImageSlot();
while (loc == nullptr && std::chrono::steady_clock::now() - t0 < WRITER_WAIT) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
loc = image_buffer.GetImageSlot();
}
return loc;
}
void JFJochReceiverRugnux::SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile) {
try {
const auto status = image_buffer.GetStatus();
message.receiver_buf_available = status.available_slots;
message.receiver_buf_in_preparation = status.preparation_slots;
message.receiver_buf_in_sending = status.sending_slots;
AddToAverages(message);
scan_result.Add(message);
if (!serialmx_filter.ApplyFilter(message))
++images_skipped;
else
ForwardImage(message, AnalysedImageSlot());
plots.Add(message, profile);
current_status.SetProgress(GetProgress());
current_status.SetStatus(GetStatus());
} catch (const JFJochException &e) {
logger.ErrorException(e);
Cancel(e);
} catch (const std::exception &e) {
const JFJochException error(JFJochExceptionCategory::WrongDAQState,
fmt::format("Image {} not written - {}", message.number, e.what()));
logger.ErrorException(error);
Cancel(error);
}
}
void JFJochReceiverRugnux::StartRugnux() {
void JFJochReceiverRugnuxRotation::StartRugnux() {
// A failure here costs the analysis only: the images go on to the writer as they would without it.
try {
// The start message as the writer gets it, so rugnux sees the dataset an HDF5 master states.
std::vector<uint8_t> start_cbor;
image_buffer.GetStartMessage(start_cbor);
reader.Open(CBORStream2Deserialize(start_cbor)->start_message.value());
rugnux_done = std::async(std::launch::async, &JFJochReceiverRugnux::RugnuxThread, this);
rugnux_done = std::async(std::launch::async, &JFJochReceiverRugnuxRotation::RugnuxThread, this);
rugnux_out = rugnux_published.get_future();
probe_done = std::async(std::launch::async, &JFJochReceiverRugnux::ProbeThread, this);
probe_done = std::async(std::launch::async, &JFJochReceiverRugnuxRotation::ProbeThread, this);
} catch (const std::exception &e) {
logger.Error("rugnux live: cannot start - {}", e.what());
live_status.Finish(RugnuxLiveState::Failed, e.what());
}
}
void JFJochReceiverRugnux::RugnuxThread() {
void JFJochReceiverRugnuxRotation::RugnuxThread() {
bool published = false;
try {
const auto dataset = reader.GetDataset();
@@ -468,7 +314,7 @@ void JFJochReceiverRugnux::RugnuxThread() {
rugnux_published.set_value();
}
void JFJochReceiverRugnux::ProbeThread() {
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
@@ -515,12 +361,8 @@ void JFJochReceiverRugnux::ProbeThread() {
logger.Debug("rugnux live: lattice probe - {}", live_status.Get().probe);
}
void JFJochReceiverRugnux::AddToEndMessage(EndMessage &message) {
void JFJochReceiverRugnuxRotation::AddToEndMessage(EndMessage &message) {
JFJochReceiverStream::AddToEndMessage(message);
if (stills) {
stills->FillEndMessage(message);
return;
}
// 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);
@@ -574,18 +416,7 @@ void JFJochReceiverRugnux::AddToEndMessage(EndMessage &message) {
}
}
void JFJochReceiverRugnux::OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) {
if (stills_run) {
HoldStillsImage(message);
return;
}
live_status.ImageReceived(received);
CountImage(message);
StampImage(message, image_puller.GetCurrentFifoUtilization());
ForwardImage(message, image_buffer.GetImageSlot());
}
void JFJochReceiverRugnux::Cancel(bool silent) {
void JFJochReceiverRugnuxRotation::Cancel(bool silent) {
detector_finished = true;
// The silent cancel is the detector reading idle at the end of every collection: rugnux goes on.
if (silent)
@@ -593,20 +424,16 @@ void JFJochReceiverRugnux::Cancel(bool silent) {
// /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.
if (!images_done)
if (!end_message_received)
JFJochReceiver::Cancel(silent);
Cut("cancel");
}
void JFJochReceiverRugnux::Cut(const std::string &reason) {
void JFJochReceiverRugnuxRotation::Cut(const std::string &reason) {
std::unique_lock ul(rugnux_m);
if (cut || rugnux_result)
return;
cut = true;
if (stills) {
stills->Cut(reason);
return;
}
reader.Ended(); // no image is waited for any more
if (rugnux != nullptr)
rugnux->Cancel();
@@ -616,13 +443,11 @@ void JFJochReceiverRugnux::Cut(const std::string &reason) {
logger.Debug("rugnux live: analysis cut ({}) - the end message carries what is established", reason);
}
void JFJochReceiverRugnux::SetSpotFindingSettings(const SpotFindingSettings &in_spot_finding_settings) {
// A rotation run takes rugnux's own spot finding; the frames of a serial run are analysed with these.
if (stills_run)
JFJochReceiver::SetSpotFindingSettings(in_spot_finding_settings);
void JFJochReceiverRugnuxRotation::SetSpotFindingSettings(const SpotFindingSettings &in_spot_finding_settings) {
// A rotation run takes rugnux's own spot finding
}
JFJochReceiverOutput JFJochReceiverRugnux::GetFinalStatistics() const {
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.
receiver/JFJochReceiverRugnuxRotation.h
+80
View File
@@ -0,0 +1,80 @@
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only
#pragma once
#include <future>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include "JFJochReceiverStream.h"
#include "RugnuxLiveStatus.h"
#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).
class JFJochReceiverRugnuxRotation : public JFJochReceiverStream {
using JFJochReceiverStream::Cancel;
JFJochImageBufferReader reader;
const int rugnux_nthreads;
const size_t host_budget; // what the main run and the probe plan their host memory against
const std::shared_ptr<const RugnuxModel> model;
const double deadline_s; // after the last image, the analysis is cut (0 = never)
RugnuxLiveStatus &live_status;
std::string skipped_warning; // why the collection was not analysed, for the broker's status
mutable std::mutex rugnux_m;
Rugnux *rugnux = nullptr; // while it runs, for Cut
Rugnux *probe = nullptr; // the same for the lattice probe
std::optional<ProcessResult> rugnux_result;
std::promise<void> rugnux_published; // rugnux_result is in place, or the run has failed (RugnuxThread)
bool cut = false;
// rugnux_out is ready once the run's result is in place, rugnux_done once the run is gone as well.
std::future<void> rugnux_out;
std::future<void> probe_done;
std::future<void> rugnux_done;
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.
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".
void Cut(const std::string &reason);
void AddToEndMessage(EndMessage &message) override;
public:
JFJochReceiverRugnuxRotation(const DiffractionExperiment &experiment,
const PixelMask &pixel_mask,
ImagePuller &image_puller,
ImagePusher &image_pusher,
Logger &logger,
int64_t forward_nthreads,
int rugnux_nthreads,
std::shared_ptr<const RugnuxModel> model,
double deadline_s,
RugnuxLiveStatus &live_status,
const SpotFindingSettings &spot_finding_settings,
JFJochReceiverCurrentStatus &current_status,
JFJochReceiverPlots &plots,
ImageBuffer &image_buffer,
ZMQPreviewSocket *zmq_preview_socket = nullptr,
ZMQMetadataSocket *zmq_metadata_socket = nullptr);
~JFJochReceiverRugnuxRotation() override;
void Cancel(bool silent) override;
JFJochReceiverOutput GetFinalStatistics() const override;
void SetSpotFindingSettings(const SpotFindingSettings &spot_finding_settings) override;
};
receiver/JFJochReceiverRugnuxStills.cpp
+211
View File
@@ -0,0 +1,211 @@
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only
#include "JFJochReceiverRugnuxStills.h"
#include <cstring>
#include <thread>
JFJochReceiverRugnuxStills::JFJochReceiverRugnuxStills(const DiffractionExperiment &in_experiment,
const PixelMask &in_pixel_mask,
ImagePuller &in_image_puller,
ImagePusher &in_image_pusher,
Logger &in_logger,
int64_t in_forward_nthreads,
std::shared_ptr<const RugnuxModel> in_model,
double in_deadline_s,
RugnuxLiveStatus &in_live_status,
const SpotFindingSettings &in_spot_finding_settings,
JFJochReceiverCurrentStatus &in_current_status,
JFJochReceiverPlots &in_plots,
ImageBuffer &in_image_buffer,
ZMQPreviewSocket *in_zmq_preview_socket,
ZMQMetadataSocket *in_zmq_metadata_socket,
IndexerThreadPool *in_indexing_thread_pool)
: JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger,
in_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots,
in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr),
model(std::move(in_model)),
deadline_s(in_deadline_s),
live_status(in_live_status),
indexing_thread_pool(in_indexing_thread_pool),
stills_nthreads(std::min<int64_t>(16, in_forward_nthreads)) {
StartThreads();
}
JFJochReceiverRugnuxStills::~JFJochReceiverRugnuxStills() {
cancelled = true;
JoinThreads();
}
void JFJochReceiverRugnuxStills::OnStart(std::chrono::steady_clock::time_point received) {
const int64_t slots = image_buffer.GetStatus().total_slots;
stills_keep_free = std::min(stills_nthreads, slots / 2);
auto analysis = std::make_unique<StillsGPUAnalysis>(
experiment, pixel_mask, MakeStartMessage(experiment, pixel_mask, *az_int_mapping),
indexing_thread_pool, stills_nthreads,
// What the pre-pass holds and fits on; the frames beyond it wait in the image buffer as well.
slots / 2, logger,
[this](const StillsGPUAnalysis::Setup &s) {
// The profiles are binned on the pre-pass's geometry.
plots.Setup(s.experiment, *s.mapping);
SendStartMessage(MakeStartMessage(s.experiment, s.mask, *s.mapping));
},
[this](DataMessage &m, const AzimuthalIntegrationProfile &p) { SendAnalysedImage(m, p); },
[this] {
// 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. With the broker's
// default fixed threshold 3 points fewer frames index.
SpotFindingSettings settings = GetSpotFindingSettings();
settings.adaptive_threshold = true;
settings.min_pix_per_spot = std::nullopt;
return settings;
},
// The start message did not go out (or the analysis could not be set up): nothing can be
// written, so the collection stops.
[this](const JFJochException &e) { Cancel(e); },
&live_status, true, model, deadline_s);
std::unique_lock ul(stills_mutex);
stills = std::move(analysis);
if (cut)
stills->Cut("cancel");
}
void JFJochReceiverRugnuxStills::OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) {
// The analysis keeps per-image results by number, for the images the collection declares; the stream is not ours.
if (message.number < 0 || message.number >= experiment.GetImageNum()) {
logger.Error("Dropping image {} - the collection has {} images", message.number, experiment.GetImageNum());
return;
}
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.
auto held = std::make_shared<HeldImage>();
held->slot = image_buffer.GetImageSlot(stills_keep_free);
if (held->slot == nullptr) {
writer_queue_full = true;
return;
}
if (size > static_cast<size_t>(experiment.GetImageBufferLocationSize())) {
logger.Error("Dropping image {} - {} B do not fit in buffer slot {} B", message.number, size,
experiment.GetImageBufferLocationSize());
return;
}
memcpy(held->slot->GetImage(), message.image.GetCompressed(), size);
held->message = message;
held->message.image = CompressedImage(held->slot->GetImage(), size, message.image.GetWidth(),
message.image.GetHeight(), message.image.GetMode(),
message.image.GetCompressionAlgorithm(), message.image.GetChannel());
StampImage(held->message, image_puller.GetCurrentFifoUtilization());
stills->Hold(held);
}
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.
std::exception_ptr error;
try {
stills->Finish();
} catch (const std::exception &) {
error = std::current_exception();
}
current_status.SetProgress(1.0);
current_status.SetStatus(GetStatus());
if (!error)
SendEndMessage();
end_time = std::chrono::system_clock::now();
CheckIfImagesSent();
if (error) {
current_status.SetProgress({});
current_status.SetStatus(GetStatus());
std::rethrow_exception(error);
}
WriterFinalize();
stills->Written();
current_status.SetProgress({});
current_status.SetStatus(GetStatus());
}
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.
constexpr auto WRITER_WAIT = std::chrono::seconds(2);
const auto t0 = std::chrono::steady_clock::now();
auto loc = image_buffer.GetImageSlot();
while (loc == nullptr && std::chrono::steady_clock::now() - t0 < WRITER_WAIT) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
loc = image_buffer.GetImageSlot();
}
return loc;
}
void JFJochReceiverRugnuxStills::SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile) {
try {
const auto status = image_buffer.GetStatus();
message.receiver_buf_available = status.available_slots;
message.receiver_buf_in_preparation = status.preparation_slots;
message.receiver_buf_in_sending = status.sending_slots;
AddToAverages(message);
scan_result.Add(message);
if (!serialmx_filter.ApplyFilter(message))
++images_skipped;
else
ForwardImage(message, AnalysedImageSlot());
plots.Add(message, profile);
current_status.SetProgress(GetProgress());
current_status.SetStatus(GetStatus());
} catch (const JFJochException &e) {
logger.ErrorException(e);
Cancel(e);
} catch (const std::exception &e) {
const JFJochException error(JFJochExceptionCategory::WrongDAQState,
fmt::format("Image {} not written - {}", message.number, e.what()));
logger.ErrorException(error);
Cancel(error);
}
}
void JFJochReceiverRugnuxStills::AddToEndMessage(EndMessage &message) {
JFJochReceiverStream::AddToEndMessage(message);
if (stills)
stills->FillEndMessage(message);
}
void JFJochReceiverRugnuxStills::Cancel(bool silent) {
detector_finished = true;
// 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.
if (!end_message_received)
JFJochReceiver::Cancel(silent);
Cut("cancel");
}
void JFJochReceiverRugnuxStills::Cut(const std::string &reason) {
std::unique_lock ul(stills_mutex);
if (cut)
return;
cut = true;
if (stills) {
stills->Cut(reason);
return;
}
live_status.Cut(reason);
logger.Debug("rugnux live: analysis cut ({}) - the end message carries what is established", reason);
}
receiver/JFJochReceiverRugnuxStills.h
+59
View File
@@ -0,0 +1,59 @@
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only
#pragma once
#include <memory>
#include <mutex>
#include <string>
#include "JFJochReceiverStream.h"
#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.
class JFJochReceiverRugnuxStills : public JFJochReceiverStream {
using JFJochReceiverStream::Cancel;
const std::shared_ptr<const RugnuxModel> model;
const double deadline_s; // after the last image, the model step is cut (0 = never)
RugnuxLiveStatus &live_status;
IndexerThreadPool *const indexing_thread_pool;
const int64_t stills_nthreads;
int64_t stills_keep_free = 0; // slots frames do not take, for the analysis threads to send with
std::mutex stills_mutex;
bool cut = false;
std::unique_ptr<StillsGPUAnalysis> stills; // under stills_mutex until the frames are in, for Cut
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;
ZeroCopyReturnValue *AnalysedImageSlot();
void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile);
void Cut(const std::string &reason);
void AddToEndMessage(EndMessage &message) override;
public:
JFJochReceiverRugnuxStills(const DiffractionExperiment &experiment,
const PixelMask &pixel_mask,
ImagePuller &image_puller,
ImagePusher &image_pusher,
Logger &logger,
int64_t forward_nthreads,
std::shared_ptr<const RugnuxModel> model,
double deadline_s,
RugnuxLiveStatus &live_status,
const SpotFindingSettings &spot_finding_settings,
JFJochReceiverCurrentStatus &current_status,
JFJochReceiverPlots &plots,
ImageBuffer &image_buffer,
ZMQPreviewSocket *zmq_preview_socket = nullptr,
ZMQMetadataSocket *zmq_metadata_socket = nullptr,
IndexerThreadPool *indexing_thread_pool = nullptr);
~JFJochReceiverRugnuxStills() override;
void Cancel(bool silent) override;
};
receiver/JFJochReceiverService.cpp
+40 -42
View File
@@ -2,7 +2,9 @@
// SPDX-License-Identifier: GPL-3.0-only
#include "JFJochReceiverService.h"
#include "JFJochReceiverRugnux.h"
#include "JFJochReceiverRugnuxRotation.h"
#include "JFJochReceiverRugnuxStills.h"
#include "../rugnux/RugnuxDefaults.h"
#include "../common/ThreadAffinity.h"
#include "JFJochReceiverFPGA.h"
#include "JFJochReceiverLite.h"
@@ -117,39 +119,22 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
preview_image.Configure(experiment, pixel_mask);
// Pedestal runs and host summation stay with JFJochReceiverFPGA
auto detector_type = experiment.GetDetectorType();
if (fpga_image_puller && (detector_type != DetectorType::DECTRIS) && (experiment.GetImageNum() > 0)) {
bool stream_input = (experiment.GetDetectorType() == DetectorType::DECTRIS);
if (fpga_image_puller && !stream_input && (experiment.GetImageNum() > 0)) {
if (experiment.IsCPUSummation())
logger.Warning("Summation above what the card can do needs JFJochReceiverFPGA: FPGA image puller not used");
else {
fpga_puller = std::make_shared<FPGAImagePuller>(experiment, pixel_mask, calibration, aq_devices, logger,
nthreads_local, fpga_image_puller_slots);
puller = fpga_puller;
detector_type = DetectorType::DECTRIS; // from here on, one stream receiver input like any other
stream_input = true;
}
}
switch (detector_type) {
case DetectorType::EIGER:
case DetectorType::JUNGFRAU:
if (experiment.IsPedestalRun()) {
receiver = std::make_unique<JFJochReceiverPedestal>(experiment, pixel_mask,
aq_devices, image_pusher,
logger,
spot_finding_settings,
receiver_status,
plots,
image_buffer,
zmq_preview_socket.get(),
zmq_metadata_socket.get(),
indexer_thread_pool.get());
break;
}
receiver = std::make_unique<JFJochReceiverFPGA>(experiment, pixel_mask,
calibration,
if (!stream_input && experiment.IsPedestalRun()) {
receiver = std::make_unique<JFJochReceiverPedestal>(experiment, pixel_mask,
aq_devices, image_pusher,
logger,
nthreads_local,
spot_finding_settings,
receiver_status,
plots,
@@ -157,23 +142,38 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
zmq_preview_socket.get(),
zmq_metadata_socket.get(),
indexer_thread_pool.get());
break;
case DetectorType::DECTRIS:
if (puller)
image_puller = puller;
else {
image_puller = std::make_shared<ZMQImagePuller>(
experiment.GetDetectorSetup().GetDECTRISStream2Addr());
}
// The dark frames of the mask sequence take Lite's mask path, with rugnux on as well.
if (rugnux_enable && experiment.GetDetectorMode() == DetectorMode::Standard) {
receiver = std::make_unique<JFJochReceiverRugnux>(
experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local,
std::max<int>(1, static_cast<int>(AvailableCpus()) - rugnux_reserved_cpus),
rugnux_model, rugnux_deadline_s, rugnux_status, spot_finding_settings, receiver_status, plots, image_buffer,
zmq_preview_socket.get(), zmq_metadata_socket.get(), indexer_thread_pool.get());
break;
}
} else if (!stream_input) {
receiver = std::make_unique<JFJochReceiverFPGA>(experiment, pixel_mask,
calibration,
aq_devices, image_pusher,
logger,
nthreads_local,
spot_finding_settings,
receiver_status,
plots,
image_buffer,
zmq_preview_socket.get(),
zmq_metadata_socket.get(),
indexer_thread_pool.get());
} else {
if (puller)
image_puller = puller;
else
image_puller = std::make_shared<ZMQImagePuller>(experiment.GetDetectorSetup().GetDECTRISStream2Addr());
// The dark frames of the mask sequence take Lite's mask path, with rugnux on as well.
if (rugnux_enable && (experiment.GetDetectorMode() == DetectorMode::Standard) && RugnuxRotationData(experiment))
receiver = std::make_unique<JFJochReceiverRugnuxRotation>(
experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local,
std::max<int>(1, static_cast<int>(AvailableCpus()) - rugnux_reserved_cpus),
rugnux_model, rugnux_deadline_s, rugnux_status, spot_finding_settings, receiver_status, plots,
image_buffer, zmq_preview_socket.get(), zmq_metadata_socket.get());
else if (rugnux_enable && (experiment.GetDetectorMode() == DetectorMode::Standard))
receiver = std::make_unique<JFJochReceiverRugnuxStills>(
experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local,
rugnux_model, rugnux_deadline_s, rugnux_status, spot_finding_settings, receiver_status, plots,
image_buffer, zmq_preview_socket.get(), zmq_metadata_socket.get(), indexer_thread_pool.get());
else
receiver = std::make_unique<JFJochReceiverLite>(experiment,
pixel_mask,
*image_puller,
@@ -187,8 +187,6 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
zmq_preview_socket.get(),
zmq_metadata_socket.get(),
indexer_thread_pool.get());
break;
}
measurement = std::async(std::launch::async, &JFJochReceiverService::FinalizeMeasurement, this);
state = ReceiverState::Running;
receiver/JFJochReceiverService.h
+1 -1
View File
@@ -27,7 +27,7 @@ class JFJochReceiverService {
int64_t nthreads = 8;
// Rotation collections from a DECTRIS detector analysed by rugnux while they arrive
// (JFJochReceiverRugnux), instead of image by image. Off unless asked for.
// (JFJochReceiverRugnuxRotation), 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
receiver/RugnuxLiveStatus.h
+1 -1
View File
@@ -110,7 +110,7 @@ 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 (JFJochReceiverRugnux::ProbeThread): an early answer
// 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.
std::string probe;
std::optional<double> probe_s; // when it answered
tests/FPGAImagePullerTest.cpp
+6 -6
View File
@@ -21,7 +21,7 @@
#include "../writer/HDF5Objects.h"
#include "../writer/StreamWriter.h"
#include "../receiver/FPGAImagePuller.h"
#include "../receiver/JFJochReceiverRugnux.h"
#include "../receiver/JFJochReceiverRugnuxRotation.h"
#include "../receiver/JFJochReceiverService.h"
#include "../rugnux/RugnuxDefaults.h"
@@ -634,7 +634,7 @@ TEST_CASE("FPGAImagePuller_NotForPedestalAndHostSummation", "[FPGAImagePuller]")
CHECK(out.status.images_collected == 2);
}
// JFJochReceiverRugnux takes the same stream, with its own forwarding thread.
// JFJochReceiverRugnuxRotation takes the same stream, with its own forwarding thread.
TEST_CASE("FPGAImagePuller_RugnuxReceiver", "[FPGAImagePuller]") {
auto x = Base(DetJF(2), 6, 3);
x.Conversion();
@@ -689,10 +689,10 @@ TEST_CASE("FPGAImagePuller_SaturatedAndErrorCountsFromTheCard", "[FPGAImagePulle
SECTION("JFJochReceiverLite") {
puller = Run(x, true, 2);
}
SECTION("JFJochReceiverRugnux, stills") {
SECTION("JFJochReceiverRugnuxStills") {
puller = Run(x, true, 2, true);
}
SECTION("JFJochReceiverRugnux, rotation") {
SECTION("JFJochReceiverRugnuxRotation") {
auto rotation = x;
rotation.Goniometer(GoniometerAxis("omega", 0, 0.1, Coord(-1, 0, 0), std::nullopt));
puller = Run(rotation, true, 2, true);
@@ -759,8 +759,8 @@ TEST_CASE("FPGAImagePuller_RugnuxReceiver_Summation", "[FPGAImagePuller]") {
const JFCalibration calib = Calibration(x);
FPGAImagePuller puller(x, mask, &calib, devices, logger, 2);
JFJochReceiverRugnux receiver(x, mask, puller, pusher, logger, 2, 2, nullptr, 30, live, Spots(), status, plots,
buffer);
JFJochReceiverRugnuxRotation receiver(x, mask, puller, pusher, logger, 2, 2, nullptr, 30, live, Spots(), status,
plots, buffer);
receiver.StopReceiver();
puller.Stop();
CHECK(receiver.GetEfficiency() == 1.0f);
tests/JFJochReceiverRugnuxTest.cpp
-1
View File
@@ -10,7 +10,6 @@
#include "../image_analysis/structure_refinement/RugnuxModel.h"
#include "../image_puller/TestImagePuller.h"
#include "../image_pusher/ImagePusher.h"
#include "../receiver/JFJochReceiverRugnux.h"
#include "../receiver/JFJochReceiverService.h"
#include "../writer/HDF5Objects.h"
tests/RugnuxResultFilesTest.cpp
+2 -2
View File
@@ -134,7 +134,7 @@ TEST_CASE("RugnuxResultFiles: rugnux's outputs -> END blocks -> CBOR -> writer,
provenance.command_line = "rugnux -N 8 input_master.h5";
provenance.input_file = "/some/where/input_master.h5";
// The files as rugnux writes them from its outputs (RugnuxOutputFiles.cpp), and the same outputs as the
// broker puts them in the END message (JFJochReceiverRugnux::AddToEndMessage).
// broker puts them in the END message (JFJochReceiverRugnuxRotation::AddToEndMessage).
EndMessage end{};
end.max_image_number = 100;
end.merged = MergedMtzTable(reflections, TEST_CELL, experiment, provenance);
@@ -369,7 +369,7 @@ TEST_CASE("RugnuxResultFiles: the refinement's model, maps and blobs reach the w
.polar_contacts = 3, .metal_contacts = 1, .anomalous_sigma = 1.25};
{ std::ofstream f(original + "_blobs.json"); f << DensityBlobsJson(DensityBlobsToMessage(blobs)); }
// As the broker puts them in the END message (JFJochReceiverRugnux::AddToEndMessage).
// As the broker puts them in the END message (JFJochReceiverRugnuxRotation::AddToEndMessage).
EndMessage end{};
end.refined_model = PlacedModelFromStructure(TestModel(), TEST_CELL, sg.xhm());
end.refined_map_coefficients = maps;