{
    "file_diffs": [
        {
            "old_file_content": "\"\"\"Bridges between the `asyncio` module and Tornado IOLoop.\n\n.. versionadded:: 3.2\n\nThis module integrates Tornado with the ``asyncio`` module introduced\nin Python 3.4. This makes it possible to combine the two libraries on\nthe same event loop.\n\n.. deprecated:: 5.0\n\n   While the code in this module is still used, it is now enabled\n   automatically when `asyncio` is available, so applications should\n   no longer need to refer to this module directly.\n\n.. note::\n\n   Tornado is designed to use a selector-based event loop. On Windows,\n   where a proactor-based event loop has been the default since Python 3.8,\n   a selector event loop is emulated by running ``select`` on a separate thread.\n   Configuring ``asyncio`` to use a selector event loop may improve performance\n   of Tornado (but may reduce performance of other ``asyncio``-based libraries\n   in the same process).\n\"\"\"\n\nimport asyncio\nimport atexit\nimport concurrent.futures\nimport errno\nimport functools\nimport select\nimport socket\nimport sys\nimport threading\nimport typing\nimport warnings\nfrom tornado.gen import convert_yielded\nfrom tornado.ioloop import IOLoop, _Selectable\n\nfrom typing import Any, TypeVar, Awaitable, Callable, Union, Optional, List, Tuple, Dict\n\nif typing.TYPE_CHECKING:\n    from typing import Set  # noqa: F401\n    from typing_extensions import Protocol\n\n    class _HasFileno(Protocol):\n        def fileno(self) -> int:\n            pass\n\n    _FileDescriptorLike = Union[int, _HasFileno]\n\n_T = TypeVar(\"_T\")\n\n\n# Collection of selector thread event loops to shut down on exit.\n_selector_loops = set()  # type: Set[AddThreadSelectorEventLoop]\n\n\ndef _atexit_callback() -> None:\n    for loop in _selector_loops:\n        with loop._select_cond:\n            loop._closing_selector = True\n            loop._select_cond.notify()\n        try:\n            loop._waker_w.send(b\"a\")\n        except BlockingIOError:\n            pass\n        # If we don't join our (daemon) thread here, we may get a deadlock\n        # during interpreter shutdown. I don't really understand why. This\n        # deadlock happens every time in CI (both travis and appveyor) but\n        # I've never been able to reproduce locally.\n        loop._thread.join()\n    _selector_loops.clear()\n\n\natexit.register(_atexit_callback)\n\nif sys.version_info >= (3, 10):\n\n    def _get_event_loop() -> asyncio.AbstractEventLoop:\n        try:\n            return asyncio.get_running_loop()\n        except RuntimeError:\n            pass\n\n        return asyncio.get_event_loop_policy().get_event_loop()\n\n\nelse:\n    from asyncio import get_event_loop as _get_event_loop\n\n\nclass BaseAsyncIOLoop(IOLoop):\n    def initialize(  # type: ignore\n        self, asyncio_loop: asyncio.AbstractEventLoop, **kwargs: Any\n    ) -> None:\n        # asyncio_loop is always the real underlying IOLoop. This is used in\n        # ioloop.py to maintain the asyncio-to-ioloop mappings.\n        self.asyncio_loop = asyncio_loop\n        # selector_loop is an event loop that implements the add_reader family of\n        # methods. Usually the same as asyncio_loop but differs on platforms such\n        # as windows where the default event loop does not implement these methods.\n        self.selector_loop = asyncio_loop\n        if hasattr(asyncio, \"ProactorEventLoop\") and isinstance(\n            asyncio_loop, asyncio.ProactorEventLoop  # type: ignore\n        ):\n            # Ignore this line for mypy because the abstract method checker\n            # doesn't understand dynamic proxies.\n            self.selector_loop = AddThreadSelectorEventLoop(asyncio_loop)  # type: ignore\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}  # type: Dict[int, Tuple[Union[int, _Selectable], Callable]]\n        # Set of fds listening for reads/writes\n        self.readers = set()  # type: Set[int]\n        self.writers = set()  # type: Set[int]\n        self.closing = False\n        # If an asyncio loop was closed through an asyncio interface\n        # instead of IOLoop.close(), we'd never hear about it and may\n        # have left a dangling reference in our map. In case an\n        # application (or, more likely, a test suite) creates and\n        # destroys a lot of event loops in this way, check here to\n        # ensure that we don't have a lot of dead loops building up in\n        # the map.\n        #\n        # TODO(bdarnell): consider making self.asyncio_loop a weakref\n        # for AsyncIOMainLoop and make _ioloop_for_asyncio a\n        # WeakKeyDictionary.\n        for loop in IOLoop._ioloop_for_asyncio.copy():\n            if loop.is_closed():\n                try:\n                    del IOLoop._ioloop_for_asyncio[loop]\n                except KeyError:\n                    pass\n        IOLoop._ioloop_for_asyncio[asyncio_loop] = self\n\n        self._thread_identity = 0\n\n        super().initialize(**kwargs)\n\n        def assign_thread_identity() -> None:\n            self._thread_identity = threading.get_ident()\n\n        self.add_callback(assign_thread_identity)\n\n    def close(self, all_fds: bool = False) -> None:\n        self.closing = True\n        for fd in list(self.handlers):\n            fileobj, handler_func = self.handlers[fd]\n            self.remove_handler(fd)\n            if all_fds:\n                self.close_fd(fileobj)\n        # Remove the mapping before closing the asyncio loop. If this\n        # happened in the other order, we could race against another\n        # initialize() call which would see the closed asyncio loop,\n        # assume it was closed from the asyncio side, and do this\n        # cleanup for us, leading to a KeyError.\n        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]\n        if self.selector_loop is not self.asyncio_loop:\n            self.selector_loop.close()\n        self.asyncio_loop.close()\n\n    def add_handler(\n        self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int\n    ) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd in self.handlers:\n            raise ValueError(\"fd %s added twice\" % fd)\n        self.handlers[fd] = (fileobj, handler)\n        if events & IOLoop.READ:\n            self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.selector_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.selector_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd: Union[int, _Selectable]) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.selector_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.selector_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd: int, events: int) -> None:\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self) -> None:\n        try:\n            old_loop = _get_event_loop()\n        except (RuntimeError, AssertionError):\n            old_loop = None  # type: ignore\n        try:\n            asyncio.set_event_loop(self.asyncio_loop)\n            self.asyncio_loop.run_forever()\n        finally:\n            asyncio.set_event_loop(old_loop)\n\n    def stop(self) -> None:\n        self.asyncio_loop.stop()\n\n    def call_at(\n        self, when: float, callback: Callable, *args: Any, **kwargs: Any\n    ) -> object:\n        # asyncio.call_at supports *args but not **kwargs, so bind them here.\n        # We do not synchronize self.time and asyncio_loop.time, so\n        # convert from absolute to relative.\n        return self.asyncio_loop.call_later(\n            max(0, when - self.time()),\n            self._run_callback,\n            functools.partial(callback, *args, **kwargs),\n        )\n\n    def remove_timeout(self, timeout: object) -> None:\n        timeout.cancel()  # type: ignore\n\n    def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:\n        if threading.get_ident() == self._thread_identity:\n            call_soon = self.asyncio_loop.call_soon\n        else:\n            call_soon = self.asyncio_loop.call_soon_threadsafe\n        try:\n            call_soon(self._run_callback, functools.partial(callback, *args, **kwargs))\n        except RuntimeError:\n            # \"Event loop is closed\". Swallow the exception for\n            # consistency with PollIOLoop (and logical consistency\n            # with the fact that we can't guarantee that an\n            # add_callback that completes without error will\n            # eventually execute).\n            pass\n        except AttributeError:\n            # ProactorEventLoop may raise this instead of RuntimeError\n            # if call_soon_threadsafe races with a call to close().\n            # Swallow it too for consistency.\n            pass\n\n    def add_callback_from_signal(\n        self, callback: Callable, *args: Any, **kwargs: Any\n    ) -> None:\n        try:\n            self.asyncio_loop.call_soon_threadsafe(\n                self._run_callback, functools.partial(callback, *args, **kwargs)\n            )\n        except RuntimeError:\n            pass\n\n    def run_in_executor(\n        self,\n        executor: Optional[concurrent.futures.Executor],\n        func: Callable[..., _T],\n        *args: Any\n    ) -> Awaitable[_T]:\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor: concurrent.futures.Executor) -> None:\n        return self.asyncio_loop.set_default_executor(executor)\n\n\nclass AsyncIOMainLoop(BaseAsyncIOLoop):\n    \"\"\"``AsyncIOMainLoop`` creates an `.IOLoop` that corresponds to the\n    current ``asyncio`` event loop (i.e. the one returned by\n    ``asyncio.get_event_loop()``).\n\n    .. deprecated:: 5.0\n\n       Now used automatically when appropriate; it is no longer necessary\n       to refer to this class directly.\n\n    .. versionchanged:: 5.0\n\n       Closing an `AsyncIOMainLoop` now closes the underlying asyncio loop.\n    \"\"\"\n\n    def initialize(self, **kwargs: Any) -> None:  # type: ignore\n        super().initialize(asyncio.get_event_loop(), **kwargs)\n\n    def make_current(self) -> None:\n        # AsyncIOMainLoop already refers to the current asyncio loop so\n        # nothing to do here.\n        pass\n\n\nclass AsyncIOLoop(BaseAsyncIOLoop):\n    \"\"\"``AsyncIOLoop`` is an `.IOLoop` that runs on an ``asyncio`` event loop.\n    This class follows the usual Tornado semantics for creating new\n    ``IOLoops``; these loops are not necessarily related to the\n    ``asyncio`` default event loop.\n\n    Each ``AsyncIOLoop`` creates a new ``asyncio.EventLoop``; this object\n    can be accessed with the ``asyncio_loop`` attribute.\n\n    .. versionchanged:: 6.2\n\n       Support explicit ``asyncio_loop`` argument\n       for specifying the asyncio loop to attach to,\n       rather than always creating a new one with the default policy.\n\n    .. versionchanged:: 5.0\n\n       When an ``AsyncIOLoop`` becomes the current `.IOLoop`, it also sets\n       the current `asyncio` event loop.\n\n    .. deprecated:: 5.0\n\n       Now used automatically when appropriate; it is no longer necessary\n       to refer to this class directly.\n    \"\"\"\n\n    def initialize(self, **kwargs: Any) -> None:  # type: ignore\n        self.is_current = False\n        loop = None\n        if \"asyncio_loop\" not in kwargs:\n            kwargs[\"asyncio_loop\"] = loop = asyncio.new_event_loop()\n        try:\n            super().initialize(**kwargs)\n        except Exception:\n            # If initialize() does not succeed (taking ownership of the loop),\n            # we have to close it.\n            if loop is not None:\n                loop.close()\n            raise\n\n    def close(self, all_fds: bool = False) -> None:\n        if self.is_current:\n            self._clear_current()\n        super().close(all_fds=all_fds)\n\n    def make_current(self) -> None:\n        warnings.warn(\n            \"make_current is deprecated; start the event loop first\",\n            DeprecationWarning,\n        )\n        if not self.is_current:\n            try:\n                self.old_asyncio = asyncio.get_event_loop()\n            except (RuntimeError, AssertionError):\n                self.old_asyncio = None  # type: ignore\n            self.is_current = True\n        asyncio.set_event_loop(self.asyncio_loop)\n\n    def _clear_current_hook(self) -> None:\n        if self.is_current:\n            asyncio.set_event_loop(self.old_asyncio)\n            self.is_current = False\n\n\ndef to_tornado_future(asyncio_future: asyncio.Future) -> asyncio.Future:\n    \"\"\"Convert an `asyncio.Future` to a `tornado.concurrent.Future`.\n\n    .. versionadded:: 4.1\n\n    .. deprecated:: 5.0\n       Tornado ``Futures`` have been merged with `asyncio.Future`,\n       so this method is now a no-op.\n    \"\"\"\n    return asyncio_future\n\n\ndef to_asyncio_future(tornado_future: asyncio.Future) -> asyncio.Future:\n    \"\"\"Convert a Tornado yieldable object to an `asyncio.Future`.\n\n    .. versionadded:: 4.1\n\n    .. versionchanged:: 4.3\n       Now accepts any yieldable object, not just\n       `tornado.concurrent.Future`.\n\n    .. deprecated:: 5.0\n       Tornado ``Futures`` have been merged with `asyncio.Future`,\n       so this method is now equivalent to `tornado.gen.convert_yielded`.\n    \"\"\"\n    return convert_yielded(tornado_future)\n\n\nif sys.platform == \"win32\" and hasattr(asyncio, \"WindowsSelectorEventLoopPolicy\"):\n    # \"Any thread\" and \"selector\" should be orthogonal, but there's not a clean\n    # interface for composing policies so pick the right base.\n    _BasePolicy = asyncio.WindowsSelectorEventLoopPolicy  # type: ignore\nelse:\n    _BasePolicy = asyncio.DefaultEventLoopPolicy\n\n\nclass AnyThreadEventLoopPolicy(_BasePolicy):  # type: ignore\n    \"\"\"Event loop policy that allows loop creation on any thread.\n\n    The default `asyncio` event loop policy only automatically creates\n    event loops in the main threads. Other threads must create event\n    loops explicitly or `asyncio.get_event_loop` (and therefore\n    `.IOLoop.current`) will fail. Installing this policy allows event\n    loops to be created automatically on any thread, matching the\n    behavior of Tornado versions prior to 5.0 (or 5.0 on Python 2).\n\n    Usage::\n\n        asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())\n\n    .. versionadded:: 5.0\n\n    .. deprecated:: 6.2\n\n        ``AnyThreadEventLoopPolicy`` affects the implicit creation\n        of an event loop, which is deprecated in Python 3.10 and\n        will be removed in a future version of Python. At that time\n        ``AnyThreadEventLoopPolicy`` will no longer be useful.\n        If you are relying on it, use `asyncio.new_event_loop`\n        or `asyncio.run` explicitly in any non-main threads that\n        need event loops.\n    \"\"\"\n\n    def __init__(self) -> None:\n        super().__init__()\n        warnings.warn(\n            \"AnyThreadEventLoopPolicy is deprecated, use asyncio.run \"\n            \"or asyncio.new_event_loop instead\",\n            DeprecationWarning,\n        )\n\n    def get_event_loop(self) -> asyncio.AbstractEventLoop:\n        try:\n            return super().get_event_loop()\n        except (RuntimeError, AssertionError):\n            # This was an AssertionError in Python 3.4.2 (which ships with Debian Jessie)\n            # and changed to a RuntimeError in 3.4.3.\n            # \"There is no current event loop in thread %r\"\n            loop = self.new_event_loop()\n            self.set_event_loop(loop)\n            return loop\n\n\nclass AddThreadSelectorEventLoop(asyncio.AbstractEventLoop):\n    \"\"\"Wrap an event loop to add implementations of the ``add_reader`` method family.\n\n    Instances of this class start a second thread to run a selector.\n    This thread is completely hidden from the user; all callbacks are\n    run on the wrapped event loop's thread.\n\n    This class is used automatically by Tornado; applications should not need\n    to refer to it directly.\n\n    It is safe to wrap any event loop with this class, although it only makes sense\n    for event loops that do not implement the ``add_reader`` family of methods\n    themselves (i.e. ``WindowsProactorEventLoop``)\n\n    Closing the ``AddThreadSelectorEventLoop`` also closes the wrapped event loop.\n\n    \"\"\"\n\n    # This class is a __getattribute__-based proxy. All attributes other than those\n    # in this set are proxied through to the underlying loop.\n    MY_ATTRIBUTES = {\n        \"_consume_waker\",\n        \"_select_cond\",\n        \"_select_args\",\n        \"_closing_selector\",\n        \"_thread\",\n        \"_handle_event\",\n        \"_readers\",\n        \"_real_loop\",\n        \"_start_select\",\n        \"_run_select\",\n        \"_handle_select\",\n        \"_wake_selector\",\n        \"_waker_r\",\n        \"_waker_w\",\n        \"_writers\",\n        \"add_reader\",\n        \"add_writer\",\n        \"close\",\n        \"remove_reader\",\n        \"remove_writer\",\n    }\n\n    def __getattribute__(self, name: str) -> Any:\n        if name in AddThreadSelectorEventLoop.MY_ATTRIBUTES:\n            return super().__getattribute__(name)\n        return getattr(self._real_loop, name)\n\n    def __init__(self, real_loop: asyncio.AbstractEventLoop) -> None:\n        self._real_loop = real_loop\n\n        # Create a thread to run the select system call. We manage this thread\n        # manually so we can trigger a clean shutdown from an atexit hook. Note\n        # that due to the order of operations at shutdown, only daemon threads\n        # can be shut down in this way (non-daemon threads would require the\n        # introduction of a new hook: https://bugs.python.org/issue41962)\n        self._select_cond = threading.Condition()\n        self._select_args = (\n            None\n        )  # type: Optional[Tuple[List[_FileDescriptorLike], List[_FileDescriptorLike]]]\n        self._closing_selector = False\n        self._thread = threading.Thread(\n            name=\"Tornado selector\",\n            daemon=True,\n            target=self._run_select,\n        )\n        self._thread.start()\n        # Start the select loop once the loop is started.\n        self._real_loop.call_soon(self._start_select)\n\n        self._readers = {}  # type: Dict[_FileDescriptorLike, Callable]\n        self._writers = {}  # type: Dict[_FileDescriptorLike, Callable]\n\n        # Writing to _waker_w will wake up the selector thread, which\n        # watches for _waker_r to be readable.\n        self._waker_r, self._waker_w = socket.socketpair()\n        self._waker_r.setblocking(False)\n        self._waker_w.setblocking(False)\n        _selector_loops.add(self)\n        self.add_reader(self._waker_r, self._consume_waker)\n\n    def __del__(self) -> None:\n        # If the top-level application code uses asyncio interfaces to\n        # start and stop the event loop, no objects created in Tornado\n        # can get a clean shutdown notification. If we're just left to\n        # be GC'd, we must explicitly close our sockets to avoid\n        # logging warnings.\n        _selector_loops.discard(self)\n        self._waker_r.close()\n        self._waker_w.close()\n\n    def close(self) -> None:\n        with self._select_cond:\n            self._closing_selector = True\n            self._select_cond.notify()\n        self._wake_selector()\n        self._thread.join()\n        _selector_loops.discard(self)\n        self._waker_r.close()\n        self._waker_w.close()\n        self._real_loop.close()\n\n    def _wake_selector(self) -> None:\n        try:\n            self._waker_w.send(b\"a\")\n        except BlockingIOError:\n            pass\n\n    def _consume_waker(self) -> None:\n        try:\n            self._waker_r.recv(1024)\n        except BlockingIOError:\n            pass\n\n    def _start_select(self) -> None:\n        # Capture reader and writer sets here in the event loop\n        # thread to avoid any problems with concurrent\n        # modification while the select loop uses them.\n        with self._select_cond:\n            assert self._select_args is None\n            self._select_args = (list(self._readers.keys()), list(self._writers.keys()))\n            self._select_cond.notify()\n\n    def _run_select(self) -> None:\n        while True:\n            with self._select_cond:\n                while self._select_args is None and not self._closing_selector:\n                    self._select_cond.wait()\n                if self._closing_selector:\n                    return\n                assert self._select_args is not None\n                to_read, to_write = self._select_args\n                self._select_args = None\n\n            # We use the simpler interface of the select module instead of\n            # the more stateful interface in the selectors module because\n            # this class is only intended for use on windows, where\n            # select.select is the only option. The selector interface\n            # does not have well-documented thread-safety semantics that\n            # we can rely on so ensuring proper synchronization would be\n            # tricky.\n            try:\n                # On windows, selecting on a socket for write will not\n                # return the socket when there is an error (but selecting\n                # for reads works). Also select for errors when selecting\n                # for writes, and merge the results.\n                #\n                # This pattern is also used in\n                # https://github.com/python/cpython/blob/v3.8.0/Lib/selectors.py#L312-L317\n                rs, ws, xs = select.select(to_read, to_write, to_write)\n                ws = ws + xs\n            except OSError as e:\n                # After remove_reader or remove_writer is called, the file\n                # descriptor may subsequently be closed on the event loop\n                # thread. It's possible that this select thread hasn't\n                # gotten into the select system call by the time that\n                # happens in which case (at least on macOS), select may\n                # raise a \"bad file descriptor\" error. If we get that\n                # error, check and see if we're also being woken up by\n                # polling the waker alone. If we are, just return to the\n                # event loop and we'll get the updated set of file\n                # descriptors on the next iteration. Otherwise, raise the\n                # original error.\n                if e.errno == getattr(errno, \"WSAENOTSOCK\", errno.EBADF):\n                    rs, _, _ = select.select([self._waker_r.fileno()], [], [], 0)\n                    if rs:\n                        ws = []\n                    else:\n                        raise\n                else:\n                    raise\n\n            try:\n                self._real_loop.call_soon_threadsafe(self._handle_select, rs, ws)\n            except RuntimeError:\n                # \"Event loop is closed\". Swallow the exception for\n                # consistency with PollIOLoop (and logical consistency\n                # with the fact that we can't guarantee that an\n                # add_callback that completes without error will\n                # eventually execute).\n                pass\n            except AttributeError:\n                # ProactorEventLoop may raise this instead of RuntimeError\n                # if call_soon_threadsafe races with a call to close().\n                # Swallow it too for consistency.\n                pass\n\n    def _handle_select(\n        self, rs: List[\"_FileDescriptorLike\"], ws: List[\"_FileDescriptorLike\"]\n    ) -> None:\n        for r in rs:\n            self._handle_event(r, self._readers)\n        for w in ws:\n            self._handle_event(w, self._writers)\n        self._start_select()\n\n    def _handle_event(\n        self,\n        fd: \"_FileDescriptorLike\",\n        cb_map: Dict[\"_FileDescriptorLike\", Callable],\n    ) -> None:\n        try:\n            callback = cb_map[fd]\n        except KeyError:\n            return\n        callback()\n\n    def add_reader(\n        self, fd: \"_FileDescriptorLike\", callback: Callable[..., None], *args: Any\n    ) -> None:\n        self._readers[fd] = functools.partial(callback, *args)\n        self._wake_selector()\n\n    def add_writer(\n        self, fd: \"_FileDescriptorLike\", callback: Callable[..., None], *args: Any\n    ) -> None:\n        self._writers[fd] = functools.partial(callback, *args)\n        self._wake_selector()\n\n    def remove_reader(self, fd: \"_FileDescriptorLike\") -> None:\n        del self._readers[fd]\n        self._wake_selector()\n\n    def remove_writer(self, fd: \"_FileDescriptorLike\") -> None:\n        del self._writers[fd]\n        self._wake_selector()\n",
            "new_file_content": "\"\"\"Bridges between the `asyncio` module and Tornado IOLoop.\n\n.. versionadded:: 3.2\n\nThis module integrates Tornado with the ``asyncio`` module introduced\nin Python 3.4. This makes it possible to combine the two libraries on\nthe same event loop.\n\n.. deprecated:: 5.0\n\n   While the code in this module is still used, it is now enabled\n   automatically when `asyncio` is available, so applications should\n   no longer need to refer to this module directly.\n\n.. note::\n\n   Tornado is designed to use a selector-based event loop. On Windows,\n   where a proactor-based event loop has been the default since Python 3.8,\n   a selector event loop is emulated by running ``select`` on a separate thread.\n   Configuring ``asyncio`` to use a selector event loop may improve performance\n   of Tornado (but may reduce performance of other ``asyncio``-based libraries\n   in the same process).\n\"\"\"\n\nimport asyncio\nimport atexit\nimport concurrent.futures\nimport errno\nimport functools\nimport select\nimport socket\nimport sys\nimport threading\nimport typing\nimport warnings\nfrom tornado.gen import convert_yielded\nfrom tornado.ioloop import IOLoop, _Selectable\n\nfrom typing import Any, TypeVar, Awaitable, Callable, Union, Optional, List, Tuple, Dict\n\nif typing.TYPE_CHECKING:\n    from typing import Set  # noqa: F401\n    from typing_extensions import Protocol\n\n    class _HasFileno(Protocol):\n        def fileno(self) -> int:\n            pass\n\n    _FileDescriptorLike = Union[int, _HasFileno]\n\n_T = TypeVar(\"_T\")\n\n\n# Collection of selector thread event loops to shut down on exit.\n_selector_loops = set()  # type: Set[AddThreadSelectorEventLoop]\n\n\ndef _atexit_callback() -> None:\n    for loop in _selector_loops:\n        with loop._select_cond:\n            loop._closing_selector = True\n            loop._select_cond.notify()\n        try:\n            loop._waker_w.send(b\"a\")\n        except BlockingIOError:\n            pass\n        # If we don't join our (daemon) thread here, we may get a deadlock\n        # during interpreter shutdown. I don't really understand why. This\n        # deadlock happens every time in CI (both travis and appveyor) but\n        # I've never been able to reproduce locally.\n        loop._thread.join()\n    _selector_loops.clear()\n\n\natexit.register(_atexit_callback)\n\nif sys.version_info >= (3, 10):\n\n    def _get_event_loop() -> asyncio.AbstractEventLoop:\n        try:\n            return asyncio.get_running_loop()\n        except RuntimeError:\n            pass\n\n        return asyncio.get_event_loop_policy().get_event_loop()\n\n\nelse:\n    from asyncio import get_event_loop as _get_event_loop\n\n\nclass BaseAsyncIOLoop(IOLoop):\n    def initialize(  # type: ignore\n        self, asyncio_loop: asyncio.AbstractEventLoop, **kwargs: Any\n    ) -> None:\n        # asyncio_loop is always the real underlying IOLoop. This is used in\n        # ioloop.py to maintain the asyncio-to-ioloop mappings.\n        self.asyncio_loop = asyncio_loop\n        # selector_loop is an event loop that implements the add_reader family of\n        # methods. Usually the same as asyncio_loop but differs on platforms such\n        # as windows where the default event loop does not implement these methods.\n        self.selector_loop = asyncio_loop\n        if hasattr(asyncio, \"ProactorEventLoop\") and isinstance(\n            asyncio_loop, asyncio.ProactorEventLoop  # type: ignore\n        ):\n            # Ignore this line for mypy because the abstract method checker\n            # doesn't understand dynamic proxies.\n            self.selector_loop = AddThreadSelectorEventLoop(asyncio_loop)  # type: ignore\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}  # type: Dict[int, Tuple[Union[int, _Selectable], Callable]]\n        # Set of fds listening for reads/writes\n        self.readers = set()  # type: Set[int]\n        self.writers = set()  # type: Set[int]\n        self.closing = False\n        # If an asyncio loop was closed through an asyncio interface\n        # instead of IOLoop.close(), we'd never hear about it and may\n        # have left a dangling reference in our map. In case an\n        # application (or, more likely, a test suite) creates and\n        # destroys a lot of event loops in this way, check here to\n        # ensure that we don't have a lot of dead loops building up in\n        # the map.\n        #\n        # TODO(bdarnell): consider making self.asyncio_loop a weakref\n        # for AsyncIOMainLoop and make _ioloop_for_asyncio a\n        # WeakKeyDictionary.\n        for loop in IOLoop._ioloop_for_asyncio.copy():\n            if loop.is_closed():\n                try:\n                    del IOLoop._ioloop_for_asyncio[loop]\n                except KeyError:\n                    pass\n\n        # Make sure we don't already have an IOLoop for this asyncio loop\n        if asyncio_loop in IOLoop._ioloop_for_asyncio:\n            existing_loop = IOLoop._ioloop_for_asyncio[asyncio_loop]\n            raise RuntimeError(\n                f\"IOLoop {existing_loop} already associated with asyncio loop {asyncio_loop}\"\n            )\n\n        IOLoop._ioloop_for_asyncio[asyncio_loop] = self\n\n        self._thread_identity = 0\n\n        super().initialize(**kwargs)\n\n        def assign_thread_identity() -> None:\n            self._thread_identity = threading.get_ident()\n\n        self.add_callback(assign_thread_identity)\n\n    def close(self, all_fds: bool = False) -> None:\n        self.closing = True\n        for fd in list(self.handlers):\n            fileobj, handler_func = self.handlers[fd]\n            self.remove_handler(fd)\n            if all_fds:\n                self.close_fd(fileobj)\n        # Remove the mapping before closing the asyncio loop. If this\n        # happened in the other order, we could race against another\n        # initialize() call which would see the closed asyncio loop,\n        # assume it was closed from the asyncio side, and do this\n        # cleanup for us, leading to a KeyError.\n        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]\n        if self.selector_loop is not self.asyncio_loop:\n            self.selector_loop.close()\n        self.asyncio_loop.close()\n\n    def add_handler(\n        self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int\n    ) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd in self.handlers:\n            raise ValueError(\"fd %s added twice\" % fd)\n        self.handlers[fd] = (fileobj, handler)\n        if events & IOLoop.READ:\n            self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.selector_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.selector_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd: Union[int, _Selectable]) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.selector_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.selector_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd: int, events: int) -> None:\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self) -> None:\n        try:\n            old_loop = _get_event_loop()\n        except (RuntimeError, AssertionError):\n            old_loop = None  # type: ignore\n        try:\n            asyncio.set_event_loop(self.asyncio_loop)\n            self.asyncio_loop.run_forever()\n        finally:\n            asyncio.set_event_loop(old_loop)\n\n    def stop(self) -> None:\n        self.asyncio_loop.stop()\n\n    def call_at(\n        self, when: float, callback: Callable, *args: Any, **kwargs: Any\n    ) -> object:\n        # asyncio.call_at supports *args but not **kwargs, so bind them here.\n        # We do not synchronize self.time and asyncio_loop.time, so\n        # convert from absolute to relative.\n        return self.asyncio_loop.call_later(\n            max(0, when - self.time()),\n            self._run_callback,\n            functools.partial(callback, *args, **kwargs),\n        )\n\n    def remove_timeout(self, timeout: object) -> None:\n        timeout.cancel()  # type: ignore\n\n    def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:\n        if threading.get_ident() == self._thread_identity:\n            call_soon = self.asyncio_loop.call_soon\n        else:\n            call_soon = self.asyncio_loop.call_soon_threadsafe\n        try:\n            call_soon(self._run_callback, functools.partial(callback, *args, **kwargs))\n        except RuntimeError:\n            # \"Event loop is closed\". Swallow the exception for\n            # consistency with PollIOLoop (and logical consistency\n            # with the fact that we can't guarantee that an\n            # add_callback that completes without error will\n            # eventually execute).\n            pass\n        except AttributeError:\n            # ProactorEventLoop may raise this instead of RuntimeError\n            # if call_soon_threadsafe races with a call to close().\n            # Swallow it too for consistency.\n            pass\n\n    def add_callback_from_signal(\n        self, callback: Callable, *args: Any, **kwargs: Any\n    ) -> None:\n        try:\n            self.asyncio_loop.call_soon_threadsafe(\n                self._run_callback, functools.partial(callback, *args, **kwargs)\n            )\n        except RuntimeError:\n            pass\n\n    def run_in_executor(\n        self,\n        executor: Optional[concurrent.futures.Executor],\n        func: Callable[..., _T],\n        *args: Any\n    ) -> Awaitable[_T]:\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor: concurrent.futures.Executor) -> None:\n        return self.asyncio_loop.set_default_executor(executor)\n\n\nclass AsyncIOMainLoop(BaseAsyncIOLoop):\n    \"\"\"``AsyncIOMainLoop`` creates an `.IOLoop` that corresponds to the\n    current ``asyncio`` event loop (i.e. the one returned by\n    ``asyncio.get_event_loop()``).\n\n    .. deprecated:: 5.0\n\n       Now used automatically when appropriate; it is no longer necessary\n       to refer to this class directly.\n\n    .. versionchanged:: 5.0\n\n       Closing an `AsyncIOMainLoop` now closes the underlying asyncio loop.\n    \"\"\"\n\n    def initialize(self, **kwargs: Any) -> None:  # type: ignore\n        super().initialize(asyncio.get_event_loop(), **kwargs)\n\n    def make_current(self) -> None:\n        # AsyncIOMainLoop already refers to the current asyncio loop so\n        # nothing to do here.\n        pass\n\n\nclass AsyncIOLoop(BaseAsyncIOLoop):\n    \"\"\"``AsyncIOLoop`` is an `.IOLoop` that runs on an ``asyncio`` event loop.\n    This class follows the usual Tornado semantics for creating new\n    ``IOLoops``; these loops are not necessarily related to the\n    ``asyncio`` default event loop.\n\n    Each ``AsyncIOLoop`` creates a new ``asyncio.EventLoop``; this object\n    can be accessed with the ``asyncio_loop`` attribute.\n\n    .. versionchanged:: 6.2\n\n       Support explicit ``asyncio_loop`` argument\n       for specifying the asyncio loop to attach to,\n       rather than always creating a new one with the default policy.\n\n    .. versionchanged:: 5.0\n\n       When an ``AsyncIOLoop`` becomes the current `.IOLoop`, it also sets\n       the current `asyncio` event loop.\n\n    .. deprecated:: 5.0\n\n       Now used automatically when appropriate; it is no longer necessary\n       to refer to this class directly.\n    \"\"\"\n\n    def initialize(self, **kwargs: Any) -> None:  # type: ignore\n        self.is_current = False\n        loop = None\n        if \"asyncio_loop\" not in kwargs:\n            kwargs[\"asyncio_loop\"] = loop = asyncio.new_event_loop()\n        try:\n            super().initialize(**kwargs)\n        except Exception:\n            # If initialize() does not succeed (taking ownership of the loop),\n            # we have to close it.\n            if loop is not None:\n                loop.close()\n            raise\n\n    def close(self, all_fds: bool = False) -> None:\n        if self.is_current:\n            self._clear_current()\n        super().close(all_fds=all_fds)\n\n    def make_current(self) -> None:\n        warnings.warn(\n            \"make_current is deprecated; start the event loop first\",\n            DeprecationWarning,\n        )\n        if not self.is_current:\n            try:\n                self.old_asyncio = asyncio.get_event_loop()\n            except (RuntimeError, AssertionError):\n                self.old_asyncio = None  # type: ignore\n            self.is_current = True\n        asyncio.set_event_loop(self.asyncio_loop)\n\n    def _clear_current_hook(self) -> None:\n        if self.is_current:\n            asyncio.set_event_loop(self.old_asyncio)\n            self.is_current = False\n\n\ndef to_tornado_future(asyncio_future: asyncio.Future) -> asyncio.Future:\n    \"\"\"Convert an `asyncio.Future` to a `tornado.concurrent.Future`.\n\n    .. versionadded:: 4.1\n\n    .. deprecated:: 5.0\n       Tornado ``Futures`` have been merged with `asyncio.Future`,\n       so this method is now a no-op.\n    \"\"\"\n    return asyncio_future\n\n\ndef to_asyncio_future(tornado_future: asyncio.Future) -> asyncio.Future:\n    \"\"\"Convert a Tornado yieldable object to an `asyncio.Future`.\n\n    .. versionadded:: 4.1\n\n    .. versionchanged:: 4.3\n       Now accepts any yieldable object, not just\n       `tornado.concurrent.Future`.\n\n    .. deprecated:: 5.0\n       Tornado ``Futures`` have been merged with `asyncio.Future`,\n       so this method is now equivalent to `tornado.gen.convert_yielded`.\n    \"\"\"\n    return convert_yielded(tornado_future)\n\n\nif sys.platform == \"win32\" and hasattr(asyncio, \"WindowsSelectorEventLoopPolicy\"):\n    # \"Any thread\" and \"selector\" should be orthogonal, but there's not a clean\n    # interface for composing policies so pick the right base.\n    _BasePolicy = asyncio.WindowsSelectorEventLoopPolicy  # type: ignore\nelse:\n    _BasePolicy = asyncio.DefaultEventLoopPolicy\n\n\nclass AnyThreadEventLoopPolicy(_BasePolicy):  # type: ignore\n    \"\"\"Event loop policy that allows loop creation on any thread.\n\n    The default `asyncio` event loop policy only automatically creates\n    event loops in the main threads. Other threads must create event\n    loops explicitly or `asyncio.get_event_loop` (and therefore\n    `.IOLoop.current`) will fail. Installing this policy allows event\n    loops to be created automatically on any thread, matching the\n    behavior of Tornado versions prior to 5.0 (or 5.0 on Python 2).\n\n    Usage::\n\n        asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())\n\n    .. versionadded:: 5.0\n\n    .. deprecated:: 6.2\n\n        ``AnyThreadEventLoopPolicy`` affects the implicit creation\n        of an event loop, which is deprecated in Python 3.10 and\n        will be removed in a future version of Python. At that time\n        ``AnyThreadEventLoopPolicy`` will no longer be useful.\n        If you are relying on it, use `asyncio.new_event_loop`\n        or `asyncio.run` explicitly in any non-main threads that\n        need event loops.\n    \"\"\"\n\n    def __init__(self) -> None:\n        super().__init__()\n        warnings.warn(\n            \"AnyThreadEventLoopPolicy is deprecated, use asyncio.run \"\n            \"or asyncio.new_event_loop instead\",\n            DeprecationWarning,\n        )\n\n    def get_event_loop(self) -> asyncio.AbstractEventLoop:\n        try:\n            return super().get_event_loop()\n        except (RuntimeError, AssertionError):\n            # This was an AssertionError in Python 3.4.2 (which ships with Debian Jessie)\n            # and changed to a RuntimeError in 3.4.3.\n            # \"There is no current event loop in thread %r\"\n            loop = self.new_event_loop()\n            self.set_event_loop(loop)\n            return loop\n\n\nclass AddThreadSelectorEventLoop(asyncio.AbstractEventLoop):\n    \"\"\"Wrap an event loop to add implementations of the ``add_reader`` method family.\n\n    Instances of this class start a second thread to run a selector.\n    This thread is completely hidden from the user; all callbacks are\n    run on the wrapped event loop's thread.\n\n    This class is used automatically by Tornado; applications should not need\n    to refer to it directly.\n\n    It is safe to wrap any event loop with this class, although it only makes sense\n    for event loops that do not implement the ``add_reader`` family of methods\n    themselves (i.e. ``WindowsProactorEventLoop``)\n\n    Closing the ``AddThreadSelectorEventLoop`` also closes the wrapped event loop.\n\n    \"\"\"\n\n    # This class is a __getattribute__-based proxy. All attributes other than those\n    # in this set are proxied through to the underlying loop.\n    MY_ATTRIBUTES = {\n        \"_consume_waker\",\n        \"_select_cond\",\n        \"_select_args\",\n        \"_closing_selector\",\n        \"_thread\",\n        \"_handle_event\",\n        \"_readers\",\n        \"_real_loop\",\n        \"_start_select\",\n        \"_run_select\",\n        \"_handle_select\",\n        \"_wake_selector\",\n        \"_waker_r\",\n        \"_waker_w\",\n        \"_writers\",\n        \"add_reader\",\n        \"add_writer\",\n        \"close\",\n        \"remove_reader\",\n        \"remove_writer\",\n    }\n\n    def __getattribute__(self, name: str) -> Any:\n        if name in AddThreadSelectorEventLoop.MY_ATTRIBUTES:\n            return super().__getattribute__(name)\n        return getattr(self._real_loop, name)\n\n    def __init__(self, real_loop: asyncio.AbstractEventLoop) -> None:\n        self._real_loop = real_loop\n\n        # Create a thread to run the select system call. We manage this thread\n        # manually so we can trigger a clean shutdown from an atexit hook. Note\n        # that due to the order of operations at shutdown, only daemon threads\n        # can be shut down in this way (non-daemon threads would require the\n        # introduction of a new hook: https://bugs.python.org/issue41962)\n        self._select_cond = threading.Condition()\n        self._select_args = (\n            None\n        )  # type: Optional[Tuple[List[_FileDescriptorLike], List[_FileDescriptorLike]]]\n        self._closing_selector = False\n        self._thread = threading.Thread(\n            name=\"Tornado selector\",\n            daemon=True,\n            target=self._run_select,\n        )\n        self._thread.start()\n        # Start the select loop once the loop is started.\n        self._real_loop.call_soon(self._start_select)\n\n        self._readers = {}  # type: Dict[_FileDescriptorLike, Callable]\n        self._writers = {}  # type: Dict[_FileDescriptorLike, Callable]\n\n        # Writing to _waker_w will wake up the selector thread, which\n        # watches for _waker_r to be readable.\n        self._waker_r, self._waker_w = socket.socketpair()\n        self._waker_r.setblocking(False)\n        self._waker_w.setblocking(False)\n        _selector_loops.add(self)\n        self.add_reader(self._waker_r, self._consume_waker)\n\n    def __del__(self) -> None:\n        # If the top-level application code uses asyncio interfaces to\n        # start and stop the event loop, no objects created in Tornado\n        # can get a clean shutdown notification. If we're just left to\n        # be GC'd, we must explicitly close our sockets to avoid\n        # logging warnings.\n        _selector_loops.discard(self)\n        self._waker_r.close()\n        self._waker_w.close()\n\n    def close(self) -> None:\n        with self._select_cond:\n            self._closing_selector = True\n            self._select_cond.notify()\n        self._wake_selector()\n        self._thread.join()\n        _selector_loops.discard(self)\n        self._waker_r.close()\n        self._waker_w.close()\n        self._real_loop.close()\n\n    def _wake_selector(self) -> None:\n        try:\n            self._waker_w.send(b\"a\")\n        except BlockingIOError:\n            pass\n\n    def _consume_waker(self) -> None:\n        try:\n            self._waker_r.recv(1024)\n        except BlockingIOError:\n            pass\n\n    def _start_select(self) -> None:\n        # Capture reader and writer sets here in the event loop\n        # thread to avoid any problems with concurrent\n        # modification while the select loop uses them.\n        with self._select_cond:\n            assert self._select_args is None\n            self._select_args = (list(self._readers.keys()), list(self._writers.keys()))\n            self._select_cond.notify()\n\n    def _run_select(self) -> None:\n        while True:\n            with self._select_cond:\n                while self._select_args is None and not self._closing_selector:\n                    self._select_cond.wait()\n                if self._closing_selector:\n                    return\n                assert self._select_args is not None\n                to_read, to_write = self._select_args\n                self._select_args = None\n\n            # We use the simpler interface of the select module instead of\n            # the more stateful interface in the selectors module because\n            # this class is only intended for use on windows, where\n            # select.select is the only option. The selector interface\n            # does not have well-documented thread-safety semantics that\n            # we can rely on so ensuring proper synchronization would be\n            # tricky.\n            try:\n                # On windows, selecting on a socket for write will not\n                # return the socket when there is an error (but selecting\n                # for reads works). Also select for errors when selecting\n                # for writes, and merge the results.\n                #\n                # This pattern is also used in\n                # https://github.com/python/cpython/blob/v3.8.0/Lib/selectors.py#L312-L317\n                rs, ws, xs = select.select(to_read, to_write, to_write)\n                ws = ws + xs\n            except OSError as e:\n                # After remove_reader or remove_writer is called, the file\n                # descriptor may subsequently be closed on the event loop\n                # thread. It's possible that this select thread hasn't\n                # gotten into the select system call by the time that\n                # happens in which case (at least on macOS), select may\n                # raise a \"bad file descriptor\" error. If we get that\n                # error, check and see if we're also being woken up by\n                # polling the waker alone. If we are, just return to the\n                # event loop and we'll get the updated set of file\n                # descriptors on the next iteration. Otherwise, raise the\n                # original error.\n                if e.errno == getattr(errno, \"WSAENOTSOCK\", errno.EBADF):\n                    rs, _, _ = select.select([self._waker_r.fileno()], [], [], 0)\n                    if rs:\n                        ws = []\n                    else:\n                        raise\n                else:\n                    raise\n\n            try:\n                self._real_loop.call_soon_threadsafe(self._handle_select, rs, ws)\n            except RuntimeError:\n                # \"Event loop is closed\". Swallow the exception for\n                # consistency with PollIOLoop (and logical consistency\n                # with the fact that we can't guarantee that an\n                # add_callback that completes without error will\n                # eventually execute).\n                pass\n            except AttributeError:\n                # ProactorEventLoop may raise this instead of RuntimeError\n                # if call_soon_threadsafe races with a call to close().\n                # Swallow it too for consistency.\n                pass\n\n    def _handle_select(\n        self, rs: List[\"_FileDescriptorLike\"], ws: List[\"_FileDescriptorLike\"]\n    ) -> None:\n        for r in rs:\n            self._handle_event(r, self._readers)\n        for w in ws:\n            self._handle_event(w, self._writers)\n        self._start_select()\n\n    def _handle_event(\n        self,\n        fd: \"_FileDescriptorLike\",\n        cb_map: Dict[\"_FileDescriptorLike\", Callable],\n    ) -> None:\n        try:\n            callback = cb_map[fd]\n        except KeyError:\n            return\n        callback()\n\n    def add_reader(\n        self, fd: \"_FileDescriptorLike\", callback: Callable[..., None], *args: Any\n    ) -> None:\n        self._readers[fd] = functools.partial(callback, *args)\n        self._wake_selector()\n\n    def add_writer(\n        self, fd: \"_FileDescriptorLike\", callback: Callable[..., None], *args: Any\n    ) -> None:\n        self._writers[fd] = functools.partial(callback, *args)\n        self._wake_selector()\n\n    def remove_reader(self, fd: \"_FileDescriptorLike\") -> None:\n        del self._readers[fd]\n        self._wake_selector()\n\n    def remove_writer(self, fd: \"_FileDescriptorLike\") -> None:\n        del self._writers[fd]\n        self._wake_selector()\n",
            "header": {
                "file": {
                    "path": "tornado/platform/asyncio.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "3b367815",
                "new_commit_hash": "c35955cc",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/platform/asyncio.py"
            },
            "plus_file": {
                "path": "b/tornado/platform/asyncio.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 129,
                            "length": 6
                        },
                        "new_range": {
                            "start": 129,
                            "length": 14
                        },
                        "section": "class BaseAsyncIOLoop(IOLoop):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "                    del IOLoop._ioloop_for_asyncio[loop]",
                                "type": "context"
                            },
                            {
                                "content": "                except KeyError:",
                                "type": "context"
                            },
                            {
                                "content": "                    pass",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        # Make sure we don't already have an IOLoop for this asyncio loop",
                                "type": "added"
                            },
                            {
                                "content": "        if asyncio_loop in IOLoop._ioloop_for_asyncio:",
                                "type": "added"
                            },
                            {
                                "content": "            existing_loop = IOLoop._ioloop_for_asyncio[asyncio_loop]",
                                "type": "added"
                            },
                            {
                                "content": "            raise RuntimeError(",
                                "type": "added"
                            },
                            {
                                "content": "                f\"IOLoop {existing_loop} already associated with asyncio loop {asyncio_loop}\"",
                                "type": "added"
                            },
                            {
                                "content": "            )",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        IOLoop._ioloop_for_asyncio[asyncio_loop] = self",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "        self._thread_identity = 0",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/platform/asyncio.py",
                            "type": "method",
                            "name": "BaseAsyncIOLoop.initialize",
                            "content": "class BaseAsyncIOLoop(IOLoop):\n    def initialize(  # type: ignore\n        self, asyncio_loop: asyncio.AbstractEventLoop, **kwargs: Any\n    ) -> None:\n        # asyncio_loop is always the real underlying IOLoop. This is used in\n        # ioloop.py to maintain the asyncio-to-ioloop mappings.\n        self.asyncio_loop = asyncio_loop\n        # selector_loop is an event loop that implements the add_reader family of\n        # methods. Usually the same as asyncio_loop but differs on platforms such\n        # as windows where the default event loop does not implement these methods.\n        self.selector_loop = asyncio_loop\n        if hasattr(asyncio, \"ProactorEventLoop\") and isinstance(\n            asyncio_loop, asyncio.ProactorEventLoop  # type: ignore\n        ):\n            # Ignore this line for mypy because the abstract method checker\n            # doesn't understand dynamic proxies.\n            self.selector_loop = AddThreadSelectorEventLoop(asyncio_loop)  # type: ignore\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}  # type: Dict[int, Tuple[Union[int, _Selectable], Callable]]\n        # Set of fds listening for reads/writes\n        self.readers = set()  # type: Set[int]\n        self.writers = set()  # type: Set[int]\n        self.closing = False\n        # If an asyncio loop was closed through an asyncio interface\n        # instead of IOLoop.close(), we'd never hear about it and may\n        # have left a dangling reference in our map. In case an\n        # application (or, more likely, a test suite) creates and\n        # destroys a lot of event loops in this way, check here to\n        # ensure that we don't have a lot of dead loops building up in\n        # the map.\n        #\n        # TODO(bdarnell): consider making self.asyncio_loop a weakref\n        # for AsyncIOMainLoop and make _ioloop_for_asyncio a\n        # WeakKeyDictionary.\n        for loop in IOLoop._ioloop_for_asyncio.copy():\n            if loop.is_closed():\n                try:\n                    del IOLoop._ioloop_for_asyncio[loop]\n                except KeyError:\n                    pass\n\n        # Make sure we don't already have an IOLoop for this asyncio loop\n        if asyncio_loop in IOLoop._ioloop_for_asyncio:\n            existing_loop = IOLoop._ioloop_for_asyncio[asyncio_loop]\n            raise RuntimeError(\n                f\"IOLoop {existing_loop} already associated with asyncio loop {asyncio_loop}\"\n            )\n\n        IOLoop._ioloop_for_asyncio[asyncio_loop] = self\n\n        self._thread_identity = 0\n\n        super().initialize(**kwargs)\n\n        def assign_thread_identity() -> None:\n            self._thread_identity = threading.get_ident()\n\n        self.add_callback(assign_thread_identity)\n\n    def close(self, all_fds: bool = False) -> None:\n        self.closing = True\n        for fd in list(self.handlers):\n            fileobj, handler_func = self.handlers[fd]\n            self.remove_handler(fd)\n            if all_fds:\n                self.close_fd(fileobj)\n        # Remove the mapping before closing the asyncio loop. If this\n        # happened in the other order, we could race against another\n        # initialize() call which would see the closed asyncio loop,\n        # assume it was closed from the asyncio side, and do this\n        # cleanup for us, leading to a KeyError.\n        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]\n        if self.selector_loop is not self.asyncio_loop:\n            self.selector_loop.close()\n        self.asyncio_loop.close()\n\n    def add_handler(\n        self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int\n    ) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd in self.handlers:\n            raise ValueError(\"fd %s added twice\" % fd)\n        self.handlers[fd] = (fileobj, handler)\n        if events & IOLoop.READ:\n            self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.selector_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.selector_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd: Union[int, _Selectable]) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.selector_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.selector_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd: int, events: int) -> None:\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self) -> None:\n        try:\n            old_loop = _get_event_loop()\n        except (RuntimeError, AssertionError):\n            old_loop = None  # type: ignore\n        try:\n            asyncio.set_event_loop(self.asyncio_loop)\n            self.asyncio_loop.run_forever()\n        finally:\n            asyncio.set_event_loop(old_loop)\n\n    def stop(self) -> None:\n        self.asyncio_loop.stop()\n\n    def call_at(\n        self, when: float, callback: Callable, *args: Any, **kwargs: Any\n    ) -> object:\n        # asyncio.call_at supports *args but not **kwargs, so bind them here.\n        # We do not synchronize self.time and asyncio_loop.time, so\n        # convert from absolute to relative.\n        return self.asyncio_loop.call_later(\n            max(0, when - self.time()),\n            self._run_callback,\n            functools.partial(callback, *args, **kwargs),\n        )\n\n    def remove_timeout(self, timeout: object) -> None:\n        timeout.cancel()  # type: ignore\n\n    def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:\n        if threading.get_ident() == self._thread_identity:\n            call_soon = self.asyncio_loop.call_soon\n        else:\n            call_soon = self.asyncio_loop.call_soon_threadsafe\n        try:\n            call_soon(self._run_callback, functools.partial(callback, *args, **kwargs))\n        except RuntimeError:\n            # \"Event loop is closed\". Swallow the exception for\n            # consistency with PollIOLoop (and logical consistency\n            # with the fact that we can't guarantee that an\n            # add_callback that completes without error will\n            # eventually execute).\n            pass\n        except AttributeError:\n            # ProactorEventLoop may raise this instead of RuntimeError\n            # if call_soon_threadsafe races with a call to close().\n            # Swallow it too for consistency.\n            pass\n\n    def add_callback_from_signal(\n        self, callback: Callable, *args: Any, **kwargs: Any\n    ) -> None:\n        try:\n            self.asyncio_loop.call_soon_threadsafe(\n                self._run_callback, functools.partial(callback, *args, **kwargs)\n            )\n        except RuntimeError:\n            pass\n\n    def run_in_executor(\n        self,\n        executor: Optional[concurrent.futures.Executor],\n        func: Callable[..., _T],\n        *args: Any\n    ) -> Awaitable[_T]:\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor: concurrent.futures.Executor) -> None:\n        return self.asyncio_loop.set_default_executor(executor)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 93,
                            "end_lineno": 149,
                            "parent": {
                                "file_name": "tornado/platform/asyncio.py",
                                "type": "class",
                                "name": "BaseAsyncIOLoop",
                                "content": "class BaseAsyncIOLoop(IOLoop):\n    def initialize(  # type: ignore\n        self, asyncio_loop: asyncio.AbstractEventLoop, **kwargs: Any\n    ) -> None:\n        # asyncio_loop is always the real underlying IOLoop. This is used in\n        # ioloop.py to maintain the asyncio-to-ioloop mappings.\n        self.asyncio_loop = asyncio_loop\n        # selector_loop is an event loop that implements the add_reader family of\n        # methods. Usually the same as asyncio_loop but differs on platforms such\n        # as windows where the default event loop does not implement these methods.\n        self.selector_loop = asyncio_loop\n        if hasattr(asyncio, \"ProactorEventLoop\") and isinstance(\n            asyncio_loop, asyncio.ProactorEventLoop  # type: ignore\n        ):\n            # Ignore this line for mypy because the abstract method checker\n            # doesn't understand dynamic proxies.\n            self.selector_loop = AddThreadSelectorEventLoop(asyncio_loop)  # type: ignore\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}  # type: Dict[int, Tuple[Union[int, _Selectable], Callable]]\n        # Set of fds listening for reads/writes\n        self.readers = set()  # type: Set[int]\n        self.writers = set()  # type: Set[int]\n        self.closing = False\n        # If an asyncio loop was closed through an asyncio interface\n        # instead of IOLoop.close(), we'd never hear about it and may\n        # have left a dangling reference in our map. In case an\n        # application (or, more likely, a test suite) creates and\n        # destroys a lot of event loops in this way, check here to\n        # ensure that we don't have a lot of dead loops building up in\n        # the map.\n        #\n        # TODO(bdarnell): consider making self.asyncio_loop a weakref\n        # for AsyncIOMainLoop and make _ioloop_for_asyncio a\n        # WeakKeyDictionary.\n        for loop in IOLoop._ioloop_for_asyncio.copy():\n            if loop.is_closed():\n                try:\n                    del IOLoop._ioloop_for_asyncio[loop]\n                except KeyError:\n                    pass\n\n        # Make sure we don't already have an IOLoop for this asyncio loop\n        if asyncio_loop in IOLoop._ioloop_for_asyncio:\n            existing_loop = IOLoop._ioloop_for_asyncio[asyncio_loop]\n            raise RuntimeError(\n                f\"IOLoop {existing_loop} already associated with asyncio loop {asyncio_loop}\"\n            )\n\n        IOLoop._ioloop_for_asyncio[asyncio_loop] = self\n\n        self._thread_identity = 0\n\n        super().initialize(**kwargs)\n\n        def assign_thread_identity() -> None:\n            self._thread_identity = threading.get_ident()\n\n        self.add_callback(assign_thread_identity)\n\n    def close(self, all_fds: bool = False) -> None:\n        self.closing = True\n        for fd in list(self.handlers):\n            fileobj, handler_func = self.handlers[fd]\n            self.remove_handler(fd)\n            if all_fds:\n                self.close_fd(fileobj)\n        # Remove the mapping before closing the asyncio loop. If this\n        # happened in the other order, we could race against another\n        # initialize() call which would see the closed asyncio loop,\n        # assume it was closed from the asyncio side, and do this\n        # cleanup for us, leading to a KeyError.\n        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]\n        if self.selector_loop is not self.asyncio_loop:\n            self.selector_loop.close()\n        self.asyncio_loop.close()\n\n    def add_handler(\n        self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int\n    ) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd in self.handlers:\n            raise ValueError(\"fd %s added twice\" % fd)\n        self.handlers[fd] = (fileobj, handler)\n        if events & IOLoop.READ:\n            self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.selector_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.selector_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd: Union[int, _Selectable]) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.selector_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.selector_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd: int, events: int) -> None:\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self) -> None:\n        try:\n            old_loop = _get_event_loop()\n        except (RuntimeError, AssertionError):\n            old_loop = None  # type: ignore\n        try:\n            asyncio.set_event_loop(self.asyncio_loop)\n            self.asyncio_loop.run_forever()\n        finally:\n            asyncio.set_event_loop(old_loop)\n\n    def stop(self) -> None:\n        self.asyncio_loop.stop()\n\n    def call_at(\n        self, when: float, callback: Callable, *args: Any, **kwargs: Any\n    ) -> object:\n        # asyncio.call_at supports *args but not **kwargs, so bind them here.\n        # We do not synchronize self.time and asyncio_loop.time, so\n        # convert from absolute to relative.\n        return self.asyncio_loop.call_later(\n            max(0, when - self.time()),\n            self._run_callback,\n            functools.partial(callback, *args, **kwargs),\n        )\n\n    def remove_timeout(self, timeout: object) -> None:\n        timeout.cancel()  # type: ignore\n\n    def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:\n        if threading.get_ident() == self._thread_identity:\n            call_soon = self.asyncio_loop.call_soon\n        else:\n            call_soon = self.asyncio_loop.call_soon_threadsafe\n        try:\n            call_soon(self._run_callback, functools.partial(callback, *args, **kwargs))\n        except RuntimeError:\n            # \"Event loop is closed\". Swallow the exception for\n            # consistency with PollIOLoop (and logical consistency\n            # with the fact that we can't guarantee that an\n            # add_callback that completes without error will\n            # eventually execute).\n            pass\n        except AttributeError:\n            # ProactorEventLoop may raise this instead of RuntimeError\n            # if call_soon_threadsafe races with a call to close().\n            # Swallow it too for consistency.\n            pass\n\n    def add_callback_from_signal(\n        self, callback: Callable, *args: Any, **kwargs: Any\n    ) -> None:\n        try:\n            self.asyncio_loop.call_soon_threadsafe(\n                self._run_callback, functools.partial(callback, *args, **kwargs)\n            )\n        except RuntimeError:\n            pass\n\n    def run_in_executor(\n        self,\n        executor: Optional[concurrent.futures.Executor],\n        func: Callable[..., _T],\n        *args: Any\n    ) -> Awaitable[_T]:\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor: concurrent.futures.Executor) -> None:\n        return self.asyncio_loop.set_default_executor(executor)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 92,
                                "end_lineno": 285,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/platform/asyncio.py",
                            "type": "class",
                            "name": "BaseAsyncIOLoop",
                            "content": "class BaseAsyncIOLoop(IOLoop):\n    def initialize(  # type: ignore\n        self, asyncio_loop: asyncio.AbstractEventLoop, **kwargs: Any\n    ) -> None:\n        # asyncio_loop is always the real underlying IOLoop. This is used in\n        # ioloop.py to maintain the asyncio-to-ioloop mappings.\n        self.asyncio_loop = asyncio_loop\n        # selector_loop is an event loop that implements the add_reader family of\n        # methods. Usually the same as asyncio_loop but differs on platforms such\n        # as windows where the default event loop does not implement these methods.\n        self.selector_loop = asyncio_loop\n        if hasattr(asyncio, \"ProactorEventLoop\") and isinstance(\n            asyncio_loop, asyncio.ProactorEventLoop  # type: ignore\n        ):\n            # Ignore this line for mypy because the abstract method checker\n            # doesn't understand dynamic proxies.\n            self.selector_loop = AddThreadSelectorEventLoop(asyncio_loop)  # type: ignore\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}  # type: Dict[int, Tuple[Union[int, _Selectable], Callable]]\n        # Set of fds listening for reads/writes\n        self.readers = set()  # type: Set[int]\n        self.writers = set()  # type: Set[int]\n        self.closing = False\n        # If an asyncio loop was closed through an asyncio interface\n        # instead of IOLoop.close(), we'd never hear about it and may\n        # have left a dangling reference in our map. In case an\n        # application (or, more likely, a test suite) creates and\n        # destroys a lot of event loops in this way, check here to\n        # ensure that we don't have a lot of dead loops building up in\n        # the map.\n        #\n        # TODO(bdarnell): consider making self.asyncio_loop a weakref\n        # for AsyncIOMainLoop and make _ioloop_for_asyncio a\n        # WeakKeyDictionary.\n        for loop in IOLoop._ioloop_for_asyncio.copy():\n            if loop.is_closed():\n                try:\n                    del IOLoop._ioloop_for_asyncio[loop]\n                except KeyError:\n                    pass\n\n        # Make sure we don't already have an IOLoop for this asyncio loop\n        if asyncio_loop in IOLoop._ioloop_for_asyncio:\n            existing_loop = IOLoop._ioloop_for_asyncio[asyncio_loop]\n            raise RuntimeError(\n                f\"IOLoop {existing_loop} already associated with asyncio loop {asyncio_loop}\"\n            )\n\n        IOLoop._ioloop_for_asyncio[asyncio_loop] = self\n\n        self._thread_identity = 0\n\n        super().initialize(**kwargs)\n\n        def assign_thread_identity() -> None:\n            self._thread_identity = threading.get_ident()\n\n        self.add_callback(assign_thread_identity)\n\n    def close(self, all_fds: bool = False) -> None:\n        self.closing = True\n        for fd in list(self.handlers):\n            fileobj, handler_func = self.handlers[fd]\n            self.remove_handler(fd)\n            if all_fds:\n                self.close_fd(fileobj)\n        # Remove the mapping before closing the asyncio loop. If this\n        # happened in the other order, we could race against another\n        # initialize() call which would see the closed asyncio loop,\n        # assume it was closed from the asyncio side, and do this\n        # cleanup for us, leading to a KeyError.\n        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]\n        if self.selector_loop is not self.asyncio_loop:\n            self.selector_loop.close()\n        self.asyncio_loop.close()\n\n    def add_handler(\n        self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int\n    ) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd in self.handlers:\n            raise ValueError(\"fd %s added twice\" % fd)\n        self.handlers[fd] = (fileobj, handler)\n        if events & IOLoop.READ:\n            self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.selector_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.selector_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd: Union[int, _Selectable]) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.selector_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.selector_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd: int, events: int) -> None:\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self) -> None:\n        try:\n            old_loop = _get_event_loop()\n        except (RuntimeError, AssertionError):\n            old_loop = None  # type: ignore\n        try:\n            asyncio.set_event_loop(self.asyncio_loop)\n            self.asyncio_loop.run_forever()\n        finally:\n            asyncio.set_event_loop(old_loop)\n\n    def stop(self) -> None:\n        self.asyncio_loop.stop()\n\n    def call_at(\n        self, when: float, callback: Callable, *args: Any, **kwargs: Any\n    ) -> object:\n        # asyncio.call_at supports *args but not **kwargs, so bind them here.\n        # We do not synchronize self.time and asyncio_loop.time, so\n        # convert from absolute to relative.\n        return self.asyncio_loop.call_later(\n            max(0, when - self.time()),\n            self._run_callback,\n            functools.partial(callback, *args, **kwargs),\n        )\n\n    def remove_timeout(self, timeout: object) -> None:\n        timeout.cancel()  # type: ignore\n\n    def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:\n        if threading.get_ident() == self._thread_identity:\n            call_soon = self.asyncio_loop.call_soon\n        else:\n            call_soon = self.asyncio_loop.call_soon_threadsafe\n        try:\n            call_soon(self._run_callback, functools.partial(callback, *args, **kwargs))\n        except RuntimeError:\n            # \"Event loop is closed\". Swallow the exception for\n            # consistency with PollIOLoop (and logical consistency\n            # with the fact that we can't guarantee that an\n            # add_callback that completes without error will\n            # eventually execute).\n            pass\n        except AttributeError:\n            # ProactorEventLoop may raise this instead of RuntimeError\n            # if call_soon_threadsafe races with a call to close().\n            # Swallow it too for consistency.\n            pass\n\n    def add_callback_from_signal(\n        self, callback: Callable, *args: Any, **kwargs: Any\n    ) -> None:\n        try:\n            self.asyncio_loop.call_soon_threadsafe(\n                self._run_callback, functools.partial(callback, *args, **kwargs)\n            )\n        except RuntimeError:\n            pass\n\n    def run_in_executor(\n        self,\n        executor: Optional[concurrent.futures.Executor],\n        func: Callable[..., _T],\n        *args: Any\n    ) -> Awaitable[_T]:\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor: concurrent.futures.Executor) -> None:\n        return self.asyncio_loop.set_default_executor(executor)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 92,
                            "end_lineno": 285,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "import asyncio\nfrom concurrent.futures import ThreadPoolExecutor\nfrom concurrent import futures\nfrom collections.abc import Generator\nimport contextlib\nimport datetime\nimport functools\nimport socket\nimport subprocess\nimport sys\nimport threading\nimport time\nimport types\nfrom unittest import mock\nimport unittest\n\nfrom tornado.escape import native_str\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop, TimeoutError, PeriodicCallback\nfrom tornado.log import app_log\nfrom tornado.testing import (\n    AsyncTestCase,\n    bind_unused_port,\n    ExpectLog,\n    gen_test,\n    setup_with_context_manager,\n)\nfrom tornado.test.util import (\n    ignore_deprecation,\n    skipIfNonUnix,\n    skipOnTravis,\n)\nfrom tornado.concurrent import Future\n\nimport typing\n\nif typing.TYPE_CHECKING:\n    from typing import List  # noqa: F401\n\n\nclass TestIOLoop(AsyncTestCase):\n    def test_add_callback_return_sequence(self):\n        # A callback returning {} or [] shouldn't spin the CPU, see Issue #1803.\n        self.calls = 0\n\n        loop = self.io_loop\n        test = self\n        old_add_callback = loop.add_callback\n\n        def add_callback(self, callback, *args, **kwargs):\n            test.calls += 1\n            old_add_callback(callback, *args, **kwargs)\n\n        loop.add_callback = types.MethodType(add_callback, loop)  # type: ignore\n        loop.add_callback(lambda: {})  # type: ignore\n        loop.add_callback(lambda: [])  # type: ignore\n        loop.add_timeout(datetime.timedelta(milliseconds=50), loop.stop)\n        loop.start()\n        self.assertLess(self.calls, 10)\n\n    @skipOnTravis\n    def test_add_callback_wakeup(self):\n        # Make sure that add_callback from inside a running IOLoop\n        # wakes up the IOLoop immediately instead of waiting for a timeout.\n        def callback():\n            self.called = True\n            self.stop()\n\n        def schedule_callback():\n            self.called = False\n            self.io_loop.add_callback(callback)\n            # Store away the time so we can check if we woke up immediately\n            self.start_time = time.time()\n\n        self.io_loop.add_timeout(self.io_loop.time(), schedule_callback)\n        self.wait()\n        self.assertAlmostEqual(time.time(), self.start_time, places=2)\n        self.assertTrue(self.called)\n\n    @skipOnTravis\n    def test_add_callback_wakeup_other_thread(self):\n        def target():\n            # sleep a bit to let the ioloop go into its poll loop\n            time.sleep(0.01)\n            self.stop_time = time.time()\n            self.io_loop.add_callback(self.stop)\n\n        thread = threading.Thread(target=target)\n        self.io_loop.add_callback(thread.start)\n        self.wait()\n        delta = time.time() - self.stop_time\n        self.assertLess(delta, 0.1)\n        thread.join()\n\n    def test_add_timeout_timedelta(self):\n        self.io_loop.add_timeout(datetime.timedelta(microseconds=1), self.stop)\n        self.wait()\n\n    def test_multiple_add(self):\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.add_handler(\n                sock.fileno(), lambda fd, events: None, IOLoop.READ\n            )\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(\n                Exception,\n                self.io_loop.add_handler,\n                sock.fileno(),\n                lambda fd, events: None,\n                IOLoop.READ,\n            )\n        finally:\n            self.io_loop.remove_handler(sock.fileno())\n            sock.close()\n\n    def test_remove_without_add(self):\n        # remove_handler should not throw an exception if called on an fd\n        # was never added.\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.remove_handler(sock.fileno())\n        finally:\n            sock.close()\n\n    def test_add_callback_from_signal(self):\n        # cheat a little bit and just run this normally, since we can't\n        # easily simulate the races that happen with real signal handlers\n        self.io_loop.add_callback_from_signal(self.stop)\n        self.wait()\n\n    def test_add_callback_from_signal_other_thread(self):\n        # Very crude test, just to make sure that we cover this case.\n        # This also happens to be the first test where we run an IOLoop in\n        # a non-main thread.\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=other_ioloop.start)\n        thread.start()\n        other_ioloop.add_callback_from_signal(other_ioloop.stop)\n        thread.join()\n        other_ioloop.close()\n\n    def test_add_callback_while_closing(self):\n        # add_callback should not fail if it races with another thread\n        # closing the IOLoop. The callbacks are dropped silently\n        # without executing.\n        closing = threading.Event()\n\n        def target():\n            other_ioloop.add_callback(other_ioloop.stop)\n            other_ioloop.start()\n            closing.set()\n            other_ioloop.close(all_fds=True)\n\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            other_ioloop.add_callback(lambda: None)\n\n    @skipIfNonUnix  # just because socketpair is so convenient\n    def test_read_while_writeable(self):\n        # Ensure that write events don't come in while we're waiting for\n        # a read and haven't asked for writeability. (the reverse is\n        # difficult to test for)\n        client, server = socket.socketpair()\n        try:\n\n            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(\n                self.io_loop.time() + 0.01, functools.partial(server.send, b\"asdf\")\n            )\n            self.wait()\n            self.io_loop.remove_handler(client.fileno())\n        finally:\n            client.close()\n            server.close()\n\n    def test_remove_timeout_after_fire(self):\n        # It is not an error to call remove_timeout after it has run.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), self.stop)\n        self.wait()\n        self.io_loop.remove_timeout(handle)\n\n    def test_remove_timeout_cleanup(self):\n        # Add and remove enough callbacks to trigger cleanup.\n        # Not a very thorough test, but it ensures that the cleanup code\n        # gets executed and doesn't blow up.  This test is only really useful\n        # on PollIOLoop subclasses, but it should run silently on any\n        # implementation.\n        for i in range(2000):\n            timeout = self.io_loop.add_timeout(self.io_loop.time() + 3600, lambda: None)\n            self.io_loop.remove_timeout(timeout)\n        # HACK: wait two IOLoop iterations for the GC to happen.\n        self.io_loop.add_callback(lambda: self.io_loop.add_callback(self.stop))\n        self.wait()\n\n    def test_remove_timeout_from_timeout(self):\n        calls = [False, False]\n\n        # Schedule several callbacks and wait for them all to come due at once.\n        # t2 should be cancelled by t1, even though it is already scheduled to\n        # be run before the ioloop even looks at it.\n        now = self.io_loop.time()\n\n        def t1():\n            calls[0] = True\n            self.io_loop.remove_timeout(t2_handle)\n\n        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\n\n        t2_handle = self.io_loop.add_timeout(now + 0.02, t2)\n        self.io_loop.add_timeout(now + 0.03, self.stop)\n        time.sleep(0.03)\n        self.wait()\n        self.assertEqual(calls, [True, False])\n\n    def test_timeout_with_arguments(self):\n        # This tests that all the timeout methods pass through *args correctly.\n        results = []  # type: List[int]\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0), results.append, 2)\n        self.io_loop.call_at(self.io_loop.time(), results.append, 3)\n        self.io_loop.call_later(0, results.append, 4)\n        self.io_loop.call_later(0, self.stop)\n        self.wait()\n        # The asyncio event loop does not guarantee the order of these\n        # callbacks.\n        self.assertEqual(sorted(results), [1, 2, 3, 4])\n\n    def test_add_timeout_return(self):\n        # All the timeout methods return non-None handles that can be\n        # passed to remove_timeout.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_at_return(self):\n        handle = self.io_loop.call_at(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_later_return(self):\n        handle = self.io_loop.call_later(0, lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_close_file_object(self):\n        \"\"\"When a file object is used instead of a numeric file descriptor,\n        the object should be closed (by IOLoop.close(all_fds=True),\n        not just the fd.\n        \"\"\"\n        # Use a socket since they are supported by IOLoop on all platforms.\n        # Unfortunately, sockets don't support the .closed attribute for\n        # inspecting their close status, so we must use a wrapper.\n        class SocketWrapper(object):\n            def __init__(self, sockobj):\n                self.sockobj = sockobj\n                self.closed = False\n\n            def fileno(self):\n                return self.sockobj.fileno()\n\n            def close(self):\n                self.closed = True\n                self.sockobj.close()\n\n        sockobj, port = bind_unused_port()\n        socket_wrapper = SocketWrapper(sockobj)\n        io_loop = IOLoop()\n        io_loop.add_handler(socket_wrapper, lambda fd, events: None, IOLoop.READ)\n        io_loop.close(all_fds=True)\n        self.assertTrue(socket_wrapper.closed)\n\n    def test_handler_callback_file_object(self):\n        \"\"\"The handler callback receives the same fd object it passed in.\"\"\"\n        server_sock, port = bind_unused_port()\n        fds = []\n\n        def handle_connection(fd, events):\n            fds.append(fd)\n            conn, addr = server_sock.accept()\n            conn.close()\n            self.stop()\n\n        self.io_loop.add_handler(server_sock, handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.io_loop.remove_handler(server_sock)\n        self.io_loop.add_handler(server_sock.fileno(), handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.assertIs(fds[0], server_sock)\n        self.assertEqual(fds[1], server_sock.fileno())\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_mixed_fd_fileobj(self):\n        server_sock, port = bind_unused_port()\n\n        def f(fd, events):\n            pass\n\n        self.io_loop.add_handler(server_sock, f, IOLoop.READ)\n        with self.assertRaises(Exception):\n            # The exact error is unspecified - some implementations use\n            # IOError, others use ValueError.\n            self.io_loop.add_handler(server_sock.fileno(), f, IOLoop.READ)\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_reentrant(self):\n        \"\"\"Calling start() twice should raise an error, not deadlock.\"\"\"\n        returned_from_start = [False]\n        got_exception = [False]\n\n        def callback():\n            try:\n                self.io_loop.start()\n                returned_from_start[0] = True\n            except Exception:\n                got_exception[0] = True\n            self.stop()\n\n        self.io_loop.add_callback(callback)\n        self.wait()\n        self.assertTrue(got_exception[0])\n        self.assertFalse(returned_from_start[0])\n\n    def test_exception_logging(self):\n        \"\"\"Uncaught exceptions get logged by the IOLoop.\"\"\"\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_future(self):\n        \"\"\"The IOLoop examines exceptions from Futures and logs them.\"\"\"\n\n        @gen.coroutine\n        def callback():\n            self.io_loop.add_callback(self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n\n        async def callback():\n            # Stop the IOLoop two iterations after raising an exception\n            # to give the exception time to be logged.\n            self.io_loop.add_callback(self.io_loop.add_callback, self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_spawn_callback(self):\n        # Both add_callback and spawn_callback run directly on the IOLoop,\n        # so their errors are logged without stopping the test.\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n        # A spawned callback is run directly on the IOLoop, so it will be\n        # logged without stopping the test.\n        self.io_loop.spawn_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    @skipIfNonUnix\n    def test_remove_handler_from_handler(self):\n        # Create two sockets with simultaneous read events.\n        client, server = socket.socketpair()\n        try:\n            client.send(b\"abc\")\n            server.send(b\"abc\")\n\n            # After reading from one fd, remove the other from the IOLoop.\n            chunks = []\n\n            def handle_read(fd, events):\n                chunks.append(fd.recv(1024))\n                if fd is client:\n                    self.io_loop.remove_handler(server)\n                else:\n                    self.io_loop.remove_handler(client)\n\n            self.io_loop.add_handler(client, handle_read, self.io_loop.READ)\n            self.io_loop.add_handler(server, handle_read, self.io_loop.READ)\n            self.io_loop.call_later(0.1, self.stop)\n            self.wait()\n\n            # Only one fd was read; the other was cleanly removed.\n            self.assertEqual(chunks, [b\"abc\"])\n        finally:\n            client.close()\n            server.close()\n\n    @skipIfNonUnix\n    @gen_test\n    def test_init_close_race(self):\n        # Regression test for #2367\n        #\n        # Skipped on windows because of what looks like a bug in the\n        # proactor event loop when started and stopped on non-main\n        # threads.\n        def f():\n            for i in range(10):\n                loop = IOLoop(make_current=False)\n                loop.close()\n\n        yield gen.multi([self.io_loop.run_in_executor(None, f) for i in range(2)])\n\n    def test_explicit_asyncio_loop(self):\n        asyncio_loop = asyncio.new_event_loop()\n        loop = IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        assert loop.asyncio_loop is asyncio_loop  # type: ignore\n        loop.close()\n\n\n# Deliberately not a subclass of AsyncTestCase so the IOLoop isn't\n# automatically set as current.\nclass TestIOLoopCurrent(unittest.TestCase):\n    def setUp(self):\n        setup_with_context_manager(self, ignore_deprecation())\n        self.io_loop = None  # type: typing.Optional[IOLoop]\n        IOLoop.clear_current()\n\n    def tearDown(self):\n        if self.io_loop is not None:\n            self.io_loop.close()\n\n    def test_default_current(self):\n        self.io_loop = IOLoop()\n        # The first IOLoop with default arguments is made current.\n        self.assertIs(self.io_loop, IOLoop.current())\n        # A second IOLoop can be created but is not made current.\n        io_loop2 = IOLoop()\n        self.assertIs(self.io_loop, IOLoop.current())\n        io_loop2.close()\n\n    def test_non_current(self):\n        self.io_loop = IOLoop(make_current=False)\n        # The new IOLoop is not initially made current.\n        self.assertIsNone(IOLoop.current(instance=False))\n        # Starting the IOLoop makes it current, and stopping the loop\n        # makes it non-current. This process is repeatable.\n        for i in range(3):\n\n            def f():\n                self.current_io_loop = IOLoop.current()\n                assert self.io_loop is not None\n                self.io_loop.stop()\n\n            self.io_loop.add_callback(f)\n            self.io_loop.start()\n            self.assertIs(self.current_io_loop, self.io_loop)\n            # Now that the loop is stopped, it is no longer current.\n            self.assertIsNone(IOLoop.current(instance=False))\n\n    def test_force_current(self):\n        self.io_loop = IOLoop(make_current=True)\n        self.assertIs(self.io_loop, IOLoop.current())\n        with self.assertRaises(RuntimeError):\n            # A second make_current=True construction cannot succeed.\n            IOLoop(make_current=True)\n        # current() was not affected by the failed construction.\n        self.assertIs(self.io_loop, IOLoop.current())\n\n\nclass TestIOLoopCurrentAsync(AsyncTestCase):\n    def setUp(self):\n        super().setUp()\n        setup_with_context_manager(self, ignore_deprecation())\n\n    @gen_test\n    def test_clear_without_current(self):\n        # If there is no current IOLoop, clear_current is a no-op (but\n        # should not fail). Use a thread so we see the threading.Local\n        # in a pristine state.\n        with ThreadPoolExecutor(1) as e:\n            yield e.submit(IOLoop.clear_current)\n\n\nclass TestIOLoopFutures(AsyncTestCase):\n    def test_add_future_threads(self):\n        with futures.ThreadPoolExecutor(1) as pool:\n\n            def dummy():\n                pass\n\n            self.io_loop.add_future(\n                pool.submit(dummy), lambda future: self.stop(future)\n            )\n            future = self.wait()\n            self.assertTrue(future.done())\n            self.assertTrue(future.result() is None)\n\n    @gen_test\n    def test_run_in_executor_gen(self):\n        event1 = threading.Event()\n        event2 = threading.Event()\n\n        def sync_func(self_event, other_event):\n            self_event.set()\n            other_event.wait()\n            # Note that return value doesn't actually do anything,\n            # it is just passed through to our final assertion to\n            # make sure it is passed through properly.\n            return self_event\n\n        # Run two synchronous functions, which would deadlock if not\n        # run in parallel.\n        res = yield [\n            IOLoop.current().run_in_executor(None, sync_func, event1, event2),\n            IOLoop.current().run_in_executor(None, sync_func, event2, event1),\n        ]\n\n        self.assertEqual([event1, event2], res)\n\n    @gen_test\n    def test_run_in_executor_native(self):\n        event1 = threading.Event()\n        event2 = threading.Event()\n\n        def sync_func(self_event, other_event):\n            self_event.set()\n            other_event.wait()\n            return self_event\n\n        # Go through an async wrapper to ensure that the result of\n        # run_in_executor works with await and not just gen.coroutine\n        # (simply passing the underlying concurrent future would do that).\n        async def async_wrapper(self_event, other_event):\n            return await IOLoop.current().run_in_executor(\n                None, sync_func, self_event, other_event\n            )\n\n        res = yield [async_wrapper(event1, event2), async_wrapper(event2, event1)]\n\n        self.assertEqual([event1, event2], res)\n\n    @gen_test\n    def test_set_default_executor(self):\n        count = [0]\n\n        class MyExecutor(futures.ThreadPoolExecutor):\n            def submit(self, func, *args):\n                count[0] += 1\n                return super().submit(func, *args)\n\n        event = threading.Event()\n\n        def sync_func():\n            event.set()\n\n        executor = MyExecutor(1)\n        loop = IOLoop.current()\n        loop.set_default_executor(executor)\n        yield loop.run_in_executor(None, sync_func)\n        self.assertEqual(1, count[0])\n        self.assertTrue(event.is_set())\n\n\nclass TestIOLoopRunSync(unittest.TestCase):\n    def setUp(self):\n        self.io_loop = IOLoop(make_current=False)\n\n    def tearDown(self):\n        self.io_loop.close()\n\n    def test_sync_result(self):\n        with self.assertRaises(gen.BadYieldError):\n            self.io_loop.run_sync(lambda: 42)\n\n    def test_sync_exception(self):\n        with self.assertRaises(ZeroDivisionError):\n            self.io_loop.run_sync(lambda: 1 / 0)\n\n    def test_async_result(self):\n        @gen.coroutine\n        def f():\n            yield gen.moment\n            raise gen.Return(42)\n\n        self.assertEqual(self.io_loop.run_sync(f), 42)\n\n    def test_async_exception(self):\n        @gen.coroutine\n        def f():\n            yield gen.moment\n            1 / 0\n\n        with self.assertRaises(ZeroDivisionError):\n            self.io_loop.run_sync(f)\n\n    def test_current(self):\n        def f():\n            self.assertIs(IOLoop.current(), self.io_loop)\n\n        self.io_loop.run_sync(f)\n\n    def test_timeout(self):\n        @gen.coroutine\n        def f():\n            yield gen.sleep(1)\n\n        self.assertRaises(TimeoutError, self.io_loop.run_sync, f, timeout=0.01)\n\n    def test_native_coroutine(self):\n        @gen.coroutine\n        def f1():\n            yield gen.moment\n\n        async def f2():\n            await f1()\n\n        self.io_loop.run_sync(f2)\n\n\nclass TestPeriodicCallbackMath(unittest.TestCase):\n    def simulate_calls(self, pc, durations):\n        \"\"\"Simulate a series of calls to the PeriodicCallback.\n\n        Pass a list of call durations in seconds (negative values\n        work to simulate clock adjustments during the call, or more or\n        less equivalently, between calls). This method returns the\n        times at which each call would be made.\n        \"\"\"\n        calls = []\n        now = 1000\n        pc._next_timeout = now\n        for d in durations:\n            pc._update_next(now)\n            calls.append(pc._next_timeout)\n            now = pc._next_timeout + d\n        return calls\n\n    def dummy(self):\n        pass\n\n    def test_basic(self):\n        pc = PeriodicCallback(self.dummy, 10000)\n        self.assertEqual(\n            self.simulate_calls(pc, [0] * 5), [1010, 1020, 1030, 1040, 1050]\n        )\n\n    def test_overrun(self):\n        # If a call runs for too long, we skip entire cycles to get\n        # back on schedule.\n        call_durations = [9, 9, 10, 11, 20, 20, 35, 35, 0, 0, 0]\n        expected = [\n            1010,\n            1020,\n            1030,  # first 3 calls on schedule\n            1050,\n            1070,  # next 2 delayed one cycle\n            1100,\n            1130,  # next 2 delayed 2 cycles\n            1170,\n            1210,  # next 2 delayed 3 cycles\n            1220,\n            1230,  # then back on schedule.\n        ]\n\n        pc = PeriodicCallback(self.dummy, 10000)\n        self.assertEqual(self.simulate_calls(pc, call_durations), expected)\n\n    def test_clock_backwards(self):\n        pc = PeriodicCallback(self.dummy, 10000)\n        # Backwards jumps are ignored, potentially resulting in a\n        # slightly slow schedule (although we assume that when\n        # time.time() and time.monotonic() are different, time.time()\n        # is getting adjusted by NTP and is therefore more accurate)\n        self.assertEqual(\n            self.simulate_calls(pc, [-2, -1, -3, -2, 0]), [1010, 1020, 1030, 1040, 1050]\n        )\n\n        # For big jumps, we should perhaps alter the schedule, but we\n        # don't currently. This trace shows that we run callbacks\n        # every 10s of time.time(), but the first and second calls are\n        # 110s of real time apart because the backwards jump is\n        # ignored.\n        self.assertEqual(self.simulate_calls(pc, [-100, 0, 0]), [1010, 1020, 1030])\n\n    def test_jitter(self):\n        random_times = [0.5, 1, 0, 0.75]\n        expected = [1010, 1022.5, 1030, 1041.25]\n        call_durations = [0] * len(random_times)\n        pc = PeriodicCallback(self.dummy, 10000, jitter=0.5)\n\n        def mock_random():\n            return random_times.pop(0)\n\n        with mock.patch(\"random.random\", mock_random):\n            self.assertEqual(self.simulate_calls(pc, call_durations), expected)\n\n    def test_timedelta(self):\n        pc = PeriodicCallback(lambda: None, datetime.timedelta(minutes=1, seconds=23))\n        expected_callback_time = 83000\n        self.assertEqual(pc.callback_time, expected_callback_time)\n\n\nclass TestPeriodicCallbackAsync(AsyncTestCase):\n    def test_periodic_plain(self):\n        count = 0\n\n        def callback() -> None:\n            nonlocal count\n            count += 1\n            if count == 3:\n                self.stop()\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        pc.stop()\n        self.assertEqual(count, 3)\n\n    def test_periodic_coro(self) -> None:\n        counts = [0, 0]\n\n        @gen.coroutine\n        def callback() -> \"Generator[Future[None], object, None]\":\n            counts[0] += 1\n            yield gen.sleep(0.025)\n            counts[1] += 1\n            if counts[1] == 3:\n                pc.stop()\n                self.io_loop.add_callback(self.stop)\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        self.assertEqual(counts[0], 3)\n        self.assertEqual(counts[1], 3)\n\n    def test_periodic_async(self) -> None:\n        counts = [0, 0]\n\n        async def callback() -> None:\n            counts[0] += 1\n            await gen.sleep(0.025)\n            counts[1] += 1\n            if counts[1] == 3:\n                pc.stop()\n                self.io_loop.add_callback(self.stop)\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        self.assertEqual(counts[0], 3)\n        self.assertEqual(counts[1], 3)\n\n\nclass TestIOLoopConfiguration(unittest.TestCase):\n    def run_python(self, *statements):\n        stmt_list = [\n            \"from tornado.ioloop import IOLoop\",\n            \"classname = lambda x: x.__class__.__name__\",\n        ] + list(statements)\n        args = [sys.executable, \"-c\", \"; \".join(stmt_list)]\n        return native_str(subprocess.check_output(args)).strip()\n\n    def test_default(self):\n        # When asyncio is available, it is used by default.\n        cls = self.run_python(\"print(classname(IOLoop.current()))\")\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n        cls = self.run_python(\"print(classname(IOLoop()))\")\n        self.assertEqual(cls, \"AsyncIOLoop\")\n\n    def test_asyncio(self):\n        cls = self.run_python(\n            'IOLoop.configure(\"tornado.platform.asyncio.AsyncIOLoop\")',\n            \"print(classname(IOLoop.current()))\",\n        )\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n\n    def test_asyncio_main(self):\n        cls = self.run_python(\n            \"from tornado.platform.asyncio import AsyncIOMainLoop\",\n            \"AsyncIOMainLoop().install()\",\n            \"print(classname(IOLoop.current()))\",\n        )\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n\n\nif __name__ == \"__main__\":\n    unittest.main()\n",
            "new_file_content": "import asyncio\nfrom concurrent.futures import ThreadPoolExecutor\nfrom concurrent import futures\nfrom collections.abc import Generator\nimport contextlib\nimport datetime\nimport functools\nimport socket\nimport subprocess\nimport sys\nimport threading\nimport time\nimport types\nfrom unittest import mock\nimport unittest\n\nfrom tornado.escape import native_str\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop, TimeoutError, PeriodicCallback\nfrom tornado.log import app_log\nfrom tornado.testing import (\n    AsyncTestCase,\n    bind_unused_port,\n    ExpectLog,\n    gen_test,\n    setup_with_context_manager,\n)\nfrom tornado.test.util import (\n    ignore_deprecation,\n    skipIfNonUnix,\n    skipOnTravis,\n)\nfrom tornado.concurrent import Future\n\nimport typing\n\nif typing.TYPE_CHECKING:\n    from typing import List  # noqa: F401\n\n\nclass TestIOLoop(AsyncTestCase):\n    def test_add_callback_return_sequence(self):\n        # A callback returning {} or [] shouldn't spin the CPU, see Issue #1803.\n        self.calls = 0\n\n        loop = self.io_loop\n        test = self\n        old_add_callback = loop.add_callback\n\n        def add_callback(self, callback, *args, **kwargs):\n            test.calls += 1\n            old_add_callback(callback, *args, **kwargs)\n\n        loop.add_callback = types.MethodType(add_callback, loop)  # type: ignore\n        loop.add_callback(lambda: {})  # type: ignore\n        loop.add_callback(lambda: [])  # type: ignore\n        loop.add_timeout(datetime.timedelta(milliseconds=50), loop.stop)\n        loop.start()\n        self.assertLess(self.calls, 10)\n\n    @skipOnTravis\n    def test_add_callback_wakeup(self):\n        # Make sure that add_callback from inside a running IOLoop\n        # wakes up the IOLoop immediately instead of waiting for a timeout.\n        def callback():\n            self.called = True\n            self.stop()\n\n        def schedule_callback():\n            self.called = False\n            self.io_loop.add_callback(callback)\n            # Store away the time so we can check if we woke up immediately\n            self.start_time = time.time()\n\n        self.io_loop.add_timeout(self.io_loop.time(), schedule_callback)\n        self.wait()\n        self.assertAlmostEqual(time.time(), self.start_time, places=2)\n        self.assertTrue(self.called)\n\n    @skipOnTravis\n    def test_add_callback_wakeup_other_thread(self):\n        def target():\n            # sleep a bit to let the ioloop go into its poll loop\n            time.sleep(0.01)\n            self.stop_time = time.time()\n            self.io_loop.add_callback(self.stop)\n\n        thread = threading.Thread(target=target)\n        self.io_loop.add_callback(thread.start)\n        self.wait()\n        delta = time.time() - self.stop_time\n        self.assertLess(delta, 0.1)\n        thread.join()\n\n    def test_add_timeout_timedelta(self):\n        self.io_loop.add_timeout(datetime.timedelta(microseconds=1), self.stop)\n        self.wait()\n\n    def test_multiple_add(self):\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.add_handler(\n                sock.fileno(), lambda fd, events: None, IOLoop.READ\n            )\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(\n                Exception,\n                self.io_loop.add_handler,\n                sock.fileno(),\n                lambda fd, events: None,\n                IOLoop.READ,\n            )\n        finally:\n            self.io_loop.remove_handler(sock.fileno())\n            sock.close()\n\n    def test_remove_without_add(self):\n        # remove_handler should not throw an exception if called on an fd\n        # was never added.\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.remove_handler(sock.fileno())\n        finally:\n            sock.close()\n\n    def test_add_callback_from_signal(self):\n        # cheat a little bit and just run this normally, since we can't\n        # easily simulate the races that happen with real signal handlers\n        self.io_loop.add_callback_from_signal(self.stop)\n        self.wait()\n\n    def test_add_callback_from_signal_other_thread(self):\n        # Very crude test, just to make sure that we cover this case.\n        # This also happens to be the first test where we run an IOLoop in\n        # a non-main thread.\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=other_ioloop.start)\n        thread.start()\n        other_ioloop.add_callback_from_signal(other_ioloop.stop)\n        thread.join()\n        other_ioloop.close()\n\n    def test_add_callback_while_closing(self):\n        # add_callback should not fail if it races with another thread\n        # closing the IOLoop. The callbacks are dropped silently\n        # without executing.\n        closing = threading.Event()\n\n        def target():\n            other_ioloop.add_callback(other_ioloop.stop)\n            other_ioloop.start()\n            closing.set()\n            other_ioloop.close(all_fds=True)\n\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            other_ioloop.add_callback(lambda: None)\n\n    @skipIfNonUnix  # just because socketpair is so convenient\n    def test_read_while_writeable(self):\n        # Ensure that write events don't come in while we're waiting for\n        # a read and haven't asked for writeability. (the reverse is\n        # difficult to test for)\n        client, server = socket.socketpair()\n        try:\n\n            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(\n                self.io_loop.time() + 0.01, functools.partial(server.send, b\"asdf\")\n            )\n            self.wait()\n            self.io_loop.remove_handler(client.fileno())\n        finally:\n            client.close()\n            server.close()\n\n    def test_remove_timeout_after_fire(self):\n        # It is not an error to call remove_timeout after it has run.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), self.stop)\n        self.wait()\n        self.io_loop.remove_timeout(handle)\n\n    def test_remove_timeout_cleanup(self):\n        # Add and remove enough callbacks to trigger cleanup.\n        # Not a very thorough test, but it ensures that the cleanup code\n        # gets executed and doesn't blow up.  This test is only really useful\n        # on PollIOLoop subclasses, but it should run silently on any\n        # implementation.\n        for i in range(2000):\n            timeout = self.io_loop.add_timeout(self.io_loop.time() + 3600, lambda: None)\n            self.io_loop.remove_timeout(timeout)\n        # HACK: wait two IOLoop iterations for the GC to happen.\n        self.io_loop.add_callback(lambda: self.io_loop.add_callback(self.stop))\n        self.wait()\n\n    def test_remove_timeout_from_timeout(self):\n        calls = [False, False]\n\n        # Schedule several callbacks and wait for them all to come due at once.\n        # t2 should be cancelled by t1, even though it is already scheduled to\n        # be run before the ioloop even looks at it.\n        now = self.io_loop.time()\n\n        def t1():\n            calls[0] = True\n            self.io_loop.remove_timeout(t2_handle)\n\n        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\n\n        t2_handle = self.io_loop.add_timeout(now + 0.02, t2)\n        self.io_loop.add_timeout(now + 0.03, self.stop)\n        time.sleep(0.03)\n        self.wait()\n        self.assertEqual(calls, [True, False])\n\n    def test_timeout_with_arguments(self):\n        # This tests that all the timeout methods pass through *args correctly.\n        results = []  # type: List[int]\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0), results.append, 2)\n        self.io_loop.call_at(self.io_loop.time(), results.append, 3)\n        self.io_loop.call_later(0, results.append, 4)\n        self.io_loop.call_later(0, self.stop)\n        self.wait()\n        # The asyncio event loop does not guarantee the order of these\n        # callbacks.\n        self.assertEqual(sorted(results), [1, 2, 3, 4])\n\n    def test_add_timeout_return(self):\n        # All the timeout methods return non-None handles that can be\n        # passed to remove_timeout.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_at_return(self):\n        handle = self.io_loop.call_at(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_later_return(self):\n        handle = self.io_loop.call_later(0, lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_close_file_object(self):\n        \"\"\"When a file object is used instead of a numeric file descriptor,\n        the object should be closed (by IOLoop.close(all_fds=True),\n        not just the fd.\n        \"\"\"\n        # Use a socket since they are supported by IOLoop on all platforms.\n        # Unfortunately, sockets don't support the .closed attribute for\n        # inspecting their close status, so we must use a wrapper.\n        class SocketWrapper(object):\n            def __init__(self, sockobj):\n                self.sockobj = sockobj\n                self.closed = False\n\n            def fileno(self):\n                return self.sockobj.fileno()\n\n            def close(self):\n                self.closed = True\n                self.sockobj.close()\n\n        sockobj, port = bind_unused_port()\n        socket_wrapper = SocketWrapper(sockobj)\n        io_loop = IOLoop()\n        io_loop.add_handler(socket_wrapper, lambda fd, events: None, IOLoop.READ)\n        io_loop.close(all_fds=True)\n        self.assertTrue(socket_wrapper.closed)\n\n    def test_handler_callback_file_object(self):\n        \"\"\"The handler callback receives the same fd object it passed in.\"\"\"\n        server_sock, port = bind_unused_port()\n        fds = []\n\n        def handle_connection(fd, events):\n            fds.append(fd)\n            conn, addr = server_sock.accept()\n            conn.close()\n            self.stop()\n\n        self.io_loop.add_handler(server_sock, handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.io_loop.remove_handler(server_sock)\n        self.io_loop.add_handler(server_sock.fileno(), handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.assertIs(fds[0], server_sock)\n        self.assertEqual(fds[1], server_sock.fileno())\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_mixed_fd_fileobj(self):\n        server_sock, port = bind_unused_port()\n\n        def f(fd, events):\n            pass\n\n        self.io_loop.add_handler(server_sock, f, IOLoop.READ)\n        with self.assertRaises(Exception):\n            # The exact error is unspecified - some implementations use\n            # IOError, others use ValueError.\n            self.io_loop.add_handler(server_sock.fileno(), f, IOLoop.READ)\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_reentrant(self):\n        \"\"\"Calling start() twice should raise an error, not deadlock.\"\"\"\n        returned_from_start = [False]\n        got_exception = [False]\n\n        def callback():\n            try:\n                self.io_loop.start()\n                returned_from_start[0] = True\n            except Exception:\n                got_exception[0] = True\n            self.stop()\n\n        self.io_loop.add_callback(callback)\n        self.wait()\n        self.assertTrue(got_exception[0])\n        self.assertFalse(returned_from_start[0])\n\n    def test_exception_logging(self):\n        \"\"\"Uncaught exceptions get logged by the IOLoop.\"\"\"\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_future(self):\n        \"\"\"The IOLoop examines exceptions from Futures and logs them.\"\"\"\n\n        @gen.coroutine\n        def callback():\n            self.io_loop.add_callback(self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n\n        async def callback():\n            # Stop the IOLoop two iterations after raising an exception\n            # to give the exception time to be logged.\n            self.io_loop.add_callback(self.io_loop.add_callback, self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_spawn_callback(self):\n        # Both add_callback and spawn_callback run directly on the IOLoop,\n        # so their errors are logged without stopping the test.\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n        # A spawned callback is run directly on the IOLoop, so it will be\n        # logged without stopping the test.\n        self.io_loop.spawn_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    @skipIfNonUnix\n    def test_remove_handler_from_handler(self):\n        # Create two sockets with simultaneous read events.\n        client, server = socket.socketpair()\n        try:\n            client.send(b\"abc\")\n            server.send(b\"abc\")\n\n            # After reading from one fd, remove the other from the IOLoop.\n            chunks = []\n\n            def handle_read(fd, events):\n                chunks.append(fd.recv(1024))\n                if fd is client:\n                    self.io_loop.remove_handler(server)\n                else:\n                    self.io_loop.remove_handler(client)\n\n            self.io_loop.add_handler(client, handle_read, self.io_loop.READ)\n            self.io_loop.add_handler(server, handle_read, self.io_loop.READ)\n            self.io_loop.call_later(0.1, self.stop)\n            self.wait()\n\n            # Only one fd was read; the other was cleanly removed.\n            self.assertEqual(chunks, [b\"abc\"])\n        finally:\n            client.close()\n            server.close()\n\n    @skipIfNonUnix\n    @gen_test\n    def test_init_close_race(self):\n        # Regression test for #2367\n        #\n        # Skipped on windows because of what looks like a bug in the\n        # proactor event loop when started and stopped on non-main\n        # threads.\n        def f():\n            for i in range(10):\n                loop = IOLoop(make_current=False)\n                loop.close()\n\n        yield gen.multi([self.io_loop.run_in_executor(None, f) for i in range(2)])\n\n    def test_explicit_asyncio_loop(self):\n        asyncio_loop = asyncio.new_event_loop()\n        loop = IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        assert loop.asyncio_loop is asyncio_loop  # type: ignore\n        with self.assertRaises(RuntimeError):\n            # Can't register two IOLoops with the same asyncio_loop\n            IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        loop.close()\n\n\n# Deliberately not a subclass of AsyncTestCase so the IOLoop isn't\n# automatically set as current.\nclass TestIOLoopCurrent(unittest.TestCase):\n    def setUp(self):\n        setup_with_context_manager(self, ignore_deprecation())\n        self.io_loop = None  # type: typing.Optional[IOLoop]\n        IOLoop.clear_current()\n\n    def tearDown(self):\n        if self.io_loop is not None:\n            self.io_loop.close()\n\n    def test_default_current(self):\n        self.io_loop = IOLoop()\n        # The first IOLoop with default arguments is made current.\n        self.assertIs(self.io_loop, IOLoop.current())\n        # A second IOLoop can be created but is not made current.\n        io_loop2 = IOLoop()\n        self.assertIs(self.io_loop, IOLoop.current())\n        io_loop2.close()\n\n    def test_non_current(self):\n        self.io_loop = IOLoop(make_current=False)\n        # The new IOLoop is not initially made current.\n        self.assertIsNone(IOLoop.current(instance=False))\n        # Starting the IOLoop makes it current, and stopping the loop\n        # makes it non-current. This process is repeatable.\n        for i in range(3):\n\n            def f():\n                self.current_io_loop = IOLoop.current()\n                assert self.io_loop is not None\n                self.io_loop.stop()\n\n            self.io_loop.add_callback(f)\n            self.io_loop.start()\n            self.assertIs(self.current_io_loop, self.io_loop)\n            # Now that the loop is stopped, it is no longer current.\n            self.assertIsNone(IOLoop.current(instance=False))\n\n    def test_force_current(self):\n        self.io_loop = IOLoop(make_current=True)\n        self.assertIs(self.io_loop, IOLoop.current())\n        with self.assertRaises(RuntimeError):\n            # A second make_current=True construction cannot succeed.\n            IOLoop(make_current=True)\n        # current() was not affected by the failed construction.\n        self.assertIs(self.io_loop, IOLoop.current())\n\n\nclass TestIOLoopCurrentAsync(AsyncTestCase):\n    def setUp(self):\n        super().setUp()\n        setup_with_context_manager(self, ignore_deprecation())\n\n    @gen_test\n    def test_clear_without_current(self):\n        # If there is no current IOLoop, clear_current is a no-op (but\n        # should not fail). Use a thread so we see the threading.Local\n        # in a pristine state.\n        with ThreadPoolExecutor(1) as e:\n            yield e.submit(IOLoop.clear_current)\n\n\nclass TestIOLoopFutures(AsyncTestCase):\n    def test_add_future_threads(self):\n        with futures.ThreadPoolExecutor(1) as pool:\n\n            def dummy():\n                pass\n\n            self.io_loop.add_future(\n                pool.submit(dummy), lambda future: self.stop(future)\n            )\n            future = self.wait()\n            self.assertTrue(future.done())\n            self.assertTrue(future.result() is None)\n\n    @gen_test\n    def test_run_in_executor_gen(self):\n        event1 = threading.Event()\n        event2 = threading.Event()\n\n        def sync_func(self_event, other_event):\n            self_event.set()\n            other_event.wait()\n            # Note that return value doesn't actually do anything,\n            # it is just passed through to our final assertion to\n            # make sure it is passed through properly.\n            return self_event\n\n        # Run two synchronous functions, which would deadlock if not\n        # run in parallel.\n        res = yield [\n            IOLoop.current().run_in_executor(None, sync_func, event1, event2),\n            IOLoop.current().run_in_executor(None, sync_func, event2, event1),\n        ]\n\n        self.assertEqual([event1, event2], res)\n\n    @gen_test\n    def test_run_in_executor_native(self):\n        event1 = threading.Event()\n        event2 = threading.Event()\n\n        def sync_func(self_event, other_event):\n            self_event.set()\n            other_event.wait()\n            return self_event\n\n        # Go through an async wrapper to ensure that the result of\n        # run_in_executor works with await and not just gen.coroutine\n        # (simply passing the underlying concurrent future would do that).\n        async def async_wrapper(self_event, other_event):\n            return await IOLoop.current().run_in_executor(\n                None, sync_func, self_event, other_event\n            )\n\n        res = yield [async_wrapper(event1, event2), async_wrapper(event2, event1)]\n\n        self.assertEqual([event1, event2], res)\n\n    @gen_test\n    def test_set_default_executor(self):\n        count = [0]\n\n        class MyExecutor(futures.ThreadPoolExecutor):\n            def submit(self, func, *args):\n                count[0] += 1\n                return super().submit(func, *args)\n\n        event = threading.Event()\n\n        def sync_func():\n            event.set()\n\n        executor = MyExecutor(1)\n        loop = IOLoop.current()\n        loop.set_default_executor(executor)\n        yield loop.run_in_executor(None, sync_func)\n        self.assertEqual(1, count[0])\n        self.assertTrue(event.is_set())\n\n\nclass TestIOLoopRunSync(unittest.TestCase):\n    def setUp(self):\n        self.io_loop = IOLoop(make_current=False)\n\n    def tearDown(self):\n        self.io_loop.close()\n\n    def test_sync_result(self):\n        with self.assertRaises(gen.BadYieldError):\n            self.io_loop.run_sync(lambda: 42)\n\n    def test_sync_exception(self):\n        with self.assertRaises(ZeroDivisionError):\n            self.io_loop.run_sync(lambda: 1 / 0)\n\n    def test_async_result(self):\n        @gen.coroutine\n        def f():\n            yield gen.moment\n            raise gen.Return(42)\n\n        self.assertEqual(self.io_loop.run_sync(f), 42)\n\n    def test_async_exception(self):\n        @gen.coroutine\n        def f():\n            yield gen.moment\n            1 / 0\n\n        with self.assertRaises(ZeroDivisionError):\n            self.io_loop.run_sync(f)\n\n    def test_current(self):\n        def f():\n            self.assertIs(IOLoop.current(), self.io_loop)\n\n        self.io_loop.run_sync(f)\n\n    def test_timeout(self):\n        @gen.coroutine\n        def f():\n            yield gen.sleep(1)\n\n        self.assertRaises(TimeoutError, self.io_loop.run_sync, f, timeout=0.01)\n\n    def test_native_coroutine(self):\n        @gen.coroutine\n        def f1():\n            yield gen.moment\n\n        async def f2():\n            await f1()\n\n        self.io_loop.run_sync(f2)\n\n\nclass TestPeriodicCallbackMath(unittest.TestCase):\n    def simulate_calls(self, pc, durations):\n        \"\"\"Simulate a series of calls to the PeriodicCallback.\n\n        Pass a list of call durations in seconds (negative values\n        work to simulate clock adjustments during the call, or more or\n        less equivalently, between calls). This method returns the\n        times at which each call would be made.\n        \"\"\"\n        calls = []\n        now = 1000\n        pc._next_timeout = now\n        for d in durations:\n            pc._update_next(now)\n            calls.append(pc._next_timeout)\n            now = pc._next_timeout + d\n        return calls\n\n    def dummy(self):\n        pass\n\n    def test_basic(self):\n        pc = PeriodicCallback(self.dummy, 10000)\n        self.assertEqual(\n            self.simulate_calls(pc, [0] * 5), [1010, 1020, 1030, 1040, 1050]\n        )\n\n    def test_overrun(self):\n        # If a call runs for too long, we skip entire cycles to get\n        # back on schedule.\n        call_durations = [9, 9, 10, 11, 20, 20, 35, 35, 0, 0, 0]\n        expected = [\n            1010,\n            1020,\n            1030,  # first 3 calls on schedule\n            1050,\n            1070,  # next 2 delayed one cycle\n            1100,\n            1130,  # next 2 delayed 2 cycles\n            1170,\n            1210,  # next 2 delayed 3 cycles\n            1220,\n            1230,  # then back on schedule.\n        ]\n\n        pc = PeriodicCallback(self.dummy, 10000)\n        self.assertEqual(self.simulate_calls(pc, call_durations), expected)\n\n    def test_clock_backwards(self):\n        pc = PeriodicCallback(self.dummy, 10000)\n        # Backwards jumps are ignored, potentially resulting in a\n        # slightly slow schedule (although we assume that when\n        # time.time() and time.monotonic() are different, time.time()\n        # is getting adjusted by NTP and is therefore more accurate)\n        self.assertEqual(\n            self.simulate_calls(pc, [-2, -1, -3, -2, 0]), [1010, 1020, 1030, 1040, 1050]\n        )\n\n        # For big jumps, we should perhaps alter the schedule, but we\n        # don't currently. This trace shows that we run callbacks\n        # every 10s of time.time(), but the first and second calls are\n        # 110s of real time apart because the backwards jump is\n        # ignored.\n        self.assertEqual(self.simulate_calls(pc, [-100, 0, 0]), [1010, 1020, 1030])\n\n    def test_jitter(self):\n        random_times = [0.5, 1, 0, 0.75]\n        expected = [1010, 1022.5, 1030, 1041.25]\n        call_durations = [0] * len(random_times)\n        pc = PeriodicCallback(self.dummy, 10000, jitter=0.5)\n\n        def mock_random():\n            return random_times.pop(0)\n\n        with mock.patch(\"random.random\", mock_random):\n            self.assertEqual(self.simulate_calls(pc, call_durations), expected)\n\n    def test_timedelta(self):\n        pc = PeriodicCallback(lambda: None, datetime.timedelta(minutes=1, seconds=23))\n        expected_callback_time = 83000\n        self.assertEqual(pc.callback_time, expected_callback_time)\n\n\nclass TestPeriodicCallbackAsync(AsyncTestCase):\n    def test_periodic_plain(self):\n        count = 0\n\n        def callback() -> None:\n            nonlocal count\n            count += 1\n            if count == 3:\n                self.stop()\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        pc.stop()\n        self.assertEqual(count, 3)\n\n    def test_periodic_coro(self) -> None:\n        counts = [0, 0]\n\n        @gen.coroutine\n        def callback() -> \"Generator[Future[None], object, None]\":\n            counts[0] += 1\n            yield gen.sleep(0.025)\n            counts[1] += 1\n            if counts[1] == 3:\n                pc.stop()\n                self.io_loop.add_callback(self.stop)\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        self.assertEqual(counts[0], 3)\n        self.assertEqual(counts[1], 3)\n\n    def test_periodic_async(self) -> None:\n        counts = [0, 0]\n\n        async def callback() -> None:\n            counts[0] += 1\n            await gen.sleep(0.025)\n            counts[1] += 1\n            if counts[1] == 3:\n                pc.stop()\n                self.io_loop.add_callback(self.stop)\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        self.assertEqual(counts[0], 3)\n        self.assertEqual(counts[1], 3)\n\n\nclass TestIOLoopConfiguration(unittest.TestCase):\n    def run_python(self, *statements):\n        stmt_list = [\n            \"from tornado.ioloop import IOLoop\",\n            \"classname = lambda x: x.__class__.__name__\",\n        ] + list(statements)\n        args = [sys.executable, \"-c\", \"; \".join(stmt_list)]\n        return native_str(subprocess.check_output(args)).strip()\n\n    def test_default(self):\n        # When asyncio is available, it is used by default.\n        cls = self.run_python(\"print(classname(IOLoop.current()))\")\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n        cls = self.run_python(\"print(classname(IOLoop()))\")\n        self.assertEqual(cls, \"AsyncIOLoop\")\n\n    def test_asyncio(self):\n        cls = self.run_python(\n            'IOLoop.configure(\"tornado.platform.asyncio.AsyncIOLoop\")',\n            \"print(classname(IOLoop.current()))\",\n        )\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n\n    def test_asyncio_main(self):\n        cls = self.run_python(\n            \"from tornado.platform.asyncio import AsyncIOMainLoop\",\n            \"AsyncIOMainLoop().install()\",\n            \"print(classname(IOLoop.current()))\",\n        )\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n\n\nif __name__ == \"__main__\":\n    unittest.main()\n",
            "header": {
                "file": {
                    "path": "tornado/test/ioloop_test.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "73ca3254",
                "new_commit_hash": "557ebc37",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/test/ioloop_test.py"
            },
            "plus_file": {
                "path": "b/tornado/test/ioloop_test.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 432,
                            "length": 6
                        },
                        "new_range": {
                            "start": 432,
                            "length": 9
                        },
                        "section": "class TestIOLoop(AsyncTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        asyncio_loop = asyncio.new_event_loop()",
                                "type": "context"
                            },
                            {
                                "content": "        loop = IOLoop(asyncio_loop=asyncio_loop, make_current=False)",
                                "type": "context"
                            },
                            {
                                "content": "        assert loop.asyncio_loop is asyncio_loop  # type: ignore",
                                "type": "context"
                            },
                            {
                                "content": "        with self.assertRaises(RuntimeError):",
                                "type": "added"
                            },
                            {
                                "content": "            # Can't register two IOLoops with the same asyncio_loop",
                                "type": "added"
                            },
                            {
                                "content": "            IOLoop(asyncio_loop=asyncio_loop, make_current=False)",
                                "type": "added"
                            },
                            {
                                "content": "        loop.close()",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/test/ioloop_test.py",
                            "type": "class",
                            "name": "TestIOLoop",
                            "content": "class TestIOLoop(AsyncTestCase):\n    def test_add_callback_return_sequence(self):\n        # A callback returning {} or [] shouldn't spin the CPU, see Issue #1803.\n        self.calls = 0\n\n        loop = self.io_loop\n        test = self\n        old_add_callback = loop.add_callback\n\n        def add_callback(self, callback, *args, **kwargs):\n            test.calls += 1\n            old_add_callback(callback, *args, **kwargs)\n\n        loop.add_callback = types.MethodType(add_callback, loop)  # type: ignore\n        loop.add_callback(lambda: {})  # type: ignore\n        loop.add_callback(lambda: [])  # type: ignore\n        loop.add_timeout(datetime.timedelta(milliseconds=50), loop.stop)\n        loop.start()\n        self.assertLess(self.calls, 10)\n\n    @skipOnTravis\n    def test_add_callback_wakeup(self):\n        # Make sure that add_callback from inside a running IOLoop\n        # wakes up the IOLoop immediately instead of waiting for a timeout.\n        def callback():\n            self.called = True\n            self.stop()\n\n        def schedule_callback():\n            self.called = False\n            self.io_loop.add_callback(callback)\n            # Store away the time so we can check if we woke up immediately\n            self.start_time = time.time()\n\n        self.io_loop.add_timeout(self.io_loop.time(), schedule_callback)\n        self.wait()\n        self.assertAlmostEqual(time.time(), self.start_time, places=2)\n        self.assertTrue(self.called)\n\n    @skipOnTravis\n    def test_add_callback_wakeup_other_thread(self):\n        def target():\n            # sleep a bit to let the ioloop go into its poll loop\n            time.sleep(0.01)\n            self.stop_time = time.time()\n            self.io_loop.add_callback(self.stop)\n\n        thread = threading.Thread(target=target)\n        self.io_loop.add_callback(thread.start)\n        self.wait()\n        delta = time.time() - self.stop_time\n        self.assertLess(delta, 0.1)\n        thread.join()\n\n    def test_add_timeout_timedelta(self):\n        self.io_loop.add_timeout(datetime.timedelta(microseconds=1), self.stop)\n        self.wait()\n\n    def test_multiple_add(self):\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.add_handler(\n                sock.fileno(), lambda fd, events: None, IOLoop.READ\n            )\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(\n                Exception,\n                self.io_loop.add_handler,\n                sock.fileno(),\n                lambda fd, events: None,\n                IOLoop.READ,\n            )\n        finally:\n            self.io_loop.remove_handler(sock.fileno())\n            sock.close()\n\n    def test_remove_without_add(self):\n        # remove_handler should not throw an exception if called on an fd\n        # was never added.\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.remove_handler(sock.fileno())\n        finally:\n            sock.close()\n\n    def test_add_callback_from_signal(self):\n        # cheat a little bit and just run this normally, since we can't\n        # easily simulate the races that happen with real signal handlers\n        self.io_loop.add_callback_from_signal(self.stop)\n        self.wait()\n\n    def test_add_callback_from_signal_other_thread(self):\n        # Very crude test, just to make sure that we cover this case.\n        # This also happens to be the first test where we run an IOLoop in\n        # a non-main thread.\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=other_ioloop.start)\n        thread.start()\n        other_ioloop.add_callback_from_signal(other_ioloop.stop)\n        thread.join()\n        other_ioloop.close()\n\n    def test_add_callback_while_closing(self):\n        # add_callback should not fail if it races with another thread\n        # closing the IOLoop. The callbacks are dropped silently\n        # without executing.\n        closing = threading.Event()\n\n        def target():\n            other_ioloop.add_callback(other_ioloop.stop)\n            other_ioloop.start()\n            closing.set()\n            other_ioloop.close(all_fds=True)\n\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            other_ioloop.add_callback(lambda: None)\n\n    @skipIfNonUnix  # just because socketpair is so convenient\n    def test_read_while_writeable(self):\n        # Ensure that write events don't come in while we're waiting for\n        # a read and haven't asked for writeability. (the reverse is\n        # difficult to test for)\n        client, server = socket.socketpair()\n        try:\n\n            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(\n                self.io_loop.time() + 0.01, functools.partial(server.send, b\"asdf\")\n            )\n            self.wait()\n            self.io_loop.remove_handler(client.fileno())\n        finally:\n            client.close()\n            server.close()\n\n    def test_remove_timeout_after_fire(self):\n        # It is not an error to call remove_timeout after it has run.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), self.stop)\n        self.wait()\n        self.io_loop.remove_timeout(handle)\n\n    def test_remove_timeout_cleanup(self):\n        # Add and remove enough callbacks to trigger cleanup.\n        # Not a very thorough test, but it ensures that the cleanup code\n        # gets executed and doesn't blow up.  This test is only really useful\n        # on PollIOLoop subclasses, but it should run silently on any\n        # implementation.\n        for i in range(2000):\n            timeout = self.io_loop.add_timeout(self.io_loop.time() + 3600, lambda: None)\n            self.io_loop.remove_timeout(timeout)\n        # HACK: wait two IOLoop iterations for the GC to happen.\n        self.io_loop.add_callback(lambda: self.io_loop.add_callback(self.stop))\n        self.wait()\n\n    def test_remove_timeout_from_timeout(self):\n        calls = [False, False]\n\n        # Schedule several callbacks and wait for them all to come due at once.\n        # t2 should be cancelled by t1, even though it is already scheduled to\n        # be run before the ioloop even looks at it.\n        now = self.io_loop.time()\n\n        def t1():\n            calls[0] = True\n            self.io_loop.remove_timeout(t2_handle)\n\n        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\n\n        t2_handle = self.io_loop.add_timeout(now + 0.02, t2)\n        self.io_loop.add_timeout(now + 0.03, self.stop)\n        time.sleep(0.03)\n        self.wait()\n        self.assertEqual(calls, [True, False])\n\n    def test_timeout_with_arguments(self):\n        # This tests that all the timeout methods pass through *args correctly.\n        results = []  # type: List[int]\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0), results.append, 2)\n        self.io_loop.call_at(self.io_loop.time(), results.append, 3)\n        self.io_loop.call_later(0, results.append, 4)\n        self.io_loop.call_later(0, self.stop)\n        self.wait()\n        # The asyncio event loop does not guarantee the order of these\n        # callbacks.\n        self.assertEqual(sorted(results), [1, 2, 3, 4])\n\n    def test_add_timeout_return(self):\n        # All the timeout methods return non-None handles that can be\n        # passed to remove_timeout.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_at_return(self):\n        handle = self.io_loop.call_at(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_later_return(self):\n        handle = self.io_loop.call_later(0, lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_close_file_object(self):\n        \"\"\"When a file object is used instead of a numeric file descriptor,\n        the object should be closed (by IOLoop.close(all_fds=True),\n        not just the fd.\n        \"\"\"\n        # Use a socket since they are supported by IOLoop on all platforms.\n        # Unfortunately, sockets don't support the .closed attribute for\n        # inspecting their close status, so we must use a wrapper.\n        class SocketWrapper(object):\n            def __init__(self, sockobj):\n                self.sockobj = sockobj\n                self.closed = False\n\n            def fileno(self):\n                return self.sockobj.fileno()\n\n            def close(self):\n                self.closed = True\n                self.sockobj.close()\n\n        sockobj, port = bind_unused_port()\n        socket_wrapper = SocketWrapper(sockobj)\n        io_loop = IOLoop()\n        io_loop.add_handler(socket_wrapper, lambda fd, events: None, IOLoop.READ)\n        io_loop.close(all_fds=True)\n        self.assertTrue(socket_wrapper.closed)\n\n    def test_handler_callback_file_object(self):\n        \"\"\"The handler callback receives the same fd object it passed in.\"\"\"\n        server_sock, port = bind_unused_port()\n        fds = []\n\n        def handle_connection(fd, events):\n            fds.append(fd)\n            conn, addr = server_sock.accept()\n            conn.close()\n            self.stop()\n\n        self.io_loop.add_handler(server_sock, handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.io_loop.remove_handler(server_sock)\n        self.io_loop.add_handler(server_sock.fileno(), handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.assertIs(fds[0], server_sock)\n        self.assertEqual(fds[1], server_sock.fileno())\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_mixed_fd_fileobj(self):\n        server_sock, port = bind_unused_port()\n\n        def f(fd, events):\n            pass\n\n        self.io_loop.add_handler(server_sock, f, IOLoop.READ)\n        with self.assertRaises(Exception):\n            # The exact error is unspecified - some implementations use\n            # IOError, others use ValueError.\n            self.io_loop.add_handler(server_sock.fileno(), f, IOLoop.READ)\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_reentrant(self):\n        \"\"\"Calling start() twice should raise an error, not deadlock.\"\"\"\n        returned_from_start = [False]\n        got_exception = [False]\n\n        def callback():\n            try:\n                self.io_loop.start()\n                returned_from_start[0] = True\n            except Exception:\n                got_exception[0] = True\n            self.stop()\n\n        self.io_loop.add_callback(callback)\n        self.wait()\n        self.assertTrue(got_exception[0])\n        self.assertFalse(returned_from_start[0])\n\n    def test_exception_logging(self):\n        \"\"\"Uncaught exceptions get logged by the IOLoop.\"\"\"\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_future(self):\n        \"\"\"The IOLoop examines exceptions from Futures and logs them.\"\"\"\n\n        @gen.coroutine\n        def callback():\n            self.io_loop.add_callback(self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n\n        async def callback():\n            # Stop the IOLoop two iterations after raising an exception\n            # to give the exception time to be logged.\n            self.io_loop.add_callback(self.io_loop.add_callback, self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_spawn_callback(self):\n        # Both add_callback and spawn_callback run directly on the IOLoop,\n        # so their errors are logged without stopping the test.\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n        # A spawned callback is run directly on the IOLoop, so it will be\n        # logged without stopping the test.\n        self.io_loop.spawn_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    @skipIfNonUnix\n    def test_remove_handler_from_handler(self):\n        # Create two sockets with simultaneous read events.\n        client, server = socket.socketpair()\n        try:\n            client.send(b\"abc\")\n            server.send(b\"abc\")\n\n            # After reading from one fd, remove the other from the IOLoop.\n            chunks = []\n\n            def handle_read(fd, events):\n                chunks.append(fd.recv(1024))\n                if fd is client:\n                    self.io_loop.remove_handler(server)\n                else:\n                    self.io_loop.remove_handler(client)\n\n            self.io_loop.add_handler(client, handle_read, self.io_loop.READ)\n            self.io_loop.add_handler(server, handle_read, self.io_loop.READ)\n            self.io_loop.call_later(0.1, self.stop)\n            self.wait()\n\n            # Only one fd was read; the other was cleanly removed.\n            self.assertEqual(chunks, [b\"abc\"])\n        finally:\n            client.close()\n            server.close()\n\n    @skipIfNonUnix\n    @gen_test\n    def test_init_close_race(self):\n        # Regression test for #2367\n        #\n        # Skipped on windows because of what looks like a bug in the\n        # proactor event loop when started and stopped on non-main\n        # threads.\n        def f():\n            for i in range(10):\n                loop = IOLoop(make_current=False)\n                loop.close()\n\n        yield gen.multi([self.io_loop.run_in_executor(None, f) for i in range(2)])\n\n    def test_explicit_asyncio_loop(self):\n        asyncio_loop = asyncio.new_event_loop()\n        loop = IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        assert loop.asyncio_loop is asyncio_loop  # type: ignore\n        with self.assertRaises(RuntimeError):\n            # Can't register two IOLoops with the same asyncio_loop\n            IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        loop.close()",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 41,
                            "end_lineno": 438,
                            "parent": null
                        },
                        {
                            "file_name": "tornado/test/ioloop_test.py",
                            "type": "method",
                            "name": "TestIOLoop.test_explicit_asyncio_loop",
                            "content": "class TestIOLoop(AsyncTestCase):\n    def test_add_callback_return_sequence(self):\n        # A callback returning {} or [] shouldn't spin the CPU, see Issue #1803.\n        self.calls = 0\n\n        loop = self.io_loop\n        test = self\n        old_add_callback = loop.add_callback\n\n        def add_callback(self, callback, *args, **kwargs):\n            test.calls += 1\n            old_add_callback(callback, *args, **kwargs)\n\n        loop.add_callback = types.MethodType(add_callback, loop)  # type: ignore\n        loop.add_callback(lambda: {})  # type: ignore\n        loop.add_callback(lambda: [])  # type: ignore\n        loop.add_timeout(datetime.timedelta(milliseconds=50), loop.stop)\n        loop.start()\n        self.assertLess(self.calls, 10)\n\n    @skipOnTravis\n    def test_add_callback_wakeup(self):\n        # Make sure that add_callback from inside a running IOLoop\n        # wakes up the IOLoop immediately instead of waiting for a timeout.\n        def callback():\n            self.called = True\n            self.stop()\n\n        def schedule_callback():\n            self.called = False\n            self.io_loop.add_callback(callback)\n            # Store away the time so we can check if we woke up immediately\n            self.start_time = time.time()\n\n        self.io_loop.add_timeout(self.io_loop.time(), schedule_callback)\n        self.wait()\n        self.assertAlmostEqual(time.time(), self.start_time, places=2)\n        self.assertTrue(self.called)\n\n    @skipOnTravis\n    def test_add_callback_wakeup_other_thread(self):\n        def target():\n            # sleep a bit to let the ioloop go into its poll loop\n            time.sleep(0.01)\n            self.stop_time = time.time()\n            self.io_loop.add_callback(self.stop)\n\n        thread = threading.Thread(target=target)\n        self.io_loop.add_callback(thread.start)\n        self.wait()\n        delta = time.time() - self.stop_time\n        self.assertLess(delta, 0.1)\n        thread.join()\n\n    def test_add_timeout_timedelta(self):\n        self.io_loop.add_timeout(datetime.timedelta(microseconds=1), self.stop)\n        self.wait()\n\n    def test_multiple_add(self):\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.add_handler(\n                sock.fileno(), lambda fd, events: None, IOLoop.READ\n            )\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(\n                Exception,\n                self.io_loop.add_handler,\n                sock.fileno(),\n                lambda fd, events: None,\n                IOLoop.READ,\n            )\n        finally:\n            self.io_loop.remove_handler(sock.fileno())\n            sock.close()\n\n    def test_remove_without_add(self):\n        # remove_handler should not throw an exception if called on an fd\n        # was never added.\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.remove_handler(sock.fileno())\n        finally:\n            sock.close()\n\n    def test_add_callback_from_signal(self):\n        # cheat a little bit and just run this normally, since we can't\n        # easily simulate the races that happen with real signal handlers\n        self.io_loop.add_callback_from_signal(self.stop)\n        self.wait()\n\n    def test_add_callback_from_signal_other_thread(self):\n        # Very crude test, just to make sure that we cover this case.\n        # This also happens to be the first test where we run an IOLoop in\n        # a non-main thread.\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=other_ioloop.start)\n        thread.start()\n        other_ioloop.add_callback_from_signal(other_ioloop.stop)\n        thread.join()\n        other_ioloop.close()\n\n    def test_add_callback_while_closing(self):\n        # add_callback should not fail if it races with another thread\n        # closing the IOLoop. The callbacks are dropped silently\n        # without executing.\n        closing = threading.Event()\n\n        def target():\n            other_ioloop.add_callback(other_ioloop.stop)\n            other_ioloop.start()\n            closing.set()\n            other_ioloop.close(all_fds=True)\n\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            other_ioloop.add_callback(lambda: None)\n\n    @skipIfNonUnix  # just because socketpair is so convenient\n    def test_read_while_writeable(self):\n        # Ensure that write events don't come in while we're waiting for\n        # a read and haven't asked for writeability. (the reverse is\n        # difficult to test for)\n        client, server = socket.socketpair()\n        try:\n\n            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(\n                self.io_loop.time() + 0.01, functools.partial(server.send, b\"asdf\")\n            )\n            self.wait()\n            self.io_loop.remove_handler(client.fileno())\n        finally:\n            client.close()\n            server.close()\n\n    def test_remove_timeout_after_fire(self):\n        # It is not an error to call remove_timeout after it has run.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), self.stop)\n        self.wait()\n        self.io_loop.remove_timeout(handle)\n\n    def test_remove_timeout_cleanup(self):\n        # Add and remove enough callbacks to trigger cleanup.\n        # Not a very thorough test, but it ensures that the cleanup code\n        # gets executed and doesn't blow up.  This test is only really useful\n        # on PollIOLoop subclasses, but it should run silently on any\n        # implementation.\n        for i in range(2000):\n            timeout = self.io_loop.add_timeout(self.io_loop.time() + 3600, lambda: None)\n            self.io_loop.remove_timeout(timeout)\n        # HACK: wait two IOLoop iterations for the GC to happen.\n        self.io_loop.add_callback(lambda: self.io_loop.add_callback(self.stop))\n        self.wait()\n\n    def test_remove_timeout_from_timeout(self):\n        calls = [False, False]\n\n        # Schedule several callbacks and wait for them all to come due at once.\n        # t2 should be cancelled by t1, even though it is already scheduled to\n        # be run before the ioloop even looks at it.\n        now = self.io_loop.time()\n\n        def t1():\n            calls[0] = True\n            self.io_loop.remove_timeout(t2_handle)\n\n        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\n\n        t2_handle = self.io_loop.add_timeout(now + 0.02, t2)\n        self.io_loop.add_timeout(now + 0.03, self.stop)\n        time.sleep(0.03)\n        self.wait()\n        self.assertEqual(calls, [True, False])\n\n    def test_timeout_with_arguments(self):\n        # This tests that all the timeout methods pass through *args correctly.\n        results = []  # type: List[int]\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0), results.append, 2)\n        self.io_loop.call_at(self.io_loop.time(), results.append, 3)\n        self.io_loop.call_later(0, results.append, 4)\n        self.io_loop.call_later(0, self.stop)\n        self.wait()\n        # The asyncio event loop does not guarantee the order of these\n        # callbacks.\n        self.assertEqual(sorted(results), [1, 2, 3, 4])\n\n    def test_add_timeout_return(self):\n        # All the timeout methods return non-None handles that can be\n        # passed to remove_timeout.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_at_return(self):\n        handle = self.io_loop.call_at(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_later_return(self):\n        handle = self.io_loop.call_later(0, lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_close_file_object(self):\n        \"\"\"When a file object is used instead of a numeric file descriptor,\n        the object should be closed (by IOLoop.close(all_fds=True),\n        not just the fd.\n        \"\"\"\n        # Use a socket since they are supported by IOLoop on all platforms.\n        # Unfortunately, sockets don't support the .closed attribute for\n        # inspecting their close status, so we must use a wrapper.\n        class SocketWrapper(object):\n            def __init__(self, sockobj):\n                self.sockobj = sockobj\n                self.closed = False\n\n            def fileno(self):\n                return self.sockobj.fileno()\n\n            def close(self):\n                self.closed = True\n                self.sockobj.close()\n\n        sockobj, port = bind_unused_port()\n        socket_wrapper = SocketWrapper(sockobj)\n        io_loop = IOLoop()\n        io_loop.add_handler(socket_wrapper, lambda fd, events: None, IOLoop.READ)\n        io_loop.close(all_fds=True)\n        self.assertTrue(socket_wrapper.closed)\n\n    def test_handler_callback_file_object(self):\n        \"\"\"The handler callback receives the same fd object it passed in.\"\"\"\n        server_sock, port = bind_unused_port()\n        fds = []\n\n        def handle_connection(fd, events):\n            fds.append(fd)\n            conn, addr = server_sock.accept()\n            conn.close()\n            self.stop()\n\n        self.io_loop.add_handler(server_sock, handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.io_loop.remove_handler(server_sock)\n        self.io_loop.add_handler(server_sock.fileno(), handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.assertIs(fds[0], server_sock)\n        self.assertEqual(fds[1], server_sock.fileno())\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_mixed_fd_fileobj(self):\n        server_sock, port = bind_unused_port()\n\n        def f(fd, events):\n            pass\n\n        self.io_loop.add_handler(server_sock, f, IOLoop.READ)\n        with self.assertRaises(Exception):\n            # The exact error is unspecified - some implementations use\n            # IOError, others use ValueError.\n            self.io_loop.add_handler(server_sock.fileno(), f, IOLoop.READ)\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_reentrant(self):\n        \"\"\"Calling start() twice should raise an error, not deadlock.\"\"\"\n        returned_from_start = [False]\n        got_exception = [False]\n\n        def callback():\n            try:\n                self.io_loop.start()\n                returned_from_start[0] = True\n            except Exception:\n                got_exception[0] = True\n            self.stop()\n\n        self.io_loop.add_callback(callback)\n        self.wait()\n        self.assertTrue(got_exception[0])\n        self.assertFalse(returned_from_start[0])\n\n    def test_exception_logging(self):\n        \"\"\"Uncaught exceptions get logged by the IOLoop.\"\"\"\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_future(self):\n        \"\"\"The IOLoop examines exceptions from Futures and logs them.\"\"\"\n\n        @gen.coroutine\n        def callback():\n            self.io_loop.add_callback(self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n\n        async def callback():\n            # Stop the IOLoop two iterations after raising an exception\n            # to give the exception time to be logged.\n            self.io_loop.add_callback(self.io_loop.add_callback, self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_spawn_callback(self):\n        # Both add_callback and spawn_callback run directly on the IOLoop,\n        # so their errors are logged without stopping the test.\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n        # A spawned callback is run directly on the IOLoop, so it will be\n        # logged without stopping the test.\n        self.io_loop.spawn_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    @skipIfNonUnix\n    def test_remove_handler_from_handler(self):\n        # Create two sockets with simultaneous read events.\n        client, server = socket.socketpair()\n        try:\n            client.send(b\"abc\")\n            server.send(b\"abc\")\n\n            # After reading from one fd, remove the other from the IOLoop.\n            chunks = []\n\n            def handle_read(fd, events):\n                chunks.append(fd.recv(1024))\n                if fd is client:\n                    self.io_loop.remove_handler(server)\n                else:\n                    self.io_loop.remove_handler(client)\n\n            self.io_loop.add_handler(client, handle_read, self.io_loop.READ)\n            self.io_loop.add_handler(server, handle_read, self.io_loop.READ)\n            self.io_loop.call_later(0.1, self.stop)\n            self.wait()\n\n            # Only one fd was read; the other was cleanly removed.\n            self.assertEqual(chunks, [b\"abc\"])\n        finally:\n            client.close()\n            server.close()\n\n    @skipIfNonUnix\n    @gen_test\n    def test_init_close_race(self):\n        # Regression test for #2367\n        #\n        # Skipped on windows because of what looks like a bug in the\n        # proactor event loop when started and stopped on non-main\n        # threads.\n        def f():\n            for i in range(10):\n                loop = IOLoop(make_current=False)\n                loop.close()\n\n        yield gen.multi([self.io_loop.run_in_executor(None, f) for i in range(2)])\n\n    def test_explicit_asyncio_loop(self):\n        asyncio_loop = asyncio.new_event_loop()\n        loop = IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        assert loop.asyncio_loop is asyncio_loop  # type: ignore\n        with self.assertRaises(RuntimeError):\n            # Can't register two IOLoops with the same asyncio_loop\n            IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        loop.close()",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 431,
                            "end_lineno": 438,
                            "parent": {
                                "file_name": "tornado/test/ioloop_test.py",
                                "type": "class",
                                "name": "TestIOLoop",
                                "content": "class TestIOLoop(AsyncTestCase):\n    def test_add_callback_return_sequence(self):\n        # A callback returning {} or [] shouldn't spin the CPU, see Issue #1803.\n        self.calls = 0\n\n        loop = self.io_loop\n        test = self\n        old_add_callback = loop.add_callback\n\n        def add_callback(self, callback, *args, **kwargs):\n            test.calls += 1\n            old_add_callback(callback, *args, **kwargs)\n\n        loop.add_callback = types.MethodType(add_callback, loop)  # type: ignore\n        loop.add_callback(lambda: {})  # type: ignore\n        loop.add_callback(lambda: [])  # type: ignore\n        loop.add_timeout(datetime.timedelta(milliseconds=50), loop.stop)\n        loop.start()\n        self.assertLess(self.calls, 10)\n\n    @skipOnTravis\n    def test_add_callback_wakeup(self):\n        # Make sure that add_callback from inside a running IOLoop\n        # wakes up the IOLoop immediately instead of waiting for a timeout.\n        def callback():\n            self.called = True\n            self.stop()\n\n        def schedule_callback():\n            self.called = False\n            self.io_loop.add_callback(callback)\n            # Store away the time so we can check if we woke up immediately\n            self.start_time = time.time()\n\n        self.io_loop.add_timeout(self.io_loop.time(), schedule_callback)\n        self.wait()\n        self.assertAlmostEqual(time.time(), self.start_time, places=2)\n        self.assertTrue(self.called)\n\n    @skipOnTravis\n    def test_add_callback_wakeup_other_thread(self):\n        def target():\n            # sleep a bit to let the ioloop go into its poll loop\n            time.sleep(0.01)\n            self.stop_time = time.time()\n            self.io_loop.add_callback(self.stop)\n\n        thread = threading.Thread(target=target)\n        self.io_loop.add_callback(thread.start)\n        self.wait()\n        delta = time.time() - self.stop_time\n        self.assertLess(delta, 0.1)\n        thread.join()\n\n    def test_add_timeout_timedelta(self):\n        self.io_loop.add_timeout(datetime.timedelta(microseconds=1), self.stop)\n        self.wait()\n\n    def test_multiple_add(self):\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.add_handler(\n                sock.fileno(), lambda fd, events: None, IOLoop.READ\n            )\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(\n                Exception,\n                self.io_loop.add_handler,\n                sock.fileno(),\n                lambda fd, events: None,\n                IOLoop.READ,\n            )\n        finally:\n            self.io_loop.remove_handler(sock.fileno())\n            sock.close()\n\n    def test_remove_without_add(self):\n        # remove_handler should not throw an exception if called on an fd\n        # was never added.\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.remove_handler(sock.fileno())\n        finally:\n            sock.close()\n\n    def test_add_callback_from_signal(self):\n        # cheat a little bit and just run this normally, since we can't\n        # easily simulate the races that happen with real signal handlers\n        self.io_loop.add_callback_from_signal(self.stop)\n        self.wait()\n\n    def test_add_callback_from_signal_other_thread(self):\n        # Very crude test, just to make sure that we cover this case.\n        # This also happens to be the first test where we run an IOLoop in\n        # a non-main thread.\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=other_ioloop.start)\n        thread.start()\n        other_ioloop.add_callback_from_signal(other_ioloop.stop)\n        thread.join()\n        other_ioloop.close()\n\n    def test_add_callback_while_closing(self):\n        # add_callback should not fail if it races with another thread\n        # closing the IOLoop. The callbacks are dropped silently\n        # without executing.\n        closing = threading.Event()\n\n        def target():\n            other_ioloop.add_callback(other_ioloop.stop)\n            other_ioloop.start()\n            closing.set()\n            other_ioloop.close(all_fds=True)\n\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            other_ioloop.add_callback(lambda: None)\n\n    @skipIfNonUnix  # just because socketpair is so convenient\n    def test_read_while_writeable(self):\n        # Ensure that write events don't come in while we're waiting for\n        # a read and haven't asked for writeability. (the reverse is\n        # difficult to test for)\n        client, server = socket.socketpair()\n        try:\n\n            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(\n                self.io_loop.time() + 0.01, functools.partial(server.send, b\"asdf\")\n            )\n            self.wait()\n            self.io_loop.remove_handler(client.fileno())\n        finally:\n            client.close()\n            server.close()\n\n    def test_remove_timeout_after_fire(self):\n        # It is not an error to call remove_timeout after it has run.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), self.stop)\n        self.wait()\n        self.io_loop.remove_timeout(handle)\n\n    def test_remove_timeout_cleanup(self):\n        # Add and remove enough callbacks to trigger cleanup.\n        # Not a very thorough test, but it ensures that the cleanup code\n        # gets executed and doesn't blow up.  This test is only really useful\n        # on PollIOLoop subclasses, but it should run silently on any\n        # implementation.\n        for i in range(2000):\n            timeout = self.io_loop.add_timeout(self.io_loop.time() + 3600, lambda: None)\n            self.io_loop.remove_timeout(timeout)\n        # HACK: wait two IOLoop iterations for the GC to happen.\n        self.io_loop.add_callback(lambda: self.io_loop.add_callback(self.stop))\n        self.wait()\n\n    def test_remove_timeout_from_timeout(self):\n        calls = [False, False]\n\n        # Schedule several callbacks and wait for them all to come due at once.\n        # t2 should be cancelled by t1, even though it is already scheduled to\n        # be run before the ioloop even looks at it.\n        now = self.io_loop.time()\n\n        def t1():\n            calls[0] = True\n            self.io_loop.remove_timeout(t2_handle)\n\n        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\n\n        t2_handle = self.io_loop.add_timeout(now + 0.02, t2)\n        self.io_loop.add_timeout(now + 0.03, self.stop)\n        time.sleep(0.03)\n        self.wait()\n        self.assertEqual(calls, [True, False])\n\n    def test_timeout_with_arguments(self):\n        # This tests that all the timeout methods pass through *args correctly.\n        results = []  # type: List[int]\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0), results.append, 2)\n        self.io_loop.call_at(self.io_loop.time(), results.append, 3)\n        self.io_loop.call_later(0, results.append, 4)\n        self.io_loop.call_later(0, self.stop)\n        self.wait()\n        # The asyncio event loop does not guarantee the order of these\n        # callbacks.\n        self.assertEqual(sorted(results), [1, 2, 3, 4])\n\n    def test_add_timeout_return(self):\n        # All the timeout methods return non-None handles that can be\n        # passed to remove_timeout.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_at_return(self):\n        handle = self.io_loop.call_at(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_later_return(self):\n        handle = self.io_loop.call_later(0, lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_close_file_object(self):\n        \"\"\"When a file object is used instead of a numeric file descriptor,\n        the object should be closed (by IOLoop.close(all_fds=True),\n        not just the fd.\n        \"\"\"\n        # Use a socket since they are supported by IOLoop on all platforms.\n        # Unfortunately, sockets don't support the .closed attribute for\n        # inspecting their close status, so we must use a wrapper.\n        class SocketWrapper(object):\n            def __init__(self, sockobj):\n                self.sockobj = sockobj\n                self.closed = False\n\n            def fileno(self):\n                return self.sockobj.fileno()\n\n            def close(self):\n                self.closed = True\n                self.sockobj.close()\n\n        sockobj, port = bind_unused_port()\n        socket_wrapper = SocketWrapper(sockobj)\n        io_loop = IOLoop()\n        io_loop.add_handler(socket_wrapper, lambda fd, events: None, IOLoop.READ)\n        io_loop.close(all_fds=True)\n        self.assertTrue(socket_wrapper.closed)\n\n    def test_handler_callback_file_object(self):\n        \"\"\"The handler callback receives the same fd object it passed in.\"\"\"\n        server_sock, port = bind_unused_port()\n        fds = []\n\n        def handle_connection(fd, events):\n            fds.append(fd)\n            conn, addr = server_sock.accept()\n            conn.close()\n            self.stop()\n\n        self.io_loop.add_handler(server_sock, handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.io_loop.remove_handler(server_sock)\n        self.io_loop.add_handler(server_sock.fileno(), handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.assertIs(fds[0], server_sock)\n        self.assertEqual(fds[1], server_sock.fileno())\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_mixed_fd_fileobj(self):\n        server_sock, port = bind_unused_port()\n\n        def f(fd, events):\n            pass\n\n        self.io_loop.add_handler(server_sock, f, IOLoop.READ)\n        with self.assertRaises(Exception):\n            # The exact error is unspecified - some implementations use\n            # IOError, others use ValueError.\n            self.io_loop.add_handler(server_sock.fileno(), f, IOLoop.READ)\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_reentrant(self):\n        \"\"\"Calling start() twice should raise an error, not deadlock.\"\"\"\n        returned_from_start = [False]\n        got_exception = [False]\n\n        def callback():\n            try:\n                self.io_loop.start()\n                returned_from_start[0] = True\n            except Exception:\n                got_exception[0] = True\n            self.stop()\n\n        self.io_loop.add_callback(callback)\n        self.wait()\n        self.assertTrue(got_exception[0])\n        self.assertFalse(returned_from_start[0])\n\n    def test_exception_logging(self):\n        \"\"\"Uncaught exceptions get logged by the IOLoop.\"\"\"\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_future(self):\n        \"\"\"The IOLoop examines exceptions from Futures and logs them.\"\"\"\n\n        @gen.coroutine\n        def callback():\n            self.io_loop.add_callback(self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n\n        async def callback():\n            # Stop the IOLoop two iterations after raising an exception\n            # to give the exception time to be logged.\n            self.io_loop.add_callback(self.io_loop.add_callback, self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_spawn_callback(self):\n        # Both add_callback and spawn_callback run directly on the IOLoop,\n        # so their errors are logged without stopping the test.\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n        # A spawned callback is run directly on the IOLoop, so it will be\n        # logged without stopping the test.\n        self.io_loop.spawn_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    @skipIfNonUnix\n    def test_remove_handler_from_handler(self):\n        # Create two sockets with simultaneous read events.\n        client, server = socket.socketpair()\n        try:\n            client.send(b\"abc\")\n            server.send(b\"abc\")\n\n            # After reading from one fd, remove the other from the IOLoop.\n            chunks = []\n\n            def handle_read(fd, events):\n                chunks.append(fd.recv(1024))\n                if fd is client:\n                    self.io_loop.remove_handler(server)\n                else:\n                    self.io_loop.remove_handler(client)\n\n            self.io_loop.add_handler(client, handle_read, self.io_loop.READ)\n            self.io_loop.add_handler(server, handle_read, self.io_loop.READ)\n            self.io_loop.call_later(0.1, self.stop)\n            self.wait()\n\n            # Only one fd was read; the other was cleanly removed.\n            self.assertEqual(chunks, [b\"abc\"])\n        finally:\n            client.close()\n            server.close()\n\n    @skipIfNonUnix\n    @gen_test\n    def test_init_close_race(self):\n        # Regression test for #2367\n        #\n        # Skipped on windows because of what looks like a bug in the\n        # proactor event loop when started and stopped on non-main\n        # threads.\n        def f():\n            for i in range(10):\n                loop = IOLoop(make_current=False)\n                loop.close()\n\n        yield gen.multi([self.io_loop.run_in_executor(None, f) for i in range(2)])\n\n    def test_explicit_asyncio_loop(self):\n        asyncio_loop = asyncio.new_event_loop()\n        loop = IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        assert loop.asyncio_loop is asyncio_loop  # type: ignore\n        with self.assertRaises(RuntimeError):\n            # Can't register two IOLoops with the same asyncio_loop\n            IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        loop.close()",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 41,
                                "end_lineno": 438,
                                "parent": null
                            }
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "44e28cf32ca120189026b55395587415e731f905^",
    "new_commit_hash": "44e28cf32ca120189026b55395587415e731f905",
    "commit_message": "Prevent multiple IOLoops for one asyncio loop",
    "commit_date": "2022-06-16T10:20:39+02:00",
    "metadata": {}
}
{
    "repo_name": "tornado",
    "new_commit_hash": "44e28cf32ca120189026b55395587415e731f905",
    "test_file_codes": [
        "import asyncio\nfrom concurrent.futures import ThreadPoolExecutor\nfrom concurrent import futures\nfrom collections.abc import Generator\nimport contextlib\nimport datetime\nimport functools\nimport socket\nimport subprocess\nimport sys\nimport threading\nimport time\nimport types\nfrom unittest import mock\nimport unittest\n\nfrom tornado.escape import native_str\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop, TimeoutError, PeriodicCallback\nfrom tornado.log import app_log\nfrom tornado.testing import (\n    AsyncTestCase,\n    bind_unused_port,\n    ExpectLog,\n    gen_test,\n    setup_with_context_manager,\n)\nfrom tornado.test.util import (\n    ignore_deprecation,\n    skipIfNonUnix,\n    skipOnTravis,\n)\nfrom tornado.concurrent import Future\n\nimport typing\n\nif typing.TYPE_CHECKING:\n    from typing import List  # noqa: F401\n\n\nclass TestIOLoop(AsyncTestCase):\n    def test_add_callback_return_sequence(self):\n        # A callback returning {} or [] shouldn't spin the CPU, see Issue #1803.\n        self.calls = 0\n\n        loop = self.io_loop\n        test = self\n        old_add_callback = loop.add_callback\n\n        def add_callback(self, callback, *args, **kwargs):\n            test.calls += 1\n            old_add_callback(callback, *args, **kwargs)\n\n        loop.add_callback = types.MethodType(add_callback, loop)  # type: ignore\n        loop.add_callback(lambda: {})  # type: ignore\n        loop.add_callback(lambda: [])  # type: ignore\n        loop.add_timeout(datetime.timedelta(milliseconds=50), loop.stop)\n        loop.start()\n        self.assertLess(self.calls, 10)\n\n    @skipOnTravis\n    def test_add_callback_wakeup(self):\n        # Make sure that add_callback from inside a running IOLoop\n        # wakes up the IOLoop immediately instead of waiting for a timeout.\n        def callback():\n            self.called = True\n            self.stop()\n\n        def schedule_callback():\n            self.called = False\n            self.io_loop.add_callback(callback)\n            # Store away the time so we can check if we woke up immediately\n            self.start_time = time.time()\n\n        self.io_loop.add_timeout(self.io_loop.time(), schedule_callback)\n        self.wait()\n        self.assertAlmostEqual(time.time(), self.start_time, places=2)\n        self.assertTrue(self.called)\n\n    @skipOnTravis\n    def test_add_callback_wakeup_other_thread(self):\n        def target():\n            # sleep a bit to let the ioloop go into its poll loop\n            time.sleep(0.01)\n            self.stop_time = time.time()\n            self.io_loop.add_callback(self.stop)\n\n        thread = threading.Thread(target=target)\n        self.io_loop.add_callback(thread.start)\n        self.wait()\n        delta = time.time() - self.stop_time\n        self.assertLess(delta, 0.1)\n        thread.join()\n\n    def test_add_timeout_timedelta(self):\n        self.io_loop.add_timeout(datetime.timedelta(microseconds=1), self.stop)\n        self.wait()\n\n    def test_multiple_add(self):\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.add_handler(\n                sock.fileno(), lambda fd, events: None, IOLoop.READ\n            )\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(\n                Exception,\n                self.io_loop.add_handler,\n                sock.fileno(),\n                lambda fd, events: None,\n                IOLoop.READ,\n            )\n        finally:\n            self.io_loop.remove_handler(sock.fileno())\n            sock.close()\n\n    def test_remove_without_add(self):\n        # remove_handler should not throw an exception if called on an fd\n        # was never added.\n        sock, port = bind_unused_port()\n        try:\n            self.io_loop.remove_handler(sock.fileno())\n        finally:\n            sock.close()\n\n    def test_add_callback_from_signal(self):\n        # cheat a little bit and just run this normally, since we can't\n        # easily simulate the races that happen with real signal handlers\n        self.io_loop.add_callback_from_signal(self.stop)\n        self.wait()\n\n    def test_add_callback_from_signal_other_thread(self):\n        # Very crude test, just to make sure that we cover this case.\n        # This also happens to be the first test where we run an IOLoop in\n        # a non-main thread.\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=other_ioloop.start)\n        thread.start()\n        other_ioloop.add_callback_from_signal(other_ioloop.stop)\n        thread.join()\n        other_ioloop.close()\n\n    def test_add_callback_while_closing(self):\n        # add_callback should not fail if it races with another thread\n        # closing the IOLoop. The callbacks are dropped silently\n        # without executing.\n        closing = threading.Event()\n\n        def target():\n            other_ioloop.add_callback(other_ioloop.stop)\n            other_ioloop.start()\n            closing.set()\n            other_ioloop.close(all_fds=True)\n\n        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            other_ioloop.add_callback(lambda: None)\n\n    @skipIfNonUnix  # just because socketpair is so convenient\n    def test_read_while_writeable(self):\n        # Ensure that write events don't come in while we're waiting for\n        # a read and haven't asked for writeability. (the reverse is\n        # difficult to test for)\n        client, server = socket.socketpair()\n        try:\n\n            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(\n                self.io_loop.time() + 0.01, functools.partial(server.send, b\"asdf\")\n            )\n            self.wait()\n            self.io_loop.remove_handler(client.fileno())\n        finally:\n            client.close()\n            server.close()\n\n    def test_remove_timeout_after_fire(self):\n        # It is not an error to call remove_timeout after it has run.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), self.stop)\n        self.wait()\n        self.io_loop.remove_timeout(handle)\n\n    def test_remove_timeout_cleanup(self):\n        # Add and remove enough callbacks to trigger cleanup.\n        # Not a very thorough test, but it ensures that the cleanup code\n        # gets executed and doesn't blow up.  This test is only really useful\n        # on PollIOLoop subclasses, but it should run silently on any\n        # implementation.\n        for i in range(2000):\n            timeout = self.io_loop.add_timeout(self.io_loop.time() + 3600, lambda: None)\n            self.io_loop.remove_timeout(timeout)\n        # HACK: wait two IOLoop iterations for the GC to happen.\n        self.io_loop.add_callback(lambda: self.io_loop.add_callback(self.stop))\n        self.wait()\n\n    def test_remove_timeout_from_timeout(self):\n        calls = [False, False]\n\n        # Schedule several callbacks and wait for them all to come due at once.\n        # t2 should be cancelled by t1, even though it is already scheduled to\n        # be run before the ioloop even looks at it.\n        now = self.io_loop.time()\n\n        def t1():\n            calls[0] = True\n            self.io_loop.remove_timeout(t2_handle)\n\n        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\n\n        t2_handle = self.io_loop.add_timeout(now + 0.02, t2)\n        self.io_loop.add_timeout(now + 0.03, self.stop)\n        time.sleep(0.03)\n        self.wait()\n        self.assertEqual(calls, [True, False])\n\n    def test_timeout_with_arguments(self):\n        # This tests that all the timeout methods pass through *args correctly.\n        results = []  # type: List[int]\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0), results.append, 2)\n        self.io_loop.call_at(self.io_loop.time(), results.append, 3)\n        self.io_loop.call_later(0, results.append, 4)\n        self.io_loop.call_later(0, self.stop)\n        self.wait()\n        # The asyncio event loop does not guarantee the order of these\n        # callbacks.\n        self.assertEqual(sorted(results), [1, 2, 3, 4])\n\n    def test_add_timeout_return(self):\n        # All the timeout methods return non-None handles that can be\n        # passed to remove_timeout.\n        handle = self.io_loop.add_timeout(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_at_return(self):\n        handle = self.io_loop.call_at(self.io_loop.time(), lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_call_later_return(self):\n        handle = self.io_loop.call_later(0, lambda: None)\n        self.assertFalse(handle is None)\n        self.io_loop.remove_timeout(handle)\n\n    def test_close_file_object(self):\n        \"\"\"When a file object is used instead of a numeric file descriptor,\n        the object should be closed (by IOLoop.close(all_fds=True),\n        not just the fd.\n        \"\"\"\n        # Use a socket since they are supported by IOLoop on all platforms.\n        # Unfortunately, sockets don't support the .closed attribute for\n        # inspecting their close status, so we must use a wrapper.\n        class SocketWrapper(object):\n            def __init__(self, sockobj):\n                self.sockobj = sockobj\n                self.closed = False\n\n            def fileno(self):\n                return self.sockobj.fileno()\n\n            def close(self):\n                self.closed = True\n                self.sockobj.close()\n\n        sockobj, port = bind_unused_port()\n        socket_wrapper = SocketWrapper(sockobj)\n        io_loop = IOLoop()\n        io_loop.add_handler(socket_wrapper, lambda fd, events: None, IOLoop.READ)\n        io_loop.close(all_fds=True)\n        self.assertTrue(socket_wrapper.closed)\n\n    def test_handler_callback_file_object(self):\n        \"\"\"The handler callback receives the same fd object it passed in.\"\"\"\n        server_sock, port = bind_unused_port()\n        fds = []\n\n        def handle_connection(fd, events):\n            fds.append(fd)\n            conn, addr = server_sock.accept()\n            conn.close()\n            self.stop()\n\n        self.io_loop.add_handler(server_sock, handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.io_loop.remove_handler(server_sock)\n        self.io_loop.add_handler(server_sock.fileno(), handle_connection, IOLoop.READ)\n        with contextlib.closing(socket.socket()) as client_sock:\n            client_sock.connect((\"127.0.0.1\", port))\n            self.wait()\n        self.assertIs(fds[0], server_sock)\n        self.assertEqual(fds[1], server_sock.fileno())\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_mixed_fd_fileobj(self):\n        server_sock, port = bind_unused_port()\n\n        def f(fd, events):\n            pass\n\n        self.io_loop.add_handler(server_sock, f, IOLoop.READ)\n        with self.assertRaises(Exception):\n            # The exact error is unspecified - some implementations use\n            # IOError, others use ValueError.\n            self.io_loop.add_handler(server_sock.fileno(), f, IOLoop.READ)\n        self.io_loop.remove_handler(server_sock.fileno())\n        server_sock.close()\n\n    def test_reentrant(self):\n        \"\"\"Calling start() twice should raise an error, not deadlock.\"\"\"\n        returned_from_start = [False]\n        got_exception = [False]\n\n        def callback():\n            try:\n                self.io_loop.start()\n                returned_from_start[0] = True\n            except Exception:\n                got_exception[0] = True\n            self.stop()\n\n        self.io_loop.add_callback(callback)\n        self.wait()\n        self.assertTrue(got_exception[0])\n        self.assertFalse(returned_from_start[0])\n\n    def test_exception_logging(self):\n        \"\"\"Uncaught exceptions get logged by the IOLoop.\"\"\"\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_future(self):\n        \"\"\"The IOLoop examines exceptions from Futures and logs them.\"\"\"\n\n        @gen.coroutine\n        def callback():\n            self.io_loop.add_callback(self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n\n        async def callback():\n            # Stop the IOLoop two iterations after raising an exception\n            # to give the exception time to be logged.\n            self.io_loop.add_callback(self.io_loop.add_callback, self.stop)\n            1 / 0\n\n        self.io_loop.add_callback(callback)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    def test_spawn_callback(self):\n        # Both add_callback and spawn_callback run directly on the IOLoop,\n        # so their errors are logged without stopping the test.\n        self.io_loop.add_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n        # A spawned callback is run directly on the IOLoop, so it will be\n        # logged without stopping the test.\n        self.io_loop.spawn_callback(lambda: 1 / 0)\n        self.io_loop.add_callback(self.stop)\n        with ExpectLog(app_log, \"Exception in callback\"):\n            self.wait()\n\n    @skipIfNonUnix\n    def test_remove_handler_from_handler(self):\n        # Create two sockets with simultaneous read events.\n        client, server = socket.socketpair()\n        try:\n            client.send(b\"abc\")\n            server.send(b\"abc\")\n\n            # After reading from one fd, remove the other from the IOLoop.\n            chunks = []\n\n            def handle_read(fd, events):\n                chunks.append(fd.recv(1024))\n                if fd is client:\n                    self.io_loop.remove_handler(server)\n                else:\n                    self.io_loop.remove_handler(client)\n\n            self.io_loop.add_handler(client, handle_read, self.io_loop.READ)\n            self.io_loop.add_handler(server, handle_read, self.io_loop.READ)\n            self.io_loop.call_later(0.1, self.stop)\n            self.wait()\n\n            # Only one fd was read; the other was cleanly removed.\n            self.assertEqual(chunks, [b\"abc\"])\n        finally:\n            client.close()\n            server.close()\n\n    @skipIfNonUnix\n    @gen_test\n    def test_init_close_race(self):\n        # Regression test for #2367\n        #\n        # Skipped on windows because of what looks like a bug in the\n        # proactor event loop when started and stopped on non-main\n        # threads.\n        def f():\n            for i in range(10):\n                loop = IOLoop(make_current=False)\n                loop.close()\n\n        yield gen.multi([self.io_loop.run_in_executor(None, f) for i in range(2)])\n\n    def test_explicit_asyncio_loop(self):\n        asyncio_loop = asyncio.new_event_loop()\n        loop = IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        assert loop.asyncio_loop is asyncio_loop  # type: ignore\n        with self.assertRaises(RuntimeError):\n            # Can't register two IOLoops with the same asyncio_loop\n            IOLoop(asyncio_loop=asyncio_loop, make_current=False)\n        loop.close()\n\n\n# Deliberately not a subclass of AsyncTestCase so the IOLoop isn't\n# automatically set as current.\nclass TestIOLoopCurrent(unittest.TestCase):\n    def setUp(self):\n        setup_with_context_manager(self, ignore_deprecation())\n        self.io_loop = None  # type: typing.Optional[IOLoop]\n        IOLoop.clear_current()\n\n    def tearDown(self):\n        if self.io_loop is not None:\n            self.io_loop.close()\n\n    def test_default_current(self):\n        self.io_loop = IOLoop()\n        # The first IOLoop with default arguments is made current.\n        self.assertIs(self.io_loop, IOLoop.current())\n        # A second IOLoop can be created but is not made current.\n        io_loop2 = IOLoop()\n        self.assertIs(self.io_loop, IOLoop.current())\n        io_loop2.close()\n\n    def test_non_current(self):\n        self.io_loop = IOLoop(make_current=False)\n        # The new IOLoop is not initially made current.\n        self.assertIsNone(IOLoop.current(instance=False))\n        # Starting the IOLoop makes it current, and stopping the loop\n        # makes it non-current. This process is repeatable.\n        for i in range(3):\n\n            def f():\n                self.current_io_loop = IOLoop.current()\n                assert self.io_loop is not None\n                self.io_loop.stop()\n\n            self.io_loop.add_callback(f)\n            self.io_loop.start()\n            self.assertIs(self.current_io_loop, self.io_loop)\n            # Now that the loop is stopped, it is no longer current.\n            self.assertIsNone(IOLoop.current(instance=False))\n\n    def test_force_current(self):\n        self.io_loop = IOLoop(make_current=True)\n        self.assertIs(self.io_loop, IOLoop.current())\n        with self.assertRaises(RuntimeError):\n            # A second make_current=True construction cannot succeed.\n            IOLoop(make_current=True)\n        # current() was not affected by the failed construction.\n        self.assertIs(self.io_loop, IOLoop.current())\n\n\nclass TestIOLoopCurrentAsync(AsyncTestCase):\n    def setUp(self):\n        super().setUp()\n        setup_with_context_manager(self, ignore_deprecation())\n\n    @gen_test\n    def test_clear_without_current(self):\n        # If there is no current IOLoop, clear_current is a no-op (but\n        # should not fail). Use a thread so we see the threading.Local\n        # in a pristine state.\n        with ThreadPoolExecutor(1) as e:\n            yield e.submit(IOLoop.clear_current)\n\n\nclass TestIOLoopFutures(AsyncTestCase):\n    def test_add_future_threads(self):\n        with futures.ThreadPoolExecutor(1) as pool:\n\n            def dummy():\n                pass\n\n            self.io_loop.add_future(\n                pool.submit(dummy), lambda future: self.stop(future)\n            )\n            future = self.wait()\n            self.assertTrue(future.done())\n            self.assertTrue(future.result() is None)\n\n    @gen_test\n    def test_run_in_executor_gen(self):\n        event1 = threading.Event()\n        event2 = threading.Event()\n\n        def sync_func(self_event, other_event):\n            self_event.set()\n            other_event.wait()\n            # Note that return value doesn't actually do anything,\n            # it is just passed through to our final assertion to\n            # make sure it is passed through properly.\n            return self_event\n\n        # Run two synchronous functions, which would deadlock if not\n        # run in parallel.\n        res = yield [\n            IOLoop.current().run_in_executor(None, sync_func, event1, event2),\n            IOLoop.current().run_in_executor(None, sync_func, event2, event1),\n        ]\n\n        self.assertEqual([event1, event2], res)\n\n    @gen_test\n    def test_run_in_executor_native(self):\n        event1 = threading.Event()\n        event2 = threading.Event()\n\n        def sync_func(self_event, other_event):\n            self_event.set()\n            other_event.wait()\n            return self_event\n\n        # Go through an async wrapper to ensure that the result of\n        # run_in_executor works with await and not just gen.coroutine\n        # (simply passing the underlying concurrent future would do that).\n        async def async_wrapper(self_event, other_event):\n            return await IOLoop.current().run_in_executor(\n                None, sync_func, self_event, other_event\n            )\n\n        res = yield [async_wrapper(event1, event2), async_wrapper(event2, event1)]\n\n        self.assertEqual([event1, event2], res)\n\n    @gen_test\n    def test_set_default_executor(self):\n        count = [0]\n\n        class MyExecutor(futures.ThreadPoolExecutor):\n            def submit(self, func, *args):\n                count[0] += 1\n                return super().submit(func, *args)\n\n        event = threading.Event()\n\n        def sync_func():\n            event.set()\n\n        executor = MyExecutor(1)\n        loop = IOLoop.current()\n        loop.set_default_executor(executor)\n        yield loop.run_in_executor(None, sync_func)\n        self.assertEqual(1, count[0])\n        self.assertTrue(event.is_set())\n\n\nclass TestIOLoopRunSync(unittest.TestCase):\n    def setUp(self):\n        self.io_loop = IOLoop(make_current=False)\n\n    def tearDown(self):\n        self.io_loop.close()\n\n    def test_sync_result(self):\n        with self.assertRaises(gen.BadYieldError):\n            self.io_loop.run_sync(lambda: 42)\n\n    def test_sync_exception(self):\n        with self.assertRaises(ZeroDivisionError):\n            self.io_loop.run_sync(lambda: 1 / 0)\n\n    def test_async_result(self):\n        @gen.coroutine\n        def f():\n            yield gen.moment\n            raise gen.Return(42)\n\n        self.assertEqual(self.io_loop.run_sync(f), 42)\n\n    def test_async_exception(self):\n        @gen.coroutine\n        def f():\n            yield gen.moment\n            1 / 0\n\n        with self.assertRaises(ZeroDivisionError):\n            self.io_loop.run_sync(f)\n\n    def test_current(self):\n        def f():\n            self.assertIs(IOLoop.current(), self.io_loop)\n\n        self.io_loop.run_sync(f)\n\n    def test_timeout(self):\n        @gen.coroutine\n        def f():\n            yield gen.sleep(1)\n\n        self.assertRaises(TimeoutError, self.io_loop.run_sync, f, timeout=0.01)\n\n    def test_native_coroutine(self):\n        @gen.coroutine\n        def f1():\n            yield gen.moment\n\n        async def f2():\n            await f1()\n\n        self.io_loop.run_sync(f2)\n\n\nclass TestPeriodicCallbackMath(unittest.TestCase):\n    def simulate_calls(self, pc, durations):\n        \"\"\"Simulate a series of calls to the PeriodicCallback.\n\n        Pass a list of call durations in seconds (negative values\n        work to simulate clock adjustments during the call, or more or\n        less equivalently, between calls). This method returns the\n        times at which each call would be made.\n        \"\"\"\n        calls = []\n        now = 1000\n        pc._next_timeout = now\n        for d in durations:\n            pc._update_next(now)\n            calls.append(pc._next_timeout)\n            now = pc._next_timeout + d\n        return calls\n\n    def dummy(self):\n        pass\n\n    def test_basic(self):\n        pc = PeriodicCallback(self.dummy, 10000)\n        self.assertEqual(\n            self.simulate_calls(pc, [0] * 5), [1010, 1020, 1030, 1040, 1050]\n        )\n\n    def test_overrun(self):\n        # If a call runs for too long, we skip entire cycles to get\n        # back on schedule.\n        call_durations = [9, 9, 10, 11, 20, 20, 35, 35, 0, 0, 0]\n        expected = [\n            1010,\n            1020,\n            1030,  # first 3 calls on schedule\n            1050,\n            1070,  # next 2 delayed one cycle\n            1100,\n            1130,  # next 2 delayed 2 cycles\n            1170,\n            1210,  # next 2 delayed 3 cycles\n            1220,\n            1230,  # then back on schedule.\n        ]\n\n        pc = PeriodicCallback(self.dummy, 10000)\n        self.assertEqual(self.simulate_calls(pc, call_durations), expected)\n\n    def test_clock_backwards(self):\n        pc = PeriodicCallback(self.dummy, 10000)\n        # Backwards jumps are ignored, potentially resulting in a\n        # slightly slow schedule (although we assume that when\n        # time.time() and time.monotonic() are different, time.time()\n        # is getting adjusted by NTP and is therefore more accurate)\n        self.assertEqual(\n            self.simulate_calls(pc, [-2, -1, -3, -2, 0]), [1010, 1020, 1030, 1040, 1050]\n        )\n\n        # For big jumps, we should perhaps alter the schedule, but we\n        # don't currently. This trace shows that we run callbacks\n        # every 10s of time.time(), but the first and second calls are\n        # 110s of real time apart because the backwards jump is\n        # ignored.\n        self.assertEqual(self.simulate_calls(pc, [-100, 0, 0]), [1010, 1020, 1030])\n\n    def test_jitter(self):\n        random_times = [0.5, 1, 0, 0.75]\n        expected = [1010, 1022.5, 1030, 1041.25]\n        call_durations = [0] * len(random_times)\n        pc = PeriodicCallback(self.dummy, 10000, jitter=0.5)\n\n        def mock_random():\n            return random_times.pop(0)\n\n        with mock.patch(\"random.random\", mock_random):\n            self.assertEqual(self.simulate_calls(pc, call_durations), expected)\n\n    def test_timedelta(self):\n        pc = PeriodicCallback(lambda: None, datetime.timedelta(minutes=1, seconds=23))\n        expected_callback_time = 83000\n        self.assertEqual(pc.callback_time, expected_callback_time)\n\n\nclass TestPeriodicCallbackAsync(AsyncTestCase):\n    def test_periodic_plain(self):\n        count = 0\n\n        def callback() -> None:\n            nonlocal count\n            count += 1\n            if count == 3:\n                self.stop()\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        pc.stop()\n        self.assertEqual(count, 3)\n\n    def test_periodic_coro(self) -> None:\n        counts = [0, 0]\n\n        @gen.coroutine\n        def callback() -> \"Generator[Future[None], object, None]\":\n            counts[0] += 1\n            yield gen.sleep(0.025)\n            counts[1] += 1\n            if counts[1] == 3:\n                pc.stop()\n                self.io_loop.add_callback(self.stop)\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        self.assertEqual(counts[0], 3)\n        self.assertEqual(counts[1], 3)\n\n    def test_periodic_async(self) -> None:\n        counts = [0, 0]\n\n        async def callback() -> None:\n            counts[0] += 1\n            await gen.sleep(0.025)\n            counts[1] += 1\n            if counts[1] == 3:\n                pc.stop()\n                self.io_loop.add_callback(self.stop)\n\n        pc = PeriodicCallback(callback, 10)\n        pc.start()\n        self.wait()\n        self.assertEqual(counts[0], 3)\n        self.assertEqual(counts[1], 3)\n\n\nclass TestIOLoopConfiguration(unittest.TestCase):\n    def run_python(self, *statements):\n        stmt_list = [\n            \"from tornado.ioloop import IOLoop\",\n            \"classname = lambda x: x.__class__.__name__\",\n        ] + list(statements)\n        args = [sys.executable, \"-c\", \"; \".join(stmt_list)]\n        return native_str(subprocess.check_output(args)).strip()\n\n    def test_default(self):\n        # When asyncio is available, it is used by default.\n        cls = self.run_python(\"print(classname(IOLoop.current()))\")\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n        cls = self.run_python(\"print(classname(IOLoop()))\")\n        self.assertEqual(cls, \"AsyncIOLoop\")\n\n    def test_asyncio(self):\n        cls = self.run_python(\n            'IOLoop.configure(\"tornado.platform.asyncio.AsyncIOLoop\")',\n            \"print(classname(IOLoop.current()))\",\n        )\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n\n    def test_asyncio_main(self):\n        cls = self.run_python(\n            \"from tornado.platform.asyncio import AsyncIOMainLoop\",\n            \"AsyncIOMainLoop().install()\",\n            \"print(classname(IOLoop.current()))\",\n        )\n        self.assertEqual(cls, \"AsyncIOMainLoop\")\n\n\nif __name__ == \"__main__\":\n    unittest.main()\n",
        "#!/usr/bin/env python\nimport unittest\nimport sys\nimport time\nimport traceback\nimport platform\nimport os\n\n\nclass PytestLikeResult(unittest.TestResult):\n    def __init__(self, *args, **kwargs):\n        super().__init__(*args, **kwargs)\n        self.start_time = None\n        # Store (test, outcome, info) for each test:\n        #   outcome ∈ { \"passed\", \"failed\", \"error\", \"skipped\" }\n        #   info can hold the exception or skip reason, etc.\n        self.results = []\n\n    def startTestRun(self):\n        \"\"\"Called once before all tests.\"\"\"\n        self.start_time = time.time()\n        # Print a header similar to pytest\n        print(\"=\" * 29 + \" test session starts \" + \"=\" * 29)\n        # You can customize the platform/versions/paths as you like:\n        print(\n            f\"platform {platform.system().lower()} -- Python {platform.python_version()}\"\n        )\n        print(f\"rootdir: {os.path.abspath(os.curdir)}\")\n\n    def stopTestRun(self):\n        \"\"\"Called once after all tests are done.\"\"\"\n        elapsed = time.time() - self.start_time\n        print()  # blank line after final dot\n\n        # Summarize\n        ntests = len(self.results)\n        print(f\"collected {ntests} items\")\n\n        # Show details of failures/errors\n        failures = [r for r in self.results if r[1] == \"failed\"]\n        errors = [r for r in self.results if r[1] == \"error\"]\n\n        if failures:\n            print(\"\\n\" + \"=\" * 35 + \" FAILURES \" + \"=\" * 35)\n            for test, outcome, err_info in failures:\n                test = test.split(\"::\")\n                if len(test) == 3:\n                    test = f\"{test[1]}.{test[2]}\"\n                else:\n                    test = f\"{test[1]}\"\n                print(f\"__________ {test} __________\")\n                self._print_traceback(err_info)\n        if errors:\n            print(\"\\n\" + \"=\" * 35 + \" ERRORS \" + \"=\" * 37)\n            for test, outcome, err_info in errors:\n                test = test.split(\"::\")\n                if len(test) == 3:\n                    test = f\"{test[1]}.{test[2]}\"\n                else:\n                    test = f\"{test[1]}\"\n                print(f\"__________ {test} __________\")\n                self._print_traceback(err_info)\n\n        # Optional short summary lines for passes/fails/skips\n        print(\"\\n==================== short test summary info ====================\")\n        for test, outcome, info in self.results:\n            if outcome in (\"passed\", \"failed\", \"error\", \"skipped\"):\n                print(f\"{outcome.upper()} {test}\")\n\n        # Final one-line summary: “1 failed, 1 errors, 10 passed in 2.04s”\n        passed_count = sum(1 for r in self.results if r[1] == \"passed\")\n        failed_count = len(failures)\n        error_count = len(errors)\n        skipped_count = sum(1 for r in self.results if r[1] == \"skipped\")\n\n        summary_bits = []\n        if failed_count:\n            summary_bits.append(f\"{failed_count} failed\")\n        if error_count:\n            summary_bits.append(f\"{error_count} error{'s' if error_count != 1 else ''}\")\n        if passed_count:\n            summary_bits.append(f\"{passed_count} passed\")\n        if skipped_count:\n            summary_bits.append(f\"{skipped_count} skipped\")\n\n        summary_str = \", \".join(summary_bits) if summary_bits else \"no tests run\"\n        print(\n            f\"=================== {summary_str} in {elapsed:.2f}s ===================\"\n        )\n\n    def startTest(self, test):\n        \"\"\"Called right before each test method.\"\"\"\n        super().startTest(test)\n        # Print a dot or some indicator\n        print(\".\", end=\"\", flush=True)\n\n    def addSuccess(self, test):\n        super().addSuccess(test)\n        self.results.append((self._test_id(test), \"passed\", None))\n\n    def addFailure(self, test, err):\n        super().addFailure(test, err)\n        self.results.append((self._test_id(test), \"failed\", err))\n\n    def addError(self, test, err):\n        super().addError(test, err)\n        self.results.append((self._test_id(test), \"error\", err))\n\n    def addSkip(self, test, reason):\n        super().addSkip(test, reason)\n        self.results.append((self._test_id(test), \"skipped\", reason))\n\n    def _print_traceback(self, err_info):\n        \"\"\"Helper to print the traceback like pytest does.\"\"\"\n        if isinstance(err_info, tuple) and len(err_info) == 3:\n            exc_type, exc_value, tb = err_info\n            traceback.print_exception(exc_type, exc_value, tb, file=sys.stdout)\n        else:\n            print(str(err_info))\n\n    def _test_id(self, test):\n        try:\n            return f\"{test.__class__.__module__}::{test.__class__.__name__}::{test._testMethodName}\"\n        except AttributeError:\n            # Fallback for _ErrorHolder objects\n            return f\"{test.__class__.__module__}::{test.__class__.__name__}\"\n\n\nclass PytestLikeRunner(unittest.TextTestRunner):\n    \"\"\"A custom TextTestRunner that uses the PytestLikeResult.\"\"\"\n\n    resultclass = PytestLikeResult\n\n    def run(self, test):\n        result = self._makeResult()\n        # Overriding run to ensure we call startTestRun/stopTestRun\n        result.startTestRun()\n        test(result)\n        result.stopTestRun()\n        return result\n\n    def tearDown(self):\n        # Reset signal handlers\n        from twisted.internet import reactor\n\n        reactor.removeAll()\n        reactor.stop()\n        super().tearDown()\n\n\ndef main():\n    loader = unittest.TestLoader()\n    suite = loader.discover(\".\", pattern=\"test_*.py\")  # discover tests in r2e_tests/\n    runner = PytestLikeRunner(verbosity=0)\n    result = runner.run(suite)\n    # Exit with code 0 if all tests passed/skipped, 1 if there were fails/errors\n    sys.exit(not result.wasSuccessful())\n\n\nif __name__ == \"__main__\":\n    main()\n"
    ],
    "test_file_names": [
        "test_1.py",
        "tornado_unittest_runner.py"
    ],
    "setup_res_code": 0,
    "setup_res_stdout": "",
    "setup_res_stderr": "Using CPython 3.9.21\nCreating virtual environment at: .venv\nActivate with: source .venv/bin/activate\nResolved 1 package in 865ms\nPrepared 1 package in 963ms\nInstalled 1 package in 1ms\n + tornado==6.2b1 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_44e28cf32ca120189026b55395587415e731f905)\nResolved 29 packages in 18ms\nInstalled 29 packages in 171ms\n + attrs==24.3.0\n + automat==24.8.1\n + cachetools==5.5.0\n + cffi==1.17.1\n + chardet==5.2.0\n + colorama==0.4.6\n + constantly==23.10.4\n + distlib==0.3.9\n + exceptiongroup==1.2.2\n + filelock==3.16.1\n + hyperlink==21.0.0\n + idna==3.10\n + incremental==24.7.2\n + iniconfig==2.0.0\n + packaging==24.2\n + platformdirs==4.3.6\n + pluggy==1.5.0\n + pycares==4.5.0\n + pycparser==2.22\n + pycurl==7.45.4\n + pyproject-api==1.8.0\n + pytest==8.3.4\n + setuptools==75.6.0\n + tomli==2.2.1\n + tox==4.23.2\n + twisted==24.11.0\n + typing-extensions==4.12.2\n + virtualenv==20.28.0\n + zope-interface==7.2\n",
    "new_commit_res_code": 0,
    "new_commit_res_stdout": "============================= test session starts =============================\nplatform linux -- Python 3.9.21\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_44e28cf32ca120189026b55395587415e731f905\n......................................................\ncollected 54 items\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_from_signal\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_from_signal_other_thread\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_return_sequence\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_wakeup\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_wakeup_other_thread\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_while_closing\nPASSED r2e_tests.test_1::TestIOLoop::test_add_timeout_return\nPASSED r2e_tests.test_1::TestIOLoop::test_add_timeout_timedelta\nPASSED r2e_tests.test_1::TestIOLoop::test_call_at_return\nPASSED r2e_tests.test_1::TestIOLoop::test_call_later_return\nPASSED r2e_tests.test_1::TestIOLoop::test_close_file_object\nPASSED r2e_tests.test_1::TestIOLoop::test_exception_logging\nPASSED r2e_tests.test_1::TestIOLoop::test_exception_logging_future\nPASSED r2e_tests.test_1::TestIOLoop::test_exception_logging_native_coro\nPASSED r2e_tests.test_1::TestIOLoop::test_explicit_asyncio_loop\nPASSED r2e_tests.test_1::TestIOLoop::test_handler_callback_file_object\nPASSED r2e_tests.test_1::TestIOLoop::test_init_close_race\nPASSED r2e_tests.test_1::TestIOLoop::test_mixed_fd_fileobj\nPASSED r2e_tests.test_1::TestIOLoop::test_multiple_add\nPASSED r2e_tests.test_1::TestIOLoop::test_read_while_writeable\nPASSED r2e_tests.test_1::TestIOLoop::test_reentrant\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_handler_from_handler\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_timeout_after_fire\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_timeout_cleanup\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_timeout_from_timeout\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_without_add\nPASSED r2e_tests.test_1::TestIOLoop::test_spawn_callback\nPASSED r2e_tests.test_1::TestIOLoop::test_timeout_with_arguments\nPASSED r2e_tests.test_1::TestIOLoopConfiguration::test_asyncio\nPASSED r2e_tests.test_1::TestIOLoopConfiguration::test_asyncio_main\nPASSED r2e_tests.test_1::TestIOLoopConfiguration::test_default\nPASSED r2e_tests.test_1::TestIOLoopCurrent::test_default_current\nPASSED r2e_tests.test_1::TestIOLoopCurrent::test_force_current\nPASSED r2e_tests.test_1::TestIOLoopCurrent::test_non_current\nPASSED r2e_tests.test_1::TestIOLoopCurrentAsync::test_clear_without_current\nPASSED r2e_tests.test_1::TestIOLoopFutures::test_add_future_threads\nPASSED r2e_tests.test_1::TestIOLoopFutures::test_run_in_executor_gen\nPASSED r2e_tests.test_1::TestIOLoopFutures::test_run_in_executor_native\nPASSED r2e_tests.test_1::TestIOLoopFutures::test_set_default_executor\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_async_exception\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_async_result\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_current\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_native_coroutine\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_sync_exception\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_sync_result\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_timeout\nPASSED r2e_tests.test_1::TestPeriodicCallbackAsync::test_periodic_async\nPASSED r2e_tests.test_1::TestPeriodicCallbackAsync::test_periodic_coro\nPASSED r2e_tests.test_1::TestPeriodicCallbackAsync::test_periodic_plain\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_basic\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_clock_backwards\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_jitter\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_overrun\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_timedelta\n=================== 54 passed in 0.77s ===================\n",
    "new_commit_res_stderr": "",
    "old_commit_res_code": 1,
    "old_commit_res_stdout": "============================= test session starts =============================\nplatform linux -- Python 3.9.21\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_44e28cf32ca120189026b55395587415e731f905\n......................................................\ncollected 54 items\n\n=================================== FAILURES ===================================\n__________ TestIOLoop.test_explicit_asyncio_loop __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 592, in run\n    self._callTestMethod(testMethod)\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 550, in _callTestMethod\n    method()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_44e28cf32ca120189026b55395587415e731f905/tornado/testing.py\", line 102, in __call__\n    result = self.orig_method(*args, **kwargs)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_44e28cf32ca120189026b55395587415e731f905/r2e_tests/test_1.py\", line 437, in test_explicit_asyncio_loop\n    IOLoop(asyncio_loop=asyncio_loop, make_current=False)\nAssertionError: RuntimeError not raised\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_from_signal\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_from_signal_other_thread\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_return_sequence\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_wakeup\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_wakeup_other_thread\nPASSED r2e_tests.test_1::TestIOLoop::test_add_callback_while_closing\nPASSED r2e_tests.test_1::TestIOLoop::test_add_timeout_return\nPASSED r2e_tests.test_1::TestIOLoop::test_add_timeout_timedelta\nPASSED r2e_tests.test_1::TestIOLoop::test_call_at_return\nPASSED r2e_tests.test_1::TestIOLoop::test_call_later_return\nPASSED r2e_tests.test_1::TestIOLoop::test_close_file_object\nPASSED r2e_tests.test_1::TestIOLoop::test_exception_logging\nPASSED r2e_tests.test_1::TestIOLoop::test_exception_logging_future\nPASSED r2e_tests.test_1::TestIOLoop::test_exception_logging_native_coro\nFAILED r2e_tests.test_1::TestIOLoop::test_explicit_asyncio_loop\nPASSED r2e_tests.test_1::TestIOLoop::test_handler_callback_file_object\nPASSED r2e_tests.test_1::TestIOLoop::test_init_close_race\nPASSED r2e_tests.test_1::TestIOLoop::test_mixed_fd_fileobj\nPASSED r2e_tests.test_1::TestIOLoop::test_multiple_add\nPASSED r2e_tests.test_1::TestIOLoop::test_read_while_writeable\nPASSED r2e_tests.test_1::TestIOLoop::test_reentrant\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_handler_from_handler\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_timeout_after_fire\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_timeout_cleanup\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_timeout_from_timeout\nPASSED r2e_tests.test_1::TestIOLoop::test_remove_without_add\nPASSED r2e_tests.test_1::TestIOLoop::test_spawn_callback\nPASSED r2e_tests.test_1::TestIOLoop::test_timeout_with_arguments\nPASSED r2e_tests.test_1::TestIOLoopConfiguration::test_asyncio\nPASSED r2e_tests.test_1::TestIOLoopConfiguration::test_asyncio_main\nPASSED r2e_tests.test_1::TestIOLoopConfiguration::test_default\nPASSED r2e_tests.test_1::TestIOLoopCurrent::test_default_current\nPASSED r2e_tests.test_1::TestIOLoopCurrent::test_force_current\nPASSED r2e_tests.test_1::TestIOLoopCurrent::test_non_current\nPASSED r2e_tests.test_1::TestIOLoopCurrentAsync::test_clear_without_current\nPASSED r2e_tests.test_1::TestIOLoopFutures::test_add_future_threads\nPASSED r2e_tests.test_1::TestIOLoopFutures::test_run_in_executor_gen\nPASSED r2e_tests.test_1::TestIOLoopFutures::test_run_in_executor_native\nPASSED r2e_tests.test_1::TestIOLoopFutures::test_set_default_executor\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_async_exception\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_async_result\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_current\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_native_coroutine\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_sync_exception\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_sync_result\nPASSED r2e_tests.test_1::TestIOLoopRunSync::test_timeout\nPASSED r2e_tests.test_1::TestPeriodicCallbackAsync::test_periodic_async\nPASSED r2e_tests.test_1::TestPeriodicCallbackAsync::test_periodic_coro\nPASSED r2e_tests.test_1::TestPeriodicCallbackAsync::test_periodic_plain\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_basic\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_clock_backwards\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_jitter\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_overrun\nPASSED r2e_tests.test_1::TestPeriodicCallbackMath::test_timedelta\n=================== 1 failed, 53 passed in 0.74s ===================\n",
    "old_commit_res_stderr": ""
}
tornado/platform/asyncio.py tornado/test/ioloop_test.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
**Issue Title:**
Creating Multiple IOLoop Instances with the Same asyncio Loop Does Not Raise an Error

**Issue Description:**

When attempting to associate multiple `IOLoop` instances with the same `asyncio` loop, no error is raised. This behavior can lead to conflicts and unexpected issues within the application.

**Example Code:**

```python
import asyncio
from tornado.ioloop import IOLoop

# Create a new asyncio event loop
asyncio_loop = asyncio.new_event_loop()

# Associate the first IOLoop with the asyncio loop
loop1 = IOLoop(asyncio_loop=asyncio_loop, make_current=False)

# Attempt to associate a second IOLoop with the same asyncio loop
loop2 = IOLoop(asyncio_loop=asyncio_loop, make_current=False)  # Should raise RuntimeError
```

**Expected Behavior:**

Creating a second `IOLoop` with the same `asyncio` loop should raise a `RuntimeError` indicating that the asyncio loop is already associated with an `IOLoop`. This prevents multiple `IOLoop` instances from interfering with each other.

**Actual Behavior:**

No error is raised when creating the second `IOLoop`, allowing multiple `IOLoop` instances to be associated with the same `asyncio` loop. This can lead to unexpected behavior and potential conflicts within the application.

**Additional Context:**

Prior to this issue, the test `test_explicit_asyncio_loop` failed because it expected a `RuntimeError` when associating multiple `IOLoop` instances with the same `asyncio` loop, but the error was not raised. The recent commit introduces a check to prevent this scenario by raising a `RuntimeError` if an `IOLoop` is already associated with the given `asyncio` loop.

[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### tornado/platform/asyncio.py
"""Bridges between the `asyncio` module and Tornado IOLoop.

.. versionadded:: 3.2

This module integrates Tornado with the ``asyncio`` module introduced
in Python 3.4. This makes it possible to combine the two libraries on
the same event loop.

.. deprecated:: 5.0

   While the code in this module is still used, it is now enabled
   automatically when `asyncio` is available, so applications should
   no longer need to refer to this module directly.

.. note::

   Tornado is designed to use a selector-based event loop. On Windows,
   where a proactor-based event loop has been the default since Python 3.8,
   a selector event loop is emulated by running ``select`` on a separate thread.
   Configuring ``asyncio`` to use a selector event loop may improve performance
   of Tornado (but may reduce performance of other ``asyncio``-based libraries
   in the same process).
"""

import asyncio
import atexit
import concurrent.futures
import errno
import functools
import select
import socket
import sys
import threading
import typing
import warnings
from tornado.gen import convert_yielded
from tornado.ioloop import IOLoop, _Selectable

from typing import Any, TypeVar, Awaitable, Callable, Union, Optional, List, Tuple, Dict

if typing.TYPE_CHECKING:
    from typing import Set  # noqa: F401
    from typing_extensions import Protocol

    class _HasFileno(Protocol):
        def fileno(self) -> int:
            pass

    _FileDescriptorLike = Union[int, _HasFileno]

_T = TypeVar("_T")


# Collection of selector thread event loops to shut down on exit.
_selector_loops = set()  # type: Set[AddThreadSelectorEventLoop]


def _atexit_callback() -> None:
    for loop in _selector_loops:
        with loop._select_cond:
            loop._closing_selector = True
            loop._select_cond.notify()
        try:
            loop._waker_w.send(b"a")
        except BlockingIOError:
            pass
        # If we don't join our (daemon) thread here, we may get a deadlock
        # during interpreter shutdown. I don't really understand why. This
        # deadlock happens every time in CI (both travis and appveyor) but
        # I've never been able to reproduce locally.
        loop._thread.join()
    _selector_loops.clear()


atexit.register(_atexit_callback)

if sys.version_info >= (3, 10):

    def _get_event_loop() -> asyncio.AbstractEventLoop:
        try:
            return asyncio.get_running_loop()
        except RuntimeError:
            pass

        return asyncio.get_event_loop_policy().get_event_loop()


else:
    from asyncio import get_event_loop as _get_event_loop


class BaseAsyncIOLoop(IOLoop):
    def initialize(  # type: ignore
        self, asyncio_loop: asyncio.AbstractEventLoop, **kwargs: Any
    ) -> None:
        # asyncio_loop is always the real underlying IOLoop. This is used in
        # ioloop.py to maintain the asyncio-to-ioloop mappings.
        self.asyncio_loop = asyncio_loop
        # selector_loop is an event loop that implements the add_reader family of
        # methods. Usually the same as asyncio_loop but differs on platforms such
        # as windows where the default event loop does not implement these methods.
        self.selector_loop = asyncio_loop
        if hasattr(asyncio, "ProactorEventLoop") and isinstance(
            asyncio_loop, asyncio.ProactorEventLoop  # type: ignore
        ):
            # Ignore this line for mypy because the abstract method checker
            # doesn't understand dynamic proxies.
            self.selector_loop = AddThreadSelectorEventLoop(asyncio_loop)  # type: ignore
        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)
        self.handlers = {}  # type: Dict[int, Tuple[Union[int, _Selectable], Callable]]
        # Set of fds listening for reads/writes
        self.readers = set()  # type: Set[int]
        self.writers = set()  # type: Set[int]
        self.closing = False
        # If an asyncio loop was closed through an asyncio interface
        # instead of IOLoop.close(), we'd never hear about it and may
        # have left a dangling reference in our map. In case an
        # application (or, more likely, a test suite) creates and
        # destroys a lot of event loops in this way, check here to
        # ensure that we don't have a lot of dead loops building up in
        # the map.
        #
        # TODO(bdarnell): consider making self.asyncio_loop a weakref
        # for AsyncIOMainLoop and make _ioloop_for_asyncio a
        # WeakKeyDictionary.
        for loop in IOLoop._ioloop_for_asyncio.copy():
            if loop.is_closed():
                try:
                    del IOLoop._ioloop_for_asyncio[loop]
                except KeyError:
                    pass
        IOLoop._ioloop_for_asyncio[asyncio_loop] = self

        self._thread_identity = 0

        super().initialize(**kwargs)

        def assign_thread_identity() -> None:
            self._thread_identity = threading.get_ident()

        self.add_callback(assign_thread_identity)

    def close(self, all_fds: bool = False) -> None:
        self.closing = True
        for fd in list(self.handlers):
            fileobj, handler_func = self.handlers[fd]
            self.remove_handler(fd)
            if all_fds:
                self.close_fd(fileobj)
        # Remove the mapping before closing the asyncio loop. If this
        # happened in the other order, we could race against another
        # initialize() call which would see the closed asyncio loop,
        # assume it was closed from the asyncio side, and do this
        # cleanup for us, leading to a KeyError.
        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]
        if self.selector_loop is not self.asyncio_loop:
            self.selector_loop.close()
        self.asyncio_loop.close()

    def add_handler(
        self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int
    ) -> None:
        fd, fileobj = self.split_fd(fd)
        if fd in self.handlers:
            raise ValueError("fd %s added twice" % fd)
        self.handlers[fd] = (fileobj, handler)
        if events & IOLoop.READ:
            self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)
            self.readers.add(fd)
        if events & IOLoop.WRITE:
            self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)
            self.writers.add(fd)

    def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:
        fd, fileobj = self.split_fd(fd)
        if events & IOLoop.READ:
            if fd not in self.readers:
                self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)
                self.readers.add(fd)
        else:
            if fd in self.readers:
                self.selector_loop.remove_reader(fd)
                self.readers.remove(fd)
        if events & IOLoop.WRITE:
            if fd not in self.writers:
                self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)
                self.writers.add(fd)
        else:
            if fd in self.writers:
                self.selector_loop.remove_writer(fd)
                self.writers.remove(fd)

    def remove_handler(self, fd: Union[int, _Selectable]) -> None:
        fd, fileobj = self.split_fd(fd)
        if fd not in self.handlers:
            return
        if fd in self.readers:
            self.selector_loop.remove_reader(fd)
            self.readers.remove(fd)
        if fd in self.writers:
            self.selector_loop.remove_writer(fd)
            self.writers.remove(fd)
        del self.handlers[fd]

    def _handle_events(self, fd: int, events: int) -> None:
        fileobj, handler_func = self.handlers[fd]
        handler_func(fileobj, events)

    def start(self) -> None:
        try:
            old_loop = _get_event_loop()
        except (RuntimeError, AssertionError):
            old_loop = None  # type: ignore
        try:
            asyncio.set_event_loop(self.asyncio_loop)
            self.asyncio_loop.run_forever()
        finally:
            asyncio.set_event_loop(old_loop)

    def stop(self) -> None:
        self.asyncio_loop.stop()

    def call_at(
        self, when: float, callback: Callable, *args: Any, **kwargs: Any
    ) -> object:
        # asyncio.call_at supports *args but not **kwargs, so bind them here.
        # We do not synchronize self.time and asyncio_loop.time, so
        # convert from absolute to relative.
        return self.asyncio_loop.call_later(
            max(0, when - self.time()),
            self._run_callback,
            functools.partial(callback, *args, **kwargs),
        )

    def remove_timeout(self, timeout: object) -> None:
        timeout.cancel()  # type: ignore

    def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:
        if threading.get_ident() == self._thread_identity:
            call_soon = self.asyncio_loop.call_soon
        else:
            call_soon = self.asyncio_loop.call_soon_threadsafe
        try:
            call_soon(self._run_callback, functools.partial(callback, *args, **kwargs))
        except RuntimeError:
            # "Event loop is closed". Swallow the exception for
            # consistency with PollIOLoop (and logical consistency
            # with the fact that we can't guarantee that an
            # add_callback that completes without error will
            # eventually execute).
            pass
        except AttributeError:
            # ProactorEventLoop may raise this instead of RuntimeError
            # if call_soon_threadsafe races with a call to close().
            # Swallow it too for consistency.
            pass

    def add_callback_from_signal(
        self, callback: Callable, *args: Any, **kwargs: Any
    ) -> None:
        try:
            self.asyncio_loop.call_soon_threadsafe(
                self._run_callback, functools.partial(callback, *args, **kwargs)
            )
        except RuntimeError:
            pass

    def run_in_executor(
        self,
        executor: Optional[concurrent.futures.Executor],
        func: Callable[..., _T],
        *args: Any
    ) -> Awaitable[_T]:
        return self.asyncio_loop.run_in_executor(executor, func, *args)

    def set_default_executor(self, executor: concurrent.futures.Executor) -> None:
        return self.asyncio_loop.set_default_executor(executor)


class AsyncIOMainLoop(BaseAsyncIOLoop):
    """``AsyncIOMainLoop`` creates an `.IOLoop` that corresponds to the
    current ``asyncio`` event loop (i.e. the one returned by
    ``asyncio.get_event_loop()``).

    .. deprecated:: 5.0

       Now used automatically when appropriate; it is no longer necessary
       to refer to this class directly.

    .. versionchanged:: 5.0

       Closing an `AsyncIOMainLoop` now closes the underlying asyncio loop.
    """

    def initialize(self, **kwargs: Any) -> None:  # type: ignore
        super().initialize(asyncio.get_event_loop(), **kwargs)

    def make_current(self) -> None:
        # AsyncIOMainLoop already refers to the current asyncio loop so
        # nothing to do here.
        pass


class AsyncIOLoop(BaseAsyncIOLoop):
    """``AsyncIOLoop`` is an `.IOLoop` that runs on an ``asyncio`` event loop.
    This class follows the usual Tornado semantics for creating new
    ``IOLoops``; these loops are not necessarily related to the
    ``asyncio`` default event loop.

    Each ``AsyncIOLoop`` creates a new ``asyncio.EventLoop``; this object
    can be accessed with the ``asyncio_loop`` attribute.

    .. versionchanged:: 6.2

       Support explicit ``asyncio_loop`` argument
       for specifying the asyncio loop to attach to,
       rather than always creating a new one with the default policy.

    .. versionchanged:: 5.0

       When an ``AsyncIOLoop`` becomes the current `.IOLoop`, it also sets
       the current `asyncio` event loop.

    .. deprecated:: 5.0

       Now used automatically when appropriate; it is no longer necessary
       to refer to this class directly.
    """

    def initialize(self, **kwargs: Any) -> None:  # type: ignore
        self.is_current = False
        loop = None
        if "asyncio_loop" not in kwargs:
            kwargs["asyncio_loop"] = loop = asyncio.new_event_loop()
        try:
            super().initialize(**kwargs)
        except Exception:
            # If initialize() does not succeed (taking ownership of the loop),
            # we have to close it.
            if loop is not None:
                loop.close()
            raise

    def close(self, all_fds: bool = False) -> None:
        if self.is_current:
            self._clear_current()
        super().close(all_fds=all_fds)

    def make_current(self) -> None:
        warnings.warn(
            "make_current is deprecated; start the event loop first",
            DeprecationWarning,
        )
        if not self.is_current:
            try:
                self.old_asyncio = asyncio.get_event_loop()
            except (RuntimeError, AssertionError):
                self.old_asyncio = None  # type: ignore
            self.is_current = True
        asyncio.set_event_loop(self.asyncio_loop)

    def _clear_current_hook(self) -> None:
        if self.is_current:
            asyncio.set_event_loop(self.old_asyncio)
            self.is_current = False


def to_tornado_future(asyncio_future: asyncio.Future) -> asyncio.Future:
    """Convert an `asyncio.Future` to a `tornado.concurrent.Future`.

    .. versionadded:: 4.1

    .. deprecated:: 5.0
       Tornado ``Futures`` have been merged with `asyncio.Future`,
       so this method is now a no-op.
    """
    return asyncio_future


def to_asyncio_future(tornado_future: asyncio.Future) -> asyncio.Future:
    """Convert a Tornado yieldable object to an `asyncio.Future`.

    .. versionadded:: 4.1

    .. versionchanged:: 4.3
       Now accepts any yieldable object, not just
       `tornado.concurrent.Future`.

    .. deprecated:: 5.0
       Tornado ``Futures`` have been merged with `asyncio.Future`,
       so this method is now equivalent to `tornado.gen.convert_yielded`.
    """
    return convert_yielded(tornado_future)


if sys.platform == "win32" and hasattr(asyncio, "WindowsSelectorEventLoopPolicy"):
    # "Any thread" and "selector" should be orthogonal, but there's not a clean
    # interface for composing policies so pick the right base.
    _BasePolicy = asyncio.WindowsSelectorEventLoopPolicy  # type: ignore
else:
    _BasePolicy = asyncio.DefaultEventLoopPolicy


class AnyThreadEventLoopPolicy(_BasePolicy):  # type: ignore
    """Event loop policy that allows loop creation on any thread.

    The default `asyncio` event loop policy only automatically creates
    event loops in the main threads. Other threads must create event
    loops explicitly or `asyncio.get_event_loop` (and therefore
    `.IOLoop.current`) will fail. Installing this policy allows event
    loops to be created automatically on any thread, matching the
    behavior of Tornado versions prior to 5.0 (or 5.0 on Python 2).

    Usage::

        asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())

    .. versionadded:: 5.0

    .. deprecated:: 6.2

        ``AnyThreadEventLoopPolicy`` affects the implicit creation
        of an event loop, which is deprecated in Python 3.10 and
        will be removed in a future version of Python. At that time
        ``AnyThreadEventLoopPolicy`` will no longer be useful.
        If you are relying on it, use `asyncio.new_event_loop`
        or `asyncio.run` explicitly in any non-main threads that
        need event loops.
    """

    def __init__(self) -> None:
        super().__init__()
        warnings.warn(
            "AnyThreadEventLoopPolicy is deprecated, use asyncio.run "
            "or asyncio.new_event_loop instead",
            DeprecationWarning,
        )

    def get_event_loop(self) -> asyncio.AbstractEventLoop:
        try:
            return super().get_event_loop()
        except (RuntimeError, AssertionError):
            # This was an AssertionError in Python 3.4.2 (which ships with Debian Jessie)
            # and changed to a RuntimeError in 3.4.3.
            # "There is no current event loop in thread %r"
            loop = self.new_event_loop()
            self.set_event_loop(loop)
            return loop


class AddThreadSelectorEventLoop(asyncio.AbstractEventLoop):
    """Wrap an event loop to add implementations of the ``add_reader`` method family.

    Instances of this class start a second thread to run a selector.
    This thread is completely hidden from the user; all callbacks are
    run on the wrapped event loop's thread.

    This class is used automatically by Tornado; applications should not need
    to refer to it directly.

    It is safe to wrap any event loop with this class, although it only makes sense
    for event loops that do not implement the ``add_reader`` family of methods
    themselves (i.e. ``WindowsProactorEventLoop``)

    Closing the ``AddThreadSelectorEventLoop`` also closes the wrapped event loop.

    """

    # This class is a __getattribute__-based proxy. All attributes other than those
    # in this set are proxied through to the underlying loop.
    MY_ATTRIBUTES = {
        "_consume_waker",
        "_select_cond",
        "_select_args",
        "_closing_selector",
        "_thread",
        "_handle_event",
        "_readers",
        "_real_loop",
        "_start_select",
        "_run_select",
        "_handle_select",
        "_wake_selector",
        "_waker_r",
        "_waker_w",
        "_writers",
        "add_reader",
        "add_writer",
        "close",
        "remove_reader",
        "remove_writer",
    }

    def __getattribute__(self, name: str) -> Any:
        if name in AddThreadSelectorEventLoop.MY_ATTRIBUTES:
            return super().__getattribute__(name)
        return getattr(self._real_loop, name)

    def __init__(self, real_loop: asyncio.AbstractEventLoop) -> None:
        self._real_loop = real_loop

        # Create a thread to run the select system call. We manage this thread
        # manually so we can trigger a clean shutdown from an atexit hook. Note
        # that due to the order of operations at shutdown, only daemon threads
        # can be shut down in this way (non-daemon threads would require the
        # introduction of a new hook: https://bugs.python.org/issue41962)
        self._select_cond = threading.Condition()
        self._select_args = (
            None
        )  # type: Optional[Tuple[List[_FileDescriptorLike], List[_FileDescriptorLike]]]
        self._closing_selector = False
        self._thread = threading.Thread(
            name="Tornado selector",
            daemon=True,
            target=self._run_select,
        )
        self._thread.start()
        # Start the select loop once the loop is started.
        self._real_loop.call_soon(self._start_select)

        self._readers = {}  # type: Dict[_FileDescriptorLike, Callable]
        self._writers = {}  # type: Dict[_FileDescriptorLike, Callable]

        # Writing to _waker_w will wake up the selector thread, which
        # watches for _waker_r to be readable.
        self._waker_r, self._waker_w = socket.socketpair()
        self._waker_r.setblocking(False)
        self._waker_w.setblocking(False)
        _selector_loops.add(self)
        self.add_reader(self._waker_r, self._consume_waker)

    def __del__(self) -> None:
        # If the top-level application code uses asyncio interfaces to
        # start and stop the event loop, no objects created in Tornado
        # can get a clean shutdown notification. If we're just left to
        # be GC'd, we must explicitly close our sockets to avoid
        # logging warnings.
        _selector_loops.discard(self)
        self._waker_r.close()
        self._waker_w.close()

    def close(self) -> None:
        with self._select_cond:
            self._closing_selector = True
            self._select_cond.notify()
        self._wake_selector()
        self._thread.join()
        _selector_loops.discard(self)
        self._waker_r.close()
        self._waker_w.close()
        self._real_loop.close()

    def _wake_selector(self) -> None:
        try:
            self._waker_w.send(b"a")
        except BlockingIOError:
            pass

    def _consume_waker(self) -> None:
        try:
            self._waker_r.recv(1024)
        except BlockingIOError:
            pass

    def _start_select(self) -> None:
        # Capture reader and writer sets here in the event loop
        # thread to avoid any problems with concurrent
        # modification while the select loop uses them.
        with self._select_cond:
            assert self._select_args is None
            self._select_args = (list(self._readers.keys()), list(self._writers.keys()))
            self._select_cond.notify()

    def _run_select(self) -> None:
        while True:
            with self._select_cond:
                while self._select_args is None and not self._closing_selector:
                    self._select_cond.wait()
                if self._closing_selector:
                    return
                assert self._select_args is not None
                to_read, to_write = self._select_args
                self._select_args = None

            # We use the simpler interface of the select module instead of
            # the more stateful interface in the selectors module because
            # this class is only intended for use on windows, where
            # select.select is the only option. The selector interface
            # does not have well-documented thread-safety semantics that
            # we can rely on so ensuring proper synchronization would be
            # tricky.
            try:
                # On windows, selecting on a socket for write will not
                # return the socket when there is an error (but selecting
                # for reads works). Also select for errors when selecting
                # for writes, and merge the results.
                #
                # This pattern is also used in
                # https://github.com/python/cpython/blob/v3.8.0/Lib/selectors.py#L312-L317
                rs, ws, xs = select.select(to_read, to_write, to_write)
                ws = ws + xs
            except OSError as e:
                # After remove_reader or remove_writer is called, the file
                # descriptor may subsequently be closed on the event loop
                # thread. It's possible that this select thread hasn't
                # gotten into the select system call by the time that
                # happens in which case (at least on macOS), select may
                # raise a "bad file descriptor" error. If we get that
                # error, check and see if we're also being woken up by
                # polling the waker alone. If we are, just return to the
                # event loop and we'll get the updated set of file
                # descriptors on the next iteration. Otherwise, raise the
                # original error.
                if e.errno == getattr(errno, "WSAENOTSOCK", errno.EBADF):
                    rs, _, _ = select.select([self._waker_r.fileno()], [], [], 0)
                    if rs:
                        ws = []
                    else:
                        raise
                else:
                    raise

            try:
                self._real_loop.call_soon_threadsafe(self._handle_select, rs, ws)
            except RuntimeError:
                # "Event loop is closed". Swallow the exception for
                # consistency with PollIOLoop (and logical consistency
                # with the fact that we can't guarantee that an
                # add_callback that completes without error will
                # eventually execute).
                pass
            except AttributeError:
                # ProactorEventLoop may raise this instead of RuntimeError
                # if call_soon_threadsafe races with a call to close().
                # Swallow it too for consistency.
                pass

    def _handle_select(
        self, rs: List["_FileDescriptorLike"], ws: List["_FileDescriptorLike"]
    ) -> None:
        for r in rs:
            self._handle_event(r, self._readers)
        for w in ws:
            self._handle_event(w, self._writers)
        self._start_select()

    def _handle_event(
        self,
        fd: "_FileDescriptorLike",
        cb_map: Dict["_FileDescriptorLike", Callable],
    ) -> None:
        try:
            callback = cb_map[fd]
        except KeyError:
            return
        callback()

    def add_reader(
        self, fd: "_FileDescriptorLike", callback: Callable[..., None], *args: Any
    ) -> None:
        self._readers[fd] = functools.partial(callback, *args)
        self._wake_selector()

    def add_writer(
        self, fd: "_FileDescriptorLike", callback: Callable[..., None], *args: Any
    ) -> None:
        self._writers[fd] = functools.partial(callback, *args)
        self._wake_selector()

    def remove_reader(self, fd: "_FileDescriptorLike") -> None:
        del self._readers[fd]
        self._wake_selector()

    def remove_writer(self, fd: "_FileDescriptorLike") -> None:
        del self._writers[fd]
        self._wake_selector()

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
**Issue Title:**
Creating Multiple IOLoop Instances with the Same asyncio Loop Does Not Raise an Error

**Issue Description:**

When attempting to associate multiple `IOLoop` instances with the same `asyncio` loop, no error is raised. This behavior can lead to conflicts and unexpected issues within the application.

**Example Code:**

```python
import asyncio
from tornado.ioloop import IOLoop

# Create a new asyncio event loop
asyncio_loop = asyncio.new_event_loop()

# Associate the first IOLoop with the asyncio loop
loop1 = IOLoop(asyncio_loop=asyncio_loop, make_current=False)

# Attempt to associate a second IOLoop with the same asyncio loop
loop2 = IOLoop(asyncio_loop=asyncio_loop, make_current=False)  # Should raise RuntimeError
```

**Expected Behavior:**

Creating a second `IOLoop` with the same `asyncio` loop should raise a `RuntimeError` indicating that the asyncio loop is already associated with an `IOLoop`. This prevents multiple `IOLoop` instances from interfering with each other.

**Actual Behavior:**

No error is raised when creating the second `IOLoop`, allowing multiple `IOLoop` instances to be associated with the same `asyncio` loop. This can lead to unexpected behavior and potential conflicts within the application.

**Additional Context:**

Prior to this issue, the test `test_explicit_asyncio_loop` failed because it expected a `RuntimeError` when associating multiple `IOLoop` instances with the same `asyncio` loop, but the error was not raised. The recent commit introduces a check to prevent this scenario by raising a `RuntimeError` if an `IOLoop` is already associated with the given `asyncio` loop.

[/ISSUE]
{
    "TestIOLoop.test_add_callback_from_signal": "PASSED",
    "TestIOLoop.test_add_callback_from_signal_other_thread": "PASSED",
    "TestIOLoop.test_add_callback_return_sequence": "PASSED",
    "TestIOLoop.test_add_callback_wakeup": "PASSED",
    "TestIOLoop.test_add_callback_wakeup_other_thread": "PASSED",
    "TestIOLoop.test_add_callback_while_closing": "PASSED",
    "TestIOLoop.test_add_timeout_return": "PASSED",
    "TestIOLoop.test_add_timeout_timedelta": "PASSED",
    "TestIOLoop.test_call_at_return": "PASSED",
    "TestIOLoop.test_call_later_return": "PASSED",
    "TestIOLoop.test_close_file_object": "PASSED",
    "TestIOLoop.test_exception_logging": "PASSED",
    "TestIOLoop.test_exception_logging_future": "PASSED",
    "TestIOLoop.test_exception_logging_native_coro": "PASSED",
    "TestIOLoop.test_explicit_asyncio_loop": "PASSED",
    "TestIOLoop.test_handler_callback_file_object": "PASSED",
    "TestIOLoop.test_init_close_race": "PASSED",
    "TestIOLoop.test_mixed_fd_fileobj": "PASSED",
    "TestIOLoop.test_multiple_add": "PASSED",
    "TestIOLoop.test_read_while_writeable": "PASSED",
    "TestIOLoop.test_reentrant": "PASSED",
    "TestIOLoop.test_remove_handler_from_handler": "PASSED",
    "TestIOLoop.test_remove_timeout_after_fire": "PASSED",
    "TestIOLoop.test_remove_timeout_cleanup": "PASSED",
    "TestIOLoop.test_remove_timeout_from_timeout": "PASSED",
    "TestIOLoop.test_remove_without_add": "PASSED",
    "TestIOLoop.test_spawn_callback": "PASSED",
    "TestIOLoop.test_timeout_with_arguments": "PASSED",
    "TestIOLoopConfiguration.test_asyncio": "PASSED",
    "TestIOLoopConfiguration.test_asyncio_main": "PASSED",
    "TestIOLoopConfiguration.test_default": "PASSED",
    "TestIOLoopCurrent.test_default_current": "PASSED",
    "TestIOLoopCurrent.test_force_current": "PASSED",
    "TestIOLoopCurrent.test_non_current": "PASSED",
    "TestIOLoopCurrentAsync.test_clear_without_current": "PASSED",
    "TestIOLoopFutures.test_add_future_threads": "PASSED",
    "TestIOLoopFutures.test_run_in_executor_gen": "PASSED",
    "TestIOLoopFutures.test_run_in_executor_native": "PASSED",
    "TestIOLoopFutures.test_set_default_executor": "PASSED",
    "TestIOLoopRunSync.test_async_exception": "PASSED",
    "TestIOLoopRunSync.test_async_result": "PASSED",
    "TestIOLoopRunSync.test_current": "PASSED",
    "TestIOLoopRunSync.test_native_coroutine": "PASSED",
    "TestIOLoopRunSync.test_sync_exception": "PASSED",
    "TestIOLoopRunSync.test_sync_result": "PASSED",
    "TestIOLoopRunSync.test_timeout": "PASSED",
    "TestPeriodicCallbackAsync.test_periodic_async": "PASSED",
    "TestPeriodicCallbackAsync.test_periodic_coro": "PASSED",
    "TestPeriodicCallbackAsync.test_periodic_plain": "PASSED",
    "TestPeriodicCallbackMath.test_basic": "PASSED",
    "TestPeriodicCallbackMath.test_clock_backwards": "PASSED",
    "TestPeriodicCallbackMath.test_jitter": "PASSED",
    "TestPeriodicCallbackMath.test_overrun": "PASSED",
    "TestPeriodicCallbackMath.test_timedelta": "PASSED"
}
diff --git a/tornado/platform/asyncio.py b/tornado/platform/asyncio.py
index 3b367815..c35955cc 100644
--- a/tornado/platform/asyncio.py
+++ b/tornado/platform/asyncio.py
@@ -129,6 +129,14 @@ class BaseAsyncIOLoop(IOLoop):
                     del IOLoop._ioloop_for_asyncio[loop]
                 except KeyError:
                     pass
+
+        # Make sure we don't already have an IOLoop for this asyncio loop
+        if asyncio_loop in IOLoop._ioloop_for_asyncio:
+            existing_loop = IOLoop._ioloop_for_asyncio[asyncio_loop]
+            raise RuntimeError(
+                f"IOLoop {existing_loop} already associated with asyncio loop {asyncio_loop}"
+            )
+
         IOLoop._ioloop_for_asyncio[asyncio_loop] = self
 
         self._thread_identity = 0

{"tornado/platform/asyncio.py": "\"\"\"Bridges between the `asyncio` module and Tornado IOLoop.\n\n.. versionadded:: 3.2\n\nThis module integrates Tornado with the ``asyncio`` module introduced\nin Python 3.4. This makes it possible to combine the two libraries on\nthe same event loop.\n\n.. deprecated:: 5.0\n\n   While the code in this module is still used, it is now enabled\n   automatically when `asyncio` is available, so applications should\n   no longer need to refer to this module directly.\n\n.. note::\n\n   Tornado is designed to use a selector-based event loop. On Windows,\n   where a proactor-based event loop has been the default since Python 3.8,\n   a selector event loop is emulated by running ``select`` on a separate thread.\n   Configuring ``asyncio`` to use a selector event loop may improve performance\n   of Tornado (but may reduce performance of other ``asyncio``-based libraries\n   in the same process).\n\"\"\"\n\nimport asyncio\nimport atexit\nimport concurrent.futures\nimport errno\nimport functools\nimport select\nimport socket\nimport sys\nimport threading\nimport typing\nimport warnings\nfrom tornado.gen import convert_yielded\nfrom tornado.ioloop import IOLoop, _Selectable\n\nfrom typing import Any, TypeVar, Awaitable, Callable, Union, Optional, List, Tuple, Dict\n\nif typing.TYPE_CHECKING:\n    from typing import Set  # noqa: F401\n    from typing_extensions import Protocol\n\n    class _HasFileno(Protocol):\n        def fileno(self) -> int:\n            pass\n\n    _FileDescriptorLike = Union[int, _HasFileno]\n\n_T = TypeVar(\"_T\")\n\n\n# Collection of selector thread event loops to shut down on exit.\n_selector_loops = set()  # type: Set[AddThreadSelectorEventLoop]\n\n\ndef _atexit_callback() -> None:\n    for loop in _selector_loops:\n        with loop._select_cond:\n            loop._closing_selector = True\n            loop._select_cond.notify()\n        try:\n            loop._waker_w.send(b\"a\")\n        except BlockingIOError:\n            pass\n        # If we don't join our (daemon) thread here, we may get a deadlock\n        # during interpreter shutdown. I don't really understand why. This\n        # deadlock happens every time in CI (both travis and appveyor) but\n        # I've never been able to reproduce locally.\n        loop._thread.join()\n    _selector_loops.clear()\n\n\natexit.register(_atexit_callback)\n\nif sys.version_info >= (3, 10):\n\n    def _get_event_loop() -> asyncio.AbstractEventLoop:\n        try:\n            return asyncio.get_running_loop()\n        except RuntimeError:\n            pass\n\n        return asyncio.get_event_loop_policy().get_event_loop()\n\n\nelse:\n    from asyncio import get_event_loop as _get_event_loop\n\n\nclass BaseAsyncIOLoop(IOLoop):\n    def initialize(  # type: ignore\n        self, asyncio_loop: asyncio.AbstractEventLoop, **kwargs: Any\n    ) -> None:\n        # asyncio_loop is always the real underlying IOLoop. This is used in\n        # ioloop.py to maintain the asyncio-to-ioloop mappings.\n        self.asyncio_loop = asyncio_loop\n        # selector_loop is an event loop that implements the add_reader family of\n        # methods. Usually the same as asyncio_loop but differs on platforms such\n        # as windows where the default event loop does not implement these methods.\n        self.selector_loop = asyncio_loop\n        if hasattr(asyncio, \"ProactorEventLoop\") and isinstance(\n            asyncio_loop, asyncio.ProactorEventLoop  # type: ignore\n        ):\n            # Ignore this line for mypy because the abstract method checker\n            # doesn't understand dynamic proxies.\n            self.selector_loop = AddThreadSelectorEventLoop(asyncio_loop)  # type: ignore\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}  # type: Dict[int, Tuple[Union[int, _Selectable], Callable]]\n        # Set of fds listening for reads/writes\n        self.readers = set()  # type: Set[int]\n        self.writers = set()  # type: Set[int]\n        self.closing = False\n        # If an asyncio loop was closed through an asyncio interface\n        # instead of IOLoop.close(), we'd never hear about it and may\n        # have left a dangling reference in our map. In case an\n        # application (or, more likely, a test suite) creates and\n        # destroys a lot of event loops in this way, check here to\n        # ensure that we don't have a lot of dead loops building up in\n        # the map.\n        #\n        # TODO(bdarnell): consider making self.asyncio_loop a weakref\n        # for AsyncIOMainLoop and make _ioloop_for_asyncio a\n        # WeakKeyDictionary.\n        for loop in IOLoop._ioloop_for_asyncio.copy():\n            if loop.is_closed():\n                try:\n                    del IOLoop._ioloop_for_asyncio[loop]\n                except KeyError:\n                    pass\n        IOLoop._ioloop_for_asyncio[asyncio_loop] = self\n\n        self._thread_identity = 0\n\n        super().initialize(**kwargs)\n\n        def assign_thread_identity() -> None:\n            self._thread_identity = threading.get_ident()\n\n        self.add_callback(assign_thread_identity)\n\n    def close(self, all_fds: bool = False) -> None:\n        self.closing = True\n        for fd in list(self.handlers):\n            fileobj, handler_func = self.handlers[fd]\n            self.remove_handler(fd)\n            if all_fds:\n                self.close_fd(fileobj)\n        # Remove the mapping before closing the asyncio loop. If this\n        # happened in the other order, we could race against another\n        # initialize() call which would see the closed asyncio loop,\n        # assume it was closed from the asyncio side, and do this\n        # cleanup for us, leading to a KeyError.\n        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]\n        if self.selector_loop is not self.asyncio_loop:\n            self.selector_loop.close()\n        self.asyncio_loop.close()\n\n    def add_handler(\n        self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int\n    ) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd in self.handlers:\n            raise ValueError(\"fd %s added twice\" % fd)\n        self.handlers[fd] = (fileobj, handler)\n        if events & IOLoop.READ:\n            self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.selector_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.selector_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd: Union[int, _Selectable]) -> None:\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.selector_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.selector_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd: int, events: int) -> None:\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self) -> None:\n        try:\n            old_loop = _get_event_loop()\n        except (RuntimeError, AssertionError):\n            old_loop = None  # type: ignore\n        try:\n            asyncio.set_event_loop(self.asyncio_loop)\n            self.asyncio_loop.run_forever()\n        finally:\n            asyncio.set_event_loop(old_loop)\n\n    def stop(self) -> None:\n        self.asyncio_loop.stop()\n\n    def call_at(\n        self, when: float, callback: Callable, *args: Any, **kwargs: Any\n    ) -> object:\n        # asyncio.call_at supports *args but not **kwargs, so bind them here.\n        # We do not synchronize self.time and asyncio_loop.time, so\n        # convert from absolute to relative.\n        return self.asyncio_loop.call_later(\n            max(0, when - self.time()),\n            self._run_callback,\n            functools.partial(callback, *args, **kwargs),\n        )\n\n    def remove_timeout(self, timeout: object) -> None:\n        timeout.cancel()  # type: ignore\n\n    def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:\n        if threading.get_ident() == self._thread_identity:\n            call_soon = self.asyncio_loop.call_soon\n        else:\n            call_soon = self.asyncio_loop.call_soon_threadsafe\n        try:\n            call_soon(self._run_callback, functools.partial(callback, *args, **kwargs))\n        except RuntimeError:\n            # \"Event loop is closed\". Swallow the exception for\n            # consistency with PollIOLoop (and logical consistency\n            # with the fact that we can't guarantee that an\n            # add_callback that completes without error will\n            # eventually execute).\n            pass\n        except AttributeError:\n            # ProactorEventLoop may raise this instead of RuntimeError\n            # if call_soon_threadsafe races with a call to close().\n            # Swallow it too for consistency.\n            pass\n\n    def add_callback_from_signal(\n        self, callback: Callable, *args: Any, **kwargs: Any\n    ) -> None:\n        try:\n            self.asyncio_loop.call_soon_threadsafe(\n                self._run_callback, functools.partial(callback, *args, **kwargs)\n            )\n        except RuntimeError:\n            pass\n\n    def run_in_executor(\n        self,\n        executor: Optional[concurrent.futures.Executor],\n        func: Callable[..., _T],\n        *args: Any\n    ) -> Awaitable[_T]:\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor: concurrent.futures.Executor) -> None:\n        return self.asyncio_loop.set_default_executor(executor)\n\n\nclass AsyncIOMainLoop(BaseAsyncIOLoop):\n    \"\"\"``AsyncIOMainLoop`` creates an `.IOLoop` that corresponds to the\n    current ``asyncio`` event loop (i.e. the one returned by\n    ``asyncio.get_event_loop()``).\n\n    .. deprecated:: 5.0\n\n       Now used automatically when appropriate; it is no longer necessary\n       to refer to this class directly.\n\n    .. versionchanged:: 5.0\n\n       Closing an `AsyncIOMainLoop` now closes the underlying asyncio loop.\n    \"\"\"\n\n    def initialize(self, **kwargs: Any) -> None:  # type: ignore\n        super().initialize(asyncio.get_event_loop(), **kwargs)\n\n    def make_current(self) -> None:\n        # AsyncIOMainLoop already refers to the current asyncio loop so\n        # nothing to do here.\n        pass\n\n\nclass AsyncIOLoop(BaseAsyncIOLoop):\n    \"\"\"``AsyncIOLoop`` is an `.IOLoop` that runs on an ``asyncio`` event loop.\n    This class follows the usual Tornado semantics for creating new\n    ``IOLoops``; these loops are not necessarily related to the\n    ``asyncio`` default event loop.\n\n    Each ``AsyncIOLoop`` creates a new ``asyncio.EventLoop``; this object\n    can be accessed with the ``asyncio_loop`` attribute.\n\n    .. versionchanged:: 6.2\n\n       Support explicit ``asyncio_loop`` argument\n       for specifying the asyncio loop to attach to,\n       rather than always creating a new one with the default policy.\n\n    .. versionchanged:: 5.0\n\n       When an ``AsyncIOLoop`` becomes the current `.IOLoop`, it also sets\n       the current `asyncio` event loop.\n\n    .. deprecated:: 5.0\n\n       Now used automatically when appropriate; it is no longer necessary\n       to refer to this class directly.\n    \"\"\"\n\n    def initialize(self, **kwargs: Any) -> None:  # type: ignore\n        self.is_current = False\n        loop = None\n        if \"asyncio_loop\" not in kwargs:\n            kwargs[\"asyncio_loop\"] = loop = asyncio.new_event_loop()\n        try:\n            super().initialize(**kwargs)\n        except Exception:\n            # If initialize() does not succeed (taking ownership of the loop),\n            # we have to close it.\n            if loop is not None:\n                loop.close()\n            raise\n\n    def close(self, all_fds: bool = False) -> None:\n        if self.is_current:\n            self._clear_current()\n        super().close(all_fds=all_fds)\n\n    def make_current(self) -> None:\n        warnings.warn(\n            \"make_current is deprecated; start the event loop first\",\n            DeprecationWarning,\n        )\n        if not self.is_current:\n            try:\n                self.old_asyncio = asyncio.get_event_loop()\n            except (RuntimeError, AssertionError):\n                self.old_asyncio = None  # type: ignore\n            self.is_current = True\n        asyncio.set_event_loop(self.asyncio_loop)\n\n    def _clear_current_hook(self) -> None:\n        if self.is_current:\n            asyncio.set_event_loop(self.old_asyncio)\n            self.is_current = False\n\n\ndef to_tornado_future(asyncio_future: asyncio.Future) -> asyncio.Future:\n    \"\"\"Convert an `asyncio.Future` to a `tornado.concurrent.Future`.\n\n    .. versionadded:: 4.1\n\n    .. deprecated:: 5.0\n       Tornado ``Futures`` have been merged with `asyncio.Future`,\n       so this method is now a no-op.\n    \"\"\"\n    return asyncio_future\n\n\ndef to_asyncio_future(tornado_future: asyncio.Future) -> asyncio.Future:\n    \"\"\"Convert a Tornado yieldable object to an `asyncio.Future`.\n\n    .. versionadded:: 4.1\n\n    .. versionchanged:: 4.3\n       Now accepts any yieldable object, not just\n       `tornado.concurrent.Future`.\n\n    .. deprecated:: 5.0\n       Tornado ``Futures`` have been merged with `asyncio.Future`,\n       so this method is now equivalent to `tornado.gen.convert_yielded`.\n    \"\"\"\n    return convert_yielded(tornado_future)\n\n\nif sys.platform == \"win32\" and hasattr(asyncio, \"WindowsSelectorEventLoopPolicy\"):\n    # \"Any thread\" and \"selector\" should be orthogonal, but there's not a clean\n    # interface for composing policies so pick the right base.\n    _BasePolicy = asyncio.WindowsSelectorEventLoopPolicy  # type: ignore\nelse:\n    _BasePolicy = asyncio.DefaultEventLoopPolicy\n\n\nclass AnyThreadEventLoopPolicy(_BasePolicy):  # type: ignore\n    \"\"\"Event loop policy that allows loop creation on any thread.\n\n    The default `asyncio` event loop policy only automatically creates\n    event loops in the main threads. Other threads must create event\n    loops explicitly or `asyncio.get_event_loop` (and therefore\n    `.IOLoop.current`) will fail. Installing this policy allows event\n    loops to be created automatically on any thread, matching the\n    behavior of Tornado versions prior to 5.0 (or 5.0 on Python 2).\n\n    Usage::\n\n        asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())\n\n    .. versionadded:: 5.0\n\n    .. deprecated:: 6.2\n\n        ``AnyThreadEventLoopPolicy`` affects the implicit creation\n        of an event loop, which is deprecated in Python 3.10 and\n        will be removed in a future version of Python. At that time\n        ``AnyThreadEventLoopPolicy`` will no longer be useful.\n        If you are relying on it, use `asyncio.new_event_loop`\n        or `asyncio.run` explicitly in any non-main threads that\n        need event loops.\n    \"\"\"\n\n    def __init__(self) -> None:\n        super().__init__()\n        warnings.warn(\n            \"AnyThreadEventLoopPolicy is deprecated, use asyncio.run \"\n            \"or asyncio.new_event_loop instead\",\n            DeprecationWarning,\n        )\n\n    def get_event_loop(self) -> asyncio.AbstractEventLoop:\n        try:\n            return super().get_event_loop()\n        except (RuntimeError, AssertionError):\n            # This was an AssertionError in Python 3.4.2 (which ships with Debian Jessie)\n            # and changed to a RuntimeError in 3.4.3.\n            # \"There is no current event loop in thread %r\"\n            loop = self.new_event_loop()\n            self.set_event_loop(loop)\n            return loop\n\n\nclass AddThreadSelectorEventLoop(asyncio.AbstractEventLoop):\n    \"\"\"Wrap an event loop to add implementations of the ``add_reader`` method family.\n\n    Instances of this class start a second thread to run a selector.\n    This thread is completely hidden from the user; all callbacks are\n    run on the wrapped event loop's thread.\n\n    This class is used automatically by Tornado; applications should not need\n    to refer to it directly.\n\n    It is safe to wrap any event loop with this class, although it only makes sense\n    for event loops that do not implement the ``add_reader`` family of methods\n    themselves (i.e. ``WindowsProactorEventLoop``)\n\n    Closing the ``AddThreadSelectorEventLoop`` also closes the wrapped event loop.\n\n    \"\"\"\n\n    # This class is a __getattribute__-based proxy. All attributes other than those\n    # in this set are proxied through to the underlying loop.\n    MY_ATTRIBUTES = {\n        \"_consume_waker\",\n        \"_select_cond\",\n        \"_select_args\",\n        \"_closing_selector\",\n        \"_thread\",\n        \"_handle_event\",\n        \"_readers\",\n        \"_real_loop\",\n        \"_start_select\",\n        \"_run_select\",\n        \"_handle_select\",\n        \"_wake_selector\",\n        \"_waker_r\",\n        \"_waker_w\",\n        \"_writers\",\n        \"add_reader\",\n        \"add_writer\",\n        \"close\",\n        \"remove_reader\",\n        \"remove_writer\",\n    }\n\n    def __getattribute__(self, name: str) -> Any:\n        if name in AddThreadSelectorEventLoop.MY_ATTRIBUTES:\n            return super().__getattribute__(name)\n        return getattr(self._real_loop, name)\n\n    def __init__(self, real_loop: asyncio.AbstractEventLoop) -> None:\n        self._real_loop = real_loop\n\n        # Create a thread to run the select system call. We manage this thread\n        # manually so we can trigger a clean shutdown from an atexit hook. Note\n        # that due to the order of operations at shutdown, only daemon threads\n        # can be shut down in this way (non-daemon threads would require the\n        # introduction of a new hook: https://bugs.python.org/issue41962)\n        self._select_cond = threading.Condition()\n        self._select_args = (\n            None\n        )  # type: Optional[Tuple[List[_FileDescriptorLike], List[_FileDescriptorLike]]]\n        self._closing_selector = False\n        self._thread = threading.Thread(\n            name=\"Tornado selector\",\n            daemon=True,\n            target=self._run_select,\n        )\n        self._thread.start()\n        # Start the select loop once the loop is started.\n        self._real_loop.call_soon(self._start_select)\n\n        self._readers = {}  # type: Dict[_FileDescriptorLike, Callable]\n        self._writers = {}  # type: Dict[_FileDescriptorLike, Callable]\n\n        # Writing to _waker_w will wake up the selector thread, which\n        # watches for _waker_r to be readable.\n        self._waker_r, self._waker_w = socket.socketpair()\n        self._waker_r.setblocking(False)\n        self._waker_w.setblocking(False)\n        _selector_loops.add(self)\n        self.add_reader(self._waker_r, self._consume_waker)\n\n    def __del__(self) -> None:\n        # If the top-level application code uses asyncio interfaces to\n        # start and stop the event loop, no objects created in Tornado\n        # can get a clean shutdown notification. If we're just left to\n        # be GC'd, we must explicitly close our sockets to avoid\n        # logging warnings.\n        _selector_loops.discard(self)\n        self._waker_r.close()\n        self._waker_w.close()\n\n    def close(self) -> None:\n        with self._select_cond:\n            self._closing_selector = True\n            self._select_cond.notify()\n        self._wake_selector()\n        self._thread.join()\n        _selector_loops.discard(self)\n        self._waker_r.close()\n        self._waker_w.close()\n        self._real_loop.close()\n\n    def _wake_selector(self) -> None:\n        try:\n            self._waker_w.send(b\"a\")\n        except BlockingIOError:\n            pass\n\n    def _consume_waker(self) -> None:\n        try:\n            self._waker_r.recv(1024)\n        except BlockingIOError:\n            pass\n\n    def _start_select(self) -> None:\n        # Capture reader and writer sets here in the event loop\n        # thread to avoid any problems with concurrent\n        # modification while the select loop uses them.\n        with self._select_cond:\n            assert self._select_args is None\n            self._select_args = (list(self._readers.keys()), list(self._writers.keys()))\n            self._select_cond.notify()\n\n    def _run_select(self) -> None:\n        while True:\n            with self._select_cond:\n                while self._select_args is None and not self._closing_selector:\n                    self._select_cond.wait()\n                if self._closing_selector:\n                    return\n                assert self._select_args is not None\n                to_read, to_write = self._select_args\n                self._select_args = None\n\n            # We use the simpler interface of the select module instead of\n            # the more stateful interface in the selectors module because\n            # this class is only intended for use on windows, where\n            # select.select is the only option. The selector interface\n            # does not have well-documented thread-safety semantics that\n            # we can rely on so ensuring proper synchronization would be\n            # tricky.\n            try:\n                # On windows, selecting on a socket for write will not\n                # return the socket when there is an error (but selecting\n                # for reads works). Also select for errors when selecting\n                # for writes, and merge the results.\n                #\n                # This pattern is also used in\n                # https://github.com/python/cpython/blob/v3.8.0/Lib/selectors.py#L312-L317\n                rs, ws, xs = select.select(to_read, to_write, to_write)\n                ws = ws + xs\n            except OSError as e:\n                # After remove_reader or remove_writer is called, the file\n                # descriptor may subsequently be closed on the event loop\n                # thread. It's possible that this select thread hasn't\n                # gotten into the select system call by the time that\n                # happens in which case (at least on macOS), select may\n                # raise a \"bad file descriptor\" error. If we get that\n                # error, check and see if we're also being woken up by\n                # polling the waker alone. If we are, just return to the\n                # event loop and we'll get the updated set of file\n                # descriptors on the next iteration. Otherwise, raise the\n                # original error.\n                if e.errno == getattr(errno, \"WSAENOTSOCK\", errno.EBADF):\n                    rs, _, _ = select.select([self._waker_r.fileno()], [], [], 0)\n                    if rs:\n                        ws = []\n                    else:\n                        raise\n                else:\n                    raise\n\n            try:\n                self._real_loop.call_soon_threadsafe(self._handle_select, rs, ws)\n            except RuntimeError:\n                # \"Event loop is closed\". Swallow the exception for\n                # consistency with PollIOLoop (and logical consistency\n                # with the fact that we can't guarantee that an\n                # add_callback that completes without error will\n                # eventually execute).\n                pass\n            except AttributeError:\n                # ProactorEventLoop may raise this instead of RuntimeError\n                # if call_soon_threadsafe races with a call to close().\n                # Swallow it too for consistency.\n                pass\n\n    def _handle_select(\n        self, rs: List[\"_FileDescriptorLike\"], ws: List[\"_FileDescriptorLike\"]\n    ) -> None:\n        for r in rs:\n            self._handle_event(r, self._readers)\n        for w in ws:\n            self._handle_event(w, self._writers)\n        self._start_select()\n\n    def _handle_event(\n        self,\n        fd: \"_FileDescriptorLike\",\n        cb_map: Dict[\"_FileDescriptorLike\", Callable],\n    ) -> None:\n        try:\n            callback = cb_map[fd]\n        except KeyError:\n            return\n        callback()\n\n    def add_reader(\n        self, fd: \"_FileDescriptorLike\", callback: Callable[..., None], *args: Any\n    ) -> None:\n        self._readers[fd] = functools.partial(callback, *args)\n        self._wake_selector()\n\n    def add_writer(\n        self, fd: \"_FileDescriptorLike\", callback: Callable[..., None], *args: Any\n    ) -> None:\n        self._writers[fd] = functools.partial(callback, *args)\n        self._wake_selector()\n\n    def remove_reader(self, fd: \"_FileDescriptorLike\") -> None:\n        del self._readers[fd]\n        self._wake_selector()\n\n    def remove_writer(self, fd: \"_FileDescriptorLike\") -> None:\n        del self._writers[fd]\n        self._wake_selector()\n"}
A user will ask you to solve a task. You should first draft your thinking process (inner monologue). Then, generate the solution.

Your response format must follow the template below:
<think>
Your thoughts or/and draft, like working through an exercise on scratch paper. Be as casual and as long as you want until you are confident to generate a correct solution.
</think>
<solution>
Final solution presented to the user.
</solution>
system
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
**Issue Title:**
Creating Multiple IOLoop Instances with the Same asyncio Loop Does Not Raise an Error

**Issue Description:**

When attempting to associate multiple `IOLoop` instances with the same `asyncio` loop, no error is raised. This behavior can lead to conflicts and unexpected issues within the application.

**Example Code:**

```python
import asyncio
from tornado.ioloop import IOLoop

# Create a new asyncio event loop
asyncio_loop = asyncio.new_event_loop()

# Associate the first IOLoop with the asyncio loop
loop1 = IOLoop(asyncio_loop=asyncio_loop, make_current=False)

# Attempt to associate a second IOLoop with the same asyncio loop
loop2 = IOLoop(asyncio_loop=asyncio_loop, make_current=False)  # Should raise RuntimeError
```

**Expected Behavior:**

Creating a second `IOLoop` with the same `asyncio` loop should raise a `RuntimeError` indicating that the asyncio loop is already associated with an `IOLoop`. This prevents multiple `IOLoop` instances from interfering with each other.

**Actual Behavior:**

No error is raised when creating the second `IOLoop`, allowing multiple `IOLoop` instances to be associated with the same `asyncio` loop. This can lead to unexpected behavior and potential conflicts within the application.

**Additional Context:**

Prior to this issue, the test `test_explicit_asyncio_loop` failed because it expected a `RuntimeError` when associating multiple `IOLoop` instances with the same `asyncio` loop, but the error was not raised. The recent commit introduces a check to prevent this scenario by raising a `RuntimeError` if an `IOLoop` is already associated with the given `asyncio` loop.

[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### tornado/platform/asyncio.py
"""Bridges between the `asyncio` module and Tornado IOLoop.

.. versionadded:: 3.2

This module integrates Tornado with the ``asyncio`` module introduced
in Python 3.4. This makes it possible to combine the two libraries on
the same event loop.

.. deprecated:: 5.0

   While the code in this module is still used, it is now enabled
   automatically when `asyncio` is available, so applications should
   no longer need to refer to this module directly.

.. note::

   Tornado is designed to use a selector-based event loop. On Windows,
   where a proactor-based event loop has been the default since Python 3.8,
   a selector event loop is emulated by running ``select`` on a separate thread.
   Configuring ``asyncio`` to use a selector event loop may improve performance
   of Tornado (but may reduce performance of other ``asyncio``-based libraries
   in the same process).
"""

import asyncio
import atexit
import concurrent.futures
import errno
import functools
import select
import socket
import sys
import threading
import typing
import warnings
from tornado.gen import convert_yielded
from tornado.ioloop import IOLoop, _Selectable

from typing import Any, TypeVar, Awaitable, Callable, Union, Optional, List, Tuple, Dict

if typing.TYPE_CHECKING:
    from typing import Set  # noqa: F401
    from typing_extensions import Protocol

    class _HasFileno(Protocol):
        def fileno(self) -> int:
            pass

    _FileDescriptorLike = Union[int, _HasFileno]

_T = TypeVar("_T")


# Collection of selector thread event loops to shut down on exit.
_selector_loops = set()  # type: Set[AddThreadSelectorEventLoop]


def _atexit_callback() -> None:
    for loop in _selector_loops:
        with loop._select_cond:
            loop._closing_selector = True
            loop._select_cond.notify()
        try:
            loop._waker_w.send(b"a")
        except BlockingIOError:
            pass
        # If we don't join our (daemon) thread here, we may get a deadlock
        # during interpreter shutdown. I don't really understand why. This
        # deadlock happens every time in CI (both travis and appveyor) but
        # I've never been able to reproduce locally.
        loop._thread.join()
    _selector_loops.clear()


atexit.register(_atexit_callback)

if sys.version_info >= (3, 10):

    def _get_event_loop() -> asyncio.AbstractEventLoop:
        try:
            return asyncio.get_running_loop()
        except RuntimeError:
            pass

        return asyncio.get_event_loop_policy().get_event_loop()


else:
    from asyncio import get_event_loop as _get_event_loop


class BaseAsyncIOLoop(IOLoop):
    def initialize(  # type: ignore
        self, asyncio_loop: asyncio.AbstractEventLoop, **kwargs: Any
    ) -> None:
        # asyncio_loop is always the real underlying IOLoop. This is used in
        # ioloop.py to maintain the asyncio-to-ioloop mappings.
        self.asyncio_loop = asyncio_loop
        # selector_loop is an event loop that implements the add_reader family of
        # methods. Usually the same as asyncio_loop but differs on platforms such
        # as windows where the default event loop does not implement these methods.
        self.selector_loop = asyncio_loop
        if hasattr(asyncio, "ProactorEventLoop") and isinstance(
            asyncio_loop, asyncio.ProactorEventLoop  # type: ignore
        ):
            # Ignore this line for mypy because the abstract method checker
            # doesn't understand dynamic proxies.
            self.selector_loop = AddThreadSelectorEventLoop(asyncio_loop)  # type: ignore
        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)
        self.handlers = {}  # type: Dict[int, Tuple[Union[int, _Selectable], Callable]]
        # Set of fds listening for reads/writes
        self.readers = set()  # type: Set[int]
        self.writers = set()  # type: Set[int]
        self.closing = False
        # If an asyncio loop was closed through an asyncio interface
        # instead of IOLoop.close(), we'd never hear about it and may
        # have left a dangling reference in our map. In case an
        # application (or, more likely, a test suite) creates and
        # destroys a lot of event loops in this way, check here to
        # ensure that we don't have a lot of dead loops building up in
        # the map.
        #
        # TODO(bdarnell): consider making self.asyncio_loop a weakref
        # for AsyncIOMainLoop and make _ioloop_for_asyncio a
        # WeakKeyDictionary.
        for loop in IOLoop._ioloop_for_asyncio.copy():
            if loop.is_closed():
                try:
                    del IOLoop._ioloop_for_asyncio[loop]
                except KeyError:
                    pass
        IOLoop._ioloop_for_asyncio[asyncio_loop] = self

        self._thread_identity = 0

        super().initialize(**kwargs)

        def assign_thread_identity() -> None:
            self._thread_identity = threading.get_ident()

        self.add_callback(assign_thread_identity)

    def close(self, all_fds: bool = False) -> None:
        self.closing = True
        for fd in list(self.handlers):
            fileobj, handler_func = self.handlers[fd]
            self.remove_handler(fd)
            if all_fds:
                self.close_fd(fileobj)
        # Remove the mapping before closing the asyncio loop. If this
        # happened in the other order, we could race against another
        # initialize() call which would see the closed asyncio loop,
        # assume it was closed from the asyncio side, and do this
        # cleanup for us, leading to a KeyError.
        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]
        if self.selector_loop is not self.asyncio_loop:
            self.selector_loop.close()
        self.asyncio_loop.close()

    def add_handler(
        self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int
    ) -> None:
        fd, fileobj = self.split_fd(fd)
        if fd in self.handlers:
            raise ValueError("fd %s added twice" % fd)
        self.handlers[fd] = (fileobj, handler)
        if events & IOLoop.READ:
            self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)
            self.readers.add(fd)
        if events & IOLoop.WRITE:
            self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)
            self.writers.add(fd)

    def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:
        fd, fileobj = self.split_fd(fd)
        if events & IOLoop.READ:
            if fd not in self.readers:
                self.selector_loop.add_reader(fd, self._handle_events, fd, IOLoop.READ)
                self.readers.add(fd)
        else:
            if fd in self.readers:
                self.selector_loop.remove_reader(fd)
                self.readers.remove(fd)
        if events & IOLoop.WRITE:
            if fd not in self.writers:
                self.selector_loop.add_writer(fd, self._handle_events, fd, IOLoop.WRITE)
                self.writers.add(fd)
        else:
            if fd in self.writers:
                self.selector_loop.remove_writer(fd)
                self.writers.remove(fd)

    def remove_handler(self, fd: Union[int, _Selectable]) -> None:
        fd, fileobj = self.split_fd(fd)
        if fd not in self.handlers:
            return
        if fd in self.readers:
            self.selector_loop.remove_reader(fd)
            self.readers.remove(fd)
        if fd in self.writers:
            self.selector_loop.remove_writer(fd)
            self.writers.remove(fd)
        del self.handlers[fd]

    def _handle_events(self, fd: int, events: int) -> None:
        fileobj, handler_func = self.handlers[fd]
        handler_func(fileobj, events)

    def start(self) -> None:
        try:
            old_loop = _get_event_loop()
        except (RuntimeError, AssertionError):
            old_loop = None  # type: ignore
        try:
            asyncio.set_event_loop(self.asyncio_loop)
            self.asyncio_loop.run_forever()
        finally:
            asyncio.set_event_loop(old_loop)

    def stop(self) -> None:
        self.asyncio_loop.stop()

    def call_at(
        self, when: float, callback: Callable, *args: Any, **kwargs: Any
    ) -> object:
        # asyncio.call_at supports *args but not **kwargs, so bind them here.
        # We do not synchronize self.time and asyncio_loop.time, so
        # convert from absolute to relative.
        return self.asyncio_loop.call_later(
            max(0, when - self.time()),
            self._run_callback,
            functools.partial(callback, *args, **kwargs),
        )

    def remove_timeout(self, timeout: object) -> None:
        timeout.cancel()  # type: ignore

    def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:
        if threading.get_ident() == self._thread_identity:
            call_soon = self.asyncio_loop.call_soon
        else:
            call_soon = self.asyncio_loop.call_soon_threadsafe
        try:
            call_soon(self._run_callback, functools.partial(callback, *args, **kwargs))
        except RuntimeError:
            # "Event loop is closed". Swallow the exception for
            # consistency with PollIOLoop (and logical consistency
            # with the fact that we can't guarantee that an
            # add_callback that completes without error will
            # eventually execute).
            pass
        except AttributeError:
            # ProactorEventLoop may raise this instead of RuntimeError
            # if call_soon_threadsafe races with a call to close().
            # Swallow it too for consistency.
            pass

    def add_callback_from_signal(
        self, callback: Callable, *args: Any, **kwargs: Any
    ) -> None:
        try:
            self.asyncio_loop.call_soon_threadsafe(
                self._run_callback, functools.partial(callback, *args, **kwargs)
            )
        except RuntimeError:
            pass

    def run_in_executor(
        self,
        executor: Optional[concurrent.futures.Executor],
        func: Callable[..., _T],
        *args: Any
    ) -> Awaitable[_T]:
        return self.asyncio_loop.run_in_executor(executor, func, *args)

    def set_default_executor(self, executor: concurrent.futures.Executor) -> None:
        return self.asyncio_loop.set_default_executor(executor)


class AsyncIOMainLoop(BaseAsyncIOLoop):
    """``AsyncIOMainLoop`` creates an `.IOLoop` that corresponds to the
    current ``asyncio`` event loop (i.e. the one returned by
    ``asyncio.get_event_loop()``).

    .. deprecated:: 5.0

       Now used automatically when appropriate; it is no longer necessary
       to refer to this class directly.

    .. versionchanged:: 5.0

       Closing an `AsyncIOMainLoop` now closes the underlying asyncio loop.
    """

    def initialize(self, **kwargs: Any) -> None:  # type: ignore
        super().initialize(asyncio.get_event_loop(), **kwargs)

    def make_current(self) -> None:
        # AsyncIOMainLoop already refers to the current asyncio loop so
        # nothing to do here.
        pass


class AsyncIOLoop(BaseAsyncIOLoop):
    """``AsyncIOLoop`` is an `.IOLoop` that runs on an ``asyncio`` event loop.
    This class follows the usual Tornado semantics for creating new
    ``IOLoops``; these loops are not necessarily related to the
    ``asyncio`` default event loop.

    Each ``AsyncIOLoop`` creates a new ``asyncio.EventLoop``; this object
    can be accessed with the ``asyncio_loop`` attribute.

    .. versionchanged:: 6.2

       Support explicit ``asyncio_loop`` argument
       for specifying the asyncio loop to attach to,
       rather than always creating a new one with the default policy.

    .. versionchanged:: 5.0

       When an ``AsyncIOLoop`` becomes the current `.IOLoop`, it also sets
       the current `asyncio` event loop.

    .. deprecated:: 5.0

       Now used automatically when appropriate; it is no longer necessary
       to refer to this class directly.
    """

    def initialize(self, **kwargs: Any) -> None:  # type: ignore
        self.is_current = False
        loop = None
        if "asyncio_loop" not in kwargs:
            kwargs["asyncio_loop"] = loop = asyncio.new_event_loop()
        try:
            super().initialize(**kwargs)
        except Exception:
            # If initialize() does not succeed (taking ownership of the loop),
            # we have to close it.
            if loop is not None:
                loop.close()
            raise

    def close(self, all_fds: bool = False) -> None:
        if self.is_current:
            self._clear_current()
        super().close(all_fds=all_fds)

    def make_current(self) -> None:
        warnings.warn(
            "make_current is deprecated; start the event loop first",
            DeprecationWarning,
        )
        if not self.is_current:
            try:
                self.old_asyncio = asyncio.get_event_loop()
            except (RuntimeError, AssertionError):
                self.old_asyncio = None  # type: ignore
            self.is_current = True
        asyncio.set_event_loop(self.asyncio_loop)

    def _clear_current_hook(self) -> None:
        if self.is_current:
            asyncio.set_event_loop(self.old_asyncio)
            self.is_current = False


def to_tornado_future(asyncio_future: asyncio.Future) -> asyncio.Future:
    """Convert an `asyncio.Future` to a `tornado.concurrent.Future`.

    .. versionadded:: 4.1

    .. deprecated:: 5.0
       Tornado ``Futures`` have been merged with `asyncio.Future`,
       so this method is now a no-op.
    """
    return asyncio_future


def to_asyncio_future(tornado_future: asyncio.Future) -> asyncio.Future:
    """Convert a Tornado yieldable object to an `asyncio.Future`.

    .. versionadded:: 4.1

    .. versionchanged:: 4.3
       Now accepts any yieldable object, not just
       `tornado.concurrent.Future`.

    .. deprecated:: 5.0
       Tornado ``Futures`` have been merged with `asyncio.Future`,
       so this method is now equivalent to `tornado.gen.convert_yielded`.
    """
    return convert_yielded(tornado_future)


if sys.platform == "win32" and hasattr(asyncio, "WindowsSelectorEventLoopPolicy"):
    # "Any thread" and "selector" should be orthogonal, but there's not a clean
    # interface for composing policies so pick the right base.
    _BasePolicy = asyncio.WindowsSelectorEventLoopPolicy  # type: ignore
else:
    _BasePolicy = asyncio.DefaultEventLoopPolicy


class AnyThreadEventLoopPolicy(_BasePolicy):  # type: ignore
    """Event loop policy that allows loop creation on any thread.

    The default `asyncio` event loop policy only automatically creates
    event loops in the main threads. Other threads must create event
    loops explicitly or `asyncio.get_event_loop` (and therefore
    `.IOLoop.current`) will fail. Installing this policy allows event
    loops to be created automatically on any thread, matching the
    behavior of Tornado versions prior to 5.0 (or 5.0 on Python 2).

    Usage::

        asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())

    .. versionadded:: 5.0

    .. deprecated:: 6.2

        ``AnyThreadEventLoopPolicy`` affects the implicit creation
        of an event loop, which is deprecated in Python 3.10 and
        will be removed in a future version of Python. At that time
        ``AnyThreadEventLoopPolicy`` will no longer be useful.
        If you are relying on it, use `asyncio.new_event_loop`
        or `asyncio.run` explicitly in any non-main threads that
        need event loops.
    """

    def __init__(self) -> None:
        super().__init__()
        warnings.warn(
            "AnyThreadEventLoopPolicy is deprecated, use asyncio.run "
            "or asyncio.new_event_loop instead",
            DeprecationWarning,
        )

    def get_event_loop(self) -> asyncio.AbstractEventLoop:
        try:
            return super().get_event_loop()
        except (RuntimeError, AssertionError):
            # This was an AssertionError in Python 3.4.2 (which ships with Debian Jessie)
            # and changed to a RuntimeError in 3.4.3.
            # "There is no current event loop in thread %r"
            loop = self.new_event_loop()
            self.set_event_loop(loop)
            return loop


class AddThreadSelectorEventLoop(asyncio.AbstractEventLoop):
    """Wrap an event loop to add implementations of the ``add_reader`` method family.

    Instances of this class start a second thread to run a selector.
    This thread is completely hidden from the user; all callbacks are
    run on the wrapped event loop's thread.

    This class is used automatically by Tornado; applications should not need
    to refer to it directly.

    It is safe to wrap any event loop with this class, although it only makes sense
    for event loops that do not implement the ``add_reader`` family of methods
    themselves (i.e. ``WindowsProactorEventLoop``)

    Closing the ``AddThreadSelectorEventLoop`` also closes the wrapped event loop.

    """

    # This class is a __getattribute__-based proxy. All attributes other than those
    # in this set are proxied through to the underlying loop.
    MY_ATTRIBUTES = {
        "_consume_waker",
        "_select_cond",
        "_select_args",
        "_closing_selector",
        "_thread",
        "_handle_event",
        "_readers",
        "_real_loop",
        "_start_select",
        "_run_select",
        "_handle_select",
        "_wake_selector",
        "_waker_r",
        "_waker_w",
        "_writers",
        "add_reader",
        "add_writer",
        "close",
        "remove_reader",
        "remove_writer",
    }

    def __getattribute__(self, name: str) -> Any:
        if name in AddThreadSelectorEventLoop.MY_ATTRIBUTES:
            return super().__getattribute__(name)
        return getattr(self._real_loop, name)

    def __init__(self, real_loop: asyncio.AbstractEventLoop) -> None:
        self._real_loop = real_loop

        # Create a thread to run the select system call. We manage this thread
        # manually so we can trigger a clean shutdown from an atexit hook. Note
        # that due to the order of operations at shutdown, only daemon threads
        # can be shut down in this way (non-daemon threads would require the
        # introduction of a new hook: https://bugs.python.org/issue41962)
        self._select_cond = threading.Condition()
        self._select_args = (
            None
        )  # type: Optional[Tuple[List[_FileDescriptorLike], List[_FileDescriptorLike]]]
        self._closing_selector = False
        self._thread = threading.Thread(
            name="Tornado selector",
            daemon=True,
            target=self._run_select,
        )
        self._thread.start()
        # Start the select loop once the loop is started.
        self._real_loop.call_soon(self._start_select)

        self._readers = {}  # type: Dict[_FileDescriptorLike, Callable]
        self._writers = {}  # type: Dict[_FileDescriptorLike, Callable]

        # Writing to _waker_w will wake up the selector thread, which
        # watches for _waker_r to be readable.
        self._waker_r, self._waker_w = socket.socketpair()
        self._waker_r.setblocking(False)
        self._waker_w.setblocking(False)
        _selector_loops.add(self)
        self.add_reader(self._waker_r, self._consume_waker)

    def __del__(self) -> None:
        # If the top-level application code uses asyncio interfaces to
        # start and stop the event loop, no objects created in Tornado
        # can get a clean shutdown notification. If we're just left to
        # be GC'd, we must explicitly close our sockets to avoid
        # logging warnings.
        _selector_loops.discard(self)
        self._waker_r.close()
        self._waker_w.close()

    def close(self) -> None:
        with self._select_cond:
            self._closing_selector = True
            self._select_cond.notify()
        self._wake_selector()
        self._thread.join()
        _selector_loops.discard(self)
        self._waker_r.close()
        self._waker_w.close()
        self._real_loop.close()

    def _wake_selector(self) -> None:
        try:
            self._waker_w.send(b"a")
        except BlockingIOError:
            pass

    def _consume_waker(self) -> None:
        try:
            self._waker_r.recv(1024)
        except BlockingIOError:
            pass

    def _start_select(self) -> None:
        # Capture reader and writer sets here in the event loop
        # thread to avoid any problems with concurrent
        # modification while the select loop uses them.
        with self._select_cond:
            assert self._select_args is None
            self._select_args = (list(self._readers.keys()), list(self._writers.keys()))
            self._select_cond.notify()

    def _run_select(self) -> None:
        while True:
            with self._select_cond:
                while self._select_args is None and not self._closing_selector:
                    self._select_cond.wait()
                if self._closing_selector:
                    return
                assert self._select_args is not None
                to_read, to_write = self._select_args
                self._select_args = None

            # We use the simpler interface of the select module instead of
            # the more stateful interface in the selectors module because
            # this class is only intended for use on windows, where
            # select.select is the only option. The selector interface
            # does not have well-documented thread-safety semantics that
            # we can rely on so ensuring proper synchronization would be
            # tricky.
            try:
                # On windows, selecting on a socket for write will not
                # return the socket when there is an error (but selecting
                # for reads works). Also select for errors when selecting
                # for writes, and merge the results.
                #
                # This pattern is also used in
                # https://github.com/python/cpython/blob/v3.8.0/Lib/selectors.py#L312-L317
                rs, ws, xs = select.select(to_read, to_write, to_write)
                ws = ws + xs
            except OSError as e:
                # After remove_reader or remove_writer is called, the file
                # descriptor may subsequently be closed on the event loop
                # thread. It's possible that this select thread hasn't
                # gotten into the select system call by the time that
                # happens in which case (at least on macOS), select may
                # raise a "bad file descriptor" error. If we get that
                # error, check and see if we're also being woken up by
                # polling the waker alone. If we are, just return to the
                # event loop and we'll get the updated set of file
                # descriptors on the next iteration. Otherwise, raise the
                # original error.
                if e.errno == getattr(errno, "WSAENOTSOCK", errno.EBADF):
                    rs, _, _ = select.select([self._waker_r.fileno()], [], [], 0)
                    if rs:
                        ws = []
                    else:
                        raise
                else:
                    raise

            try:
                self._real_loop.call_soon_threadsafe(self._handle_select, rs, ws)
            except RuntimeError:
                # "Event loop is closed". Swallow the exception for
                # consistency with PollIOLoop (and logical consistency
                # with the fact that we can't guarantee that an
                # add_callback that completes without error will
                # eventually execute).
                pass
            except AttributeError:
                # ProactorEventLoop may raise this instead of RuntimeError
                # if call_soon_threadsafe races with a call to close().
                # Swallow it too for consistency.
                pass

    def _handle_select(
        self, rs: List["_FileDescriptorLike"], ws: List["_FileDescriptorLike"]
    ) -> None:
        for r in rs:
            self._handle_event(r, self._readers)
        for w in ws:
            self._handle_event(w, self._writers)
        self._start_select()

    def _handle_event(
        self,
        fd: "_FileDescriptorLike",
        cb_map: Dict["_FileDescriptorLike", Callable],
    ) -> None:
        try:
            callback = cb_map[fd]
        except KeyError:
            return
        callback()

    def add_reader(
        self, fd: "_FileDescriptorLike", callback: Callable[..., None], *args: Any
    ) -> None:
        self._readers[fd] = functools.partial(callback, *args)
        self._wake_selector()

    def add_writer(
        self, fd: "_FileDescriptorLike", callback: Callable[..., None], *args: Any
    ) -> None:
        self._writers[fd] = functools.partial(callback, *args)
        self._wake_selector()

    def remove_reader(self, fd: "_FileDescriptorLike") -> None:
        del self._readers[fd]
        self._wake_selector()

    def remove_writer(self, fd: "_FileDescriptorLike") -> None:
        del self._writers[fd]
        self._wake_selector()

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
user