{
    "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 requires the `~asyncio.AbstractEventLoop.add_reader` family of\n   methods, so it is not compatible with the `~asyncio.ProactorEventLoop` on\n   Windows. Use the `~asyncio.SelectorEventLoop` instead.\n\"\"\"\n\nfrom __future__ import absolute_import, division, print_function\nimport functools\n\nimport tornado.concurrent\nfrom tornado.gen import convert_yielded\nfrom tornado.ioloop import IOLoop\nfrom tornado import stack_context\n\nimport asyncio\n\n\nclass BaseAsyncIOLoop(IOLoop):\n    def initialize(self, asyncio_loop, close_loop=False, **kwargs):\n        self.asyncio_loop = asyncio_loop\n        self.close_loop = close_loop\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}\n        # Set of fds listening for reads/writes\n        self.readers = set()\n        self.writers = set()\n        self.closing = False\n        super(BaseAsyncIOLoop, self).initialize(**kwargs)\n\n    def close(self, all_fds=False):\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        if self.close_loop:\n            self.asyncio_loop.close()\n\n    def add_handler(self, fd, handler, events):\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, stack_context.wrap(handler))\n        if events & IOLoop.READ:\n            self.asyncio_loop.add_reader(\n                fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.asyncio_loop.add_writer(\n                fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd, events):\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.asyncio_loop.add_reader(\n                    fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.asyncio_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.asyncio_loop.add_writer(\n                    fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.asyncio_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd):\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.asyncio_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.asyncio_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd, events):\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self):\n        old_current = IOLoop.current(instance=False)\n        try:\n            self._setup_logging()\n            self.make_current()\n            self.asyncio_loop.run_forever()\n        finally:\n            if old_current is None:\n                IOLoop.clear_current()\n            else:\n                old_current.make_current()\n\n    def stop(self):\n        self.asyncio_loop.stop()\n\n    def call_at(self, when, callback, *args, **kwargs):\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()), self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    def remove_timeout(self, timeout):\n        timeout.cancel()\n\n    def add_callback(self, callback, *args, **kwargs):\n        if self.closing:\n            # TODO: this is racy; we need a lock to ensure that the\n            # loop isn't closed during call_soon_threadsafe.\n            raise RuntimeError(\"IOLoop is closing\")\n        self.asyncio_loop.call_soon_threadsafe(\n            self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    add_callback_from_signal = add_callback\n\n    def run_in_executor(self, executor, func, *args):\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor):\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    def initialize(self, **kwargs):\n        super(AsyncIOMainLoop, self).initialize(asyncio.get_event_loop(),\n                                                close_loop=False, **kwargs)\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:: 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    def initialize(self, **kwargs):\n        loop = asyncio.new_event_loop()\n        try:\n            super(AsyncIOLoop, self).initialize(loop, close_loop=True, **kwargs)\n        except Exception:\n            # If initialize() does not succeed (taking ownership of the loop),\n            # we have to close it.\n            loop.close()\n            raise\n\n    def make_current(self):\n        super(AsyncIOLoop, self).make_current()\n        try:\n            self.old_asyncio = asyncio.get_event_loop()\n        except RuntimeError:\n            self.old_asyncio = None\n        asyncio.set_event_loop(self.asyncio_loop)\n\n    def _clear_current_hook(self):\n        asyncio.set_event_loop(self.old_asyncio)\n\n\ndef to_tornado_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):\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\nclass AnyThreadEventLoopPolicy(asyncio.DefaultEventLoopPolicy):\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    \"\"\"\n    def get_event_loop(self):\n        try:\n            return super().get_event_loop()\n        except RuntimeError:\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",
            "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 requires the `~asyncio.AbstractEventLoop.add_reader` family of\n   methods, so it is not compatible with the `~asyncio.ProactorEventLoop` on\n   Windows. Use the `~asyncio.SelectorEventLoop` instead.\n\"\"\"\n\nfrom __future__ import absolute_import, division, print_function\nimport functools\n\nimport tornado.concurrent\nfrom tornado.gen import convert_yielded\nfrom tornado.ioloop import IOLoop\nfrom tornado import stack_context\n\nimport asyncio\n\n\nclass BaseAsyncIOLoop(IOLoop):\n    def initialize(self, asyncio_loop, close_loop=False, **kwargs):\n        self.asyncio_loop = asyncio_loop\n        self.close_loop = close_loop\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}\n        # Set of fds listening for reads/writes\n        self.readers = set()\n        self.writers = set()\n        self.closing = False\n        super(BaseAsyncIOLoop, self).initialize(**kwargs)\n\n    def close(self, all_fds=False):\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        if self.close_loop:\n            self.asyncio_loop.close()\n\n    def add_handler(self, fd, handler, events):\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, stack_context.wrap(handler))\n        if events & IOLoop.READ:\n            self.asyncio_loop.add_reader(\n                fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.asyncio_loop.add_writer(\n                fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd, events):\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.asyncio_loop.add_reader(\n                    fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.asyncio_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.asyncio_loop.add_writer(\n                    fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.asyncio_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd):\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.asyncio_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.asyncio_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd, events):\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self):\n        old_current = IOLoop.current(instance=False)\n        try:\n            self._setup_logging()\n            self.make_current()\n            self.asyncio_loop.run_forever()\n        finally:\n            if old_current is None:\n                IOLoop.clear_current()\n            else:\n                old_current.make_current()\n\n    def stop(self):\n        self.asyncio_loop.stop()\n\n    def call_at(self, when, callback, *args, **kwargs):\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()), self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    def remove_timeout(self, timeout):\n        timeout.cancel()\n\n    def add_callback(self, callback, *args, **kwargs):\n        try:\n            self.asyncio_loop.call_soon_threadsafe(\n                self._run_callback,\n                functools.partial(stack_context.wrap(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\n    add_callback_from_signal = add_callback\n\n    def run_in_executor(self, executor, func, *args):\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor):\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    def initialize(self, **kwargs):\n        super(AsyncIOMainLoop, self).initialize(asyncio.get_event_loop(),\n                                                close_loop=False, **kwargs)\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:: 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    def initialize(self, **kwargs):\n        loop = asyncio.new_event_loop()\n        try:\n            super(AsyncIOLoop, self).initialize(loop, close_loop=True, **kwargs)\n        except Exception:\n            # If initialize() does not succeed (taking ownership of the loop),\n            # we have to close it.\n            loop.close()\n            raise\n\n    def make_current(self):\n        super(AsyncIOLoop, self).make_current()\n        try:\n            self.old_asyncio = asyncio.get_event_loop()\n        except RuntimeError:\n            self.old_asyncio = None\n        asyncio.set_event_loop(self.asyncio_loop)\n\n    def _clear_current_hook(self):\n        asyncio.set_event_loop(self.old_asyncio)\n\n\ndef to_tornado_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):\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\nclass AnyThreadEventLoopPolicy(asyncio.DefaultEventLoopPolicy):\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    \"\"\"\n    def get_event_loop(self):\n        try:\n            return super().get_event_loop()\n        except RuntimeError:\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",
            "header": {
                "file": {
                    "path": "tornado/platform/asyncio.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "adc7c1ae",
                "new_commit_hash": "42c8a800",
                "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": 130,
                            "length": 13
                        },
                        "new_range": {
                            "start": 130,
                            "length": 17
                        },
                        "section": "class BaseAsyncIOLoop(IOLoop):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        timeout.cancel()",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def add_callback(self, callback, *args, **kwargs):",
                                "type": "context"
                            },
                            {
                                "content": "        if self.closing:",
                                "type": "deleted"
                            },
                            {
                                "content": "            # TODO: this is racy; we need a lock to ensure that the",
                                "type": "deleted"
                            },
                            {
                                "content": "            # loop isn't closed during call_soon_threadsafe.",
                                "type": "deleted"
                            },
                            {
                                "content": "            raise RuntimeError(\"IOLoop is closing\")",
                                "type": "deleted"
                            },
                            {
                                "content": "        self.asyncio_loop.call_soon_threadsafe(",
                                "type": "deleted"
                            },
                            {
                                "content": "            self._run_callback,",
                                "type": "deleted"
                            },
                            {
                                "content": "            functools.partial(stack_context.wrap(callback), *args, **kwargs))",
                                "type": "deleted"
                            },
                            {
                                "content": "        try:",
                                "type": "added"
                            },
                            {
                                "content": "            self.asyncio_loop.call_soon_threadsafe(",
                                "type": "added"
                            },
                            {
                                "content": "                self._run_callback,",
                                "type": "added"
                            },
                            {
                                "content": "                functools.partial(stack_context.wrap(callback), *args, **kwargs))",
                                "type": "added"
                            },
                            {
                                "content": "        except RuntimeError:",
                                "type": "added"
                            },
                            {
                                "content": "            # \"Event loop is closed\". Swallow the exception for",
                                "type": "added"
                            },
                            {
                                "content": "            # consistency with PollIOLoop (and logical consistency",
                                "type": "added"
                            },
                            {
                                "content": "            # with the fact that we can't guarantee that an",
                                "type": "added"
                            },
                            {
                                "content": "            # add_callback that completes without error will",
                                "type": "added"
                            },
                            {
                                "content": "            # eventually execute).",
                                "type": "added"
                            },
                            {
                                "content": "            pass",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    add_callback_from_signal = add_callback",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/platform/asyncio.py",
                            "type": "method",
                            "name": "BaseAsyncIOLoop.add_callback",
                            "content": "class BaseAsyncIOLoop(IOLoop):\n    def initialize(self, asyncio_loop, close_loop=False, **kwargs):\n        self.asyncio_loop = asyncio_loop\n        self.close_loop = close_loop\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}\n        # Set of fds listening for reads/writes\n        self.readers = set()\n        self.writers = set()\n        self.closing = False\n        super(BaseAsyncIOLoop, self).initialize(**kwargs)\n\n    def close(self, all_fds=False):\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        if self.close_loop:\n            self.asyncio_loop.close()\n\n    def add_handler(self, fd, handler, events):\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, stack_context.wrap(handler))\n        if events & IOLoop.READ:\n            self.asyncio_loop.add_reader(\n                fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.asyncio_loop.add_writer(\n                fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd, events):\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.asyncio_loop.add_reader(\n                    fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.asyncio_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.asyncio_loop.add_writer(\n                    fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.asyncio_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd):\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.asyncio_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.asyncio_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd, events):\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self):\n        old_current = IOLoop.current(instance=False)\n        try:\n            self._setup_logging()\n            self.make_current()\n            self.asyncio_loop.run_forever()\n        finally:\n            if old_current is None:\n                IOLoop.clear_current()\n            else:\n                old_current.make_current()\n\n    def stop(self):\n        self.asyncio_loop.stop()\n\n    def call_at(self, when, callback, *args, **kwargs):\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()), self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    def remove_timeout(self, timeout):\n        timeout.cancel()\n\n    def add_callback(self, callback, *args, **kwargs):\n        if self.closing:\n            # TODO: this is racy; we need a lock to ensure that the\n            # loop isn't closed during call_soon_threadsafe.\n            raise RuntimeError(\"IOLoop is closing\")\n        self.asyncio_loop.call_soon_threadsafe(\n            self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    add_callback_from_signal = add_callback\n\n    def run_in_executor(self, executor, func, *args):\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor):\n        return self.asyncio_loop.set_default_executor(executor)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 132,
                            "end_lineno": 139,
                            "parent": {
                                "file_name": "tornado/platform/asyncio.py",
                                "type": "class",
                                "name": "BaseAsyncIOLoop",
                                "content": "class BaseAsyncIOLoop(IOLoop):\n    def initialize(self, asyncio_loop, close_loop=False, **kwargs):\n        self.asyncio_loop = asyncio_loop\n        self.close_loop = close_loop\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}\n        # Set of fds listening for reads/writes\n        self.readers = set()\n        self.writers = set()\n        self.closing = False\n        super(BaseAsyncIOLoop, self).initialize(**kwargs)\n\n    def close(self, all_fds=False):\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        if self.close_loop:\n            self.asyncio_loop.close()\n\n    def add_handler(self, fd, handler, events):\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, stack_context.wrap(handler))\n        if events & IOLoop.READ:\n            self.asyncio_loop.add_reader(\n                fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.asyncio_loop.add_writer(\n                fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd, events):\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.asyncio_loop.add_reader(\n                    fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.asyncio_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.asyncio_loop.add_writer(\n                    fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.asyncio_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd):\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.asyncio_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.asyncio_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd, events):\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self):\n        old_current = IOLoop.current(instance=False)\n        try:\n            self._setup_logging()\n            self.make_current()\n            self.asyncio_loop.run_forever()\n        finally:\n            if old_current is None:\n                IOLoop.clear_current()\n            else:\n                old_current.make_current()\n\n    def stop(self):\n        self.asyncio_loop.stop()\n\n    def call_at(self, when, callback, *args, **kwargs):\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()), self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    def remove_timeout(self, timeout):\n        timeout.cancel()\n\n    def add_callback(self, callback, *args, **kwargs):\n        if self.closing:\n            # TODO: this is racy; we need a lock to ensure that the\n            # loop isn't closed during call_soon_threadsafe.\n            raise RuntimeError(\"IOLoop is closing\")\n        self.asyncio_loop.call_soon_threadsafe(\n            self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    add_callback_from_signal = add_callback\n\n    def run_in_executor(self, executor, func, *args):\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor):\n        return self.asyncio_loop.set_default_executor(executor)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 33,
                                "end_lineno": 147,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/platform/asyncio.py",
                            "type": "class",
                            "name": "BaseAsyncIOLoop",
                            "content": "class BaseAsyncIOLoop(IOLoop):\n    def initialize(self, asyncio_loop, close_loop=False, **kwargs):\n        self.asyncio_loop = asyncio_loop\n        self.close_loop = close_loop\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}\n        # Set of fds listening for reads/writes\n        self.readers = set()\n        self.writers = set()\n        self.closing = False\n        super(BaseAsyncIOLoop, self).initialize(**kwargs)\n\n    def close(self, all_fds=False):\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        if self.close_loop:\n            self.asyncio_loop.close()\n\n    def add_handler(self, fd, handler, events):\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, stack_context.wrap(handler))\n        if events & IOLoop.READ:\n            self.asyncio_loop.add_reader(\n                fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.asyncio_loop.add_writer(\n                fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd, events):\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.asyncio_loop.add_reader(\n                    fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.asyncio_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.asyncio_loop.add_writer(\n                    fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.asyncio_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd):\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.asyncio_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.asyncio_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd, events):\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self):\n        old_current = IOLoop.current(instance=False)\n        try:\n            self._setup_logging()\n            self.make_current()\n            self.asyncio_loop.run_forever()\n        finally:\n            if old_current is None:\n                IOLoop.clear_current()\n            else:\n                old_current.make_current()\n\n    def stop(self):\n        self.asyncio_loop.stop()\n\n    def call_at(self, when, callback, *args, **kwargs):\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()), self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    def remove_timeout(self, timeout):\n        timeout.cancel()\n\n    def add_callback(self, callback, *args, **kwargs):\n        if self.closing:\n            # TODO: this is racy; we need a lock to ensure that the\n            # loop isn't closed during call_soon_threadsafe.\n            raise RuntimeError(\"IOLoop is closing\")\n        self.asyncio_loop.call_soon_threadsafe(\n            self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    add_callback_from_signal = add_callback\n\n    def run_in_executor(self, executor, func, *args):\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor):\n        return self.asyncio_loop.set_default_executor(executor)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 33,
                            "end_lineno": 147,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "#!/usr/bin/env python\n\n\nfrom __future__ import absolute_import, division, print_function\nimport contextlib\nimport datetime\nimport functools\nimport socket\nimport subprocess\nimport sys\nimport threading\nimport time\nimport types\n\nfrom tornado.escape import native_str\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop, TimeoutError, PollIOLoop, PeriodicCallback\nfrom tornado.log import app_log\nfrom tornado.platform.select import _Select\nfrom tornado.stack_context import ExceptionStackContext, StackContext, wrap, NullContext\nfrom tornado.testing import AsyncTestCase, bind_unused_port, ExpectLog, gen_test\nfrom tornado.test.util import unittest, skipIfNonUnix, skipOnTravis, skipBefore35, exec_test\nfrom tornado.concurrent import Future\n\ntry:\n    from concurrent import futures\nexcept ImportError:\n    futures = None\n\ntry:\n    import asyncio\nexcept ImportError:\n    asyncio = None\n\ntry:\n    import twisted\nexcept ImportError:\n    twisted = None\n\n\nclass FakeTimeSelect(_Select):\n    def __init__(self):\n        self._time = 1000\n        super(FakeTimeSelect, self).__init__()\n\n    def time(self):\n        return self._time\n\n    def sleep(self, t):\n        self._time += t\n\n    def poll(self, timeout):\n        events = super(FakeTimeSelect, self).poll(0)\n        if events:\n            return events\n        self._time += timeout\n        return []\n\n\nclass FakeTimeIOLoop(PollIOLoop):\n    \"\"\"IOLoop implementation with a fake and deterministic clock.\n\n    The clock advances as needed to trigger timeouts immediately.\n    For use when testing code that involves the passage of time\n    and no external dependencies.\n    \"\"\"\n    def initialize(self):\n        self.fts = FakeTimeSelect()\n        super(FakeTimeIOLoop, self).initialize(impl=self.fts,\n                                               time_func=self.fts.time)\n\n    def sleep(self, t):\n        \"\"\"Simulate a blocking sleep by advancing the clock.\"\"\"\n        self.fts.sleep(t)\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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        # Issue #635: add_callback() should raise a clean exception\n        # if called while another thread is closing the IOLoop.\n        if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'):\n            raise unittest.SkipTest(\"AsyncIOMainLoop shutdown not thread safe\")\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        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            try:\n                other_ioloop.add_callback(lambda: None)\n            except RuntimeError as e:\n                self.assertEqual(\"IOLoop is closing\", str(e))\n                break\n\n    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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\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        self.io_loop = None\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            def f():\n                self.current_io_loop = IOLoop.current()\n                self.io_loop.stop()\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 TestIOLoopAddCallback(AsyncTestCase):\n    def setUp(self):\n        super(TestIOLoopAddCallback, self).setUp()\n        self.active_contexts = []\n\n    def add_callback(self, callback, *args, **kwargs):\n        self.io_loop.add_callback(callback, *args, **kwargs)\n\n    @contextlib.contextmanager\n    def context(self, name):\n        self.active_contexts.append(name)\n        yield\n        self.assertEqual(self.active_contexts.pop(), name)\n\n    def test_pre_wrap(self):\n        # A pre-wrapped callback is run in the context in which it was\n        # wrapped, not when it was added to the IOLoop.\n        def f1():\n            self.assertIn('c1', self.active_contexts)\n            self.assertNotIn('c2', self.active_contexts)\n            self.stop()\n\n        with StackContext(functools.partial(self.context, 'c1')):\n            wrapped = wrap(f1)\n\n        with StackContext(functools.partial(self.context, 'c2')):\n            self.add_callback(wrapped)\n\n        self.wait()\n\n    def test_pre_wrap_with_args(self):\n        # Same as test_pre_wrap, but the function takes arguments.\n        # Implementation note: The function must not be wrapped in a\n        # functools.partial until after it has been passed through\n        # stack_context.wrap\n        def f1(foo, bar):\n            self.assertIn('c1', self.active_contexts)\n            self.assertNotIn('c2', self.active_contexts)\n            self.stop((foo, bar))\n\n        with StackContext(functools.partial(self.context, 'c1')):\n            wrapped = wrap(f1)\n\n        with StackContext(functools.partial(self.context, 'c2')):\n            self.add_callback(wrapped, 1, bar=2)\n\n        result = self.wait()\n        self.assertEqual(result, (1, 2))\n\n\nclass TestIOLoopAddCallbackFromSignal(TestIOLoopAddCallback):\n    # Repeat the add_callback tests using add_callback_from_signal\n    def add_callback(self, callback, *args, **kwargs):\n        self.io_loop.add_callback_from_signal(callback, *args, **kwargs)\n\n\n@unittest.skipIf(futures is None, \"futures module not present\")\nclass TestIOLoopFutures(AsyncTestCase):\n    def test_add_future_threads(self):\n        with futures.ThreadPoolExecutor(1) as pool:\n            self.io_loop.add_future(pool.submit(lambda: None),\n                                    lambda future: self.stop(future))\n            future = self.wait()\n            self.assertTrue(future.done())\n            self.assertTrue(future.result() is None)\n\n    def test_add_future_stack_context(self):\n        ready = threading.Event()\n\n        def task():\n            # we must wait for the ioloop callback to be scheduled before\n            # the task completes to ensure that add_future adds the callback\n            # asynchronously (which is the scenario in which capturing\n            # the stack_context matters)\n            ready.wait(1)\n            assert ready.isSet(), \"timed out\"\n            raise Exception(\"worker\")\n\n        def callback(future):\n            self.future = future\n            raise Exception(\"callback\")\n\n        def handle_exception(typ, value, traceback):\n            self.exception = value\n            self.stop()\n            return True\n\n        # stack_context propagates to the ioloop callback, but the worker\n        # task just has its exceptions caught and saved in the Future.\n        with futures.ThreadPoolExecutor(1) as pool:\n            with ExceptionStackContext(handle_exception):\n                self.io_loop.add_future(pool.submit(task), callback)\n            ready.set()\n        self.wait()\n\n        self.assertEqual(self.exception.args[0], \"callback\")\n        self.assertEqual(self.future.exception().args[0], \"worker\")\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    @skipBefore35\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 concurrrent future would do that).\n        namespace = exec_test(globals(), locals(), \"\"\"\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 [\n            namespace[\"async_wrapper\"](event1, event2),\n            namespace[\"async_wrapper\"](event2, event1)\n            ]\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(MyExecutor, self).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()\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.Task(self.io_loop.add_callback)\n            raise gen.Return(42)\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.Task(self.io_loop.add_callback)\n            1 / 0\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        self.io_loop.run_sync(f)\n\n    def test_timeout(self):\n        @gen.coroutine\n        def f():\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n        self.assertRaises(TimeoutError, self.io_loop.run_sync, f, timeout=0.01)\n\n    @skipBefore35\n    def test_native_coroutine(self):\n        namespace = exec_test(globals(), locals(), \"\"\"\n        async def f():\n            await gen.Task(self.io_loop.add_callback)\n        \"\"\")\n        self.io_loop.run_sync(namespace['f'])\n\n\nclass TestPeriodicCallback(unittest.TestCase):\n    def setUp(self):\n        self.io_loop = FakeTimeIOLoop()\n        self.io_loop.make_current()\n\n    def tearDown(self):\n        self.io_loop.close()\n\n    def test_basic(self):\n        calls = []\n\n        def cb():\n            calls.append(self.io_loop.time())\n        pc = PeriodicCallback(cb, 10000)\n        pc.start()\n        self.io_loop.call_later(50, self.io_loop.stop)\n        self.io_loop.start()\n        self.assertEqual(calls, [1010, 1020, 1030, 1040, 1050])\n\n    def test_overrun(self):\n        sleep_durations = [9, 9, 10, 11, 20, 20, 35, 35, 0, 0]\n        expected = [\n            1010, 1020, 1030,  # first 3 calls on schedule\n            1050, 1070,  # next 2 delayed one cycle\n            1100, 1130,  # next 2 delayed 2 cycles\n            1170, 1210,  # next 2 delayed 3 cycles\n            1220, 1230,  # then back on schedule.\n        ]\n        calls = []\n\n        def cb():\n            calls.append(self.io_loop.time())\n            if not sleep_durations:\n                self.io_loop.stop()\n                return\n            self.io_loop.sleep(sleep_durations.pop(0))\n        pc = PeriodicCallback(cb, 10000)\n        pc.start()\n        self.io_loop.start()\n        self.assertEqual(calls, expected)\n\n    def test_io_loop_set_at_start(self):\n        # Check PeriodicCallback uses the current IOLoop at start() time,\n        # not at instantiation time.\n        calls = []\n        io_loop = FakeTimeIOLoop()\n\n        def cb():\n            calls.append(io_loop.time())\n        pc = PeriodicCallback(cb, 10000)\n        io_loop.make_current()\n        pc.start()\n        io_loop.call_later(50, io_loop.stop)\n        io_loop.start()\n        self.assertEqual(calls, [1010, 1020, 1030, 1040, 1050])\n        io_loop.close()\n\n\nclass TestIOLoopConfiguration(unittest.TestCase):\n    def run_python(self, *statements):\n        statements = [\n            'from tornado.ioloop import IOLoop, PollIOLoop',\n            'classname = lambda x: x.__class__.__name__',\n        ] + list(statements)\n        args = [sys.executable, '-c', '; '.join(statements)]\n        return native_str(subprocess.check_output(args)).strip()\n\n    def test_default(self):\n        if asyncio is not None:\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        else:\n            # Otherwise, the default is a subclass of PollIOLoop\n            is_poll = self.run_python(\n                'print(isinstance(IOLoop.current(), PollIOLoop))')\n            self.assertEqual(is_poll, 'True')\n\n    @unittest.skipIf(asyncio is not None,\n                     \"IOLoop configuration not available\")\n    def test_explicit_select(self):\n        # SelectIOLoop can always be configured explicitly.\n        default_class = self.run_python(\n            'IOLoop.configure(\"tornado.platform.select.SelectIOLoop\")',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(default_class, 'SelectIOLoop')\n\n    @unittest.skipIf(asyncio is None, \"asyncio module not present\")\n    def test_asyncio(self):\n        cls = self.run_python(\n            'IOLoop.configure(\"tornado.platform.asyncio.AsyncIOLoop\")',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(cls, 'AsyncIOMainLoop')\n\n    @unittest.skipIf(asyncio is None, \"asyncio module not present\")\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        self.assertEqual(cls, 'AsyncIOMainLoop')\n\n    @unittest.skipIf(twisted is None, \"twisted module not present\")\n    def test_twisted(self):\n        cls = self.run_python(\n            'from tornado.platform.twisted import TwistedIOLoop',\n            'TwistedIOLoop().install()',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(cls, 'TwistedIOLoop')\n\n\nif __name__ == \"__main__\":\n    unittest.main()\n",
            "new_file_content": "#!/usr/bin/env python\n\n\nfrom __future__ import absolute_import, division, print_function\nimport contextlib\nimport datetime\nimport functools\nimport socket\nimport subprocess\nimport sys\nimport threading\nimport time\nimport types\n\nfrom tornado.escape import native_str\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop, TimeoutError, PollIOLoop, PeriodicCallback\nfrom tornado.log import app_log\nfrom tornado.platform.select import _Select\nfrom tornado.stack_context import ExceptionStackContext, StackContext, wrap, NullContext\nfrom tornado.testing import AsyncTestCase, bind_unused_port, ExpectLog, gen_test\nfrom tornado.test.util import unittest, skipIfNonUnix, skipOnTravis, skipBefore35, exec_test\nfrom tornado.concurrent import Future\n\ntry:\n    from concurrent import futures\nexcept ImportError:\n    futures = None\n\ntry:\n    import asyncio\nexcept ImportError:\n    asyncio = None\n\ntry:\n    import twisted\nexcept ImportError:\n    twisted = None\n\n\nclass FakeTimeSelect(_Select):\n    def __init__(self):\n        self._time = 1000\n        super(FakeTimeSelect, self).__init__()\n\n    def time(self):\n        return self._time\n\n    def sleep(self, t):\n        self._time += t\n\n    def poll(self, timeout):\n        events = super(FakeTimeSelect, self).poll(0)\n        if events:\n            return events\n        self._time += timeout\n        return []\n\n\nclass FakeTimeIOLoop(PollIOLoop):\n    \"\"\"IOLoop implementation with a fake and deterministic clock.\n\n    The clock advances as needed to trigger timeouts immediately.\n    For use when testing code that involves the passage of time\n    and no external dependencies.\n    \"\"\"\n    def initialize(self):\n        self.fts = FakeTimeSelect()\n        super(FakeTimeIOLoop, self).initialize(impl=self.fts,\n                                               time_func=self.fts.time)\n\n    def sleep(self, t):\n        \"\"\"Simulate a blocking sleep by advancing the clock.\"\"\"\n        self.fts.sleep(t)\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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        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    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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\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        self.io_loop = None\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            def f():\n                self.current_io_loop = IOLoop.current()\n                self.io_loop.stop()\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 TestIOLoopAddCallback(AsyncTestCase):\n    def setUp(self):\n        super(TestIOLoopAddCallback, self).setUp()\n        self.active_contexts = []\n\n    def add_callback(self, callback, *args, **kwargs):\n        self.io_loop.add_callback(callback, *args, **kwargs)\n\n    @contextlib.contextmanager\n    def context(self, name):\n        self.active_contexts.append(name)\n        yield\n        self.assertEqual(self.active_contexts.pop(), name)\n\n    def test_pre_wrap(self):\n        # A pre-wrapped callback is run in the context in which it was\n        # wrapped, not when it was added to the IOLoop.\n        def f1():\n            self.assertIn('c1', self.active_contexts)\n            self.assertNotIn('c2', self.active_contexts)\n            self.stop()\n\n        with StackContext(functools.partial(self.context, 'c1')):\n            wrapped = wrap(f1)\n\n        with StackContext(functools.partial(self.context, 'c2')):\n            self.add_callback(wrapped)\n\n        self.wait()\n\n    def test_pre_wrap_with_args(self):\n        # Same as test_pre_wrap, but the function takes arguments.\n        # Implementation note: The function must not be wrapped in a\n        # functools.partial until after it has been passed through\n        # stack_context.wrap\n        def f1(foo, bar):\n            self.assertIn('c1', self.active_contexts)\n            self.assertNotIn('c2', self.active_contexts)\n            self.stop((foo, bar))\n\n        with StackContext(functools.partial(self.context, 'c1')):\n            wrapped = wrap(f1)\n\n        with StackContext(functools.partial(self.context, 'c2')):\n            self.add_callback(wrapped, 1, bar=2)\n\n        result = self.wait()\n        self.assertEqual(result, (1, 2))\n\n\nclass TestIOLoopAddCallbackFromSignal(TestIOLoopAddCallback):\n    # Repeat the add_callback tests using add_callback_from_signal\n    def add_callback(self, callback, *args, **kwargs):\n        self.io_loop.add_callback_from_signal(callback, *args, **kwargs)\n\n\n@unittest.skipIf(futures is None, \"futures module not present\")\nclass TestIOLoopFutures(AsyncTestCase):\n    def test_add_future_threads(self):\n        with futures.ThreadPoolExecutor(1) as pool:\n            self.io_loop.add_future(pool.submit(lambda: None),\n                                    lambda future: self.stop(future))\n            future = self.wait()\n            self.assertTrue(future.done())\n            self.assertTrue(future.result() is None)\n\n    def test_add_future_stack_context(self):\n        ready = threading.Event()\n\n        def task():\n            # we must wait for the ioloop callback to be scheduled before\n            # the task completes to ensure that add_future adds the callback\n            # asynchronously (which is the scenario in which capturing\n            # the stack_context matters)\n            ready.wait(1)\n            assert ready.isSet(), \"timed out\"\n            raise Exception(\"worker\")\n\n        def callback(future):\n            self.future = future\n            raise Exception(\"callback\")\n\n        def handle_exception(typ, value, traceback):\n            self.exception = value\n            self.stop()\n            return True\n\n        # stack_context propagates to the ioloop callback, but the worker\n        # task just has its exceptions caught and saved in the Future.\n        with futures.ThreadPoolExecutor(1) as pool:\n            with ExceptionStackContext(handle_exception):\n                self.io_loop.add_future(pool.submit(task), callback)\n            ready.set()\n        self.wait()\n\n        self.assertEqual(self.exception.args[0], \"callback\")\n        self.assertEqual(self.future.exception().args[0], \"worker\")\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    @skipBefore35\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 concurrrent future would do that).\n        namespace = exec_test(globals(), locals(), \"\"\"\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 [\n            namespace[\"async_wrapper\"](event1, event2),\n            namespace[\"async_wrapper\"](event2, event1)\n            ]\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(MyExecutor, self).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()\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.Task(self.io_loop.add_callback)\n            raise gen.Return(42)\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.Task(self.io_loop.add_callback)\n            1 / 0\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        self.io_loop.run_sync(f)\n\n    def test_timeout(self):\n        @gen.coroutine\n        def f():\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n        self.assertRaises(TimeoutError, self.io_loop.run_sync, f, timeout=0.01)\n\n    @skipBefore35\n    def test_native_coroutine(self):\n        namespace = exec_test(globals(), locals(), \"\"\"\n        async def f():\n            await gen.Task(self.io_loop.add_callback)\n        \"\"\")\n        self.io_loop.run_sync(namespace['f'])\n\n\nclass TestPeriodicCallback(unittest.TestCase):\n    def setUp(self):\n        self.io_loop = FakeTimeIOLoop()\n        self.io_loop.make_current()\n\n    def tearDown(self):\n        self.io_loop.close()\n\n    def test_basic(self):\n        calls = []\n\n        def cb():\n            calls.append(self.io_loop.time())\n        pc = PeriodicCallback(cb, 10000)\n        pc.start()\n        self.io_loop.call_later(50, self.io_loop.stop)\n        self.io_loop.start()\n        self.assertEqual(calls, [1010, 1020, 1030, 1040, 1050])\n\n    def test_overrun(self):\n        sleep_durations = [9, 9, 10, 11, 20, 20, 35, 35, 0, 0]\n        expected = [\n            1010, 1020, 1030,  # first 3 calls on schedule\n            1050, 1070,  # next 2 delayed one cycle\n            1100, 1130,  # next 2 delayed 2 cycles\n            1170, 1210,  # next 2 delayed 3 cycles\n            1220, 1230,  # then back on schedule.\n        ]\n        calls = []\n\n        def cb():\n            calls.append(self.io_loop.time())\n            if not sleep_durations:\n                self.io_loop.stop()\n                return\n            self.io_loop.sleep(sleep_durations.pop(0))\n        pc = PeriodicCallback(cb, 10000)\n        pc.start()\n        self.io_loop.start()\n        self.assertEqual(calls, expected)\n\n    def test_io_loop_set_at_start(self):\n        # Check PeriodicCallback uses the current IOLoop at start() time,\n        # not at instantiation time.\n        calls = []\n        io_loop = FakeTimeIOLoop()\n\n        def cb():\n            calls.append(io_loop.time())\n        pc = PeriodicCallback(cb, 10000)\n        io_loop.make_current()\n        pc.start()\n        io_loop.call_later(50, io_loop.stop)\n        io_loop.start()\n        self.assertEqual(calls, [1010, 1020, 1030, 1040, 1050])\n        io_loop.close()\n\n\nclass TestIOLoopConfiguration(unittest.TestCase):\n    def run_python(self, *statements):\n        statements = [\n            'from tornado.ioloop import IOLoop, PollIOLoop',\n            'classname = lambda x: x.__class__.__name__',\n        ] + list(statements)\n        args = [sys.executable, '-c', '; '.join(statements)]\n        return native_str(subprocess.check_output(args)).strip()\n\n    def test_default(self):\n        if asyncio is not None:\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        else:\n            # Otherwise, the default is a subclass of PollIOLoop\n            is_poll = self.run_python(\n                'print(isinstance(IOLoop.current(), PollIOLoop))')\n            self.assertEqual(is_poll, 'True')\n\n    @unittest.skipIf(asyncio is not None,\n                     \"IOLoop configuration not available\")\n    def test_explicit_select(self):\n        # SelectIOLoop can always be configured explicitly.\n        default_class = self.run_python(\n            'IOLoop.configure(\"tornado.platform.select.SelectIOLoop\")',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(default_class, 'SelectIOLoop')\n\n    @unittest.skipIf(asyncio is None, \"asyncio module not present\")\n    def test_asyncio(self):\n        cls = self.run_python(\n            'IOLoop.configure(\"tornado.platform.asyncio.AsyncIOLoop\")',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(cls, 'AsyncIOMainLoop')\n\n    @unittest.skipIf(asyncio is None, \"asyncio module not present\")\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        self.assertEqual(cls, 'AsyncIOMainLoop')\n\n    @unittest.skipIf(twisted is None, \"twisted module not present\")\n    def test_twisted(self):\n        cls = self.run_python(\n            'from tornado.platform.twisted import TwistedIOLoop',\n            'TwistedIOLoop().install()',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(cls, 'TwistedIOLoop')\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": "61609945",
                "new_commit_hash": "0ca52c66",
                "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": 171,
                            "length": 10
                        },
                        "new_range": {
                            "start": 171,
                            "length": 9
                        },
                        "section": "class TestIOLoop(AsyncTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        other_ioloop.close()",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def test_add_callback_while_closing(self):",
                                "type": "context"
                            },
                            {
                                "content": "        # Issue #635: add_callback() should raise a clean exception",
                                "type": "deleted"
                            },
                            {
                                "content": "        # if called while another thread is closing the IOLoop.",
                                "type": "deleted"
                            },
                            {
                                "content": "        if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'):",
                                "type": "deleted"
                            },
                            {
                                "content": "            raise unittest.SkipTest(\"AsyncIOMainLoop shutdown not thread safe\")",
                                "type": "deleted"
                            },
                            {
                                "content": "        # add_callback should not fail if it races with another thread",
                                "type": "added"
                            },
                            {
                                "content": "        # closing the IOLoop. The callbacks are dropped silently",
                                "type": "added"
                            },
                            {
                                "content": "        # without executing.",
                                "type": "added"
                            },
                            {
                                "content": "        closing = threading.Event()",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "        def target():",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/test/ioloop_test.py",
                            "type": "method",
                            "name": "TestIOLoop.test_add_callback_while_closing",
                            "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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        # Issue #635: add_callback() should raise a clean exception\n        # if called while another thread is closing the IOLoop.\n        if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'):\n            raise unittest.SkipTest(\"AsyncIOMainLoop shutdown not thread safe\")\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        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            try:\n                other_ioloop.add_callback(lambda: None)\n            except RuntimeError as e:\n                self.assertEqual(\"IOLoop is closing\", str(e))\n                break\n\n    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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()",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 173,
                            "end_lineno": 194,
                            "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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        # Issue #635: add_callback() should raise a clean exception\n        # if called while another thread is closing the IOLoop.\n        if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'):\n            raise unittest.SkipTest(\"AsyncIOMainLoop shutdown not thread safe\")\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        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            try:\n                other_ioloop.add_callback(lambda: None)\n            except RuntimeError as e:\n                self.assertEqual(\"IOLoop is closing\", str(e))\n                break\n\n    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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()",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 77,
                                "end_lineno": 457,
                                "parent": null
                            }
                        },
                        {
                            "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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        # Issue #635: add_callback() should raise a clean exception\n        # if called while another thread is closing the IOLoop.\n        if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'):\n            raise unittest.SkipTest(\"AsyncIOMainLoop shutdown not thread safe\")\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        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            try:\n                other_ioloop.add_callback(lambda: None)\n            except RuntimeError as e:\n                self.assertEqual(\"IOLoop is closing\", str(e))\n                break\n\n    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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()",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 77,
                            "end_lineno": 457,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 187,
                            "length": 11
                        },
                        "new_range": {
                            "start": 186,
                            "length": 7
                        },
                        "section": "class TestIOLoop(AsyncTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        thread.start()",
                                "type": "context"
                            },
                            {
                                "content": "        closing.wait()",
                                "type": "context"
                            },
                            {
                                "content": "        for i in range(1000):",
                                "type": "context"
                            },
                            {
                                "content": "            try:",
                                "type": "deleted"
                            },
                            {
                                "content": "                other_ioloop.add_callback(lambda: None)",
                                "type": "deleted"
                            },
                            {
                                "content": "            except RuntimeError as e:",
                                "type": "deleted"
                            },
                            {
                                "content": "                self.assertEqual(\"IOLoop is closing\", str(e))",
                                "type": "deleted"
                            },
                            {
                                "content": "                break",
                                "type": "deleted"
                            },
                            {
                                "content": "            other_ioloop.add_callback(lambda: None)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def test_handle_callback_exception(self):",
                                "type": "context"
                            },
                            {
                                "content": "        # IOLoop.handle_callback_exception can be overridden to catch",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/test/ioloop_test.py",
                            "type": "method",
                            "name": "TestIOLoop.test_add_callback_while_closing",
                            "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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        # Issue #635: add_callback() should raise a clean exception\n        # if called while another thread is closing the IOLoop.\n        if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'):\n            raise unittest.SkipTest(\"AsyncIOMainLoop shutdown not thread safe\")\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        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            try:\n                other_ioloop.add_callback(lambda: None)\n            except RuntimeError as e:\n                self.assertEqual(\"IOLoop is closing\", str(e))\n                break\n\n    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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()",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 173,
                            "end_lineno": 194,
                            "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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        # Issue #635: add_callback() should raise a clean exception\n        # if called while another thread is closing the IOLoop.\n        if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'):\n            raise unittest.SkipTest(\"AsyncIOMainLoop shutdown not thread safe\")\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        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            try:\n                other_ioloop.add_callback(lambda: None)\n            except RuntimeError as e:\n                self.assertEqual(\"IOLoop is closing\", str(e))\n                break\n\n    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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()",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 77,
                                "end_lineno": 457,
                                "parent": null
                            }
                        },
                        {
                            "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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        # Issue #635: add_callback() should raise a clean exception\n        # if called while another thread is closing the IOLoop.\n        if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'):\n            raise unittest.SkipTest(\"AsyncIOMainLoop shutdown not thread safe\")\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        other_ioloop = IOLoop()\n        thread = threading.Thread(target=target)\n        thread.start()\n        closing.wait()\n        for i in range(1000):\n            try:\n                other_ioloop.add_callback(lambda: None)\n            except RuntimeError as e:\n                self.assertEqual(\"IOLoop is closing\", str(e))\n                break\n\n    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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()",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 77,
                            "end_lineno": 457,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "207ff24d273195043cc8d8e874da890160619b90^",
    "new_commit_hash": "207ff24d273195043cc8d8e874da890160619b90",
    "commit_message": "asyncio: Don't raise \"IOLoop is closing\" in add_callback\n\nThis mirrors a previous change to PollIOLoop. Fix a test that had\nrotted and become meaningless.\n\nFixes #2191",
    "commit_date": "2017-12-31T17:05:05-05:00",
    "metadata": {}
}
{
    "repo_name": "tornado",
    "new_commit_hash": "207ff24d273195043cc8d8e874da890160619b90",
    "test_file_codes": [
        "#!/usr/bin/env python\n\n\nfrom __future__ import absolute_import, division, print_function\nimport contextlib\nimport datetime\nimport functools\nimport socket\nimport subprocess\nimport sys\nimport threading\nimport time\nimport types\n\nfrom tornado.escape import native_str\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop, TimeoutError, PollIOLoop, PeriodicCallback\nfrom tornado.log import app_log\nfrom tornado.platform.select import _Select\nfrom tornado.stack_context import ExceptionStackContext, StackContext, wrap, NullContext\nfrom tornado.testing import AsyncTestCase, bind_unused_port, ExpectLog, gen_test\nfrom tornado.test.util import unittest, skipIfNonUnix, skipOnTravis, skipBefore35, exec_test\nfrom tornado.concurrent import Future\n\ntry:\n    from concurrent import futures\nexcept ImportError:\n    futures = None\n\ntry:\n    import asyncio\nexcept ImportError:\n    asyncio = None\n\ntry:\n    import twisted\nexcept ImportError:\n    twisted = None\n\n\nclass FakeTimeSelect(_Select):\n    def __init__(self):\n        self._time = 1000\n        super(FakeTimeSelect, self).__init__()\n\n    def time(self):\n        return self._time\n\n    def sleep(self, t):\n        self._time += t\n\n    def poll(self, timeout):\n        events = super(FakeTimeSelect, self).poll(0)\n        if events:\n            return events\n        self._time += timeout\n        return []\n\n\nclass FakeTimeIOLoop(PollIOLoop):\n    \"\"\"IOLoop implementation with a fake and deterministic clock.\n\n    The clock advances as needed to trigger timeouts immediately.\n    For use when testing code that involves the passage of time\n    and no external dependencies.\n    \"\"\"\n    def initialize(self):\n        self.fts = FakeTimeSelect()\n        super(FakeTimeIOLoop, self).initialize(impl=self.fts,\n                                               time_func=self.fts.time)\n\n    def sleep(self, t):\n        \"\"\"Simulate a blocking sleep by advancing the clock.\"\"\"\n        self.fts.sleep(t)\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)\n        loop.add_callback(lambda: {})\n        loop.add_callback(lambda: [])\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        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        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(sock.fileno(), lambda fd, events: None,\n                                     IOLoop.READ)\n            # Attempting to add the same handler twice fails\n            # (with a platform-dependent exception)\n            self.assertRaises(Exception, self.io_loop.add_handler,\n                              sock.fileno(), lambda fd, events: None,\n                              IOLoop.READ)\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        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    def test_handle_callback_exception(self):\n        # IOLoop.handle_callback_exception can be overridden to catch\n        # exceptions in callbacks.\n        def handle_callback_exception(callback):\n            self.assertIs(sys.exc_info()[0], ZeroDivisionError)\n            self.stop()\n        self.io_loop.handle_callback_exception = handle_callback_exception\n        with NullContext():\n            # remove the test StackContext that would see this uncaught\n            # exception as a test failure.\n            self.io_loop.add_callback(lambda: 1 / 0)\n        self.wait()\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            def handler(fd, events):\n                self.assertEqual(events, IOLoop.READ)\n                self.stop()\n            self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ)\n            self.io_loop.add_timeout(self.io_loop.time() + 0.01,\n                                     functools.partial(server.send, b'asdf'))\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,\n                                               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        self.io_loop.add_timeout(now + 0.01, t1)\n\n        def t2():\n            calls[1] = True\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 = []\n        self.io_loop.add_timeout(self.io_loop.time(), results.append, 1)\n        self.io_loop.add_timeout(datetime.timedelta(seconds=0),\n                                 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, but PollIOLoop does.\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        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,\n                            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        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,\n                                 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        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        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        # Use a NullContext to keep the exception from being caught by\n        # AsyncTestCase.\n        with NullContext():\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        with NullContext():\n            @gen.coroutine\n            def callback():\n                self.io_loop.add_callback(self.stop)\n                1 / 0\n            self.io_loop.add_callback(callback)\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    @skipBefore35\n    def test_exception_logging_native_coro(self):\n        \"\"\"The IOLoop examines exceptions from awaitables and logs them.\"\"\"\n        namespace = exec_test(globals(), locals(), \"\"\"\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        with NullContext():\n            self.io_loop.add_callback(namespace[\"callback\"])\n            with ExpectLog(app_log, \"Exception in callback\"):\n                self.wait()\n\n    def test_spawn_callback(self):\n        # An added callback runs in the test's stack_context, so will be\n        # re-raised in wait().\n        self.io_loop.add_callback(lambda: 1 / 0)\n        with self.assertRaises(ZeroDivisionError):\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            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\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        self.io_loop = None\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            def f():\n                self.current_io_loop = IOLoop.current()\n                self.io_loop.stop()\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 TestIOLoopAddCallback(AsyncTestCase):\n    def setUp(self):\n        super(TestIOLoopAddCallback, self).setUp()\n        self.active_contexts = []\n\n    def add_callback(self, callback, *args, **kwargs):\n        self.io_loop.add_callback(callback, *args, **kwargs)\n\n    @contextlib.contextmanager\n    def context(self, name):\n        self.active_contexts.append(name)\n        yield\n        self.assertEqual(self.active_contexts.pop(), name)\n\n    def test_pre_wrap(self):\n        # A pre-wrapped callback is run in the context in which it was\n        # wrapped, not when it was added to the IOLoop.\n        def f1():\n            self.assertIn('c1', self.active_contexts)\n            self.assertNotIn('c2', self.active_contexts)\n            self.stop()\n\n        with StackContext(functools.partial(self.context, 'c1')):\n            wrapped = wrap(f1)\n\n        with StackContext(functools.partial(self.context, 'c2')):\n            self.add_callback(wrapped)\n\n        self.wait()\n\n    def test_pre_wrap_with_args(self):\n        # Same as test_pre_wrap, but the function takes arguments.\n        # Implementation note: The function must not be wrapped in a\n        # functools.partial until after it has been passed through\n        # stack_context.wrap\n        def f1(foo, bar):\n            self.assertIn('c1', self.active_contexts)\n            self.assertNotIn('c2', self.active_contexts)\n            self.stop((foo, bar))\n\n        with StackContext(functools.partial(self.context, 'c1')):\n            wrapped = wrap(f1)\n\n        with StackContext(functools.partial(self.context, 'c2')):\n            self.add_callback(wrapped, 1, bar=2)\n\n        result = self.wait()\n        self.assertEqual(result, (1, 2))\n\n\nclass TestIOLoopAddCallbackFromSignal(TestIOLoopAddCallback):\n    # Repeat the add_callback tests using add_callback_from_signal\n    def add_callback(self, callback, *args, **kwargs):\n        self.io_loop.add_callback_from_signal(callback, *args, **kwargs)\n\n\n@unittest.skipIf(futures is None, \"futures module not present\")\nclass TestIOLoopFutures(AsyncTestCase):\n    def test_add_future_threads(self):\n        with futures.ThreadPoolExecutor(1) as pool:\n            self.io_loop.add_future(pool.submit(lambda: None),\n                                    lambda future: self.stop(future))\n            future = self.wait()\n            self.assertTrue(future.done())\n            self.assertTrue(future.result() is None)\n\n    def test_add_future_stack_context(self):\n        ready = threading.Event()\n\n        def task():\n            # we must wait for the ioloop callback to be scheduled before\n            # the task completes to ensure that add_future adds the callback\n            # asynchronously (which is the scenario in which capturing\n            # the stack_context matters)\n            ready.wait(1)\n            assert ready.isSet(), \"timed out\"\n            raise Exception(\"worker\")\n\n        def callback(future):\n            self.future = future\n            raise Exception(\"callback\")\n\n        def handle_exception(typ, value, traceback):\n            self.exception = value\n            self.stop()\n            return True\n\n        # stack_context propagates to the ioloop callback, but the worker\n        # task just has its exceptions caught and saved in the Future.\n        with futures.ThreadPoolExecutor(1) as pool:\n            with ExceptionStackContext(handle_exception):\n                self.io_loop.add_future(pool.submit(task), callback)\n            ready.set()\n        self.wait()\n\n        self.assertEqual(self.exception.args[0], \"callback\")\n        self.assertEqual(self.future.exception().args[0], \"worker\")\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    @skipBefore35\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 concurrrent future would do that).\n        namespace = exec_test(globals(), locals(), \"\"\"\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 [\n            namespace[\"async_wrapper\"](event1, event2),\n            namespace[\"async_wrapper\"](event2, event1)\n            ]\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(MyExecutor, self).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()\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.Task(self.io_loop.add_callback)\n            raise gen.Return(42)\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.Task(self.io_loop.add_callback)\n            1 / 0\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        self.io_loop.run_sync(f)\n\n    def test_timeout(self):\n        @gen.coroutine\n        def f():\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n        self.assertRaises(TimeoutError, self.io_loop.run_sync, f, timeout=0.01)\n\n    @skipBefore35\n    def test_native_coroutine(self):\n        namespace = exec_test(globals(), locals(), \"\"\"\n        async def f():\n            await gen.Task(self.io_loop.add_callback)\n        \"\"\")\n        self.io_loop.run_sync(namespace['f'])\n\n\nclass TestPeriodicCallback(unittest.TestCase):\n    def setUp(self):\n        self.io_loop = FakeTimeIOLoop()\n        self.io_loop.make_current()\n\n    def tearDown(self):\n        self.io_loop.close()\n\n    def test_basic(self):\n        calls = []\n\n        def cb():\n            calls.append(self.io_loop.time())\n        pc = PeriodicCallback(cb, 10000)\n        pc.start()\n        self.io_loop.call_later(50, self.io_loop.stop)\n        self.io_loop.start()\n        self.assertEqual(calls, [1010, 1020, 1030, 1040, 1050])\n\n    def test_overrun(self):\n        sleep_durations = [9, 9, 10, 11, 20, 20, 35, 35, 0, 0]\n        expected = [\n            1010, 1020, 1030,  # first 3 calls on schedule\n            1050, 1070,  # next 2 delayed one cycle\n            1100, 1130,  # next 2 delayed 2 cycles\n            1170, 1210,  # next 2 delayed 3 cycles\n            1220, 1230,  # then back on schedule.\n        ]\n        calls = []\n\n        def cb():\n            calls.append(self.io_loop.time())\n            if not sleep_durations:\n                self.io_loop.stop()\n                return\n            self.io_loop.sleep(sleep_durations.pop(0))\n        pc = PeriodicCallback(cb, 10000)\n        pc.start()\n        self.io_loop.start()\n        self.assertEqual(calls, expected)\n\n    def test_io_loop_set_at_start(self):\n        # Check PeriodicCallback uses the current IOLoop at start() time,\n        # not at instantiation time.\n        calls = []\n        io_loop = FakeTimeIOLoop()\n\n        def cb():\n            calls.append(io_loop.time())\n        pc = PeriodicCallback(cb, 10000)\n        io_loop.make_current()\n        pc.start()\n        io_loop.call_later(50, io_loop.stop)\n        io_loop.start()\n        self.assertEqual(calls, [1010, 1020, 1030, 1040, 1050])\n        io_loop.close()\n\n\nclass TestIOLoopConfiguration(unittest.TestCase):\n    def run_python(self, *statements):\n        statements = [\n            'from tornado.ioloop import IOLoop, PollIOLoop',\n            'classname = lambda x: x.__class__.__name__',\n        ] + list(statements)\n        args = [sys.executable, '-c', '; '.join(statements)]\n        return native_str(subprocess.check_output(args)).strip()\n\n    def test_default(self):\n        if asyncio is not None:\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        else:\n            # Otherwise, the default is a subclass of PollIOLoop\n            is_poll = self.run_python(\n                'print(isinstance(IOLoop.current(), PollIOLoop))')\n            self.assertEqual(is_poll, 'True')\n\n    @unittest.skipIf(asyncio is not None,\n                     \"IOLoop configuration not available\")\n    def test_explicit_select(self):\n        # SelectIOLoop can always be configured explicitly.\n        default_class = self.run_python(\n            'IOLoop.configure(\"tornado.platform.select.SelectIOLoop\")',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(default_class, 'SelectIOLoop')\n\n    @unittest.skipIf(asyncio is None, \"asyncio module not present\")\n    def test_asyncio(self):\n        cls = self.run_python(\n            'IOLoop.configure(\"tornado.platform.asyncio.AsyncIOLoop\")',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(cls, 'AsyncIOMainLoop')\n\n    @unittest.skipIf(asyncio is None, \"asyncio module not present\")\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        self.assertEqual(cls, 'AsyncIOMainLoop')\n\n    @unittest.skipIf(twisted is None, \"twisted module not present\")\n    def test_twisted(self):\n        cls = self.run_python(\n            'from tornado.platform.twisted import TwistedIOLoop',\n            'TwistedIOLoop().install()',\n            'print(classname(IOLoop.current()))')\n        self.assertEqual(cls, 'TwistedIOLoop')\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 988ms\nPrepared 1 package in 1.43s\nInstalled 1 package in 5ms\n + tornado==5.0a1 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_207ff24d273195043cc8d8e874da890160619b90)\nResolved 29 packages in 15ms\nInstalled 29 packages in 122ms\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_207ff24d273195043cc8d8e874da890160619b90\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_handle_callback_exception\nPASSED r2e_tests.test_1::TestIOLoop::test_handler_callback_file_object\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::TestIOLoopAddCallback::test_pre_wrap\nPASSED r2e_tests.test_1::TestIOLoopAddCallback::test_pre_wrap_with_args\nPASSED r2e_tests.test_1::TestIOLoopAddCallbackFromSignal::test_pre_wrap\nPASSED r2e_tests.test_1::TestIOLoopAddCallbackFromSignal::test_pre_wrap_with_args\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\nSKIPPED r2e_tests.test_1::TestIOLoopConfiguration::test_explicit_select\nPASSED r2e_tests.test_1::TestIOLoopConfiguration::test_twisted\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::TestIOLoopFutures::test_add_future_stack_context\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::TestPeriodicCallback::test_basic\nPASSED r2e_tests.test_1::TestPeriodicCallback::test_io_loop_set_at_start\nPASSED r2e_tests.test_1::TestPeriodicCallback::test_overrun\n=================== 53 passed, 1 skipped in 0.89s ===================\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_207ff24d273195043cc8d8e874da890160619b90\n......................................................\ncollected 54 items\n\n=================================== ERRORS =====================================\n__________ TestIOLoop.test_add_callback_while_closing __________\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_207ff24d273195043cc8d8e874da890160619b90/tornado/testing.py\", line 124, in __call__\n    result = self.orig_method(*args, **kwargs)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_207ff24d273195043cc8d8e874da890160619b90/r2e_tests/test_1.py\", line 189, in test_add_callback_while_closing\n    other_ioloop.add_callback(lambda: None)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_207ff24d273195043cc8d8e874da890160619b90/tornado/platform/asyncio.py\", line 136, in add_callback\n    raise RuntimeError(\"IOLoop is closing\")\nRuntimeError: IOLoop is closing\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\nERROR 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_handle_callback_exception\nPASSED r2e_tests.test_1::TestIOLoop::test_handler_callback_file_object\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::TestIOLoopAddCallback::test_pre_wrap\nPASSED r2e_tests.test_1::TestIOLoopAddCallback::test_pre_wrap_with_args\nPASSED r2e_tests.test_1::TestIOLoopAddCallbackFromSignal::test_pre_wrap\nPASSED r2e_tests.test_1::TestIOLoopAddCallbackFromSignal::test_pre_wrap_with_args\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\nSKIPPED r2e_tests.test_1::TestIOLoopConfiguration::test_explicit_select\nPASSED r2e_tests.test_1::TestIOLoopConfiguration::test_twisted\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::TestIOLoopFutures::test_add_future_stack_context\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::TestPeriodicCallback::test_basic\nPASSED r2e_tests.test_1::TestPeriodicCallback::test_io_loop_set_at_start\nPASSED r2e_tests.test_1::TestPeriodicCallback::test_overrun\n=================== 1 error, 52 passed, 1 skipped in 0.71s ===================\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**: Adding a Callback While IOLoop Is Closing Raises RuntimeError

**Description**:
When the `IOLoop` is in the process of closing, attempting to add a callback results in a `RuntimeError` with the message "IOLoop is closing". This behavior is unexpected because callbacks added during the shutdown process should be silently ignored to ensure a smooth termination.

**Example Code**:
```python
import threading
from tornado.ioloop import IOLoop

def shutdown_loop(loop):
    loop.stop()
    loop.close()

ioloop = IOLoop.current()
thread = threading.Thread(target=shutdown_loop, args=(ioloop,))
thread.start()
thread.join()

# Attempting to add a callback while the loop is closing
ioloop.add_callback(lambda: print("This should not raise an error"))
```

**Expected Behavior**:
Adding a callback while the `IOLoop` is closing should not raise an exception. Instead, the callback should be silently ignored to allow the loop to shut down gracefully.

**Actual Behavior**:
A `RuntimeError` with the message "IOLoop is closing" is raised when trying to add a callback during the shutdown process.
[/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 requires the `~asyncio.AbstractEventLoop.add_reader` family of
   methods, so it is not compatible with the `~asyncio.ProactorEventLoop` on
   Windows. Use the `~asyncio.SelectorEventLoop` instead.
"""

from __future__ import absolute_import, division, print_function
import functools

import tornado.concurrent
from tornado.gen import convert_yielded
from tornado.ioloop import IOLoop
from tornado import stack_context

import asyncio


class BaseAsyncIOLoop(IOLoop):
    def initialize(self, asyncio_loop, close_loop=False, **kwargs):
        self.asyncio_loop = asyncio_loop
        self.close_loop = close_loop
        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)
        self.handlers = {}
        # Set of fds listening for reads/writes
        self.readers = set()
        self.writers = set()
        self.closing = False
        super(BaseAsyncIOLoop, self).initialize(**kwargs)

    def close(self, all_fds=False):
        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)
        if self.close_loop:
            self.asyncio_loop.close()

    def add_handler(self, fd, handler, events):
        fd, fileobj = self.split_fd(fd)
        if fd in self.handlers:
            raise ValueError("fd %s added twice" % fd)
        self.handlers[fd] = (fileobj, stack_context.wrap(handler))
        if events & IOLoop.READ:
            self.asyncio_loop.add_reader(
                fd, self._handle_events, fd, IOLoop.READ)
            self.readers.add(fd)
        if events & IOLoop.WRITE:
            self.asyncio_loop.add_writer(
                fd, self._handle_events, fd, IOLoop.WRITE)
            self.writers.add(fd)

    def update_handler(self, fd, events):
        fd, fileobj = self.split_fd(fd)
        if events & IOLoop.READ:
            if fd not in self.readers:
                self.asyncio_loop.add_reader(
                    fd, self._handle_events, fd, IOLoop.READ)
                self.readers.add(fd)
        else:
            if fd in self.readers:
                self.asyncio_loop.remove_reader(fd)
                self.readers.remove(fd)
        if events & IOLoop.WRITE:
            if fd not in self.writers:
                self.asyncio_loop.add_writer(
                    fd, self._handle_events, fd, IOLoop.WRITE)
                self.writers.add(fd)
        else:
            if fd in self.writers:
                self.asyncio_loop.remove_writer(fd)
                self.writers.remove(fd)

    def remove_handler(self, fd):
        fd, fileobj = self.split_fd(fd)
        if fd not in self.handlers:
            return
        if fd in self.readers:
            self.asyncio_loop.remove_reader(fd)
            self.readers.remove(fd)
        if fd in self.writers:
            self.asyncio_loop.remove_writer(fd)
            self.writers.remove(fd)
        del self.handlers[fd]

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

    def start(self):
        old_current = IOLoop.current(instance=False)
        try:
            self._setup_logging()
            self.make_current()
            self.asyncio_loop.run_forever()
        finally:
            if old_current is None:
                IOLoop.clear_current()
            else:
                old_current.make_current()

    def stop(self):
        self.asyncio_loop.stop()

    def call_at(self, when, callback, *args, **kwargs):
        # 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(stack_context.wrap(callback), *args, **kwargs))

    def remove_timeout(self, timeout):
        timeout.cancel()

    def add_callback(self, callback, *args, **kwargs):
        if self.closing:
            # TODO: this is racy; we need a lock to ensure that the
            # loop isn't closed during call_soon_threadsafe.
            raise RuntimeError("IOLoop is closing")
        self.asyncio_loop.call_soon_threadsafe(
            self._run_callback,
            functools.partial(stack_context.wrap(callback), *args, **kwargs))

    add_callback_from_signal = add_callback

    def run_in_executor(self, executor, func, *args):
        return self.asyncio_loop.run_in_executor(executor, func, *args)

    def set_default_executor(self, executor):
        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.
    """
    def initialize(self, **kwargs):
        super(AsyncIOMainLoop, self).initialize(asyncio.get_event_loop(),
                                                close_loop=False, **kwargs)


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:: 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):
        loop = asyncio.new_event_loop()
        try:
            super(AsyncIOLoop, self).initialize(loop, close_loop=True, **kwargs)
        except Exception:
            # If initialize() does not succeed (taking ownership of the loop),
            # we have to close it.
            loop.close()
            raise

    def make_current(self):
        super(AsyncIOLoop, self).make_current()
        try:
            self.old_asyncio = asyncio.get_event_loop()
        except RuntimeError:
            self.old_asyncio = None
        asyncio.set_event_loop(self.asyncio_loop)

    def _clear_current_hook(self):
        asyncio.set_event_loop(self.old_asyncio)


def to_tornado_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):
    """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)


class AnyThreadEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    """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

    """
    def get_event_loop(self):
        try:
            return super().get_event_loop()
        except RuntimeError:
            # "There is no current event loop in thread %r"
            loop = self.new_event_loop()
            self.set_event_loop(loop)
            return loop

```
--- 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**: Adding a Callback While IOLoop Is Closing Raises RuntimeError

**Description**:
When the `IOLoop` is in the process of closing, attempting to add a callback results in a `RuntimeError` with the message "IOLoop is closing". This behavior is unexpected because callbacks added during the shutdown process should be silently ignored to ensure a smooth termination.

**Example Code**:
```python
import threading
from tornado.ioloop import IOLoop

def shutdown_loop(loop):
    loop.stop()
    loop.close()

ioloop = IOLoop.current()
thread = threading.Thread(target=shutdown_loop, args=(ioloop,))
thread.start()
thread.join()

# Attempting to add a callback while the loop is closing
ioloop.add_callback(lambda: print("This should not raise an error"))
```

**Expected Behavior**:
Adding a callback while the `IOLoop` is closing should not raise an exception. Instead, the callback should be silently ignored to allow the loop to shut down gracefully.

**Actual Behavior**:
A `RuntimeError` with the message "IOLoop is closing" is raised when trying to add a callback during the shutdown process.
[/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_handle_callback_exception": "PASSED",
    "TestIOLoop.test_handler_callback_file_object": "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",
    "TestIOLoopAddCallback.test_pre_wrap": "PASSED",
    "TestIOLoopAddCallback.test_pre_wrap_with_args": "PASSED",
    "TestIOLoopAddCallbackFromSignal.test_pre_wrap": "PASSED",
    "TestIOLoopAddCallbackFromSignal.test_pre_wrap_with_args": "PASSED",
    "TestIOLoopConfiguration.test_asyncio": "PASSED",
    "TestIOLoopConfiguration.test_asyncio_main": "PASSED",
    "TestIOLoopConfiguration.test_default": "PASSED",
    "TestIOLoopConfiguration.test_twisted": "PASSED",
    "TestIOLoopCurrent.test_default_current": "PASSED",
    "TestIOLoopCurrent.test_force_current": "PASSED",
    "TestIOLoopCurrent.test_non_current": "PASSED",
    "TestIOLoopFutures.test_add_future_stack_context": "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",
    "TestPeriodicCallback.test_basic": "PASSED",
    "TestPeriodicCallback.test_io_loop_set_at_start": "PASSED",
    "TestPeriodicCallback.test_overrun": "PASSED"
}
diff --git a/tornado/platform/asyncio.py b/tornado/platform/asyncio.py
index adc7c1ae..42c8a800 100644
--- a/tornado/platform/asyncio.py
+++ b/tornado/platform/asyncio.py
@@ -130,13 +130,17 @@ class BaseAsyncIOLoop(IOLoop):
         timeout.cancel()
 
     def add_callback(self, callback, *args, **kwargs):
-        if self.closing:
-            # TODO: this is racy; we need a lock to ensure that the
-            # loop isn't closed during call_soon_threadsafe.
-            raise RuntimeError("IOLoop is closing")
-        self.asyncio_loop.call_soon_threadsafe(
-            self._run_callback,
-            functools.partial(stack_context.wrap(callback), *args, **kwargs))
+        try:
+            self.asyncio_loop.call_soon_threadsafe(
+                self._run_callback,
+                functools.partial(stack_context.wrap(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
 
     add_callback_from_signal = add_callback
 

{"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 requires the `~asyncio.AbstractEventLoop.add_reader` family of\n   methods, so it is not compatible with the `~asyncio.ProactorEventLoop` on\n   Windows. Use the `~asyncio.SelectorEventLoop` instead.\n\"\"\"\n\nfrom __future__ import absolute_import, division, print_function\nimport functools\n\nimport tornado.concurrent\nfrom tornado.gen import convert_yielded\nfrom tornado.ioloop import IOLoop\nfrom tornado import stack_context\n\nimport asyncio\n\n\nclass BaseAsyncIOLoop(IOLoop):\n    def initialize(self, asyncio_loop, close_loop=False, **kwargs):\n        self.asyncio_loop = asyncio_loop\n        self.close_loop = close_loop\n        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)\n        self.handlers = {}\n        # Set of fds listening for reads/writes\n        self.readers = set()\n        self.writers = set()\n        self.closing = False\n        super(BaseAsyncIOLoop, self).initialize(**kwargs)\n\n    def close(self, all_fds=False):\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        if self.close_loop:\n            self.asyncio_loop.close()\n\n    def add_handler(self, fd, handler, events):\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, stack_context.wrap(handler))\n        if events & IOLoop.READ:\n            self.asyncio_loop.add_reader(\n                fd, self._handle_events, fd, IOLoop.READ)\n            self.readers.add(fd)\n        if events & IOLoop.WRITE:\n            self.asyncio_loop.add_writer(\n                fd, self._handle_events, fd, IOLoop.WRITE)\n            self.writers.add(fd)\n\n    def update_handler(self, fd, events):\n        fd, fileobj = self.split_fd(fd)\n        if events & IOLoop.READ:\n            if fd not in self.readers:\n                self.asyncio_loop.add_reader(\n                    fd, self._handle_events, fd, IOLoop.READ)\n                self.readers.add(fd)\n        else:\n            if fd in self.readers:\n                self.asyncio_loop.remove_reader(fd)\n                self.readers.remove(fd)\n        if events & IOLoop.WRITE:\n            if fd not in self.writers:\n                self.asyncio_loop.add_writer(\n                    fd, self._handle_events, fd, IOLoop.WRITE)\n                self.writers.add(fd)\n        else:\n            if fd in self.writers:\n                self.asyncio_loop.remove_writer(fd)\n                self.writers.remove(fd)\n\n    def remove_handler(self, fd):\n        fd, fileobj = self.split_fd(fd)\n        if fd not in self.handlers:\n            return\n        if fd in self.readers:\n            self.asyncio_loop.remove_reader(fd)\n            self.readers.remove(fd)\n        if fd in self.writers:\n            self.asyncio_loop.remove_writer(fd)\n            self.writers.remove(fd)\n        del self.handlers[fd]\n\n    def _handle_events(self, fd, events):\n        fileobj, handler_func = self.handlers[fd]\n        handler_func(fileobj, events)\n\n    def start(self):\n        old_current = IOLoop.current(instance=False)\n        try:\n            self._setup_logging()\n            self.make_current()\n            self.asyncio_loop.run_forever()\n        finally:\n            if old_current is None:\n                IOLoop.clear_current()\n            else:\n                old_current.make_current()\n\n    def stop(self):\n        self.asyncio_loop.stop()\n\n    def call_at(self, when, callback, *args, **kwargs):\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()), self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    def remove_timeout(self, timeout):\n        timeout.cancel()\n\n    def add_callback(self, callback, *args, **kwargs):\n        if self.closing:\n            # TODO: this is racy; we need a lock to ensure that the\n            # loop isn't closed during call_soon_threadsafe.\n            raise RuntimeError(\"IOLoop is closing\")\n        self.asyncio_loop.call_soon_threadsafe(\n            self._run_callback,\n            functools.partial(stack_context.wrap(callback), *args, **kwargs))\n\n    add_callback_from_signal = add_callback\n\n    def run_in_executor(self, executor, func, *args):\n        return self.asyncio_loop.run_in_executor(executor, func, *args)\n\n    def set_default_executor(self, executor):\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    def initialize(self, **kwargs):\n        super(AsyncIOMainLoop, self).initialize(asyncio.get_event_loop(),\n                                                close_loop=False, **kwargs)\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:: 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    def initialize(self, **kwargs):\n        loop = asyncio.new_event_loop()\n        try:\n            super(AsyncIOLoop, self).initialize(loop, close_loop=True, **kwargs)\n        except Exception:\n            # If initialize() does not succeed (taking ownership of the loop),\n            # we have to close it.\n            loop.close()\n            raise\n\n    def make_current(self):\n        super(AsyncIOLoop, self).make_current()\n        try:\n            self.old_asyncio = asyncio.get_event_loop()\n        except RuntimeError:\n            self.old_asyncio = None\n        asyncio.set_event_loop(self.asyncio_loop)\n\n    def _clear_current_hook(self):\n        asyncio.set_event_loop(self.old_asyncio)\n\n\ndef to_tornado_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):\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\nclass AnyThreadEventLoopPolicy(asyncio.DefaultEventLoopPolicy):\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    \"\"\"\n    def get_event_loop(self):\n        try:\n            return super().get_event_loop()\n        except RuntimeError:\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"}
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**: Adding a Callback While IOLoop Is Closing Raises RuntimeError

**Description**:
When the `IOLoop` is in the process of closing, attempting to add a callback results in a `RuntimeError` with the message "IOLoop is closing". This behavior is unexpected because callbacks added during the shutdown process should be silently ignored to ensure a smooth termination.

**Example Code**:
```python
import threading
from tornado.ioloop import IOLoop

def shutdown_loop(loop):
    loop.stop()
    loop.close()

ioloop = IOLoop.current()
thread = threading.Thread(target=shutdown_loop, args=(ioloop,))
thread.start()
thread.join()

# Attempting to add a callback while the loop is closing
ioloop.add_callback(lambda: print("This should not raise an error"))
```

**Expected Behavior**:
Adding a callback while the `IOLoop` is closing should not raise an exception. Instead, the callback should be silently ignored to allow the loop to shut down gracefully.

**Actual Behavior**:
A `RuntimeError` with the message "IOLoop is closing" is raised when trying to add a callback during the shutdown process.
[/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 requires the `~asyncio.AbstractEventLoop.add_reader` family of
   methods, so it is not compatible with the `~asyncio.ProactorEventLoop` on
   Windows. Use the `~asyncio.SelectorEventLoop` instead.
"""

from __future__ import absolute_import, division, print_function
import functools

import tornado.concurrent
from tornado.gen import convert_yielded
from tornado.ioloop import IOLoop
from tornado import stack_context

import asyncio


class BaseAsyncIOLoop(IOLoop):
    def initialize(self, asyncio_loop, close_loop=False, **kwargs):
        self.asyncio_loop = asyncio_loop
        self.close_loop = close_loop
        # Maps fd to (fileobj, handler function) pair (as in IOLoop.add_handler)
        self.handlers = {}
        # Set of fds listening for reads/writes
        self.readers = set()
        self.writers = set()
        self.closing = False
        super(BaseAsyncIOLoop, self).initialize(**kwargs)

    def close(self, all_fds=False):
        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)
        if self.close_loop:
            self.asyncio_loop.close()

    def add_handler(self, fd, handler, events):
        fd, fileobj = self.split_fd(fd)
        if fd in self.handlers:
            raise ValueError("fd %s added twice" % fd)
        self.handlers[fd] = (fileobj, stack_context.wrap(handler))
        if events & IOLoop.READ:
            self.asyncio_loop.add_reader(
                fd, self._handle_events, fd, IOLoop.READ)
            self.readers.add(fd)
        if events & IOLoop.WRITE:
            self.asyncio_loop.add_writer(
                fd, self._handle_events, fd, IOLoop.WRITE)
            self.writers.add(fd)

    def update_handler(self, fd, events):
        fd, fileobj = self.split_fd(fd)
        if events & IOLoop.READ:
            if fd not in self.readers:
                self.asyncio_loop.add_reader(
                    fd, self._handle_events, fd, IOLoop.READ)
                self.readers.add(fd)
        else:
            if fd in self.readers:
                self.asyncio_loop.remove_reader(fd)
                self.readers.remove(fd)
        if events & IOLoop.WRITE:
            if fd not in self.writers:
                self.asyncio_loop.add_writer(
                    fd, self._handle_events, fd, IOLoop.WRITE)
                self.writers.add(fd)
        else:
            if fd in self.writers:
                self.asyncio_loop.remove_writer(fd)
                self.writers.remove(fd)

    def remove_handler(self, fd):
        fd, fileobj = self.split_fd(fd)
        if fd not in self.handlers:
            return
        if fd in self.readers:
            self.asyncio_loop.remove_reader(fd)
            self.readers.remove(fd)
        if fd in self.writers:
            self.asyncio_loop.remove_writer(fd)
            self.writers.remove(fd)
        del self.handlers[fd]

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

    def start(self):
        old_current = IOLoop.current(instance=False)
        try:
            self._setup_logging()
            self.make_current()
            self.asyncio_loop.run_forever()
        finally:
            if old_current is None:
                IOLoop.clear_current()
            else:
                old_current.make_current()

    def stop(self):
        self.asyncio_loop.stop()

    def call_at(self, when, callback, *args, **kwargs):
        # 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(stack_context.wrap(callback), *args, **kwargs))

    def remove_timeout(self, timeout):
        timeout.cancel()

    def add_callback(self, callback, *args, **kwargs):
        if self.closing:
            # TODO: this is racy; we need a lock to ensure that the
            # loop isn't closed during call_soon_threadsafe.
            raise RuntimeError("IOLoop is closing")
        self.asyncio_loop.call_soon_threadsafe(
            self._run_callback,
            functools.partial(stack_context.wrap(callback), *args, **kwargs))

    add_callback_from_signal = add_callback

    def run_in_executor(self, executor, func, *args):
        return self.asyncio_loop.run_in_executor(executor, func, *args)

    def set_default_executor(self, executor):
        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.
    """
    def initialize(self, **kwargs):
        super(AsyncIOMainLoop, self).initialize(asyncio.get_event_loop(),
                                                close_loop=False, **kwargs)


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:: 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):
        loop = asyncio.new_event_loop()
        try:
            super(AsyncIOLoop, self).initialize(loop, close_loop=True, **kwargs)
        except Exception:
            # If initialize() does not succeed (taking ownership of the loop),
            # we have to close it.
            loop.close()
            raise

    def make_current(self):
        super(AsyncIOLoop, self).make_current()
        try:
            self.old_asyncio = asyncio.get_event_loop()
        except RuntimeError:
            self.old_asyncio = None
        asyncio.set_event_loop(self.asyncio_loop)

    def _clear_current_hook(self):
        asyncio.set_event_loop(self.old_asyncio)


def to_tornado_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):
    """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)


class AnyThreadEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    """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

    """
    def get_event_loop(self):
        try:
            return super().get_event_loop()
        except RuntimeError:
            # "There is no current event loop in thread %r"
            loop = self.new_event_loop()
            self.set_event_loop(loop)
            return loop

```
--- 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