receiver: remove dead code left from earlier receiver designs

- JFJochReceiverFPGA no longer attaches the uncompressed image before the
  analysis; only the removed CPU integrator read it.
- JFJochReceiverFPGA drops its branches for a run with no images; pedestal
  runs have their own receiver and a standard run has at least one image.
- SummationThread returns nothing instead of an unused 0.
- JFJochReceiverOutput loses master_file_name and images_written (never set
  or never read), the base receiver its always-empty images_written.
- JFJochReceiverCurrentStatus::SetEfficiency is gone: its only call was
  overwritten on the next line.
- ImageMetadata::IsBunchIDConsistent had no caller; an inconsistent frame
  still gets pulse ID UINT64_MAX.
- DarkMaskAnalysis::AnalyzeImage takes the decompression buffer by
  reference, so the per-thread buffer of the dark mask receiver is reused.
- The pedestal and dark mask receivers no longer take the indexer pool.

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-08 10:44:09 +01:00
1 parent 6e33269322
commit 51b2ee8804
16 files changed
+39 -89

No files matched your search

@@ -19,7 +19,7 @@ void DarkMaskAnalysis::check(const T *ptr) {
}
}
void DarkMaskAnalysis::AnalyzeImage(const DataMessage &data, std::vector<uint8_t> buffer) {
void DarkMaskAnalysis::AnalyzeImage(const DataMessage &data, std::vector<uint8_t> &buffer) {
if (data.image.GetWidth() * data.image.GetHeight() != mask.size())
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Invalid size of input image");
@@ -21,7 +21,7 @@ class DarkMaskAnalysis {
template <class T> void check(const T *ptr);
public:
DarkMaskAnalysis(const DarkMaskSettings& settings, size_t det_size);
void AnalyzeImage(const DataMessage &msg, std::vector<uint8_t> buffer);
void AnalyzeImage(const DataMessage &msg, std::vector<uint8_t> &buffer);
std::vector<uint32_t> GetMask() const;
};
+1 -7
View File
@@ -39,10 +39,8 @@ void ImageMetadata::Process(const DeviceOutput *output) {
first_module_loaded = true;
} else {
if (pulsed_source && (module_pulse_id != xfel_pulse_id)) {
metadata_consistent = false;
if (pulsed_source && (module_pulse_id != xfel_pulse_id))
xfel_pulse_id = UINT64_MAX; // mark clearly inconsistent frames
}
}
saturated_pixels += output->module_statistics.saturated_pixels;
error_pixels += output->module_statistics.err_pixels;
@@ -83,10 +81,6 @@ void ImageMetadata::Process(const DeviceOutput *output) {
}
}
bool ImageMetadata::IsBunchIDConsistent() const {
return metadata_consistent;
}
void ImageMetadata::Export(DataMessage &message, uint64_t packets_expected_per_image) const {
std::unique_lock ul(m);
-2
View File
@@ -16,7 +16,6 @@ class ImageMetadata {
bool pulsed_source;
bool first_module_loaded = false;
bool metadata_consistent = true;
uint64_t xfel_pulse_id = UINT64_MAX;
uint64_t xfel_event_code = 0;
@@ -43,5 +42,4 @@ public:
explicit ImageMetadata(const DiffractionExperiment &experiment);
void Process(const DeviceOutput *output);
void Export(DataMessage &message, uint64_t packets_expected_per_image) const;
bool IsBunchIDConsistent() const;
};
-2
View File
@@ -40,7 +40,6 @@ JFJochReceiver::JFJochReceiver(const DiffractionExperiment &in_experiment,
logger.Debug("Image buffer from previous run finalized");
current_status.SetProgress(0);
current_status.SetEfficiency({});
current_status.SetStatus(JFJochReceiverStatus{}); // GetStatus() is virtual function and cannot be called yet!
auto start_time_point = std::chrono::steady_clock::now();
@@ -316,7 +315,6 @@ JFJochReceiverOutput JFJochReceiver::GetFinalStatistics() const {
ret.status = GetStatus();
ret.writer_err = writer_error;
ret.scan_result = scan_result.GetResult();
ret.images_written = images_written;
ret.processing_time = plots.GetMeanProcessingTime();
return ret;
}
-2
View File
@@ -76,8 +76,6 @@ protected:
MovingAverage roi_beam_npixel{1000};
MovingAverage roi_beam_sum{1000};
std::optional<int64_t> images_written;
std::vector<std::unique_ptr<ADUHistogram>> adu_histogram_module;
PixelMask pixel_mask;
-6
View File
@@ -19,12 +19,6 @@ void JFJochReceiverCurrentStatus::SetStatus(const JFJochReceiverStatus &in_statu
status = in_status;
}
void JFJochReceiverCurrentStatus::SetEfficiency(const std::optional<float> &e) {
std::unique_lock ul(m);
if (status.has_value())
status->efficiency = e;
}
std::optional<float> JFJochReceiverCurrentStatus::GetProgress() const {
float tmp = progress;
if (std::isnan(tmp))
-1
View File
@@ -36,7 +36,6 @@ public:
std::optional<JFJochReceiverStatus> GetStatus() const;
void Clear();
void SetStatus(const JFJochReceiverStatus &status);
void SetEfficiency(const std::optional<float> &e);
void SetProgress(std::optional<float> input);
std::optional<float> GetProgress() const;
};
+2 -4
View File
@@ -15,8 +15,7 @@ JFJochReceiverDarkMask::JFJochReceiverDarkMask(const DiffractionExperiment &in_e
JFJochReceiverPlots &in_plots,
ImageBuffer &in_image_buffer,
ZMQPreviewSocket *in_zmq_preview_socket,
ZMQMetadataSocket *in_zmq_metadata_socket,
IndexerThreadPool *indexing_thread_pool)
ZMQMetadataSocket *in_zmq_metadata_socket)
: JFJochReceiverStream(in_experiment,
in_pixel_mask,
in_image_puller,
@@ -28,8 +27,7 @@ JFJochReceiverDarkMask::JFJochReceiverDarkMask(const DiffractionExperiment &in_e
in_plots,
in_image_buffer,
in_zmq_preview_socket,
in_zmq_metadata_socket,
indexing_thread_pool),
in_zmq_metadata_socket),
mask_buffer(nthreads),
dark_mask_analysis(in_experiment.GetDarkMaskSettings(), in_experiment.GetPixelsNum()) {
StartThreads();
+1 -2
View File
@@ -26,8 +26,7 @@ public:
JFJochReceiverPlots &plots,
ImageBuffer &image_buffer,
ZMQPreviewSocket *zmq_preview_socket = nullptr,
ZMQMetadataSocket *zmq_metadata_socket = nullptr,
IndexerThreadPool *indexing_thread_pool = nullptr);
ZMQMetadataSocket *zmq_metadata_socket = nullptr);
~JFJochReceiverDarkMask() override;
JFJochReceiverOutput GetFinalStatistics() const override;
+27 -48
View File
@@ -37,7 +37,7 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen
frame_transformation_nthreads(
in_forward_and_sum_nthreads),
summation_nthreads(2),
frame_transformation_ready((experiment.GetImageNum() > 0) ? frame_transformation_nthreads : 0),
frame_transformation_ready(frame_transformation_nthreads),
data_acquisition_ready(ndatastreams) {
// The FPGA integration core has a fixed bin-count limit. Reject configurations it cannot
// handle before starting acquisition, unless azimuthal integration is forced onto the CPU.
@@ -80,25 +80,23 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen
logger.Debug("Data acquisition devices ready");
if (experiment.GetImageNum() > 0) {
SendStartMessage();
SendCalibration();
SendStartMessage();
SendCalibration();
for (int i = 0; i < experiment.GetImageNum(); i++)
images_to_go.Put(i);
for (int i = 0; i < experiment.GetImageNum(); i++)
images_to_go.Put(i);
// Setup frames summation and forwarding
for (uint32_t i = 0; i < frame_transformation_nthreads; i++) {
auto handle = std::async(std::launch::async, &JFJochReceiverFPGA::FrameTransformationThread,
this, i);
frame_transformation_futures.emplace_back(std::move(handle));
}
logger.Debug("Image compression/forwarding threads started ({} threads)", frame_transformation_nthreads);
frame_transformation_ready.wait();
logger.Debug("Image compression/forwarding threads ready");
// Setup frames summation and forwarding
for (uint32_t i = 0; i < frame_transformation_nthreads; i++) {
auto handle = std::async(std::launch::async, &JFJochReceiverFPGA::FrameTransformationThread,
this, i);
frame_transformation_futures.emplace_back(std::move(handle));
}
logger.Debug("Image compression/forwarding threads started ({} threads)", frame_transformation_nthreads);
frame_transformation_ready.wait();
logger.Debug("Image compression/forwarding threads ready");
for (int d = 0; d < ndatastreams; d++)
data_acquisition_futures.emplace_back(std::async(std::launch::async, &JFJochReceiverFPGA::AcquireThread,
this, d));
@@ -172,11 +170,11 @@ void JFJochReceiverFPGA::AcquireThread(uint16_t data_stream) {
}
int64_t JFJochReceiverFPGA::SummationThread(uint16_t data_stream,
int64_t image_number,
uint16_t module_number,
uint32_t threadid,
ModuleSummation &summation) {
void JFJochReceiverFPGA::SummationThread(uint16_t data_stream,
int64_t image_number,
uint16_t module_number,
uint32_t threadid,
ModuleSummation &summation) {
ModuleSummation local_summation(experiment);
int64_t starting_frame = image_number * experiment.GetSummation();
@@ -200,7 +198,6 @@ int64_t JFJochReceiverFPGA::SummationThread(uint16_t data_stream,
if (!summation.empty())
summation.AddFPGAOutput(local_summation.GetOutput(), 4);
return 0;
}
void JFJochReceiverFPGA::FrameTransformationThread(uint32_t threadid) {
@@ -263,7 +260,7 @@ void JFJochReceiverFPGA::FrameTransformationThread(uint32_t threadid) {
for (int i = 0; i < experiment.GetModulesNum(); i++)
summation.emplace_back(std::make_unique<ModuleSummation>(experiment));
std::vector<std::future<int64_t> > futures;
std::vector<std::future<void>> futures;
for (int d = 0; d < ndatastreams; d++) {
for (int m = 0; m < experiment.GetModulesNum(d); m++) {
@@ -338,13 +335,6 @@ void JFJochReceiverFPGA::FrameTransformationThread(uint32_t threadid) {
continue;
}
message.image = CompressedImage(transformation.GetImage(),
experiment.GetPixelsNum() * experiment.GetByteDepthImage(),
experiment.GetXPixelsNum(),
experiment.GetYPixelsNum(),
experiment.GetImageMode(),
CompressionAlgorithm::NO_COMPRESSION);
// No-op unless the CPU backend is forced; fills az_int_profile_image from the assembled image
analyzer->RunAzimuthalIntegration(transformation.GetImage(), az_int_profile_image);
@@ -471,12 +461,7 @@ void JFJochReceiverFPGA::FrameTransformationThread(uint32_t threadid) {
}
float JFJochReceiverFPGA::GetEfficiency() const {
uint64_t expected_packets;
if (experiment.GetImageNum() == 0)
expected_packets = expected_packets_per_image * experiment.GetFrameNum();
else
expected_packets = expected_packets_per_image * experiment.GetImageNum();
uint64_t expected_packets = expected_packets_per_image * experiment.GetImageNum();
uint64_t received_packets = 0;
for (int d = 0; d < ndatastreams; d++) {
@@ -507,31 +492,25 @@ float JFJochReceiverFPGA::GetProgress() const {
if (experiment.IsCPUSummation())
frames *= experiment.GetSummation();
if (frames == 0)
frames = experiment.GetFrameNum();
if ((frames == 0) || (acquisition_device[0].Counters().IsAcquisitionFinished()))
if (acquisition_device[0].Counters().IsAcquisitionFinished())
return 1.0;
return static_cast<float>(acquisition_device[0].Counters().GetSlowestFrameNumber()) / static_cast<float>(frames);
}
void JFJochReceiverFPGA::FinalizeMeasurement() {
if (!frame_transformation_futures.empty()) {
for (auto &future: frame_transformation_futures)
future.get();
for (auto &future: frame_transformation_futures)
future.get();
logger.Debug("All processing threads done");
}
logger.Debug("All processing threads done");
current_status.SetProgress(1.0);
current_status.SetStatus(GetStatus());
SendEndMessage();
if (experiment.GetImageNum() > 0) {
for (int d = 0; d < ndatastreams; d++)
acquisition_device[d].Cancel();
}
for (int d = 0; d < ndatastreams; d++)
acquisition_device[d].Cancel();
end_time = std::chrono::system_clock::now();
+1 -1
View File
@@ -68,7 +68,7 @@ class JFJochReceiverFPGA : public JFJochReceiver {
void FrameTransformationThread(uint32_t threadid);
int64_t SummationThread(uint16_t data_stream,
void SummationThread(uint16_t data_stream,
int64_t image_number,
uint16_t module_number,
uint32_t threadid,
-2
View File
@@ -17,14 +17,12 @@ struct JFJochReceiverOutput {
ScanResult scan_result;
std::vector<uint64_t> received_packets;
std::vector<uint64_t> expected_packets;
std::string master_file_name;
uint64_t start_time_ms;
uint64_t end_time_ms;
float efficiency;
bool writer_queue_full_warning;
JFJochReceiverStatus status;
std::string writer_err;
std::optional<int64_t> images_written;
MeanProcessingTime processing_time;
// Why a live rugnux run was skipped, for the broker's status
std::string analysis_warning;
+2 -4
View File
@@ -13,8 +13,7 @@ JFJochReceiverPedestal::JFJochReceiverPedestal(const DiffractionExperiment &in_e
JFJochReceiverPlots &in_plots,
ImageBuffer &in_send_buf_ctrl,
ZMQPreviewSocket *in_zmq_preview_socket,
ZMQMetadataSocket *in_zmq_metadata_socket,
IndexerThreadPool *indexing_thread_pool)
ZMQMetadataSocket *in_zmq_metadata_socket)
: JFJochReceiver(in_experiment,
in_send_buf_ctrl,
in_image_sender,
@@ -24,8 +23,7 @@ JFJochReceiverPedestal::JFJochReceiverPedestal(const DiffractionExperiment &in_e
in_logger,
in_pixel_mask,
in_zmq_preview_socket,
in_zmq_metadata_socket,
indexing_thread_pool),
in_zmq_metadata_socket),
acquisition_device(in_aq_device),
ndatastreams(experiment.GetDataStreamsNum()),
pedestal_nthreads((experiment.GetStorageCellNumber() > 2) ? 1 : 4),
+1 -2
View File
@@ -50,8 +50,7 @@ public:
JFJochReceiverPlots &plots,
ImageBuffer &image_buffer,
ZMQPreviewSocket *zmq_preview_socket,
ZMQMetadataSocket *zmq_metadata_socket,
IndexerThreadPool *indexing_thread_pool);
ZMQMetadataSocket *zmq_metadata_socket);
~JFJochReceiverPedestal() override;
JFJochReceiverPedestal(const JFJochReceiverPedestal &other) = delete;
JFJochReceiverPedestal &operator=(const JFJochReceiverPedestal &other) = delete;
+2 -4
View File
@@ -134,8 +134,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
plots,
image_buffer,
zmq_preview_socket.get(),
zmq_metadata_socket.get(),
indexer_thread_pool.get());
zmq_metadata_socket.get());
} else if (!stream_input) {
receiver = std::make_unique<JFJochReceiverFPGA>(experiment, pixel_mask,
calibration,
@@ -168,8 +167,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
plots,
image_buffer,
zmq_preview_socket.get(),
zmq_metadata_socket.get(),
indexer_thread_pool.get());
zmq_metadata_socket.get());
else if (rugnux_enable && (experiment.GetDetectorMode() == DetectorMode::Standard) && RugnuxRotationData(experiment))
receiver = std::make_unique<JFJochReceiverRugnuxRotation>(
experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local,