always bind the Scheduler to the IOLoop.current() - and deprecate the Scheduler(loop= kwarg

distributed/cli/dask_scheduler.py

@@ -9,7 +9,6 @@
 import warnings
 
 import click
-from tornado.ioloop import IOLoop
 
 from distributed import Scheduler
 from distributed._signals import wait_for_signals
@@ -186,11 +185,9 @@ def del_pid_file():
         resource.setrlimit(resource.RLIMIT_NOFILE, (limit, hard))
 
     async def run():
-        loop = IOLoop.current()
         logger.info("-" * 47)
 
         scheduler = Scheduler(
-            loop=loop,
             security=sec,
             host=host,
             port=port,

distributed/scheduler.py

@@ -2882,6 +2882,14 @@ def __init__(
         transition_counter_max=False,
         **kwargs,
     ):
+        if loop is not None:
+            warnings.warn(
+                "the loop kwarg to Scheduler is deprecated",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+        self.loop = IOLoop.current()
         self._setup_logging(logger)
 
         # Attributes
@@ -2961,7 +2969,6 @@ def __init__(
             )
 
         # Communication state
-        self.loop = loop or IOLoop.current()
         self.client_comms = {}
         self.stream_comms = {}
         self._worker_coroutines = []

distributed/tests/test_scheduler.py

@@ -15,6 +15,7 @@
 import psutil
 import pytest
 from tlz import concat, first, merge, valmap
+from tornado.ioloop import IOLoop
 
 import dask
 from dask import delayed
@@ -779,8 +780,14 @@ async def test_update_graph_culls(s, a, b):
 
 
 def test_io_loop(loop):
-    s = Scheduler(loop=loop, dashboard_address=":0", validate=True)
-    assert s.io_loop is loop
+    async def main():
+        with pytest.warns(
+            DeprecationWarning, match=r"the loop kwarg to Scheduler is deprecated"
+        ):
+            s = Scheduler(loop=loop, dashboard_address=":0", validate=True)
+        assert s.io_loop is IOLoop.current()
+
+    asyncio.run(main())
 
 
 @gen_cluster(client=True)