From 431f4eff76f9f3d45122d60957796d0a6b3b8c6d Mon Sep 17 00:00:00 2001 From: Tristan Rice Date: Tue, 24 Mar 2026 10:31:19 -0700 Subject: [PATCH 1/2] Fix NCCL non-blocking test failures by adding cuda.synchronize() The QuantizedAllReduceTest and QuantizedReduceScatterTest tests were failing with "NCCL Error 7: NCCL operation in progress" because non-blocking NCCL work.wait() returns before NCCL is ready for the next operation. Adding cuda.synchronize() between iterations ensures operations are fully complete. --- torchft/collectives_test.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/torchft/collectives_test.py b/torchft/collectives_test.py index 1062796a..bc521217 100644 --- a/torchft/collectives_test.py +++ b/torchft/collectives_test.py @@ -95,9 +95,13 @@ def _run_all_reduce_collective( work = allreduce_quantized(tensors, reduce_op, pg) work.wait() + # Synchronize to ensure non-blocking NCCL operations are + # fully complete before the next iteration. + cuda.synchronize() work = pg.allreduce([expected], reduce_op) work.get_future().wait() + cuda.synchronize() _check_result_tolerance(actual, expected, tolerance) @@ -142,6 +146,7 @@ def _run_reduce_scatter_collective( work = reduce_scatter_quantized(actual_output, tensors, opts, pg) work.get_future().wait() + cuda.synchronize() padded_sizes = get_padded_sizes(tensors, world_size) padded_numel = sum(s.numel() for s in padded_sizes) @@ -157,6 +162,7 @@ def _run_reduce_scatter_collective( work = pg.reduce_scatter([expected_output], [[padded_input]], opts) work.get_future().wait() + cuda.synchronize() _check_result_tolerance(actual_output, expected_output, tolerance) From d2888ad5b9e0a38383579fbff9b4dd4e17f18e8c Mon Sep 17 00:00:00 2001 From: Tristan Rice Date: Tue, 24 Mar 2026 12:59:26 -0700 Subject: [PATCH 2/2] Fix NCCL Error 7 by removing eager_connect_single_device In non-blocking NCCL mode (blocking=False), eager_connect_single_device starts an async communicator init that cannot be waited on through Python APIs. When the first collective is subsequently called, NCCL returns "Error 7: operation in progress" because the init hasn't completed. Remove the eager_connect call and let the communicator be lazily initialized by the first collective, which properly handles the async init inside its ncclGroupStart/ncclGroupEnd context. Also reverts the cuda.synchronize() workaround in collectives_test.py which was insufficient. --- torchft/collectives_test.py | 6 ------ torchft/process_group.py | 10 +++++++--- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/torchft/collectives_test.py b/torchft/collectives_test.py index bc521217..1062796a 100644 --- a/torchft/collectives_test.py +++ b/torchft/collectives_test.py @@ -95,13 +95,9 @@ def _run_all_reduce_collective( work = allreduce_quantized(tensors, reduce_op, pg) work.wait() - # Synchronize to ensure non-blocking NCCL operations are - # fully complete before the next iteration. - cuda.synchronize() work = pg.allreduce([expected], reduce_op) work.get_future().wait() - cuda.synchronize() _check_result_tolerance(actual, expected, tolerance) @@ -146,7 +142,6 @@ def _run_reduce_scatter_collective( work = reduce_scatter_quantized(actual_output, tensors, opts, pg) work.get_future().wait() - cuda.synchronize() padded_sizes = get_padded_sizes(tensors, world_size) padded_numel = sum(s.numel() for s in padded_sizes) @@ -162,7 +157,6 @@ def _run_reduce_scatter_collective( work = pg.reduce_scatter([expected_output], [[padded_input]], opts) work.get_future().wait() - cuda.synchronize() _check_result_tolerance(actual_output, expected_output, tolerance) diff --git a/torchft/process_group.py b/torchft/process_group.py index 87fd5991..574c5b22 100644 --- a/torchft/process_group.py +++ b/torchft/process_group.py @@ -864,9 +864,13 @@ def _create_pg(self, store: Store, rank: int, world_size: int) -> BaseProcessGro # pyre-fixme[16]: no attribute ProcessGroupNCCL backend_class = BaseProcessGroupNCCL(store, rank, world_size, opts) backend_class._set_sequence_number_for_group() - backend_class.eager_connect_single_device( - torch.device(torch.accelerator.current_device_index()) - ) + # NOTE: We intentionally do NOT call eager_connect_single_device here. + # In non-blocking mode (blocking=False), eager_connect starts an async + # communicator init that cannot be waited on through Python APIs. The + # first collective would then fail with "NCCL Error 7: operation in + # progress". Instead, we let the communicator be lazily initialized by + # the first collective, which properly handles the async init inside + # its ncclGroupStart/ncclGroupEnd context. pg._register_backend( torch.device("cuda"), ProcessGroup.BackendType.NCCL, backend_class )