[BUG][AscendP2P][InProcess] Fix Host-Staging H2H Copy non-draining when cancelled

[matthewygf]

Overview

The AscendP2P host-staging path can cancel the H2H copy coroutine, but the underlying
thread pool keeps executing. Cancelling an asyncio future does not stop a
ThreadPoolExecutor worker that has already started — concurrent.futures.Future.cancel()
returns False once the work is running.

The awaiting coroutine therefore resumed immediately while copy threads were still
touching the staging pages, and the caller went on to release_staged() them. This is now
fixed: we drain the copy first, then re-raise the cancellation.

Introduced with H2H host staging (#259).

Impact

release_staged() returns pages to the staging arena free list, so a page could be handed
to another in-flight transfer while a copy thread was still using it. Both peers are
affected, because both funnel through the same helper:

• Consumer (copy_receiver_staging_to): the page is freed while threads still read it,
  so a concurrent transfer can allocate it and DMA new data in — the final KV buffer ends
  up holding another request's bytes.
• Producer (stage): the page is freed while threads still write into it, so the next
  stage() can take the page and have it overwritten underneath. stage() also caught only
  Exception, so a CancelledError skipped the release entirely and leaked the page until
  channel teardown.

Trigger: use_host_staging with p2p_delay_pull: False and p2p_use_npu: False, when a
sync-get timeout (p2p_sync_get_timeout_s) cancels the pull during the post-DMA host copy.

What changed

All in _async_h2h_copy / stage in hccl_channel.py:

• Submit slices with executor.submit and keep the concurrent.futures handles locally, so
  task cancellation cannot detach the handle from work that is still running.
  run_in_executor marks its future cancelled even when the thread cannot be stopped.
• New _await_settled helper: re-await under asyncio.shield until the work actually
  settles, then re-raise the cancellation. A single shielded await is not enough — shield
  keeps the inner operation alive but still resumes the caller immediately, so repeated
  cancels (e.g. loop teardown after a timeout) would still free pages under live writers.
• gather(..., return_exceptions=True) so a failing slice cannot resume the caller while
  sibling threads are still writing.
• asyncio.wrap_future to observe thread completion on the event loop instead of blocking an
  extra executor thread per in-flight copy.
• stage() releases arena pages on CancelledError as well as Exception, which is safe
  now that the drain is guaranteed.

Implications

We wait slightly longer on the abort path for the in-flight copy to finish. That is one
chunk-sized host memcpy — sub-millisecond to a couple of milliseconds in practice. It is
also off the request critical path: the blocking sync-get has already returned
TimeoutError to the worker by then, so the only thing deferred is reuse of the staging
page.

The copies themselves still run off the event loop; only the completion notification touches
it. At teardown close() already does _staging_copy_pool.shutdown(wait=True) before
closing the arena, so threads are joined before the arena memory goes away.

Tests

tests/v1/transfer_channel/test_staging_h2h_copy.py replaces torch._foreach_copy_ with a
fake that parks inside the copy body and counts the threads currently in it, so each test
asserts both "N threads are still writing" and "the caller has not resumed" at the same
instant, across a 500 ms window.

test	guards against
test_cancel_waits_for_executor_workers	a single cancel freeing the page under a live copy
test_repeated_cancel_still_drains	a second or third cancel doing the same
test_exception_in_one_slice_drains_siblings	one slice's error resuming the caller while siblings write
test_stage_cancelled_error_releases_staged	producer stage() leaking its arena page on cancel

All four fail against the pre-fix code (caller resumed while 2 copy thread(s) were still writing) and pass with the fix. They need no NPU device and no peers, but are skipped when
the HCCL extension is not built.

Scope note

This addresses the host-side copy pool only. Draining the HCCL transport stream before
releasing buffers on an aborted transfer is a separate concern and is not part of this PR.