diff --git a/receiver/JFJochReceiverLite.cpp b/receiver/JFJochReceiverLite.cpp index a85098e85..f02aff211 100644 --- a/receiver/JFJochReceiverLite.cpp +++ b/receiver/JFJochReceiverLite.cpp @@ -4,6 +4,30 @@ #include "JFJochReceiverLite.h" #include "../common/CUDAWrapper.h" +using namespace std::chrono_literals; + +int64_t JFJochReceiverLite::NumberOfThreads(int64_t requested_thread_number, + const DiffractionExperiment &in_experiment) { + int64_t number_of_images = in_experiment.GetImageNum(); + auto image_time = in_experiment.GetImageTime(); + + if (requested_thread_number <= 0) + return 1; + + // For very small datasets no need to go multithreaded + if (number_of_images < 4) + return 1; + + if (number_of_images < 16) + return std::min(2, requested_thread_number); + + // For 100 Hz, there is no reason to go for a very large number of threads + if (number_of_images < 256 || (image_time >= 10ms)) + return std::min(16, requested_thread_number); + + return requested_thread_number; +} + JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experiment, const PixelMask &in_pixel_mask, ImagePuller &in_image_puller, @@ -22,7 +46,7 @@ JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experimen in_image_puller, in_image_pusher, in_logger, - forward_and_sum_nthreads, + NumberOfThreads(forward_and_sum_nthreads, in_experiment), in_spot_finding_settings, in_current_status, in_plots, diff --git a/receiver/JFJochReceiverLite.h b/receiver/JFJochReceiverLite.h index df2c8fa93..f228d7899 100644 --- a/receiver/JFJochReceiverLite.h +++ b/receiver/JFJochReceiverLite.h @@ -12,6 +12,9 @@ class JFJochReceiverLite : public JFJochReceiverStream { std::vector> analysis; // per image thread + // Each thread builds a GPU analysis of its own + static int64_t NumberOfThreads(int64_t requested_thread_number, const DiffractionExperiment &experiment); + void ImageThread(uint32_t id) override; void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override; public: diff --git a/receiver/JFJochReceiverService.cpp b/receiver/JFJochReceiverService.cpp index abe2e08d2..f792343d3 100644 --- a/receiver/JFJochReceiverService.cpp +++ b/receiver/JFJochReceiverService.cpp @@ -99,10 +99,6 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Receiver not idle, cannot start"); try { - auto nthreads_local = nthreads; - if (experiment.IsCPUSummation()) - nthreads_local = 4; - // First clean-up old measurement receiver.reset(); if (fpga_puller) { @@ -119,7 +115,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, logger.Warning("Summation above what the card can do needs JFJochReceiverFPGA: FPGA image puller not used"); else { fpga_puller = std::make_shared(experiment, pixel_mask, calibration, aq_devices, logger, - nthreads_local, fpga_image_puller_slots); + nthreads, fpga_image_puller_slots); puller = fpga_puller; stream_input = true; } @@ -136,11 +132,15 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, zmq_preview_socket.get(), zmq_metadata_socket.get()); } else if (!stream_input) { + // Summation on the host runs threads of its own for each image + auto fpga_nthreads = nthreads; + if (experiment.IsCPUSummation()) + fpga_nthreads = 4; receiver = std::make_unique(experiment, pixel_mask, calibration, aq_devices, image_pusher, logger, - nthreads_local, + fpga_nthreads, spot_finding_settings, receiver_status, plots, @@ -161,7 +161,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, *image_puller, image_pusher, logger, - nthreads_local, + nthreads, spot_finding_settings, receiver_status, plots, @@ -170,13 +170,13 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, 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, + experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads, std::max(1, static_cast(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( - experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local, + experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads, 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 @@ -185,7 +185,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, *image_puller, image_pusher, logger, - nthreads_local, + nthreads, spot_finding_settings, receiver_status, plots, diff --git a/receiver/JFJochReceiverStream.cpp b/receiver/JFJochReceiverStream.cpp index 2ab22a398..6c569fb12 100644 --- a/receiver/JFJochReceiverStream.cpp +++ b/receiver/JFJochReceiverStream.cpp @@ -3,30 +3,6 @@ #include "JFJochReceiverStream.h" -using namespace std::chrono_literals; - -int64_t JFJochReceiverStream::NumberOfThreads(int64_t requested_thread_number, - const DiffractionExperiment &in_experiment) { - int64_t number_of_images = in_experiment.GetImageNum(); - auto image_time = in_experiment.GetImageTime(); - - if (requested_thread_number <= 0) - return 1; - - // For very small datasets no need to go multithreaded - if (number_of_images < 4) - return 1; - - if (number_of_images < 16) - return std::min(2, requested_thread_number); - - // For 100 Hz, there is no reason to go for a very large number of threads - if (number_of_images < 256 || (image_time >= 10ms)) - return std::min(16, requested_thread_number); - - return requested_thread_number; -} - void JFJochReceiverStream::ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &msg) { // A PSI detector's images are what the experiment set the card up to produce: nothing to learn here. if (experiment.GetDetectorType() != DetectorType::DECTRIS) @@ -62,7 +38,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper ImagePuller &in_image_puller, ImagePusher &in_image_pusher, Logger &in_logger, - int64_t requested_nthreads, + int64_t in_nthreads, const SpotFindingSettings &in_spot_finding_settings, JFJochReceiverCurrentStatus &in_current_status, JFJochReceiverPlots &in_plots, @@ -82,7 +58,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper in_zmq_metadata_socket, indexing_thread_pool), image_puller(in_image_puller), - nthreads(NumberOfThreads(requested_nthreads, in_experiment)), + nthreads(in_nthreads), measurement_started(1), images_started(nthreads) { } diff --git a/receiver/JFJochReceiverStream.h b/receiver/JFJochReceiverStream.h index 01b2b8fe0..0617d0fd7 100644 --- a/receiver/JFJochReceiverStream.h +++ b/receiver/JFJochReceiverStream.h @@ -56,14 +56,13 @@ protected: public: // Takes what the detector's start message decides into the experiment. static void ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &message); - static int64_t NumberOfThreads(int64_t requested_thread_number, const DiffractionExperiment &experiment); JFJochReceiverStream(const DiffractionExperiment &experiment, const PixelMask &pixel_mask, ImagePuller &image_puller, ImagePusher &image_pusher, Logger &logger, - int64_t requested_nthreads, + int64_t nthreads, const SpotFindingSettings &spot_finding_settings, JFJochReceiverCurrentStatus ¤t_status, JFJochReceiverPlots &plots,