CUDA: the engines' setup copies belong on the engine's stream

Making the worker streams non-blocking removed the implicit ordering that the
constructors were still relying on. Each engine uploads its static inputs - the
pixel mask, the pixel-to-bin map, the corrections, the ROI map - with a blocking
NULL-stream cudaMemcpy, and then reads them from kernels on its own stream. A
pageable host-to-device cudaMemcpy returns once the source has been staged, with
the DMA still in flight, and a non-blocking stream no longer waits for the NULL
stream. The failure mode is a silently unapplied mask or a stale mapping, not a
crash, so it would not have announced itself.

Put them on the stream the engine already owns, and synchronise once at the end of
the constructor - that is required for the preprocessor, whose source is a local
vector, and leaves the others settled rather than in flight for the cost of one
one-time sync. The GPU spot-finder test uploaded its image the same way.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

image_analysis/azint/AzIntEngineGPU.cu

@@ -108,10 +108,14 @@ AzIntEngineGPU::AzIntEngineGPU(const AzimuthalIntegrationMapping &integration, s
     shared_size = prop.sharedMemPerBlock;
     shared_needed = azint_bins * (2 * sizeof(float) + sizeof(uint32_t));
 
-    cuda_err(cudaMemcpy(gpu_azint_correction, integration.Corrections().data(), sizeof(float) * npixel,
-               cudaMemcpyHostToDevice));
-    cuda_err(cudaMemcpy(gpu_pixel_to_bin, integration.GetPixelToBin().data(), sizeof(uint16_t) * npixel,
-               cudaMemcpyHostToDevice));
+    // On this engine's stream, like every other operation it issues: the streams are non-blocking, so a
+    // NULL-stream copy is no longer ordered against the kernels that read what it uploads. The one-time
+    // synchronise leaves the constructor with the uploads settled rather than in flight.
+    cuda_err(cudaMemcpyAsync(gpu_azint_correction, integration.Corrections().data(), sizeof(float) * npixel,
+                             cudaMemcpyHostToDevice, *stream));
+    cuda_err(cudaMemcpyAsync(gpu_pixel_to_bin, integration.GetPixelToBin().data(), sizeof(uint16_t) * npixel,
+                             cudaMemcpyHostToDevice, *stream));
+    cuda_err(cudaStreamSynchronize(*stream));
 }
 
 void AzIntEngineGPU::Run(const ImagePreprocessorBuffer &image, AzimuthalIntegrationProfile &profile) {

image_analysis/image_preprocessing/ImagePreprocessorGPU.cu

@@ -103,7 +103,11 @@ ImagePreprocessorGPU::ImagePreprocessorGPU(const DiffractionExperiment &experime
     for (int i = 0; i < npixels; i++)
         mask_vec[i] = (mask.GetMask().at(i) != 0);
 
-    cudaMemcpy(gpu_mask, mask_vec.data(), npixels, cudaMemcpyHostToDevice);
+    // On this engine's stream, like every other operation it issues: the streams are non-blocking, so a
+    // NULL-stream copy is no longer ordered against the kernels that read the mask. Synchronise before
+    // leaving the constructor - mask_vec is a local and the copy must not outlive it.
+    cudaMemcpyAsync(gpu_mask, mask_vec.data(), npixels, cudaMemcpyHostToDevice, *stream);
+    cudaStreamSynchronize(*stream);
 
     // Setup GPU settings
     cudaDeviceProp prop{};

image_analysis/roi/ROIIntegrationGPU.cu

@@ -113,7 +113,12 @@ ROIIntegrationGPU::ROIIntegrationGPU(const DiffractionExperiment &experiment, st
     blocks = 4 * prop.multiProcessorCount;
     shared_needed = roi_count * (5 * sizeof(unsigned long long) + sizeof(int));
 
-    cuda_err(cudaMemcpy(gpu_roi_map, roi_map.data(), sizeof(uint16_t) * npixel, cudaMemcpyHostToDevice));
+    // On this engine's stream, like every other operation it issues: the streams are non-blocking, so a
+    // NULL-stream copy is no longer ordered against the kernels that read the map. The one-time
+    // synchronise leaves the constructor with the upload settled rather than in flight.
+    cuda_err(cudaMemcpyAsync(gpu_roi_map, roi_map.data(), sizeof(uint16_t) * npixel,
+                             cudaMemcpyHostToDevice, *stream));
+    cuda_err(cudaStreamSynchronize(*stream));
 }
 
 void ROIIntegrationGPU::Run(const ImagePreprocessorBuffer &image, std::map<std::string, ROIMessage> &out) {

image_analysis/spot_finding/AdaptiveSpotFinderGPU.cu

@@ -203,10 +203,14 @@ AdaptiveSpotFinderGPU::AdaptiveSpotFinderGPU(const AzimuthalIntegrationMapping &
     shared_clip = static_cast<size_t>(nbins) * (2 * sizeof(float) + sizeof(uint32_t));
     use_shared = (shared_plain < prop.sharedMemPerBlock);
 
-    cuda_err(cudaMemcpy(gpu_pixel_to_bin, mapping.GetPixelToBin().data(), sizeof(uint16_t) * npix,
-                        cudaMemcpyHostToDevice));
-    cuda_err(cudaMemcpy(gpu_corrections, mapping.Corrections().data(), sizeof(float) * npix,
-                        cudaMemcpyHostToDevice));
+    // On this engine's stream, like every other operation it issues: the streams are non-blocking, so a
+    // NULL-stream copy is no longer ordered against the kernels that read what it uploads. The one-time
+    // synchronise leaves the constructor with the uploads settled rather than in flight.
+    cuda_err(cudaMemcpyAsync(gpu_pixel_to_bin, mapping.GetPixelToBin().data(), sizeof(uint16_t) * npix,
+                             cudaMemcpyHostToDevice, *stream));
+    cuda_err(cudaMemcpyAsync(gpu_corrections, mapping.Corrections().data(), sizeof(float) * npix,
+                             cudaMemcpyHostToDevice, *stream));
+    cuda_err(cudaStreamSynchronize(*stream));
 }
 
 void AdaptiveSpotFinderGPU::ReducePass(const ImagePreprocessorBuffer &image, float clip_k,

tests/AdaptiveSpotFinderGPUTest.cpp

@@ -87,6 +87,9 @@ TEST_CASE("AdaptiveSpotFinderGPU_SpotFindingParity", "[AdaptiveSpotFinderGPU]")
     FillTestImage(buffer, x);
     REQUIRE(cudaMemcpy(buffer.getGPUBuffer(), buffer.getBuffer().data(),
                        x.GetPixelsNum() * sizeof(int32_t), cudaMemcpyHostToDevice) == cudaSuccess);
+    // The engines run on non-blocking streams, which do not wait for this NULL-stream copy: a pageable
+    // H2D cudaMemcpy returns once the source is staged, with the DMA still in flight.
+    REQUIRE(cudaDeviceSynchronize() == cudaSuccess);
 
     std::vector<bool> res_mask(x.GetPixelsNum(), false);
     const SpotFindingSettings settings = AdaptiveSettings();
@@ -120,6 +123,9 @@ TEST_CASE("AdaptiveSpotFinderGPU_AzimuthalIntegration", "[AdaptiveSpotFinderGPU]
     FillTestImage(buffer, x);
     REQUIRE(cudaMemcpy(buffer.getGPUBuffer(), buffer.getBuffer().data(),
                        x.GetPixelsNum() * sizeof(int32_t), cudaMemcpyHostToDevice) == cudaSuccess);
+    // The engines run on non-blocking streams, which do not wait for this NULL-stream copy: a pageable
+    // H2D cudaMemcpy returns once the source is staged, with the DMA still in flight.
+    REQUIRE(cudaDeviceSynchronize() == cudaSuccess);
 
     std::vector<bool> res_mask(x.GetPixelsNum(), false);
     const SpotFindingSettings settings = AdaptiveSettings();
@@ -165,6 +171,9 @@ TEST_CASE("AdaptiveSpotFinderGPU_RunToRunReproducible", "[AdaptiveSpotFinderGPU]
     FillTestImage(buffer, x);
     REQUIRE(cudaMemcpy(buffer.getGPUBuffer(), buffer.getBuffer().data(),
                        x.GetPixelsNum() * sizeof(int32_t), cudaMemcpyHostToDevice) == cudaSuccess);
+    // The engines run on non-blocking streams, which do not wait for this NULL-stream copy: a pageable
+    // H2D cudaMemcpy returns once the source is staged, with the DMA still in flight.
+    REQUIRE(cudaDeviceSynchronize() == cudaSuccess);
 
     std::vector<bool> res_mask(x.GetPixelsNum(), false);
     const SpotFindingSettings settings = AdaptiveSettings();
@@ -195,6 +204,9 @@ TEST_CASE("AdaptiveSpotFinderGPU_Speed", "[AdaptiveSpotFinderGPU][.benchmark]")
     FillTestImage(buffer, x);
     REQUIRE(cudaMemcpy(buffer.getGPUBuffer(), buffer.getBuffer().data(),
                        x.GetPixelsNum() * sizeof(int32_t), cudaMemcpyHostToDevice) == cudaSuccess);
+    // The engines run on non-blocking streams, which do not wait for this NULL-stream copy: a pageable
+    // H2D cudaMemcpy returns once the source is staged, with the DMA still in flight.
+    REQUIRE(cudaDeviceSynchronize() == cudaSuccess);
 
     std::vector<bool> res_mask(x.GetPixelsNum(), false);
     const SpotFindingSettings settings = AdaptiveSettings();