receiver: thread-count rules belong to the receivers they were made for
The 100 Hz / 16-thread rule came with the Lite receiver, where each thread builds a GPU analysis; it now lives there again. The dark mask, rugnux rotation and rugnux stills receivers take the requested thread count. For the dark mask receiver this means several threads instead of one: the rule saw a dark mask run as zero images. Its frames are decompressed in parallel and counted into the mask under the analysis' lock. The four threads for summation on the host apply to JFJochReceiverFPGA only, which is the receiver that sums; before, they were applied before the receiver was chosen. 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
1 parent
b146cd6f89
commit
89fb963bd9
5 files changed
+41
-39
No files matched your search
@@ -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<int64_t>(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<int64_t>(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,
|
||||
|
||||
@@ -12,6 +12,9 @@
|
||||
class JFJochReceiverLite : public JFJochReceiverStream {
|
||||
std::vector<std::unique_ptr<MXAnalysisWithoutFPGA>> 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:
|
||||
|
||||
@@ -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<FPGAImagePuller>(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<JFJochReceiverFPGA>(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<JFJochReceiverRugnuxRotation>(
|
||||
experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local,
|
||||
experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads,
|
||||
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,
|
||||
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,
|
||||
|
||||
@@ -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<int64_t>(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<int64_t>(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) {
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user