Move wait_for_signals to private module and deprecate distributed.cli…

….utils

distributed/_signals.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import signal
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+async def wait_for_signals(signals: list[signal.Signals]) -> None:
+    """Wait for the passed signals by setting global signal handlers"""
+    loop = asyncio.get_running_loop()
+    event = asyncio.Event()
+
+    old_handlers: dict[int, Any] = {}
+
+    def handle_signal(signum, frame):
+        # Restore old signal handler to allow for quicker exit
+        # if the user sends the signal again.
+        signal.signal(signum, old_handlers[signum])
+        logger.info("Received signal %s (%d)", signal.Signals(signum).name, signum)
+        loop.call_soon_threadsafe(event.set)
+
+    for sig in signals:
+        old_handlers[sig] = signal.signal(sig, handle_signal)
+
+    await event.wait()

distributed/cli/dask_scheduler.py

@@ -12,7 +12,7 @@
 from tornado.ioloop import IOLoop
 
 from distributed import Scheduler
-from distributed.cli.utils import wait_for_signals
+from distributed._signals import wait_for_signals
 from distributed.preloading import validate_preload_argv
 from distributed.proctitle import (
     enable_proctitle_on_children,

distributed/cli/dask_worker.py

@@ -20,7 +20,7 @@
 from dask.system import CPU_COUNT
 
 from distributed import Nanny
-from distributed.cli.utils import wait_for_signals
+from distributed._signals import wait_for_signals
 from distributed.comm import get_address_host_port
 from distributed.deploy.utils import nprocesses_nthreads
 from distributed.preloading import validate_preload_argv

distributed/cli/utils.py

@@ -1,34 +1,15 @@
 from __future__ import annotations
 
-import asyncio
 import logging
-import signal
 import warnings
-from typing import Any
 
 from tornado.ioloop import IOLoop
 
-logger = logging.getLogger(__name__)
-
-
-async def wait_for_signals(signals: list[signal.Signals]) -> None:
-    """Wait for the passed signals by setting global signal handlers"""
-    loop = asyncio.get_running_loop()
-    event = asyncio.Event()
+warnings.warn(
+    "the distributed.cli.utils module is deprecated", DeprecationWarning, stacklevel=2
+)
 
-    old_handlers: dict[int, Any] = {}
-
-    def handle_signal(signum, frame):
-        # Restore old signal handler to allow for quicker exit
-        # if the user sends the signal again.
-        signal.signal(signum, old_handlers[signum])
-        logger.info("Received signal %s (%d)", signal.Signals(signum).name, signum)
-        loop.call_soon_threadsafe(event.set)
-
-    for sig in signals:
-        old_handlers[sig] = signal.signal(sig, handle_signal)
-
-    await event.wait()
+logger = logging.getLogger(__name__)
 
 
 def install_signal_handlers(loop=None, cleanup=None):
@@ -37,9 +18,6 @@ def install_signal_handlers(loop=None, cleanup=None):
     a SIGINT or SIGTERM.  *cleanup* is an optional callback called,
     before the loop stops, with a single signal number argument.
     """
-    warnings.warn(
-        "install_signal_handlers is deprecated", DeprecationWarning, stacklevel=2
-    )
     import signal
 
     loop = loop or IOLoop.current()