Fixes hangs when Python future notifier is enabled (#590)

Author: pentschev

ci/run_python.sh

@@ -10,13 +10,6 @@ source "$(dirname "$0")/test_common.sh"
 # Support invoking run_pytests.sh outside the script directory
 cd "$(dirname "$(realpath "${BASH_SOURCE[0]}")")"/../
 
-# Check if Python version >= 3.13.12 (has asyncio bug)
-# Returns 0 if version >= 3.13.12, 1 otherwise
-is_python_313_12_or_higher() {
-  python -c "import sys; exit(0 if sys.version_info >= (3, 13, 12) else 1)"
-  return $?
-}
-
 run_py_tests() {
   if [ "${DISABLE_CYTHON:-0}" -eq 0 ]; then
     ARGS=("--run-cython")
@@ -85,12 +78,7 @@ run_py_tests
 log_message "Python Async Tests"
 # run_py_tests_async PROGRESS_MODE   ENABLE_DELAYED_SUBMISSION ENABLE_PYTHON_FUTURE SKIP
 run_py_tests_async   thread          0                         0                    0
-# Skip Python futures tests on Python 3.13.12+ due to asyncio bug: https://github.com/rapidsai/ucxx/issues/586
-if is_python_313_12_or_higher; then
-  run_py_tests_async   thread          1                         1                    1
-else
-  run_py_tests_async   thread          1                         1                    0
-fi
+run_py_tests_async   thread          1                         1                    0
 run_py_tests_async   blocking        0                         0                    0
 
 log_message "Python Benchmarks"

ci/run_python_distributed.sh

@@ -10,13 +10,6 @@ source "$(dirname "$0")/test_common.sh"
 # Support invoking run_pytests.sh outside the script directory
 cd "$(dirname "$(realpath "${BASH_SOURCE[0]}")")"/../
 
-# Check if Python version >= 3.13.12 (has asyncio bug)
-# Returns 0 if version >= 3.13.12, 1 otherwise
-is_python_313_12_or_higher() {
-  python -c "import sys; exit(0 if sys.version_info >= (3, 13, 12) else 1)"
-  return $?
-}
-
 install_distributed_dev_mode() {
   # Running Distributed tests which access its internals requires installing it in
   # developer mode. This isn't a great solution but it's what we can currently do
@@ -84,28 +77,20 @@ run_distributed_ucxx_tests_internal() {
   fi
 }
 
-# Determine if we should skip Python futures tests (Python 3.13.12+ has asyncio bug)
-# See: https://github.com/rapidsai/ucxx/issues/586
-if is_python_313_12_or_higher; then
-  SKIP_PYTHON_FUTURES=1
-else
-  SKIP_PYTHON_FUTURES=0
-fi
-
 # run_distributed_ucxx_tests    PROGRESS_MODE   ENABLE_DELAYED_SUBMISSION   ENABLE_PYTHON_FUTURE    SKIP
 run_distributed_ucxx_tests      blocking        0                           0                       0
 run_distributed_ucxx_tests      polling         0                           0                       0
 run_distributed_ucxx_tests      thread          0                           0                       0
-run_distributed_ucxx_tests      thread          0                           1                       ${SKIP_PYTHON_FUTURES}
+run_distributed_ucxx_tests      thread          0                           1                       0
 run_distributed_ucxx_tests      thread          1                           0                       0
-run_distributed_ucxx_tests      thread          1                           1                       ${SKIP_PYTHON_FUTURES}
+run_distributed_ucxx_tests      thread          1                           1                       0
 
 install_distributed_dev_mode
 
 # run_distributed_ucxx_tests_internal   PROGRESS_MODE   ENABLE_DELAYED_SUBMISSION   ENABLE_PYTHON_FUTURE    SKIP
 run_distributed_ucxx_tests_internal     blocking        0                           0                       0
 run_distributed_ucxx_tests_internal     polling         0                           0                       0
 run_distributed_ucxx_tests_internal     thread          0                           0                       0
-run_distributed_ucxx_tests_internal     thread          0                           1                       ${SKIP_PYTHON_FUTURES}
+run_distributed_ucxx_tests_internal     thread          0                           1                       0
 run_distributed_ucxx_tests_internal     thread          1                           0                       0
-run_distributed_ucxx_tests_internal     thread          1                           1                       ${SKIP_PYTHON_FUTURES}
+run_distributed_ucxx_tests_internal     thread          1                           1                       0

python/ucxx/ucxx/_lib_async/notifier_thread.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2022-2023, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: BSD-3-Clause
 
 
@@ -58,8 +58,10 @@ def _notifierThread(event_loop, worker, q):
         try:
             task.result(0.01)
         except TimeoutError:
-            task.cancel()
-            logger.debug("Notifier Thread Result Timeout")
+            # Do NOT cancel the task here. run_request_notifier() sets asyncio
+            # Futures for completed UCX requests; cancelling leaves them unset and
+            # causes send/recv to hang.
+            logger.debug("Notifier Thread Result Timeout (task left scheduled)")
         except Exception as e:
             logger.debug(f"Notifier Thread Result Exception: {e}")
 

python/ucxx/ucxx/_lib_async/utils_test.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2022-2025, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: BSD-3-Clause
 
 import asyncio
@@ -185,8 +185,10 @@ async def am_recv(ep):
 
 
 async def wait_listener_client_handlers(listener):
-    pass
     while listener.active_clients > 0:
-        await asyncio.sleep(0)
+        # Minimal delay to yield to the event loop so call_soon_threadsafe callbacks
+        # run. Using a very short positive sleep ensures pending callbacks are
+        # processed and significantly reduces "coroutine never awaited" warnings.
+        await asyncio.sleep(1e-9)
         if not ucxx.core._get_ctx().progress_mode.startswith("thread"):
             ucxx.progress()