drop deprecated tornado.netutil.ExecutorResolver (#6031)

ExecutorResolver was deprecated in tornado v5.0 and causes
a number of problems for us:

- it results in a thread leak as the executor is never shutdown
and so requires a .warmup() from the resource tracker
- .warmup() instantiates the resolver and attaches an io_loop
  tornadoweb/tornado@43bd9ce
  this results in a DeprecationWarning on 3.10 because there's no
  loop running

this swaps it for a backport of DefaultLoopResolver from tornado v6.2
that uses get_running_loop().getaddrinfo for resolving DNS

Author: graingert

distributed/comm/tcp.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+import asyncio
 import ctypes
 import errno
 import functools
@@ -7,7 +10,7 @@
 import sys
 import weakref
 from ssl import SSLCertVerificationError, SSLError
-from typing import ClassVar
+from typing import Any, ClassVar
 
 from tornado import gen
 
@@ -43,7 +46,6 @@
 )
 from distributed.protocol.utils import pack_frames_prelude, unpack_frames
 from distributed.system import MEMORY_LIMIT
-from distributed.threadpoolexecutor import ThreadPoolExecutor
 from distributed.utils import ensure_ip, get_ip, get_ipv6, nbytes
 
 logger = logging.getLogger(__name__)
@@ -402,38 +404,31 @@ def _check_encryption(self, address, connection_args):
             )
 
 
-class BaseTCPConnector(Connector, RequireEncryptionMixin):
-    _executor: ClassVar[ThreadPoolExecutor] = ThreadPoolExecutor(
-        2, thread_name_prefix="TCP-Executor"
-    )
-    _client: ClassVar[TCPClient]
-
-    @classmethod
-    def warmup(cls) -> None:
-        """Pre-start threads and sockets to avoid catching them in checks for thread and
-        fd leaks
-        """
-        ex = cls._executor
-        while len(ex._threads) < ex._max_workers:
-            ex._adjust_thread_count()
-        cls._get_client()
-
-    @classmethod
-    def _get_client(cls):
-        if not hasattr(cls, "_client"):
-            resolver = netutil.ExecutorResolver(
-                close_executor=False, executor=cls._executor
+class _DefaultLoopResolver(netutil.Resolver):
+    """
+    Resolver implementation using `asyncio.loop.getaddrinfo`.
+    backport from Tornado 6.2+
+    https://github.com/tornadoweb/tornado/blob/3de78b7a15ba7134917a18b0755ea24d7f8fde94/tornado/netutil.py#L416-L432
+    """
+
+    async def resolve(
+        self, host: str, port: int, family: socket.AddressFamily = socket.AF_UNSPEC
+    ) -> list[tuple[int, Any]]:
+        # On Solaris, getaddrinfo fails if the given port is not found
+        # in /etc/services and no socket type is given, so we must pass
+        # one here.  The socket type used here doesn't seem to actually
+        # matter (we discard the one we get back in the results),
+        # so the addresses we return should still be usable with SOCK_DGRAM.
+        return [
+            (fam, address)
+            for fam, _, _, _, address in await asyncio.get_running_loop().getaddrinfo(
+                host, port, family=family, type=socket.SOCK_STREAM
             )
-            cls._client = TCPClient(resolver=resolver)
-        return cls._client
+        ]
 
-    @property
-    def client(self):
-        # The `TCPClient` is cached on the class itself to avoid creating
-        # excess `ThreadPoolExecutor`s. We delay creation until inside an async
-        # function to avoid accessing an IOLoop from a context where a backing
-        # event loop doesn't exist.
-        return self._get_client()
+
+class BaseTCPConnector(Connector, RequireEncryptionMixin):
+    client: ClassVar[TCPClient] = TCPClient(resolver=_DefaultLoopResolver())
 
     async def connect(self, address, deserialize=True, **connection_args):
         self._check_encryption(address, connection_args)

distributed/pytest_resourceleaks.py

@@ -31,9 +31,7 @@ def test1():
   flag would solve this issue. See pytest_rerunfailures code for inspiration.
 
 - The @gen_cluster fixture leaks 2 fds on the first test decorated with it within a test
-  suite; this is likely caused by an incomplete warmup routine of
-  distributed.comm.tcp.BaseTCPConnector.
-  This issue would also be fixed by rerunning failing tests.
+  suite; This issue would also be fixed by rerunning failing tests.
 
 - The @pytest.mark.flaky decorator (pytest_rerunfailures) completely disables this
   plugin for the decorated tests.
@@ -57,7 +55,6 @@ def test1():
 import psutil
 import pytest
 
-from distributed.comm.tcp import BaseTCPConnector
 from distributed.compatibility import WINDOWS
 from distributed.metrics import time
 
@@ -157,9 +154,6 @@ def format(self, before: int, after: int) -> str:
 
 
 class FDChecker(ResourceChecker, name="fds"):
-    def __init__(self):
-        BaseTCPConnector.warmup()
-
     def measure(self) -> int:
         if WINDOWS:
             # Don't use num_handles(); you'll get tens of thousands of reported leaks
@@ -187,9 +181,6 @@ def format(self, before: int, after: int) -> str:
 
 
 class ActiveThreadsChecker(ResourceChecker, name="threads"):
-    def __init__(self):
-        BaseTCPConnector.warmup()
-
     def measure(self) -> set[threading.Thread]:
         return set(threading.enumerate())
 

distributed/utils_test.py

@@ -47,7 +47,7 @@
 from distributed import versions as version_module
 from distributed.client import Client, _global_clients, default_client
 from distributed.comm import Comm
-from distributed.comm.tcp import TCP, BaseTCPConnector
+from distributed.comm.tcp import TCP
 from distributed.compatibility import WINDOWS
 from distributed.config import initialize_logging
 from distributed.core import CommClosedError, ConnectionPool, Status, connect, rpc
@@ -1589,9 +1589,6 @@ def save_sys_modules():
 @contextmanager
 def check_thread_leak():
     """Context manager to ensure we haven't leaked any threads"""
-    # "TCP-Executor" threads are never stopped once they are started
-    BaseTCPConnector.warmup()
-
     active_threads_start = threading.enumerate()
 
     yield