From 1f1853b760bd5380be5a770d431feeadcc7de293 Mon Sep 17 00:00:00 2001
From: Thomas Grainger <tagrain@gmail.com>
Date: Tue, 26 Apr 2022 10:48:42 +0100
Subject: [PATCH] avoid pristine_loop/loop.start in loop_in_thread

---
 distributed/utils_test.py | 33 +++++++++++++++++++++++----------
 1 file changed, 23 insertions(+), 10 deletions(-)

diff --git a/distributed/utils_test.py b/distributed/utils_test.py
index dff357189ec..f4985ce636c 100644
--- a/distributed/utils_test.py
+++ b/distributed/utils_test.py
@@ -170,16 +170,29 @@ def start():
 
 @pytest.fixture
 def loop_in_thread(cleanup):
-    with pristine_loop() as loop:
-        thread = threading.Thread(target=loop.start, name="test IOLoop")
-        thread.daemon = True
-        thread.start()
-        loop_started = threading.Event()
-        loop.add_callback(loop_started.set)
-        loop_started.wait()
-        yield loop
-        loop.add_callback(loop.stop)
-        thread.join(timeout=5)
+    loop_started = concurrent.futures.Future()
+    with concurrent.futures.ThreadPoolExecutor(
+        1, thread_name_prefix="test IOLoop"
+    ) as tpe:
+
+        async def run():
+            io_loop = IOLoop.current()
+            stop_event = asyncio.Event()
+            loop_started.set_result((io_loop, stop_event))
+            await stop_event.wait()
+
+        ran = tpe.submit(_run_and_close_tornado, run)
+        for f in concurrent.futures.as_completed((loop_started, ran)):
+            if f is loop_started:
+                io_loop, stop_event = loop_started.result()
+                try:
+                    yield io_loop
+                finally:
+                    io_loop.add_callback(stop_event.set)
+
+            elif f is ran:
+                ran.result()
+                return
 
 
 @pytest.fixture