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:
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;
|
||||
};
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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();
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user