Sync stream before destroying GPU representation in convert_to

[mbrobbel]

Summary

• Conversions from GPU in data_batch::convert_to may enqueue async operations on the provided CUDA stream that read from the source GPU memory. Previously, the source GPU representation was destroyed immediately after reassigning _data, risking use-after-free or leaking buffers into process-lifetime static storage.
• Synchronize the stream before the old GPU representation is destroyed to ensure async operations complete.
• Use unique_lock instead of lock_guard to release the mutex before synchronizing, avoiding blocking other threads during the stream sync.
• The sync condition triggers for any conversion from a GPU source, not just GPU→HOST — this covers future converters (e.g. GPU→DISK) and custom converters that may enqueue async reads without synchronizing internally.

Test plan

• New test: stream is synchronized before old GPU representation is destroyed
  • Passes with fix, fails without fix (synced_before_destroy = false)
• New test: mutex is released before stream sync, allowing concurrent access
  • Passes with early unlock, fails when sync is inside the lock (accessed_during_sync = false)
  • Uses cudaLaunchHostFunc callback (50 ms sleep) for a deterministic window independent of GPU speed
• Without any fix: first test fails (no sync), second test passes (no lock contention since no sync)
• Verify full test suite passes

Local test results

Configuration	Test 1 (sync before destroy)	Test 2 (mutex released during sync)
With fix (early unlock + sync)	✅ Pass	✅ Pass
No fix (original code)	❌ Fail — no sync at all	✅ Pass — no lock contention since no sync
Sync inside lock (naive fix)	✅ Pass — sync happens	❌ Fail — mutex held during 50 ms sync

Only the correct implementation (early unlock + sync) passes both tests.

Raw test output

With fix (early unlock + sync):

Filters: [convert_to]
===============================================================================
All tests passed (3 assertions in 2 test cases)

Without fix (original code, no sync at all):

/home/matthijs/sirius/cucascade/test/data/test_data_batch.cpp:1753: FAILED:
  REQUIRE( observer.synced_before_destroy )
with expansion:
  false

test cases: 2 | 1 passed | 1 failed
assertions: 3 | 2 passed | 1 failed

With sync inside lock (correct sync, but naive lock_guard):

/home/matthijs/sirius/cucascade/test/data/test_data_batch.cpp:1849: FAILED:
  REQUIRE( accessed_during_sync )
with expansion:
  false

test cases: 2 | 1 passed | 1 failed
assertions: 2 | 1 passed | 1 failed

How the tests work

Test 1 — sync before destroy: Registers a custom converter that enqueues a 4 MB async cudaMemcpyDeviceToHost into pinned memory and records a CUDA event, but does not synchronize. A custom observed_gpu_representation checks cudaEventQuery in its destructor. Without the sync, the event is still pending at destruction time.

Test 2 — mutex released before sync: Same converter setup, plus a cudaLaunchHostFunc callback that sleeps for 50 ms on the stream (creating a deterministic delay independent of GPU speed). A second thread calls get_state() (which acquires the mutex) and checks the event. With the early unlock, get_state() returns immediately while the 50 ms callback is still running. With the sync inside the lock, get_state() is blocked for the full 50 ms.

🤖 Created by Claude

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

[dhruv9vats]

Thanks for catching this!

[dhruv9vats]

do we not want to cover the to GPU memory case too?

[felipeblazing]

/merge