HIP: accelerate PSF accumulation and restore parallel producers

Cooperate across 32 lanes per PSF and use two completion-protected staging slots with complete batch timing. Restore coarse OpenMP event production while serializing shared GPU submissions and direct fallback boundaries.

Add bounded benchmarks, streaming and renderer regressions, and preserve validation evidence and ownership documentation.

src/frame.c

@@ -133,6 +133,9 @@ typedef struct {
   const PsfKernelCache *cache;
 #ifdef PSF_BACKEND_HIP
   HipPsfSink *hip;
+  omp_lock_t *hip_lock; /* Shared sink lock; acquired only per chunk/fallback. */
+  int borrowed_hip;
+  double submit_seconds;
   char hip_message[256];
   HipPsfTiming hip_timing;
 #endif
@@ -176,7 +179,7 @@ static int psf_event_sink_init(PsfEventSink *sink, double *hdr, int width,
   return 0;
 }
 
-static int psf_event_sink_flush(PsfEventSink *sink) {
+static int psf_event_sink_flush_unlocked(PsfEventSink *sink) {
   if (sink->failed)
     return -1;
 #ifdef PSF_BACKEND_HIP
@@ -197,8 +200,27 @@ static int psf_event_sink_flush(PsfEventSink *sink) {
   return 0;
 }
 
+static int psf_event_sink_flush(PsfEventSink *sink) {
+#ifdef PSF_BACKEND_HIP
+  const double start = omp_get_wtime();
+  if (sink->hip_lock) omp_set_lock(sink->hip_lock);
+#endif
+  const int result = psf_event_sink_flush_unlocked(sink);
+#ifdef PSF_BACKEND_HIP
+  if (sink->hip_lock) omp_unset_lock(sink->hip_lock);
+  sink->submit_seconds += omp_get_wtime() - start;
+#endif
+  return result;
+}
+
+/* A borrowed sink calls this only while holding hip_lock across the entire
+ * download -> CPU fallback -> upload transaction. */
 static int psf_event_sink_finish_for_cpu(PsfEventSink *sink) {
+#ifdef PSF_BACKEND_HIP
+  if (psf_event_sink_flush_unlocked(sink))
+#else
   if (psf_event_sink_flush(sink))
+#endif
     return -1;
 #ifdef PSF_BACKEND_HIP
   if (sink->hip != NULL && hip_psf_sink_finish(sink->hip, sink->hdr,
@@ -226,6 +248,13 @@ static int psf_event_sink_resume_gpu(PsfEventSink *sink) {
 }
 
 static int psf_event_sink_destroy(PsfEventSink *sink) {
+#ifdef PSF_BACKEND_HIP
+  if (sink->borrowed_hip) {
+    const int result = psf_event_sink_flush(sink);
+    free(sink->events);
+    return result;
+  }
+#endif
   int result = psf_event_sink_finish_for_cpu(sink);
 #ifdef PSF_BACKEND_HIP
   if (sink->hip != NULL && hip_psf_sink_get_timing(sink->hip, &sink->hip_timing)) {
@@ -238,7 +267,7 @@ static int psf_event_sink_destroy(PsfEventSink *sink) {
   return result;
 }
 
-#if FRAME_PSF_EVENT_SINK
+#if FRAME_PSF_EVENT_SINK || defined(PSF_BACKEND_HIP)
 static void psf_event_sink_emit(PsfEventSink *sink, const PsfCachedEvent *event) {
   if (sink->failed)
     return;
@@ -1086,16 +1115,27 @@ static int splat_catalog_tile(const Star *stars, size_t count,
         &event, image_x, image_y, color, flux, context->psf, context->psf_cache,
         context->max_cache_psf_flux, context->psf_relative_tail, context->psf_min_y);
     if (direct_fallback == 1) {
-      /* A direct fallback forms an explicit ordered CPU/GPU boundary. */
-      if (psf_event_sink_finish_for_cpu(context->event_sink))
-        return -1;
-      splat_moffat_direct(context->hdr, context->width, context->height, image_x,
-                          image_y, color, flux, context->psf,
-                          context->psf_relative_tail, context->psf_min_y);
-      if (psf_event_sink_resume_gpu(context->event_sink))
-        return -1;
+      /* Exclude every other submit and fallback until HDR is back on device. */
+#ifdef PSF_BACKEND_HIP
+      const double submit_start = omp_get_wtime();
+      if (context->event_sink->hip_lock)
+        omp_set_lock(context->event_sink->hip_lock);
+#endif
+      int failed = psf_event_sink_finish_for_cpu(context->event_sink);
+      if (!failed) {
+        splat_moffat_direct(context->hdr, context->width, context->height, image_x,
+                            image_y, color, flux, context->psf,
+                            context->psf_relative_tail, context->psf_min_y);
+        failed = psf_event_sink_resume_gpu(context->event_sink);
+      }
+#ifdef PSF_BACKEND_HIP
+      if (context->event_sink->hip_lock)
+        omp_unset_lock(context->event_sink->hip_lock);
+      context->event_sink->submit_seconds += omp_get_wtime() - submit_start;
+#endif
+      if (failed) return -1;
     } else if (direct_fallback != 3) {
-#if FRAME_PSF_EVENT_SINK
+#if FRAME_PSF_EVENT_SINK || defined(PSF_BACKEND_HIP)
       psf_event_sink_emit(context->event_sink, &event);
 #else
       splat_prepared_cached_event(context->hdr, context->width, context->height,
@@ -1261,20 +1301,91 @@ size_t frame_splat_catalog(const FrameLensMesh *mesh,
                        mesh->triangle_count);
 
 #ifdef PSF_BACKEND_HIP
-  /* A single GPU HDR framebuffer owns all cached events.  Keep catalog/lens
-   * work serial for this first direct-atomic integration; the CPU parallel
-   * private-HDR path remains the PSF_BACKEND=cpu implementation. */
-  const CatalogSplatStats hip_stats = splat_catalog_triangles(
-      mesh, catalog, hdr, width, height, exposure, psf, psf_cache,
-      max_magnification, max_cache_psf_flux, psf_relative_tail, psf_min_y,
-      0, mesh->triangle_count, NULL);
-  copy_psf_splat_stats(psf_stats, hip_stats);
-  if (hip_stats.failed)
-    return SIZE_MAX;
+  /* CPU workers own only bounded event chunks. A single device cache/HDR is
+   * borrowed under a coarse submission lock, never duplicated per worker. */
+  PsfEventSink owner;
+  if (psf_event_sink_init(&owner, hdr, width, height, psf_cache)) return SIZE_MAX;
+  omp_lock_t submit_lock;
+  omp_init_lock(&submit_lock);
+  size_t hip_images = 0, hip_direct = 0, hip_clipped = 0, hip_discarded = 0;
+  int hip_failed = 0, hip_workers = 0;
+  double hip_generate_seconds = 0.0, hip_submit_seconds = 0.0;
+  const double hip_start = omp_get_wtime();
+#ifdef GR_DEBUG
+  double hip_max_magnification = 0.0;
+  size_t hip_clamped = 0;
+#pragma omp parallel reduction(+ : hip_images, hip_direct, hip_clipped, hip_discarded, hip_clamped, hip_generate_seconds, hip_submit_seconds) reduction(max : hip_failed, hip_max_magnification)
+#else
+#pragma omp parallel reduction(+ : hip_images, hip_direct, hip_clipped, hip_discarded, hip_generate_seconds, hip_submit_seconds) reduction(max : hip_failed)
+#endif
+  {
+    const double worker_start = omp_get_wtime();
+#pragma omp single
+    hip_workers = omp_get_num_threads();
+    PsfEventSink local = {.hdr = hdr, .width = width, .height = height,
+        .cache = psf_cache, .hip = owner.hip, .hip_lock = &submit_lock,
+        .borrowed_hip = 1};
+    local.events = malloc(PSF_EVENT_SINK_CAPACITY * sizeof *local.events);
+    if (!local.events) {
+      snprintf(local.hip_message, sizeof local.hip_message, "worker event allocation failed");
+      psf_event_sink_mark_failed(&local, "producer initialization");
+    }
+    const size_t worker = (size_t)omp_get_thread_num();
+    const size_t workers = (size_t)omp_get_num_threads();
+    size_t completed = 0, next_report = 8;
+    if (progress && progress->worker_callback)
+      progress->worker_callback(progress->context, worker, workers, 0, 0);
+#pragma omp for schedule(dynamic, 1) nowait
+    for (size_t t = 0; t < mesh->triangle_count; ++t) {
+      if (local.failed) continue;
+      const CatalogSplatStats stats = splat_catalog_triangles(
+          mesh, catalog, hdr, width, height, exposure, psf, psf_cache,
+          max_magnification, max_cache_psf_flux, psf_relative_tail, psf_min_y,
+          t, t + 1, &local);
+      hip_images += stats.images;
+      hip_direct += stats.direct_fallbacks;
+      hip_clipped += stats.cached_wing_clipped;
+      hip_discarded += stats.discarded_below_min_y;
+      hip_failed |= stats.failed;
+#ifdef GR_DEBUG
+      hip_max_magnification = fmax(hip_max_magnification, stats.max_raw_magnification);
+      hip_clamped += stats.magnification_clamped_triangles;
+#endif
+      ++completed;
+      if (progress && progress->worker_callback && completed == next_report) {
+        progress->worker_callback(progress->context, worker, workers, completed, 0);
+        if (next_report <= SIZE_MAX / 2) next_report *= 2;
+      }
+    }
+    if (psf_event_sink_destroy(&local)) hip_failed = 1;
+    hip_submit_seconds += local.submit_seconds;
+    hip_generate_seconds += omp_get_wtime() - worker_start - local.submit_seconds;
+    if (progress && progress->worker_callback)
+      progress->worker_callback(progress->context, worker, workers, completed, 1);
+  }
+  omp_destroy_lock(&submit_lock);
+  if (psf_event_sink_destroy(&owner)) hip_failed = 1;
+  fprintf(stderr, "HIP producers: %d workers; summed generation %.3f s, submission/fallback %.3f s; wall %.3f s\n",
+      hip_workers, hip_generate_seconds, hip_submit_seconds, omp_get_wtime() - hip_start);
+  copy_psf_splat_stats(psf_stats, (CatalogSplatStats){
+      .images = hip_images, .direct_fallbacks = hip_direct,
+      .cached_wing_clipped = hip_clipped, .discarded_below_min_y = hip_discarded,
+      .gpu_event_count = owner.hip_timing.event_count,
+      .gpu_batch_count = owner.hip_timing.batch_count,
+      .gpu_timed_batch_count = owner.hip_timing.timed_batch_count,
+      .gpu_upload_seconds = owner.hip_timing.upload_seconds,
+      .gpu_kernel_seconds = owner.hip_timing.kernel_seconds,
+      .gpu_download_seconds = owner.hip_timing.download_seconds,
+#ifdef GR_DEBUG
+      .max_raw_magnification = hip_max_magnification,
+      .magnification_clamped_triangles = hip_clamped,
+#endif
+  });
+  if (hip_failed) return SIZE_MAX;
   if (progress != NULL && progress->callback != NULL)
     progress->callback(progress->context, FRAME_SPLAT_PROGRESS_END,
                        mesh->triangle_count, mesh->triangle_count);
-  return hip_stats.images;
+  return hip_images;
 #endif
 
   const size_t pixel_count = (size_t)width * height * 3;

src/hip_psf.h

@@ -32,16 +32,18 @@ int hip_psf_sink_create(HipPsfSink **sink, int width, int height,
 int hip_psf_sink_load_hdr(HipPsfSink *sink, const double *hdr,
                           char *message, size_t message_size);
 
-/* Queues a cached-event chunk. Direct fallbacks and min-Y discards stay with
- * the caller; event_count must not exceed the creation capacity. */
+/* Copies a chunk into one of two owned pinned staging slots; caller memory
+ * may be reused on return. Slot reuse waits for its previous kernel. Calls
+ * sharing a sink must be serialized by the owner. Direct fallbacks and min-Y
+ * discards stay with the caller; event_count must not exceed capacity. */
 int hip_psf_sink_submit(HipPsfSink *sink, const PsfCachedEvent *events,
                         size_t event_count, char *message, size_t message_size);
 
 /* Completes all work and overwrites hdr with double linear RGB device HDR. */
 int hip_psf_sink_finish(HipPsfSink *sink, double *hdr, char *message,
                         size_t message_size);
-/* Valid after hip_psf_sink_finish(); timings are accumulated without a
- * per-batch host synchronization. */
+/* Complete after finish(); otherwise includes recycled/completed slots only.
+ * Every batch is timed with a fixed-size reusable event pool. */
 int hip_psf_sink_get_timing(const HipPsfSink *sink, HipPsfTiming *timing);
 void hip_psf_sink_destroy(HipPsfSink *sink);
 

src/hip_psf.hip

@@ -5,27 +5,34 @@
 #include <cmath>
 #include <cstdio>
 #include <limits>
-#include <vector>
+#include <chrono>
+#include <cstring>
 
-struct HipPsfBatchTiming {
+enum { PSF_LANES_PER_EVENT = 32, PSF_THREADS_PER_BLOCK = 128 };
+
+/* Two reusable staging slots bound host/device queue memory. A slot may be
+ * overwritten only after its kernel_end event completes. All API calls for a
+ * sink are serialized by its owner, including calls from CPU producer workers. */
+struct HipPsfBatch {
+  PsfCachedEvent *host = nullptr, *device = nullptr;
   hipEvent_t upload_start = nullptr, upload_end = nullptr;
   hipEvent_t kernel_start = nullptr, kernel_end = nullptr;
+  size_t count = 0;
 };
 
-enum { HIP_PSF_MAX_TIMED_BATCHES = 64 };
-
 struct HipPsfSink {
-  PsfCachedEvent *events = nullptr;
+  HipPsfBatch slots[2];
   float *weights = nullptr;
   double *hdr = nullptr;
   hipStream_t stream = nullptr;
   size_t event_capacity = 0, hdr_values = 0;
+  size_t completed_events = 0;
+  int device = 0;
   int width = 0, height = 0, phase_resolution = 0, radius_pixels = 0;
   double max_radius_pixels = 0.0;
   hipEvent_t download_start = nullptr, download_end = nullptr;
-  std::vector<HipPsfBatchTiming> batches;
-  size_t measured_batches = 0;
   HipPsfTiming timing = {};
+  std::chrono::steady_clock::time_point last_report = std::chrono::steady_clock::now();
 };
 
 static int report(hipError_t status, char *message, size_t message_size) {
@@ -37,13 +44,19 @@ static void ok(char *message, size_t message_size) {
   if (message && message_size) std::snprintf(message, message_size, "ok");
 }
 
-static void destroy_batch_timing(HipPsfBatchTiming *timing) {
-  if (!timing) return;
-  (void)hipEventDestroy(timing->upload_start);
-  (void)hipEventDestroy(timing->upload_end);
-  (void)hipEventDestroy(timing->kernel_start);
-  (void)hipEventDestroy(timing->kernel_end);
-  *timing = {};
+static int collect_batch(HipPsfSink *sink, HipPsfBatch *batch,
+                         char *message, size_t message_size) {
+  if (!batch->count) return 0;
+  float upload_ms = 0.0f, kernel_ms = 0.0f;
+  if (report(hipEventSynchronize(batch->kernel_end), message, message_size) ||
+      report(hipEventElapsedTime(&upload_ms, batch->upload_start, batch->upload_end), message, message_size) ||
+      report(hipEventElapsedTime(&kernel_ms, batch->kernel_start, batch->kernel_end), message, message_size)) return -1;
+  sink->timing.upload_seconds += upload_ms * 1e-3;
+  sink->timing.kernel_seconds += kernel_ms * 1e-3;
+  ++sink->timing.timed_batch_count;
+  sink->completed_events += batch->count;
+  batch->count = 0;
+  return 0;
 }
 
 __device__ static size_t weight_index(int phase_resolution, int radius_pixels,
@@ -71,7 +84,11 @@ __global__ static void splat_kernel(const PsfCachedEvent *events, size_t count,
                                     int width, int height, const float *weights,
                                     int phase_resolution, int radius_pixels,
                                     double max_radius_pixels, double *hdr) {
-  const size_t index = (size_t)blockIdx.x * blockDim.x + threadIdx.x;
+  /* Consecutive lanes visit consecutive pixels of the same phase plane.
+   * Fixed 32-lane groups also work on wave64 devices without wave intrinsics. */
+  const size_t thread = (size_t)blockIdx.x * blockDim.x + threadIdx.x;
+  const size_t index = thread / PSF_LANES_PER_EVENT;
+  const int lane = thread % PSF_LANES_PER_EVENT;
   if (index >= count) return;
   const PsfCachedEvent event = events[index];
   const int base_x = (int)floor(event.x), base_y = (int)floor(event.y);
@@ -88,7 +105,7 @@ __global__ static void splat_kernel(const PsfCachedEvent *events, size_t count,
     if (!row_range(event.support_radius, fx, fy, support, offset_y, &first, &last)) continue;
     if (first < -base_x) first = -base_x;
     if (last >= width - base_x) last = width - base_x - 1;
-    for (int offset_x = first; offset_x <= last; ++offset_x) {
+    for (int offset_x = first + lane; offset_x <= last; offset_x += PSF_LANES_PER_EVENT) {
       const int px = base_x + offset_x;
       const double w00 = weights[weight_index(phase_resolution, radius_pixels, x0, y0, offset_x, offset_y)];
       const double w10 = weights[weight_index(phase_resolution, radius_pixels, x1, y0, offset_x, offset_y)];
@@ -119,15 +136,18 @@ extern "C" int hip_psf_sink_create(HipPsfSink **out, int width, int height,
                                      const PsfKernelCache *cache, size_t event_capacity,
                                      char *message, size_t message_size) {
   if (!out || width <= 0 || height <= 0 || !cache || !cache->ready || !cache->weights ||
-      cache->phase_resolution <= 0 || cache->radius_pixels < 0 || event_capacity == 0) {
+      cache->phase_resolution <= 0 || cache->radius_pixels < 0 || event_capacity == 0 ||
+      event_capacity > std::numeric_limits<size_t>::max() / sizeof(PsfCachedEvent)) {
     if (message && message_size) std::snprintf(message, message_size, "invalid HIP PSF sink arguments");
     return -1;
   }
   const size_t nodes = (size_t)cache->phase_resolution + 1;
   const size_t side = (size_t)cache->radius_pixels * 2 + 1;
   if (nodes > std::numeric_limits<size_t>::max() / nodes || nodes * nodes > std::numeric_limits<size_t>::max() / side ||
-      nodes * nodes * side > std::numeric_limits<size_t>::max() / side || (size_t)width > std::numeric_limits<size_t>::max() / (size_t)height ||
-      (size_t)width * (size_t)height > std::numeric_limits<size_t>::max() / 3) {
+      nodes * nodes * side > std::numeric_limits<size_t>::max() / side ||
+      nodes * nodes * side * side > std::numeric_limits<size_t>::max() / sizeof(float) ||
+      (size_t)width > std::numeric_limits<size_t>::max() / (size_t)height ||
+      (size_t)width * (size_t)height > std::numeric_limits<size_t>::max() / (3 * sizeof(double))) {
     if (message && message_size) std::snprintf(message, message_size, "HIP PSF sink size overflow");
     return -1;
   }
@@ -138,16 +158,30 @@ extern "C" int hip_psf_sink_create(HipPsfSink **out, int width, int height,
   sink->width = width; sink->height = height; sink->phase_resolution = cache->phase_resolution;
   sink->radius_pixels = cache->radius_pixels; sink->max_radius_pixels = cache->max_radius_pixels;
   const size_t weight_count = nodes * nodes * side * side;
-  if (report(hipStreamCreate(&sink->stream), message, message_size) ||
+  if (report(hipGetDevice(&sink->device), message, message_size) ||
+      report(hipStreamCreate(&sink->stream), message, message_size) ||
       report(hipEventCreate(&sink->download_start), message, message_size) ||
       report(hipEventCreate(&sink->download_end), message, message_size) ||
-      report(hipMalloc(&sink->events, event_capacity * sizeof *sink->events), message, message_size) ||
       report(hipMalloc(&sink->weights, weight_count * sizeof *sink->weights), message, message_size) ||
       report(hipMalloc(&sink->hdr, sink->hdr_values * sizeof *sink->hdr), message, message_size) ||
       report(hipMemcpyAsync(sink->weights, cache->weights, weight_count * sizeof *sink->weights, hipMemcpyHostToDevice, sink->stream), message, message_size) ||
       report(hipMemsetAsync(sink->hdr, 0, sink->hdr_values * sizeof *sink->hdr, sink->stream), message, message_size)) {
     hip_psf_sink_destroy(sink); return -1;
   }
+  for (HipPsfBatch &batch : sink->slots) {
+    if (report(hipHostMalloc(&batch.host, event_capacity * sizeof(PsfCachedEvent)), message, message_size) ||
+        report(hipMalloc(&batch.device, event_capacity * sizeof(PsfCachedEvent)), message, message_size) ||
+        report(hipEventCreate(&batch.upload_start), message, message_size) ||
+        report(hipEventCreate(&batch.upload_end), message, message_size) ||
+        report(hipEventCreate(&batch.kernel_start), message, message_size) ||
+        report(hipEventCreate(&batch.kernel_end), message, message_size)) {
+      hip_psf_sink_destroy(sink); return -1;
+    }
+  }
+  /* The caller may release the immutable host cache after creation. */
+  if (report(hipStreamSynchronize(sink->stream), message, message_size)) {
+    hip_psf_sink_destroy(sink); return -1;
+  }
   *out = sink; ok(message, message_size); return 0;
 }
 
@@ -158,37 +192,37 @@ extern "C" int hip_psf_sink_submit(HipPsfSink *sink, const PsfCachedEvent *event
     return -1;
   }
   if (!event_count) { ok(message, message_size); return 0; }
-  const unsigned int threads = 128;
-  const size_t blocks = (event_count + threads - 1) / threads;
-  HipPsfBatchTiming timing;
-  const int timed = sink->batches.size() < HIP_PSF_MAX_TIMED_BATCHES;
-  if (blocks > std::numeric_limits<unsigned int>::max() ||
-      (timed && (report(hipEventCreate(&timing.upload_start), message, message_size) ||
-                 report(hipEventCreate(&timing.upload_end), message, message_size) ||
-                 report(hipEventCreate(&timing.kernel_start), message, message_size) ||
-                 report(hipEventCreate(&timing.kernel_end), message, message_size) ||
-                 report(hipEventRecord(timing.upload_start, sink->stream), message, message_size))) ||
-      report(hipMemcpyAsync(sink->events, events, event_count * sizeof *events,
-                            hipMemcpyHostToDevice, sink->stream), message, message_size) ||
-      (timed && (report(hipEventRecord(timing.upload_end, sink->stream), message, message_size) ||
-                 report(hipEventRecord(timing.kernel_start, sink->stream), message, message_size)))) {
-    destroy_batch_timing(&timing);
+  if (report(hipSetDevice(sink->device), message, message_size)) return -1;
+  const unsigned int threads = PSF_THREADS_PER_BLOCK;
+  const size_t per_block = PSF_THREADS_PER_BLOCK / PSF_LANES_PER_EVENT;
+  const size_t blocks = event_count / per_block + (event_count % per_block != 0);
+  if (blocks > std::numeric_limits<unsigned int>::max()) {
+    if (message && message_size) std::snprintf(message, message_size, "HIP PSF grid too large");
     return -1;
   }
+  HipPsfBatch &batch = sink->slots[sink->timing.batch_count % 2];
+  if (collect_batch(sink, &batch, message, message_size)) return -1;
+  std::memcpy(batch.host, events, event_count * sizeof *events);
+  if (report(hipEventRecord(batch.upload_start, sink->stream), message, message_size) ||
+      report(hipMemcpyAsync(batch.device, batch.host, event_count * sizeof *events,
+                            hipMemcpyHostToDevice, sink->stream), message, message_size) ||
+      report(hipEventRecord(batch.upload_end, sink->stream), message, message_size) ||
+      report(hipEventRecord(batch.kernel_start, sink->stream), message, message_size)) return -1;
   hipLaunchKernelGGL(splat_kernel, dim3((unsigned int)blocks), dim3(threads), 0, sink->stream,
-                     sink->events, event_count, sink->width, sink->height, sink->weights,
+                     batch.device, event_count, sink->width, sink->height, sink->weights,
                      sink->phase_resolution, sink->radius_pixels, sink->max_radius_pixels, sink->hdr);
   if (report(hipGetLastError(), message, message_size) ||
-      (timed && report(hipEventRecord(timing.kernel_end, sink->stream), message, message_size))) {
-    destroy_batch_timing(&timing);
-    return -1;
-  }
-  if (timed) {
-    sink->batches.push_back(timing);
-    ++sink->timing.timed_batch_count;
-  }
+      report(hipEventRecord(batch.kernel_end, sink->stream), message, message_size)) return -1;
+  batch.count = event_count;
   ++sink->timing.batch_count;
   sink->timing.event_count += event_count;
+  const auto now = std::chrono::steady_clock::now();
+  if (std::chrono::duration<double>(now - sink->last_report).count() >= 5.0) {
+    std::fprintf(stderr, "HIP PSF progress: %zu submitted / %zu completed events; %zu / %zu batches timed; kernel %.3f s\n",
+        sink->timing.event_count, sink->completed_events, sink->timing.timed_batch_count,
+        sink->timing.batch_count, sink->timing.kernel_seconds);
+    sink->last_report = now;
+  }
   ok(message, message_size); return 0;
 }
 
@@ -198,29 +232,24 @@ extern "C" int hip_psf_sink_load_hdr(HipPsfSink *sink, const double *hdr,
     if (message && message_size) std::snprintf(message, message_size, "invalid HIP PSF HDR upload");
     return -1;
   }
-  if (report(hipMemcpyAsync(sink->hdr, hdr, sink->hdr_values * sizeof *hdr,
-                             hipMemcpyHostToDevice, sink->stream), message, message_size))
+  if (report(hipSetDevice(sink->device), message, message_size) ||
+      report(hipMemcpyAsync(sink->hdr, hdr, sink->hdr_values * sizeof *hdr,
+                             hipMemcpyHostToDevice, sink->stream), message, message_size) ||
+      report(hipStreamSynchronize(sink->stream), message, message_size))
     return -1;
   ok(message, message_size); return 0;
 }
 
 extern "C" int hip_psf_sink_finish(HipPsfSink *sink, double *hdr, char *message, size_t message_size) {
   if (!sink || !hdr) { if (message && message_size) std::snprintf(message, message_size, "invalid HIP PSF HDR download"); return -1; }
-  if (report(hipEventRecord(sink->download_start, sink->stream), message, message_size) ||
+  if (report(hipSetDevice(sink->device), message, message_size) ||
+      report(hipEventRecord(sink->download_start, sink->stream), message, message_size) ||
       report(hipMemcpyAsync(hdr, sink->hdr, sink->hdr_values * sizeof *hdr, hipMemcpyDeviceToHost, sink->stream), message, message_size) ||
       report(hipEventRecord(sink->download_end, sink->stream), message, message_size) ||
       report(hipStreamSynchronize(sink->stream), message, message_size)) return -1;
   float milliseconds = 0.0f;
-  for (size_t i = sink->measured_batches; i < sink->batches.size(); ++i) {
-    const HipPsfBatchTiming &batch = sink->batches[i];
-    if (report(hipEventElapsedTime(&milliseconds, batch.upload_start, batch.upload_end),
-               message, message_size)) return -1;
-    sink->timing.upload_seconds += milliseconds * 1e-3;
-    if (report(hipEventElapsedTime(&milliseconds, batch.kernel_start, batch.kernel_end),
-               message, message_size)) return -1;
-    sink->timing.kernel_seconds += milliseconds * 1e-3;
-  }
-  sink->measured_batches = sink->batches.size();
+  for (HipPsfBatch &batch : sink->slots)
+    if (collect_batch(sink, &batch, message, message_size)) return -1;
   if (report(hipEventElapsedTime(&milliseconds, sink->download_start, sink->download_end),
              message, message_size)) return -1;
   sink->timing.download_seconds += milliseconds * 1e-3;
@@ -235,10 +264,20 @@ extern "C" int hip_psf_sink_get_timing(const HipPsfSink *sink, HipPsfTiming *tim
 
 extern "C" void hip_psf_sink_destroy(HipPsfSink *sink) {
   if (!sink) return;
-  for (HipPsfBatchTiming &timing : sink->batches)
-    destroy_batch_timing(&timing);
-  (void)hipEventDestroy(sink->download_start);
-  (void)hipEventDestroy(sink->download_end);
-  (void)hipFree(sink->events); (void)hipFree(sink->weights); (void)hipFree(sink->hdr);
-  (void)hipStreamDestroy(sink->stream); delete sink;
+  (void)hipSetDevice(sink->device);
+  if (sink->stream) (void)hipStreamSynchronize(sink->stream);
+  for (HipPsfBatch &batch : sink->slots) {
+    if (batch.upload_start) (void)hipEventDestroy(batch.upload_start);
+    if (batch.upload_end) (void)hipEventDestroy(batch.upload_end);
+    if (batch.kernel_start) (void)hipEventDestroy(batch.kernel_start);
+    if (batch.kernel_end) (void)hipEventDestroy(batch.kernel_end);
+    if (batch.host) (void)hipHostFree(batch.host);
+    if (batch.device) (void)hipFree(batch.device);
+  }
+  if (sink->download_start) (void)hipEventDestroy(sink->download_start);
+  if (sink->download_end) (void)hipEventDestroy(sink->download_end);
+  if (sink->weights) (void)hipFree(sink->weights);
+  if (sink->hdr) (void)hipFree(sink->hdr);
+  if (sink->stream) (void)hipStreamDestroy(sink->stream);
+  delete sink;
 }