fix test_inproc_specific_different_threads and test_io_loop_alternate…

…_loop

distributed/comm/tests/test_comms.py

@@ -27,7 +27,13 @@
 from distributed.comm.registry import backends, get_backend
 from distributed.metrics import time
 from distributed.protocol import Serialized, deserialize, serialize, to_serialize
-from distributed.utils import get_ip, get_ipv6, get_mp_context, wait_for
+from distributed.utils import (
+    get_ip,
+    get_ipv6,
+    get_mp_context,
+    run_and_close_tornado,
+    wait_for,
+)
 from distributed.utils_test import (
     gen_test,
     get_cert,
@@ -438,7 +444,7 @@ async def run_with_timeout():
         t = asyncio.create_task(func(*args, **kwargs))
         return await wait_for(t, timeout=10)
 
-    return await asyncio.to_thread(asyncio.run, run_with_timeout())
+    return await asyncio.to_thread(run_and_close_tornado, run_with_timeout)
 
 
 @gen_test()

distributed/tests/test_worker.py

@@ -592,16 +592,13 @@ async def test_io_loop(s):
 
 @gen_cluster(nthreads=[])
 async def test_io_loop_alternate_loop(s, loop):
-    async def main():
-        with pytest.warns(
-            DeprecationWarning,
-            match=r"The `loop` argument to `Worker` is ignored, and will be "
-            r"removed in a future release. The Worker always binds to the current loop",
-        ):
-            async with Worker(s.address, loop=loop) as w:
-                assert w.io_loop is w.loop is IOLoop.current()
-
-    await asyncio.to_thread(asyncio.run, main())
+    with pytest.warns(
+        DeprecationWarning,
+        match=r"The `loop` argument to `Worker` is ignored, and will be "
+        r"removed in a future release. The Worker always binds to the current loop",
+    ):
+        async with Worker(s.address, loop=loop) as w:
+            assert w.io_loop is w.loop is IOLoop.current()
 
 
 @gen_cluster(client=True)