diff --git a/image_analysis/dark_mask_analysis/DarkMaskAnalysis.cpp b/image_analysis/dark_mask_analysis/DarkMaskAnalysis.cpp index 3a53bc3d3..f12cc17b1 100644 --- a/image_analysis/dark_mask_analysis/DarkMaskAnalysis.cpp +++ b/image_analysis/dark_mask_analysis/DarkMaskAnalysis.cpp @@ -19,7 +19,7 @@ void DarkMaskAnalysis::check(const T *ptr) { } } -void DarkMaskAnalysis::AnalyzeImage(const DataMessage &data, std::vector buffer) { +void DarkMaskAnalysis::AnalyzeImage(const DataMessage &data, std::vector &buffer) { if (data.image.GetWidth() * data.image.GetHeight() != mask.size()) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Invalid size of input image"); diff --git a/image_analysis/dark_mask_analysis/DarkMaskAnalysis.h b/image_analysis/dark_mask_analysis/DarkMaskAnalysis.h index c9e8bb4e3..90e1a9eaf 100644 --- a/image_analysis/dark_mask_analysis/DarkMaskAnalysis.h +++ b/image_analysis/dark_mask_analysis/DarkMaskAnalysis.h @@ -21,7 +21,7 @@ class DarkMaskAnalysis { template void check(const T *ptr); public: DarkMaskAnalysis(const DarkMaskSettings& settings, size_t det_size); - void AnalyzeImage(const DataMessage &msg, std::vector buffer); + void AnalyzeImage(const DataMessage &msg, std::vector &buffer); std::vector GetMask() const; }; diff --git a/receiver/ImageMetadata.cpp b/receiver/ImageMetadata.cpp index 3c9c52147..add1d9117 100644 --- a/receiver/ImageMetadata.cpp +++ b/receiver/ImageMetadata.cpp @@ -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); diff --git a/receiver/ImageMetadata.h b/receiver/ImageMetadata.h index 8107fbe7e..a01f827dd 100644 --- a/receiver/ImageMetadata.h +++ b/receiver/ImageMetadata.h @@ -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; }; diff --git a/receiver/JFJochReceiver.cpp b/receiver/JFJochReceiver.cpp index 6546168a0..4be32b99c 100644 --- a/receiver/JFJochReceiver.cpp +++ b/receiver/JFJochReceiver.cpp @@ -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; } diff --git a/receiver/JFJochReceiver.h b/receiver/JFJochReceiver.h index fa6570a1e..166e364bc 100644 --- a/receiver/JFJochReceiver.h +++ b/receiver/JFJochReceiver.h @@ -76,8 +76,6 @@ protected: MovingAverage roi_beam_npixel{1000}; MovingAverage roi_beam_sum{1000}; - std::optional images_written; - std::vector> adu_histogram_module; PixelMask pixel_mask; diff --git a/receiver/JFJochReceiverCurrentStatus.cpp b/receiver/JFJochReceiverCurrentStatus.cpp index 1ec646f21..b4f350b6f 100644 --- a/receiver/JFJochReceiverCurrentStatus.cpp +++ b/receiver/JFJochReceiverCurrentStatus.cpp @@ -19,12 +19,6 @@ void JFJochReceiverCurrentStatus::SetStatus(const JFJochReceiverStatus &in_statu status = in_status; } -void JFJochReceiverCurrentStatus::SetEfficiency(const std::optional &e) { - std::unique_lock ul(m); - if (status.has_value()) - status->efficiency = e; -} - std::optional JFJochReceiverCurrentStatus::GetProgress() const { float tmp = progress; if (std::isnan(tmp)) diff --git a/receiver/JFJochReceiverCurrentStatus.h b/receiver/JFJochReceiverCurrentStatus.h index 0ef2160a6..b9b0e4190 100644 --- a/receiver/JFJochReceiverCurrentStatus.h +++ b/receiver/JFJochReceiverCurrentStatus.h @@ -36,7 +36,6 @@ public: std::optional GetStatus() const; void Clear(); void SetStatus(const JFJochReceiverStatus &status); - void SetEfficiency(const std::optional &e); void SetProgress(std::optional input); std::optional GetProgress() const; }; diff --git a/receiver/JFJochReceiverDarkMask.cpp b/receiver/JFJochReceiverDarkMask.cpp index a156f1cca..c9e53d95b 100644 --- a/receiver/JFJochReceiverDarkMask.cpp +++ b/receiver/JFJochReceiverDarkMask.cpp @@ -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(); diff --git a/receiver/JFJochReceiverDarkMask.h b/receiver/JFJochReceiverDarkMask.h index 064a2580b..c3c6b84dd 100644 --- a/receiver/JFJochReceiverDarkMask.h +++ b/receiver/JFJochReceiverDarkMask.h @@ -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; diff --git a/receiver/JFJochReceiverFPGA.cpp b/receiver/JFJochReceiverFPGA.cpp index 0bea442ac..d9d69ebe0 100644 --- a/receiver/JFJochReceiverFPGA.cpp +++ b/receiver/JFJochReceiverFPGA.cpp @@ -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(experiment)); - std::vector > futures; + std::vector> 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(acquisition_device[0].Counters().GetSlowestFrameNumber()) / static_cast(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(); diff --git a/receiver/JFJochReceiverFPGA.h b/receiver/JFJochReceiverFPGA.h index 87ba1cd95..32f1d99ef 100644 --- a/receiver/JFJochReceiverFPGA.h +++ b/receiver/JFJochReceiverFPGA.h @@ -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, diff --git a/receiver/JFJochReceiverOutput.h b/receiver/JFJochReceiverOutput.h index c9d0bd196..adc35aebf 100644 --- a/receiver/JFJochReceiverOutput.h +++ b/receiver/JFJochReceiverOutput.h @@ -17,14 +17,12 @@ struct JFJochReceiverOutput { ScanResult scan_result; std::vector received_packets; std::vector 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 images_written; MeanProcessingTime processing_time; // Why a live rugnux run was skipped, for the broker's status std::string analysis_warning; diff --git a/receiver/JFJochReceiverPedestal.cpp b/receiver/JFJochReceiverPedestal.cpp index 0009a32b3..22dcf129d 100644 --- a/receiver/JFJochReceiverPedestal.cpp +++ b/receiver/JFJochReceiverPedestal.cpp @@ -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), diff --git a/receiver/JFJochReceiverPedestal.h b/receiver/JFJochReceiverPedestal.h index cb8e894a6..59b97a9b2 100644 --- a/receiver/JFJochReceiverPedestal.h +++ b/receiver/JFJochReceiverPedestal.h @@ -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; diff --git a/receiver/JFJochReceiverService.cpp b/receiver/JFJochReceiverService.cpp index ec1c575c7..aa8327f79 100644 --- a/receiver/JFJochReceiverService.cpp +++ b/receiver/JFJochReceiverService.cpp @@ -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(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( experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local,