diff --git a/rugnux/RugnuxFirstPass.cpp b/rugnux/RugnuxFirstPass.cpp index cc44a332b..d2f814092 100644 --- a/rugnux/RugnuxFirstPass.cpp +++ b/rugnux/RugnuxFirstPass.cpp @@ -134,7 +134,7 @@ namespace { start_image(start_image), stride(stride), engines(std::max(nworkers, 1)) {} ~SpotFindingWorkers() { - OnWorkers(engines.size(), [&](size_t t) { engines[t] = Engine{}; }); + OnWorkers(engines.size(), [&](size_t t) { engines[t].Release(); }); } // The spots of each ordinal, in the order given. @@ -164,6 +164,13 @@ namespace { std::unique_ptr indexer; std::unique_ptr analysis; std::unique_ptr profile; + // In the order destruction takes, not the order a move-assignment would: see SpotEngine. + void Release() { + profile.reset(); + analysis.reset(); + indexer.reset(); + raw_image = {}; + } }; // fn(t) on n threads of its own, thread t on the card its engine lives on. @@ -568,7 +575,7 @@ void Rugnux::FirstPassRun::ReleaseSpotEngines() { for (size_t t = 0; t < engines.size(); t++) futures.emplace_back(std::async(std::launch::async, [&, t] { UseGpuOfWorker(t); - engines[t] = SpotEngine{}; + engines[t].Release(); })); for (auto &f : futures) f.get(); diff --git a/rugnux/RugnuxFirstPass.h b/rugnux/RugnuxFirstPass.h index c6fdb2d88..7ca66ce90 100644 --- a/rugnux/RugnuxFirstPass.h +++ b/rugnux/RugnuxFirstPass.h @@ -200,6 +200,16 @@ struct Rugnux::FirstPassRun { JFJochReaderRawImage raw_image; std::unique_ptr analysis; std::unique_ptr profile; + // Give the engine back with the buffer last. Not by assigning a fresh SpotEngine over it: a + // move-assignment goes in declaration order, so it freed raw_image's bytes while the engine + // still had them page-locked, and unregistered them only afterwards - by then the address + // could belong to another thread's buffer, whose copies the driver then took through the + // stale registration (and whose cudaMallocHost it refused). + void Release() { + profile.reset(); + analysis.reset(); + raw_image = {}; + } }; std::vector engines; int prefetch_batches = 0; diff --git a/rugnux/RugnuxImagePass.cpp b/rugnux/RugnuxImagePass.cpp index 63def847a..7d0828cfc 100644 --- a/rugnux/RugnuxImagePass.cpp +++ b/rugnux/RugnuxImagePass.cpp @@ -221,6 +221,8 @@ void Rugnux::ImagePass(PipelineLocals &p) { // Before the analysis engine, which page-locks these bytes for its uploads and unregisters // them when it is destroyed: it must not outlive the buffer it was given. std::shared_ptr img; + // The image the writer holds whose bytes the engine still has page-locked (see below). + std::shared_ptr pinned; MXAnalysisWithoutFPGA analysis(experiment_, mapping, pixel_mask_, *indexer, /*enable_fused_adaptive_gpu=*/true); AzimuthalIntegrationProfile profile(mapping); @@ -235,8 +237,12 @@ void Rugnux::ImagePass(PipelineLocals &p) { // Reuse the raw image frame to frame - a fresh 4-byte 16 Mpx frame is 72 MB of // untouched pages the decoder faults in one by one - except while the writer thread // still holds the previous one's pixels. - if (!img || img.use_count() > 1) + // The engine unregisters the old bytes when it pins the new ones (Analyze), so until then + // the old image is kept here: the writer would otherwise free bytes still registered. + if (!img || img.use_count() > 1) { + pinned = std::move(img); img = std::make_shared(); + } bool read = false; try { read = reader_.ReadRawImage(image_idx, *img); @@ -263,6 +269,7 @@ void Rugnux::ImagePass(PipelineLocals &p) { logger.Error("Error analyzing image {}: {}", image_idx, e.what()); continue; } + pinned.reset(); msg.processing_time_s = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); msg.run_number = experiment_.GetRunNumber(); msg.run_name = experiment_.GetRunName();