{
    "file_diffs": [
        {
            "old_file_content": "# Copyright 2015 The Tornado Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\n__all__ = ['Queue', 'QueueFull', 'QueueEmpty']\n\nimport collections\n\nfrom tornado import gen, ioloop\nfrom tornado.concurrent import Future\nfrom tornado.locks import Event\n\n\nclass QueueEmpty(Exception):\n    \"\"\"Raised by `.Queue.get_nowait` when the queue has no items.\"\"\"\n    pass\n\n\nclass QueueFull(Exception):\n    \"\"\"Raised by `.Queue.put_nowait` when a queue is at its maximum size.\"\"\"\n    pass\n\n\nclass Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result\n",
            "new_file_content": "# Copyright 2015 The Tornado Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\n__all__ = ['Queue', 'QueueFull', 'QueueEmpty']\n\nimport collections\n\nfrom tornado import gen, ioloop\nfrom tornado.concurrent import Future\nfrom tornado.locks import Event\n\n\nclass QueueEmpty(Exception):\n    \"\"\"Raised by `.Queue.get_nowait` when the queue has no items.\"\"\"\n    pass\n\n\nclass QueueFull(Exception):\n    \"\"\"Raised by `.Queue.put_nowait` when a queue is at its maximum size.\"\"\"\n    pass\n\n\ndef _set_timeout(future, timeout):\n    if timeout:\n        def on_timeout():\n            future.set_exception(gen.TimeoutError())\n        io_loop = ioloop.IOLoop.current()\n        timeout_handle = io_loop.add_timeout(timeout, on_timeout)\n        future.add_done_callback(\n            lambda _: io_loop.remove_timeout(timeout_handle))\n\n\nclass Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            _set_timeout(future, timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            _set_timeout(future, timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result\n",
            "header": {
                "file": {
                    "path": "tornado/queues.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "1e4a31ab",
                "new_commit_hash": "5986ccc9",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/queues.py"
            },
            "plus_file": {
                "path": "b/tornado/queues.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 33,
                            "length": 6
                        },
                        "new_range": {
                            "start": 33,
                            "length": 16
                        },
                        "section": "class QueueFull(Exception):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "    pass",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "def _set_timeout(future, timeout):",
                                "type": "added"
                            },
                            {
                                "content": "    if timeout:",
                                "type": "added"
                            },
                            {
                                "content": "        def on_timeout():",
                                "type": "added"
                            },
                            {
                                "content": "            future.set_exception(gen.TimeoutError())",
                                "type": "added"
                            },
                            {
                                "content": "        io_loop = ioloop.IOLoop.current()",
                                "type": "added"
                            },
                            {
                                "content": "        timeout_handle = io_loop.add_timeout(timeout, on_timeout)",
                                "type": "added"
                            },
                            {
                                "content": "        future.add_done_callback(",
                                "type": "added"
                            },
                            {
                                "content": "            lambda _: io_loop.remove_timeout(timeout_handle))",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "class Queue(object):",
                                "type": "context"
                            },
                            {
                                "content": "    \"\"\"Coordinate producer and consumer coroutines.",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "tornado/queues.py",
                            "type": "function",
                            "name": "_set_timeout",
                            "content": "def _set_timeout(future, timeout):\n    if timeout:\n        def on_timeout():\n            future.set_exception(gen.TimeoutError())\n        io_loop = ioloop.IOLoop.current()\n        timeout_handle = io_loop.add_timeout(timeout, on_timeout)\n        future.add_done_callback(\n            lambda _: io_loop.remove_timeout(timeout_handle))",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 36,
                            "end_lineno": 43,
                            "parent": null
                        }
                    ],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 82,
                            "length": 10
                        },
                        "new_range": {
                            "start": 92,
                            "length": 7
                        },
                        "section": "class Queue(object):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        except QueueFull:",
                                "type": "context"
                            },
                            {
                                "content": "            future = Future()",
                                "type": "context"
                            },
                            {
                                "content": "            self._putters.append((item, future))",
                                "type": "context"
                            },
                            {
                                "content": "            if timeout:",
                                "type": "deleted"
                            },
                            {
                                "content": "                def on_timeout():",
                                "type": "deleted"
                            },
                            {
                                "content": "                    future.set_exception(gen.TimeoutError())",
                                "type": "deleted"
                            },
                            {
                                "content": "                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)",
                                "type": "deleted"
                            },
                            {
                                "content": "            _set_timeout(future, timeout)",
                                "type": "added"
                            },
                            {
                                "content": "            return future",
                                "type": "context"
                            },
                            {
                                "content": "        else:",
                                "type": "context"
                            },
                            {
                                "content": "            return gen._null_future",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/queues.py",
                            "type": "method",
                            "name": "Queue.put",
                            "content": "class Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 74,
                            "end_lineno": 91,
                            "parent": {
                                "file_name": "tornado/queues.py",
                                "type": "class",
                                "name": "Queue",
                                "content": "class Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 36,
                                "end_lineno": 206,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/queues.py",
                            "type": "class",
                            "name": "Queue",
                            "content": "class Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 36,
                            "end_lineno": 206,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 117,
                            "length": 10
                        },
                        "new_range": {
                            "start": 124,
                            "length": 7
                        },
                        "section": "class Queue(object):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "            future.set_result(self.get_nowait())",
                                "type": "context"
                            },
                            {
                                "content": "        except QueueEmpty:",
                                "type": "context"
                            },
                            {
                                "content": "            self._getters.append(future)",
                                "type": "context"
                            },
                            {
                                "content": "            if timeout:",
                                "type": "deleted"
                            },
                            {
                                "content": "                def on_timeout():",
                                "type": "deleted"
                            },
                            {
                                "content": "                    future.set_exception(gen.TimeoutError())",
                                "type": "deleted"
                            },
                            {
                                "content": "                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)",
                                "type": "deleted"
                            },
                            {
                                "content": "            _set_timeout(future, timeout)",
                                "type": "added"
                            },
                            {
                                "content": "        return future",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def get_nowait(self):",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/queues.py",
                            "type": "class",
                            "name": "Queue",
                            "content": "class Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 36,
                            "end_lineno": 206,
                            "parent": null
                        },
                        {
                            "file_name": "tornado/queues.py",
                            "type": "method",
                            "name": "Queue.get",
                            "content": "class Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 109,
                            "end_lineno": 124,
                            "parent": {
                                "file_name": "tornado/queues.py",
                                "type": "class",
                                "name": "Queue",
                                "content": "class Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 36,
                                "end_lineno": 206,
                                "parent": null
                            }
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nfrom datetime import timedelta\nfrom random import random\n\nfrom tornado import gen, queues\nfrom tornado.gen import TimeoutError\nfrom tornado.testing import gen_test, AsyncTestCase\nfrom tornado.test.util import unittest\n\n\nclass QueueBasicTest(AsyncTestCase):\n    def test_repr_and_str(self):\n        q = queues.Queue(maxsize=1)\n        self.assertIn(hex(id(q)), repr(q))\n        self.assertNotIn(hex(id(q)), str(q))\n        q.get()\n\n        for q_str in repr(q), str(q):\n            self.assertTrue(q_str.startswith('<Queue'))\n            self.assertIn('maxsize=1', q_str)\n            self.assertIn('getters[1]', q_str)\n            self.assertNotIn('putters', q_str)\n            self.assertNotIn('tasks', q_str)\n\n        q.put(None)\n        q.put(None)\n        # Now the queue is full, this putter blocks.\n        q.put(None)\n\n        for q_str in repr(q), str(q):\n            self.assertNotIn('getters', q_str)\n            self.assertIn('putters[1]', q_str)\n            self.assertIn('tasks=2', q_str)\n\n    def test_order(self):\n        q = queues.Queue()\n        for i in [1, 3, 2]:\n            q.put_nowait(i)\n\n        items = [q.get_nowait() for _ in range(3)]\n        self.assertEqual([1, 3, 2], items)\n\n    @gen_test\n    def test_maxsize(self):\n        self.assertRaises(TypeError, queues.Queue, maxsize=None)\n        self.assertRaises(ValueError, queues.Queue, maxsize=-1)\n\n        q = queues.Queue(maxsize=2)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        self.assertEqual(2, q.maxsize)\n        self.assertTrue(q.put(0).done())\n        self.assertTrue(q.put(1).done())\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        put2 = q.put(2)\n        self.assertFalse(put2.done())\n        self.assertEqual(0, (yield q.get()))  # Make room.\n        self.assertTrue(put2.done())\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n\n\nclass QueueGetTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, (yield q.get()))\n\n    def test_nonblocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_get_exception(self):\n        q = queues.Queue()\n        self.assertRaises(queues.QueueEmpty, q.get_nowait)\n\n    @gen_test\n    def test_get_with_putters(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1)\n        self.assertEqual(0, (yield q.get()))\n        self.assertIsNone((yield put))\n\n    @gen_test\n    def test_blocking_get_wait(self):\n        q = queues.Queue()\n        q.put(0)\n        self.io_loop.call_later(0.01, q.put, 1)\n        self.io_loop.call_later(0.02, q.put, 2)\n        self.assertEqual(0, (yield q.get(timeout=timedelta(seconds=1))))\n        self.assertEqual(1, (yield q.get(timeout=timedelta(seconds=1))))\n\n    @gen_test\n    def test_get_timeout(self):\n        q = queues.Queue()\n        get_timeout = q.get(timeout=timedelta(seconds=0.01))\n        get = q.get()\n        with self.assertRaises(TimeoutError):\n            yield get_timeout\n        \n        q.put_nowait(0)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)\n        self.assertEqual(0, (yield q.get()))  # get() clears the waiters.\n        self.assertEqual(1, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_get_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        self.assertEqual(11, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(11, len(q._getters))\n        self.assertFalse(get.done())  # Final waiter is still active.\n        q.get()  # get() clears the waiters.\n        self.assertEqual(2, len(q._getters))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n\nclass QueuePutTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_put(self):\n        q = queues.Queue()\n        q.put(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_put_exception(self):\n        q = queues.Queue(1)\n        q.put(0)\n        self.assertRaises(queues.QueueFull, q.put_nowait, 1)\n\n    @gen_test\n    def test_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        yield q.put(0)\n        self.assertEqual(0, (yield get0))\n        yield q.put(1)\n        self.assertEqual(1, (yield get1))\n        \n    @gen_test\n    def test_nonblocking_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        q.put_nowait(0)\n        # put_nowait does *not* immediately unblock getters.\n        yield gen.moment\n        self.assertEqual(0, (yield get0))\n        q.put_nowait(1)\n        yield gen.moment\n        self.assertEqual(1, (yield get1))\n\n    @gen_test\n    def test_blocking_put_wait(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        self.io_loop.call_later(0.01, q.get)\n        self.io_loop.call_later(0.02, q.get)\n        futures = [q.put(0), q.put(1)]\n        self.assertFalse(any(f.done() for f in futures))\n        yield futures\n\n    @gen_test\n    def test_put_timeout(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)  # Now it's full.\n        put_timeout = q.put(1, timeout=timedelta(seconds=0.01))\n        put = q.put(2)\n        with self.assertRaises(TimeoutError):\n            yield put_timeout\n        \n        self.assertEqual(0, q.get_nowait())\n        # 1 was never put in the queue.\n        self.assertEqual(2, (yield q.get()))\n\n        # Final get() unblocked this putter.\n        yield put\n\n    @gen_test\n    def test_put_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)  # put() clears the waiters.\n        self.assertEqual(2, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_put_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        q.get()\n        self.assertEqual(12, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(12, len(q._getters))\n        self.assertFalse(get.done())  # Final waiters still active.\n        q.put(0)  # put() clears the waiters.\n        self.assertEqual(1, len(q._getters))\n        self.assertEqual(0, (yield get))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n    @gen_test\n    def test_float_maxsize(self):\n        # Non-int maxsize must round down: http://bugs.python.org/issue21723\n        q = queues.Queue(maxsize=1.3)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        q.put_nowait(0)\n        q.put_nowait(1)\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        self.assertRaises(queues.QueueFull, q.put_nowait, 2)\n        self.assertEqual(0, q.get_nowait())\n        self.assertFalse(q.empty())\n        self.assertFalse(q.full())\n\n        yield q.put(2)\n        put = q.put(3)\n        self.assertFalse(put.done())\n        self.assertEqual(1, (yield q.get()))\n        yield put\n        self.assertTrue(q.full())\n\n\nclass QueueJoinTest(AsyncTestCase):\n    def test_task_done_underflow(self):\n        q = queues.Queue()\n        self.assertRaises(ValueError, q.task_done)\n\n    @gen_test\n    def test_task_done(self):\n        q = queues.Queue()\n        for i in range(100):\n            q.put_nowait(i)\n\n        self.accumulator = 0\n\n        @gen.coroutine\n        def worker():\n            while True:\n                item = yield q.get()\n                self.accumulator += item\n                q.task_done()\n                yield gen.sleep(random() * 0.01)\n\n        # Two coroutines share work.\n        worker()\n        worker()\n        yield q.join()\n        self.assertEqual(sum(range(100)), self.accumulator)\n\n    @gen_test\n    def test_task_done_delay(self):\n        # Verify it is task_done(), not get(), that unblocks join().\n        q = queues.Queue()\n        q.put_nowait(0)\n        join = q.join()\n        self.assertFalse(join.done())\n        yield q.get()\n        self.assertFalse(join.done())\n        yield gen.moment\n        self.assertFalse(join.done())\n        q.task_done()\n        self.assertTrue(join.done())\n\n    @gen_test\n    def test_join_empty_queue(self):\n        q = queues.Queue()\n        yield q.join()\n        yield q.join()\n\n    @gen_test\n    def test_join_timeout(self):\n        q = queues.Queue()\n        q.put(0)\n        with self.assertRaises(TimeoutError):\n            yield q.join(timeout=timedelta(seconds=0.01))\n\n    @gen_test\n    def test_producer_consumer(self):\n        q = queues.Queue(maxsize=3)\n        history = []\n\n        # We don't yield between get() and task_done(), so get() must wait for\n        # the next tick. Otherwise we'd immediately call task_done and unblock\n        # join() before q.put() resumes, and we'd only process the first four\n        # items.\n        @gen.coroutine\n        def consumer():\n            while True:\n                history.append((yield q.get()))\n                q.task_done()\n\n        @gen.coroutine\n        def producer():\n            for item in range(10):\n                yield q.put(item)\n\n        consumer()\n        yield producer()\n        yield q.join()\n        self.assertEqual(list(range(10)), history)\n\n\nif __name__ == '__main__':\n    unittest.main()\n",
            "new_file_content": "# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nfrom datetime import timedelta\nfrom random import random\n\nfrom tornado import gen, queues\nfrom tornado.gen import TimeoutError\nfrom tornado.testing import gen_test, AsyncTestCase\nfrom tornado.test.util import unittest\n\n\nclass QueueBasicTest(AsyncTestCase):\n    def test_repr_and_str(self):\n        q = queues.Queue(maxsize=1)\n        self.assertIn(hex(id(q)), repr(q))\n        self.assertNotIn(hex(id(q)), str(q))\n        q.get()\n\n        for q_str in repr(q), str(q):\n            self.assertTrue(q_str.startswith('<Queue'))\n            self.assertIn('maxsize=1', q_str)\n            self.assertIn('getters[1]', q_str)\n            self.assertNotIn('putters', q_str)\n            self.assertNotIn('tasks', q_str)\n\n        q.put(None)\n        q.put(None)\n        # Now the queue is full, this putter blocks.\n        q.put(None)\n\n        for q_str in repr(q), str(q):\n            self.assertNotIn('getters', q_str)\n            self.assertIn('putters[1]', q_str)\n            self.assertIn('tasks=2', q_str)\n\n    def test_order(self):\n        q = queues.Queue()\n        for i in [1, 3, 2]:\n            q.put_nowait(i)\n\n        items = [q.get_nowait() for _ in range(3)]\n        self.assertEqual([1, 3, 2], items)\n\n    @gen_test\n    def test_maxsize(self):\n        self.assertRaises(TypeError, queues.Queue, maxsize=None)\n        self.assertRaises(ValueError, queues.Queue, maxsize=-1)\n\n        q = queues.Queue(maxsize=2)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        self.assertEqual(2, q.maxsize)\n        self.assertTrue(q.put(0).done())\n        self.assertTrue(q.put(1).done())\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        put2 = q.put(2)\n        self.assertFalse(put2.done())\n        self.assertEqual(0, (yield q.get()))  # Make room.\n        self.assertTrue(put2.done())\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n\n\nclass QueueGetTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, (yield q.get()))\n\n    def test_nonblocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_get_exception(self):\n        q = queues.Queue()\n        self.assertRaises(queues.QueueEmpty, q.get_nowait)\n\n    @gen_test\n    def test_get_with_putters(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1)\n        self.assertEqual(0, (yield q.get()))\n        self.assertIsNone((yield put))\n\n    @gen_test\n    def test_blocking_get_wait(self):\n        q = queues.Queue()\n        q.put(0)\n        self.io_loop.call_later(0.01, q.put, 1)\n        self.io_loop.call_later(0.02, q.put, 2)\n        self.assertEqual(0, (yield q.get(timeout=timedelta(seconds=1))))\n        self.assertEqual(1, (yield q.get(timeout=timedelta(seconds=1))))\n\n    @gen_test\n    def test_get_timeout(self):\n        q = queues.Queue()\n        get_timeout = q.get(timeout=timedelta(seconds=0.01))\n        get = q.get()\n        with self.assertRaises(TimeoutError):\n            yield get_timeout\n        \n        q.put_nowait(0)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_timeout_preempted(self):\n        q = queues.Queue()\n        get = q.get(timeout=timedelta(seconds=0.01))\n        q.put(0)\n        yield gen.sleep(0.02)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)\n        self.assertEqual(0, (yield q.get()))  # get() clears the waiters.\n        self.assertEqual(1, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_get_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        self.assertEqual(11, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(11, len(q._getters))\n        self.assertFalse(get.done())  # Final waiter is still active.\n        q.get()  # get() clears the waiters.\n        self.assertEqual(2, len(q._getters))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n\nclass QueuePutTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_put(self):\n        q = queues.Queue()\n        q.put(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_put_exception(self):\n        q = queues.Queue(1)\n        q.put(0)\n        self.assertRaises(queues.QueueFull, q.put_nowait, 1)\n\n    @gen_test\n    def test_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        yield q.put(0)\n        self.assertEqual(0, (yield get0))\n        yield q.put(1)\n        self.assertEqual(1, (yield get1))\n        \n    @gen_test\n    def test_nonblocking_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        q.put_nowait(0)\n        # put_nowait does *not* immediately unblock getters.\n        yield gen.moment\n        self.assertEqual(0, (yield get0))\n        q.put_nowait(1)\n        yield gen.moment\n        self.assertEqual(1, (yield get1))\n\n    @gen_test\n    def test_blocking_put_wait(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        self.io_loop.call_later(0.01, q.get)\n        self.io_loop.call_later(0.02, q.get)\n        futures = [q.put(0), q.put(1)]\n        self.assertFalse(any(f.done() for f in futures))\n        yield futures\n\n    @gen_test\n    def test_put_timeout(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)  # Now it's full.\n        put_timeout = q.put(1, timeout=timedelta(seconds=0.01))\n        put = q.put(2)\n        with self.assertRaises(TimeoutError):\n            yield put_timeout\n        \n        self.assertEqual(0, q.get_nowait())\n        # 1 was never put in the queue.\n        self.assertEqual(2, (yield q.get()))\n\n        # Final get() unblocked this putter.\n        yield put\n\n    @gen_test\n    def test_put_timeout_preempted(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1, timeout=timedelta(seconds=0.01))\n        q.get()\n        yield gen.sleep(0.02)\n        yield put  # No TimeoutError.\n\n    @gen_test\n    def test_put_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)  # put() clears the waiters.\n        self.assertEqual(2, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_put_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        q.get()\n        self.assertEqual(12, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(12, len(q._getters))\n        self.assertFalse(get.done())  # Final waiters still active.\n        q.put(0)  # put() clears the waiters.\n        self.assertEqual(1, len(q._getters))\n        self.assertEqual(0, (yield get))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n    @gen_test\n    def test_float_maxsize(self):\n        # Non-int maxsize must round down: http://bugs.python.org/issue21723\n        q = queues.Queue(maxsize=1.3)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        q.put_nowait(0)\n        q.put_nowait(1)\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        self.assertRaises(queues.QueueFull, q.put_nowait, 2)\n        self.assertEqual(0, q.get_nowait())\n        self.assertFalse(q.empty())\n        self.assertFalse(q.full())\n\n        yield q.put(2)\n        put = q.put(3)\n        self.assertFalse(put.done())\n        self.assertEqual(1, (yield q.get()))\n        yield put\n        self.assertTrue(q.full())\n\n\nclass QueueJoinTest(AsyncTestCase):\n    def test_task_done_underflow(self):\n        q = queues.Queue()\n        self.assertRaises(ValueError, q.task_done)\n\n    @gen_test\n    def test_task_done(self):\n        q = queues.Queue()\n        for i in range(100):\n            q.put_nowait(i)\n\n        self.accumulator = 0\n\n        @gen.coroutine\n        def worker():\n            while True:\n                item = yield q.get()\n                self.accumulator += item\n                q.task_done()\n                yield gen.sleep(random() * 0.01)\n\n        # Two coroutines share work.\n        worker()\n        worker()\n        yield q.join()\n        self.assertEqual(sum(range(100)), self.accumulator)\n\n    @gen_test\n    def test_task_done_delay(self):\n        # Verify it is task_done(), not get(), that unblocks join().\n        q = queues.Queue()\n        q.put_nowait(0)\n        join = q.join()\n        self.assertFalse(join.done())\n        yield q.get()\n        self.assertFalse(join.done())\n        yield gen.moment\n        self.assertFalse(join.done())\n        q.task_done()\n        self.assertTrue(join.done())\n\n    @gen_test\n    def test_join_empty_queue(self):\n        q = queues.Queue()\n        yield q.join()\n        yield q.join()\n\n    @gen_test\n    def test_join_timeout(self):\n        q = queues.Queue()\n        q.put(0)\n        with self.assertRaises(TimeoutError):\n            yield q.join(timeout=timedelta(seconds=0.01))\n\n    @gen_test\n    def test_producer_consumer(self):\n        q = queues.Queue(maxsize=3)\n        history = []\n\n        # We don't yield between get() and task_done(), so get() must wait for\n        # the next tick. Otherwise we'd immediately call task_done and unblock\n        # join() before q.put() resumes, and we'd only process the first four\n        # items.\n        @gen.coroutine\n        def consumer():\n            while True:\n                history.append((yield q.get()))\n                q.task_done()\n\n        @gen.coroutine\n        def producer():\n            for item in range(10):\n                yield q.put(item)\n\n        consumer()\n        yield producer()\n        yield q.join()\n        self.assertEqual(list(range(10)), history)\n\n\nif __name__ == '__main__':\n    unittest.main()\n",
            "header": {
                "file": {
                    "path": "tornado/test/queues_test.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "34b61131",
                "new_commit_hash": "ac211833",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/test/queues_test.py"
            },
            "plus_file": {
                "path": "b/tornado/test/queues_test.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 116,
                            "length": 6
                        },
                        "new_range": {
                            "start": 116,
                            "length": 14
                        },
                        "section": "class QueueGetTest(AsyncTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        q.put_nowait(0)",
                                "type": "context"
                            },
                            {
                                "content": "        self.assertEqual(0, (yield get))",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    @gen_test",
                                "type": "added"
                            },
                            {
                                "content": "    def test_get_timeout_preempted(self):",
                                "type": "added"
                            },
                            {
                                "content": "        q = queues.Queue()",
                                "type": "added"
                            },
                            {
                                "content": "        get = q.get(timeout=timedelta(seconds=0.01))",
                                "type": "added"
                            },
                            {
                                "content": "        q.put(0)",
                                "type": "added"
                            },
                            {
                                "content": "        yield gen.sleep(0.02)",
                                "type": "added"
                            },
                            {
                                "content": "        self.assertEqual(0, (yield get))",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    @gen_test",
                                "type": "context"
                            },
                            {
                                "content": "    def test_get_clears_timed_out_putters(self):",
                                "type": "context"
                            },
                            {
                                "content": "        q = queues.Queue(1)",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/test/queues_test.py",
                            "type": "class",
                            "name": "QueueGetTest",
                            "content": "class QueueGetTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, (yield q.get()))\n\n    def test_nonblocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_get_exception(self):\n        q = queues.Queue()\n        self.assertRaises(queues.QueueEmpty, q.get_nowait)\n\n    @gen_test\n    def test_get_with_putters(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1)\n        self.assertEqual(0, (yield q.get()))\n        self.assertIsNone((yield put))\n\n    @gen_test\n    def test_blocking_get_wait(self):\n        q = queues.Queue()\n        q.put(0)\n        self.io_loop.call_later(0.01, q.put, 1)\n        self.io_loop.call_later(0.02, q.put, 2)\n        self.assertEqual(0, (yield q.get(timeout=timedelta(seconds=1))))\n        self.assertEqual(1, (yield q.get(timeout=timedelta(seconds=1))))\n\n    @gen_test\n    def test_get_timeout(self):\n        q = queues.Queue()\n        get_timeout = q.get(timeout=timedelta(seconds=0.01))\n        get = q.get()\n        with self.assertRaises(TimeoutError):\n            yield get_timeout\n        \n        q.put_nowait(0)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_timeout_preempted(self):\n        q = queues.Queue()\n        get = q.get(timeout=timedelta(seconds=0.01))\n        q.put(0)\n        yield gen.sleep(0.02)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)\n        self.assertEqual(0, (yield q.get()))  # get() clears the waiters.\n        self.assertEqual(1, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_get_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        self.assertEqual(11, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(11, len(q._getters))\n        self.assertFalse(get.done())  # Final waiter is still active.\n        q.get()  # get() clears the waiters.\n        self.assertEqual(2, len(q._getters))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 75,
                            "end_lineno": 155,
                            "parent": null
                        }
                    ],
                    "added_entities": [
                        {
                            "file_name": "tornado/test/queues_test.py",
                            "type": "method",
                            "name": "QueueGetTest.test_get_timeout_preempted",
                            "content": "class QueueGetTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, (yield q.get()))\n\n    def test_nonblocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_get_exception(self):\n        q = queues.Queue()\n        self.assertRaises(queues.QueueEmpty, q.get_nowait)\n\n    @gen_test\n    def test_get_with_putters(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1)\n        self.assertEqual(0, (yield q.get()))\n        self.assertIsNone((yield put))\n\n    @gen_test\n    def test_blocking_get_wait(self):\n        q = queues.Queue()\n        q.put(0)\n        self.io_loop.call_later(0.01, q.put, 1)\n        self.io_loop.call_later(0.02, q.put, 2)\n        self.assertEqual(0, (yield q.get(timeout=timedelta(seconds=1))))\n        self.assertEqual(1, (yield q.get(timeout=timedelta(seconds=1))))\n\n    @gen_test\n    def test_get_timeout(self):\n        q = queues.Queue()\n        get_timeout = q.get(timeout=timedelta(seconds=0.01))\n        get = q.get()\n        with self.assertRaises(TimeoutError):\n            yield get_timeout\n        \n        q.put_nowait(0)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_timeout_preempted(self):\n        q = queues.Queue()\n        get = q.get(timeout=timedelta(seconds=0.01))\n        q.put(0)\n        yield gen.sleep(0.02)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)\n        self.assertEqual(0, (yield q.get()))  # get() clears the waiters.\n        self.assertEqual(1, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_get_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        self.assertEqual(11, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(11, len(q._getters))\n        self.assertFalse(get.done())  # Final waiter is still active.\n        q.get()  # get() clears the waiters.\n        self.assertEqual(2, len(q._getters))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 120,
                            "end_lineno": 125,
                            "parent": {
                                "file_name": "tornado/test/queues_test.py",
                                "type": "class",
                                "name": "QueueGetTest",
                                "content": "class QueueGetTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, (yield q.get()))\n\n    def test_nonblocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_get_exception(self):\n        q = queues.Queue()\n        self.assertRaises(queues.QueueEmpty, q.get_nowait)\n\n    @gen_test\n    def test_get_with_putters(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1)\n        self.assertEqual(0, (yield q.get()))\n        self.assertIsNone((yield put))\n\n    @gen_test\n    def test_blocking_get_wait(self):\n        q = queues.Queue()\n        q.put(0)\n        self.io_loop.call_later(0.01, q.put, 1)\n        self.io_loop.call_later(0.02, q.put, 2)\n        self.assertEqual(0, (yield q.get(timeout=timedelta(seconds=1))))\n        self.assertEqual(1, (yield q.get(timeout=timedelta(seconds=1))))\n\n    @gen_test\n    def test_get_timeout(self):\n        q = queues.Queue()\n        get_timeout = q.get(timeout=timedelta(seconds=0.01))\n        get = q.get()\n        with self.assertRaises(TimeoutError):\n            yield get_timeout\n        \n        q.put_nowait(0)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_timeout_preempted(self):\n        q = queues.Queue()\n        get = q.get(timeout=timedelta(seconds=0.01))\n        q.put(0)\n        yield gen.sleep(0.02)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)\n        self.assertEqual(0, (yield q.get()))  # get() clears the waiters.\n        self.assertEqual(1, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_get_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        self.assertEqual(11, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(11, len(q._getters))\n        self.assertFalse(get.done())  # Final waiter is still active.\n        q.get()  # get() clears the waiters.\n        self.assertEqual(2, len(q._getters))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 75,
                                "end_lineno": 155,
                                "parent": null
                            }
                        }
                    ],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 208,
                            "length": 6
                        },
                        "new_range": {
                            "start": 216,
                            "length": 15
                        },
                        "section": "class QueuePutTest(AsyncTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        # Final get() unblocked this putter.",
                                "type": "context"
                            },
                            {
                                "content": "        yield put",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    @gen_test",
                                "type": "added"
                            },
                            {
                                "content": "    def test_put_timeout_preempted(self):",
                                "type": "added"
                            },
                            {
                                "content": "        q = queues.Queue(1)",
                                "type": "added"
                            },
                            {
                                "content": "        q.put_nowait(0)",
                                "type": "added"
                            },
                            {
                                "content": "        put = q.put(1, timeout=timedelta(seconds=0.01))",
                                "type": "added"
                            },
                            {
                                "content": "        q.get()",
                                "type": "added"
                            },
                            {
                                "content": "        yield gen.sleep(0.02)",
                                "type": "added"
                            },
                            {
                                "content": "        yield put  # No TimeoutError.",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    @gen_test",
                                "type": "context"
                            },
                            {
                                "content": "    def test_put_clears_timed_out_putters(self):",
                                "type": "context"
                            },
                            {
                                "content": "        q = queues.Queue(1)",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/test/queues_test.py",
                            "type": "class",
                            "name": "QueuePutTest",
                            "content": "class QueuePutTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_put(self):\n        q = queues.Queue()\n        q.put(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_put_exception(self):\n        q = queues.Queue(1)\n        q.put(0)\n        self.assertRaises(queues.QueueFull, q.put_nowait, 1)\n\n    @gen_test\n    def test_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        yield q.put(0)\n        self.assertEqual(0, (yield get0))\n        yield q.put(1)\n        self.assertEqual(1, (yield get1))\n        \n    @gen_test\n    def test_nonblocking_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        q.put_nowait(0)\n        # put_nowait does *not* immediately unblock getters.\n        yield gen.moment\n        self.assertEqual(0, (yield get0))\n        q.put_nowait(1)\n        yield gen.moment\n        self.assertEqual(1, (yield get1))\n\n    @gen_test\n    def test_blocking_put_wait(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        self.io_loop.call_later(0.01, q.get)\n        self.io_loop.call_later(0.02, q.get)\n        futures = [q.put(0), q.put(1)]\n        self.assertFalse(any(f.done() for f in futures))\n        yield futures\n\n    @gen_test\n    def test_put_timeout(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)  # Now it's full.\n        put_timeout = q.put(1, timeout=timedelta(seconds=0.01))\n        put = q.put(2)\n        with self.assertRaises(TimeoutError):\n            yield put_timeout\n        \n        self.assertEqual(0, q.get_nowait())\n        # 1 was never put in the queue.\n        self.assertEqual(2, (yield q.get()))\n\n        # Final get() unblocked this putter.\n        yield put\n\n    @gen_test\n    def test_put_timeout_preempted(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1, timeout=timedelta(seconds=0.01))\n        q.get()\n        yield gen.sleep(0.02)\n        yield put  # No TimeoutError.\n\n    @gen_test\n    def test_put_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)  # put() clears the waiters.\n        self.assertEqual(2, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_put_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        q.get()\n        self.assertEqual(12, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(12, len(q._getters))\n        self.assertFalse(get.done())  # Final waiters still active.\n        q.put(0)  # put() clears the waiters.\n        self.assertEqual(1, len(q._getters))\n        self.assertEqual(0, (yield get))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n    @gen_test\n    def test_float_maxsize(self):\n        # Non-int maxsize must round down: http://bugs.python.org/issue21723\n        q = queues.Queue(maxsize=1.3)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        q.put_nowait(0)\n        q.put_nowait(1)\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        self.assertRaises(queues.QueueFull, q.put_nowait, 2)\n        self.assertEqual(0, q.get_nowait())\n        self.assertFalse(q.empty())\n        self.assertFalse(q.full())\n\n        yield q.put(2)\n        put = q.put(3)\n        self.assertFalse(put.done())\n        self.assertEqual(1, (yield q.get()))\n        yield put\n        self.assertTrue(q.full())",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 158,
                            "end_lineno": 279,
                            "parent": null
                        }
                    ],
                    "added_entities": [
                        {
                            "file_name": "tornado/test/queues_test.py",
                            "type": "method",
                            "name": "QueuePutTest.test_put_timeout_preempted",
                            "content": "class QueuePutTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_put(self):\n        q = queues.Queue()\n        q.put(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_put_exception(self):\n        q = queues.Queue(1)\n        q.put(0)\n        self.assertRaises(queues.QueueFull, q.put_nowait, 1)\n\n    @gen_test\n    def test_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        yield q.put(0)\n        self.assertEqual(0, (yield get0))\n        yield q.put(1)\n        self.assertEqual(1, (yield get1))\n        \n    @gen_test\n    def test_nonblocking_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        q.put_nowait(0)\n        # put_nowait does *not* immediately unblock getters.\n        yield gen.moment\n        self.assertEqual(0, (yield get0))\n        q.put_nowait(1)\n        yield gen.moment\n        self.assertEqual(1, (yield get1))\n\n    @gen_test\n    def test_blocking_put_wait(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        self.io_loop.call_later(0.01, q.get)\n        self.io_loop.call_later(0.02, q.get)\n        futures = [q.put(0), q.put(1)]\n        self.assertFalse(any(f.done() for f in futures))\n        yield futures\n\n    @gen_test\n    def test_put_timeout(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)  # Now it's full.\n        put_timeout = q.put(1, timeout=timedelta(seconds=0.01))\n        put = q.put(2)\n        with self.assertRaises(TimeoutError):\n            yield put_timeout\n        \n        self.assertEqual(0, q.get_nowait())\n        # 1 was never put in the queue.\n        self.assertEqual(2, (yield q.get()))\n\n        # Final get() unblocked this putter.\n        yield put\n\n    @gen_test\n    def test_put_timeout_preempted(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1, timeout=timedelta(seconds=0.01))\n        q.get()\n        yield gen.sleep(0.02)\n        yield put  # No TimeoutError.\n\n    @gen_test\n    def test_put_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)  # put() clears the waiters.\n        self.assertEqual(2, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_put_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        q.get()\n        self.assertEqual(12, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(12, len(q._getters))\n        self.assertFalse(get.done())  # Final waiters still active.\n        q.put(0)  # put() clears the waiters.\n        self.assertEqual(1, len(q._getters))\n        self.assertEqual(0, (yield get))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n    @gen_test\n    def test_float_maxsize(self):\n        # Non-int maxsize must round down: http://bugs.python.org/issue21723\n        q = queues.Queue(maxsize=1.3)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        q.put_nowait(0)\n        q.put_nowait(1)\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        self.assertRaises(queues.QueueFull, q.put_nowait, 2)\n        self.assertEqual(0, q.get_nowait())\n        self.assertFalse(q.empty())\n        self.assertFalse(q.full())\n\n        yield q.put(2)\n        put = q.put(3)\n        self.assertFalse(put.done())\n        self.assertEqual(1, (yield q.get()))\n        yield put\n        self.assertTrue(q.full())",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 220,
                            "end_lineno": 226,
                            "parent": {
                                "file_name": "tornado/test/queues_test.py",
                                "type": "class",
                                "name": "QueuePutTest",
                                "content": "class QueuePutTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_put(self):\n        q = queues.Queue()\n        q.put(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_put_exception(self):\n        q = queues.Queue(1)\n        q.put(0)\n        self.assertRaises(queues.QueueFull, q.put_nowait, 1)\n\n    @gen_test\n    def test_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        yield q.put(0)\n        self.assertEqual(0, (yield get0))\n        yield q.put(1)\n        self.assertEqual(1, (yield get1))\n        \n    @gen_test\n    def test_nonblocking_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        q.put_nowait(0)\n        # put_nowait does *not* immediately unblock getters.\n        yield gen.moment\n        self.assertEqual(0, (yield get0))\n        q.put_nowait(1)\n        yield gen.moment\n        self.assertEqual(1, (yield get1))\n\n    @gen_test\n    def test_blocking_put_wait(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        self.io_loop.call_later(0.01, q.get)\n        self.io_loop.call_later(0.02, q.get)\n        futures = [q.put(0), q.put(1)]\n        self.assertFalse(any(f.done() for f in futures))\n        yield futures\n\n    @gen_test\n    def test_put_timeout(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)  # Now it's full.\n        put_timeout = q.put(1, timeout=timedelta(seconds=0.01))\n        put = q.put(2)\n        with self.assertRaises(TimeoutError):\n            yield put_timeout\n        \n        self.assertEqual(0, q.get_nowait())\n        # 1 was never put in the queue.\n        self.assertEqual(2, (yield q.get()))\n\n        # Final get() unblocked this putter.\n        yield put\n\n    @gen_test\n    def test_put_timeout_preempted(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1, timeout=timedelta(seconds=0.01))\n        q.get()\n        yield gen.sleep(0.02)\n        yield put  # No TimeoutError.\n\n    @gen_test\n    def test_put_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)  # put() clears the waiters.\n        self.assertEqual(2, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_put_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        q.get()\n        self.assertEqual(12, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(12, len(q._getters))\n        self.assertFalse(get.done())  # Final waiters still active.\n        q.put(0)  # put() clears the waiters.\n        self.assertEqual(1, len(q._getters))\n        self.assertEqual(0, (yield get))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n    @gen_test\n    def test_float_maxsize(self):\n        # Non-int maxsize must round down: http://bugs.python.org/issue21723\n        q = queues.Queue(maxsize=1.3)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        q.put_nowait(0)\n        q.put_nowait(1)\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        self.assertRaises(queues.QueueFull, q.put_nowait, 2)\n        self.assertEqual(0, q.get_nowait())\n        self.assertFalse(q.empty())\n        self.assertFalse(q.full())\n\n        yield q.put(2)\n        put = q.put(3)\n        self.assertFalse(put.done())\n        self.assertEqual(1, (yield q.get()))\n        yield put\n        self.assertTrue(q.full())",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 158,
                                "end_lineno": 279,
                                "parent": null
                            }
                        }
                    ],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "486abfb8a3b2e40fcfb87c5179252748913fcfca^",
    "new_commit_hash": "486abfb8a3b2e40fcfb87c5179252748913fcfca",
    "commit_message": "Cancel timeouts if Queue.get or put are resolved first.",
    "commit_date": "2015-03-15T21:09:12-04:00",
    "metadata": {}
}
{
    "repo_name": "tornado",
    "new_commit_hash": "486abfb8a3b2e40fcfb87c5179252748913fcfca",
    "test_file_codes": [
        "# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nfrom datetime import timedelta\nfrom random import random\n\nfrom tornado import gen, queues\nfrom tornado.gen import TimeoutError\nfrom tornado.testing import gen_test, AsyncTestCase\nfrom tornado.test.util import unittest\n\n\nclass QueueBasicTest(AsyncTestCase):\n    def test_repr_and_str(self):\n        q = queues.Queue(maxsize=1)\n        self.assertIn(hex(id(q)), repr(q))\n        self.assertNotIn(hex(id(q)), str(q))\n        q.get()\n\n        for q_str in repr(q), str(q):\n            self.assertTrue(q_str.startswith('<Queue'))\n            self.assertIn('maxsize=1', q_str)\n            self.assertIn('getters[1]', q_str)\n            self.assertNotIn('putters', q_str)\n            self.assertNotIn('tasks', q_str)\n\n        q.put(None)\n        q.put(None)\n        # Now the queue is full, this putter blocks.\n        q.put(None)\n\n        for q_str in repr(q), str(q):\n            self.assertNotIn('getters', q_str)\n            self.assertIn('putters[1]', q_str)\n            self.assertIn('tasks=2', q_str)\n\n    def test_order(self):\n        q = queues.Queue()\n        for i in [1, 3, 2]:\n            q.put_nowait(i)\n\n        items = [q.get_nowait() for _ in range(3)]\n        self.assertEqual([1, 3, 2], items)\n\n    @gen_test\n    def test_maxsize(self):\n        self.assertRaises(TypeError, queues.Queue, maxsize=None)\n        self.assertRaises(ValueError, queues.Queue, maxsize=-1)\n\n        q = queues.Queue(maxsize=2)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        self.assertEqual(2, q.maxsize)\n        self.assertTrue(q.put(0).done())\n        self.assertTrue(q.put(1).done())\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        put2 = q.put(2)\n        self.assertFalse(put2.done())\n        self.assertEqual(0, (yield q.get()))  # Make room.\n        self.assertTrue(put2.done())\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n\n\nclass QueueGetTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, (yield q.get()))\n\n    def test_nonblocking_get(self):\n        q = queues.Queue()\n        q.put_nowait(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_get_exception(self):\n        q = queues.Queue()\n        self.assertRaises(queues.QueueEmpty, q.get_nowait)\n\n    @gen_test\n    def test_get_with_putters(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1)\n        self.assertEqual(0, (yield q.get()))\n        self.assertIsNone((yield put))\n\n    @gen_test\n    def test_blocking_get_wait(self):\n        q = queues.Queue()\n        q.put(0)\n        self.io_loop.call_later(0.01, q.put, 1)\n        self.io_loop.call_later(0.02, q.put, 2)\n        self.assertEqual(0, (yield q.get(timeout=timedelta(seconds=1))))\n        self.assertEqual(1, (yield q.get(timeout=timedelta(seconds=1))))\n\n    @gen_test\n    def test_get_timeout(self):\n        q = queues.Queue()\n        get_timeout = q.get(timeout=timedelta(seconds=0.01))\n        get = q.get()\n        with self.assertRaises(TimeoutError):\n            yield get_timeout\n        \n        q.put_nowait(0)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_timeout_preempted(self):\n        q = queues.Queue()\n        get = q.get(timeout=timedelta(seconds=0.01))\n        q.put(0)\n        yield gen.sleep(0.02)\n        self.assertEqual(0, (yield get))\n\n    @gen_test\n    def test_get_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)\n        self.assertEqual(0, (yield q.get()))  # get() clears the waiters.\n        self.assertEqual(1, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_get_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        self.assertEqual(11, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(11, len(q._getters))\n        self.assertFalse(get.done())  # Final waiter is still active.\n        q.get()  # get() clears the waiters.\n        self.assertEqual(2, len(q._getters))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n\nclass QueuePutTest(AsyncTestCase):\n    @gen_test\n    def test_blocking_put(self):\n        q = queues.Queue()\n        q.put(0)\n        self.assertEqual(0, q.get_nowait())\n\n    def test_nonblocking_put_exception(self):\n        q = queues.Queue(1)\n        q.put(0)\n        self.assertRaises(queues.QueueFull, q.put_nowait, 1)\n\n    @gen_test\n    def test_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        yield q.put(0)\n        self.assertEqual(0, (yield get0))\n        yield q.put(1)\n        self.assertEqual(1, (yield get1))\n        \n    @gen_test\n    def test_nonblocking_put_with_getters(self):\n        q = queues.Queue()\n        get0 = q.get()\n        get1 = q.get()\n        q.put_nowait(0)\n        # put_nowait does *not* immediately unblock getters.\n        yield gen.moment\n        self.assertEqual(0, (yield get0))\n        q.put_nowait(1)\n        yield gen.moment\n        self.assertEqual(1, (yield get1))\n\n    @gen_test\n    def test_blocking_put_wait(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        self.io_loop.call_later(0.01, q.get)\n        self.io_loop.call_later(0.02, q.get)\n        futures = [q.put(0), q.put(1)]\n        self.assertFalse(any(f.done() for f in futures))\n        yield futures\n\n    @gen_test\n    def test_put_timeout(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)  # Now it's full.\n        put_timeout = q.put(1, timeout=timedelta(seconds=0.01))\n        put = q.put(2)\n        with self.assertRaises(TimeoutError):\n            yield put_timeout\n        \n        self.assertEqual(0, q.get_nowait())\n        # 1 was never put in the queue.\n        self.assertEqual(2, (yield q.get()))\n\n        # Final get() unblocked this putter.\n        yield put\n\n    @gen_test\n    def test_put_timeout_preempted(self):\n        q = queues.Queue(1)\n        q.put_nowait(0)\n        put = q.put(1, timeout=timedelta(seconds=0.01))\n        q.get()\n        yield gen.sleep(0.02)\n        yield put  # No TimeoutError.\n\n    @gen_test\n    def test_put_clears_timed_out_putters(self):\n        q = queues.Queue(1)\n        # First putter succeeds, remainder block.\n        putters = [q.put(i, timedelta(seconds=0.01)) for i in range(10)]\n        put = q.put(10)\n        self.assertEqual(10, len(q._putters))\n        yield gen.sleep(0.02)\n        self.assertEqual(10, len(q._putters))\n        self.assertFalse(put.done())  # Final waiter is still active.\n        q.put(11)  # put() clears the waiters.\n        self.assertEqual(2, len(q._putters))\n        for putter in putters[1:]:\n            self.assertRaises(TimeoutError, putter.result)\n\n    @gen_test\n    def test_put_clears_timed_out_getters(self):\n        q = queues.Queue()\n        getters = [q.get(timedelta(seconds=0.01)) for _ in range(10)]\n        get = q.get()\n        q.get()\n        self.assertEqual(12, len(q._getters))\n        yield gen.sleep(0.02)\n        self.assertEqual(12, len(q._getters))\n        self.assertFalse(get.done())  # Final waiters still active.\n        q.put(0)  # put() clears the waiters.\n        self.assertEqual(1, len(q._getters))\n        self.assertEqual(0, (yield get))\n        for getter in getters:\n            self.assertRaises(TimeoutError, getter.result)\n\n    @gen_test\n    def test_float_maxsize(self):\n        # Non-int maxsize must round down: http://bugs.python.org/issue21723\n        q = queues.Queue(maxsize=1.3)\n        self.assertTrue(q.empty())\n        self.assertFalse(q.full())\n        q.put_nowait(0)\n        q.put_nowait(1)\n        self.assertFalse(q.empty())\n        self.assertTrue(q.full())\n        self.assertRaises(queues.QueueFull, q.put_nowait, 2)\n        self.assertEqual(0, q.get_nowait())\n        self.assertFalse(q.empty())\n        self.assertFalse(q.full())\n\n        yield q.put(2)\n        put = q.put(3)\n        self.assertFalse(put.done())\n        self.assertEqual(1, (yield q.get()))\n        yield put\n        self.assertTrue(q.full())\n\n\nclass QueueJoinTest(AsyncTestCase):\n    def test_task_done_underflow(self):\n        q = queues.Queue()\n        self.assertRaises(ValueError, q.task_done)\n\n    @gen_test\n    def test_task_done(self):\n        q = queues.Queue()\n        for i in range(100):\n            q.put_nowait(i)\n\n        self.accumulator = 0\n\n        @gen.coroutine\n        def worker():\n            while True:\n                item = yield q.get()\n                self.accumulator += item\n                q.task_done()\n                yield gen.sleep(random() * 0.01)\n\n        # Two coroutines share work.\n        worker()\n        worker()\n        yield q.join()\n        self.assertEqual(sum(range(100)), self.accumulator)\n\n    @gen_test\n    def test_task_done_delay(self):\n        # Verify it is task_done(), not get(), that unblocks join().\n        q = queues.Queue()\n        q.put_nowait(0)\n        join = q.join()\n        self.assertFalse(join.done())\n        yield q.get()\n        self.assertFalse(join.done())\n        yield gen.moment\n        self.assertFalse(join.done())\n        q.task_done()\n        self.assertTrue(join.done())\n\n    @gen_test\n    def test_join_empty_queue(self):\n        q = queues.Queue()\n        yield q.join()\n        yield q.join()\n\n    @gen_test\n    def test_join_timeout(self):\n        q = queues.Queue()\n        q.put(0)\n        with self.assertRaises(TimeoutError):\n            yield q.join(timeout=timedelta(seconds=0.01))\n\n    @gen_test\n    def test_producer_consumer(self):\n        q = queues.Queue(maxsize=3)\n        history = []\n\n        # We don't yield between get() and task_done(), so get() must wait for\n        # the next tick. Otherwise we'd immediately call task_done and unblock\n        # join() before q.put() resumes, and we'd only process the first four\n        # items.\n        @gen.coroutine\n        def consumer():\n            while True:\n                history.append((yield q.get()))\n                q.task_done()\n\n        @gen.coroutine\n        def producer():\n            for item in range(10):\n                yield q.put(item)\n\n        consumer()\n        yield producer()\n        yield q.join()\n        self.assertEqual(list(range(10)), history)\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 824ms\nPrepared 1 package in 1.09s\nInstalled 1 package in 2ms\n + tornado==4.2.dev1 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca)\nResolved 29 packages in 19ms\nInstalled 29 packages in 155ms\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_486abfb8a3b2e40fcfb87c5179252748913fcfca\n............................\ncollected 28 items\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::QueueBasicTest::test_maxsize\nPASSED r2e_tests.test_1::QueueBasicTest::test_order\nPASSED r2e_tests.test_1::QueueBasicTest::test_repr_and_str\nPASSED r2e_tests.test_1::QueueGetTest::test_blocking_get\nPASSED r2e_tests.test_1::QueueGetTest::test_blocking_get_wait\nPASSED r2e_tests.test_1::QueueGetTest::test_get_clears_timed_out_getters\nPASSED r2e_tests.test_1::QueueGetTest::test_get_clears_timed_out_putters\nPASSED r2e_tests.test_1::QueueGetTest::test_get_timeout\nPASSED r2e_tests.test_1::QueueGetTest::test_get_timeout_preempted\nPASSED r2e_tests.test_1::QueueGetTest::test_get_with_putters\nPASSED r2e_tests.test_1::QueueGetTest::test_nonblocking_get\nPASSED r2e_tests.test_1::QueueGetTest::test_nonblocking_get_exception\nPASSED r2e_tests.test_1::QueueJoinTest::test_join_empty_queue\nPASSED r2e_tests.test_1::QueueJoinTest::test_join_timeout\nPASSED r2e_tests.test_1::QueueJoinTest::test_producer_consumer\nPASSED r2e_tests.test_1::QueueJoinTest::test_task_done\nPASSED r2e_tests.test_1::QueueJoinTest::test_task_done_delay\nPASSED r2e_tests.test_1::QueueJoinTest::test_task_done_underflow\nPASSED r2e_tests.test_1::QueuePutTest::test_blocking_put\nPASSED r2e_tests.test_1::QueuePutTest::test_blocking_put_wait\nPASSED r2e_tests.test_1::QueuePutTest::test_float_maxsize\nPASSED r2e_tests.test_1::QueuePutTest::test_nonblocking_put_exception\nPASSED r2e_tests.test_1::QueuePutTest::test_nonblocking_put_with_getters\nPASSED r2e_tests.test_1::QueuePutTest::test_put_clears_timed_out_getters\nPASSED r2e_tests.test_1::QueuePutTest::test_put_clears_timed_out_putters\nPASSED r2e_tests.test_1::QueuePutTest::test_put_timeout\nPASSED r2e_tests.test_1::QueuePutTest::test_put_timeout_preempted\nPASSED r2e_tests.test_1::QueuePutTest::test_put_with_getters\n=================== 28 passed in 0.45s ===================\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_486abfb8a3b2e40fcfb87c5179252748913fcfca\n............................\ncollected 29 items\n\n=================================== ERRORS =====================================\n__________ QueueGetTest.test_get_timeout_preempted __________\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 595, in run\n    self._callTearDown()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 553, in _callTearDown\n    self.tearDown()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/testing.py\", line 234, in tearDown\n    self.__rethrow()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/testing.py\", line 256, in __rethrow\n    raise_exc_info(failure)\n  File \"<string>\", line 3, in raise_exc_info\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/stack_context.py\", line 314, in wrapped\n    ret = fn(*args, **kwargs)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/queues.py\", line 122, in on_timeout\n    future.set_exception(gen.TimeoutError())\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/concurrent.py\", line 250, in set_exception\n    self.set_exc_info(\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/concurrent.py\", line 276, in set_exc_info\n    self._set_done()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/concurrent.py\", line 290, in _set_done\n    for cb in self._callbacks:\nTypeError: 'NoneType' object is not iterable\n__________ QueuePutTest.test_put_timeout_preempted __________\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_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/testing.py\", line 120, in __call__\n    result = self.orig_method(*args, **kwargs)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/testing.py\", line 499, in post_coroutine\n    return self.io_loop.run_sync(\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/ioloop.py\", line 428, in run_sync\n    return future_cell[0].result()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/concurrent.py\", line 211, in result\n    raise_exc_info(self._exc_info)\n  File \"<string>\", line 3, in raise_exc_info\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/gen.py\", line 836, in run\n    yielded = self.gen.throw(*exc_info)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/r2e_tests/test_1.py\", line 226, in test_put_timeout_preempted\n    yield put  # No TimeoutError.\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/gen.py\", line 830, in run\n    value = future.result()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/concurrent.py\", line 211, in result\n    raise_exc_info(self._exc_info)\n  File \"<string>\", line 3, in raise_exc_info\ntornado.gen.TimeoutError\n__________ QueuePutTest.test_put_timeout_preempted __________\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 595, in run\n    self._callTearDown()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 553, in _callTearDown\n    self.tearDown()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/testing.py\", line 234, in tearDown\n    self.__rethrow()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/testing.py\", line 256, in __rethrow\n    raise_exc_info(failure)\n  File \"<string>\", line 3, in raise_exc_info\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/stack_context.py\", line 314, in wrapped\n    ret = fn(*args, **kwargs)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/queues.py\", line 87, in on_timeout\n    future.set_exception(gen.TimeoutError())\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/concurrent.py\", line 250, in set_exception\n    self.set_exc_info(\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/concurrent.py\", line 276, in set_exc_info\n    self._set_done()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_486abfb8a3b2e40fcfb87c5179252748913fcfca/tornado/concurrent.py\", line 290, in _set_done\n    for cb in self._callbacks:\nTypeError: 'NoneType' object is not iterable\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::QueueBasicTest::test_maxsize\nPASSED r2e_tests.test_1::QueueBasicTest::test_order\nPASSED r2e_tests.test_1::QueueBasicTest::test_repr_and_str\nPASSED r2e_tests.test_1::QueueGetTest::test_blocking_get\nPASSED r2e_tests.test_1::QueueGetTest::test_blocking_get_wait\nPASSED r2e_tests.test_1::QueueGetTest::test_get_clears_timed_out_getters\nPASSED r2e_tests.test_1::QueueGetTest::test_get_clears_timed_out_putters\nPASSED r2e_tests.test_1::QueueGetTest::test_get_timeout\nERROR r2e_tests.test_1::QueueGetTest::test_get_timeout_preempted\nPASSED r2e_tests.test_1::QueueGetTest::test_get_with_putters\nPASSED r2e_tests.test_1::QueueGetTest::test_nonblocking_get\nPASSED r2e_tests.test_1::QueueGetTest::test_nonblocking_get_exception\nPASSED r2e_tests.test_1::QueueJoinTest::test_join_empty_queue\nPASSED r2e_tests.test_1::QueueJoinTest::test_join_timeout\nPASSED r2e_tests.test_1::QueueJoinTest::test_producer_consumer\nPASSED r2e_tests.test_1::QueueJoinTest::test_task_done\nPASSED r2e_tests.test_1::QueueJoinTest::test_task_done_delay\nPASSED r2e_tests.test_1::QueueJoinTest::test_task_done_underflow\nPASSED r2e_tests.test_1::QueuePutTest::test_blocking_put\nPASSED r2e_tests.test_1::QueuePutTest::test_blocking_put_wait\nPASSED r2e_tests.test_1::QueuePutTest::test_float_maxsize\nPASSED r2e_tests.test_1::QueuePutTest::test_nonblocking_put_exception\nPASSED r2e_tests.test_1::QueuePutTest::test_nonblocking_put_with_getters\nPASSED r2e_tests.test_1::QueuePutTest::test_put_clears_timed_out_getters\nPASSED r2e_tests.test_1::QueuePutTest::test_put_clears_timed_out_putters\nPASSED r2e_tests.test_1::QueuePutTest::test_put_timeout\nERROR r2e_tests.test_1::QueuePutTest::test_put_timeout_preempted\nERROR r2e_tests.test_1::QueuePutTest::test_put_timeout_preempted\nPASSED r2e_tests.test_1::QueuePutTest::test_put_with_getters\n=================== 3 errors, 26 passed in 0.47s ===================\n",
    "old_commit_res_stderr": ""
}
tornado/queues.py tornado/test/queues_test.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title:** Timeouts Not Properly Canceled in Queue Operations Leading to TypeError

**Description:**
When using `Queue.get` or `Queue.put` with a timeout, if the operation completes before the timeout duration, the timeout is not correctly canceled. This results in a `TypeError` because the internal callbacks are improperly handled.

**Example Buggy Code:**
```python
from tornado.queues import Queue
from tornado import gen
from datetime import timedelta

@gen.coroutine
def example():
    q = Queue()
    # Attempt to get an item with a short timeout
    get_future = q.get(timeout=timedelta(seconds=0.01))
    # Put an item into the queue before the timeout
    q.put_nowait(0)
    yield gen.sleep(0.02)
    result = yield get_future  # This line raises TypeError
```

**Expected Behavior:**
The `get` operation should successfully retrieve the item (`0`) without any errors, as the item was put into the queue before the timeout expired.

**Actual Behavior:**
A `TypeError` is raised with the message `'NoneType' object is not iterable` when the `get` operation attempts to handle the already resolved future. This occurs because the timeout callback was not properly canceled, leaving the future in an inconsistent state.

[/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/queues.py
# Copyright 2015 The Tornado Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

from __future__ import absolute_import, division, print_function, with_statement

__all__ = ['Queue', 'QueueFull', 'QueueEmpty']

import collections

from tornado import gen, ioloop
from tornado.concurrent import Future
from tornado.locks import Event


class QueueEmpty(Exception):
    """Raised by `.Queue.get_nowait` when the queue has no items."""
    pass


class QueueFull(Exception):
    """Raised by `.Queue.put_nowait` when a queue is at its maximum size."""
    pass


class Queue(object):
    """Coordinate producer and consumer coroutines.

    If maxsize is 0 (the default) the queue size is unbounded.
    """
    def __init__(self, maxsize=0):
        if maxsize is None:
            raise TypeError("maxsize can't be None")

        if maxsize < 0:
            raise ValueError("maxsize can't be negative")

        self._maxsize = maxsize
        self._init()
        self._getters = collections.deque([])  # Futures.
        self._putters = collections.deque([])  # Pairs of (item, Future).
        self._unfinished_tasks = 0
        self._finished = Event()
        self._finished.set()

    @property
    def maxsize(self):
        """Number of items allowed in the queue."""
        return self._maxsize

    def qsize(self):
        """Number of items in the queue."""
        return len(self._queue)

    def empty(self):
        return not self._queue

    def full(self):
        if self.maxsize == 0:
            return False
        else:
            return self.qsize() >= self.maxsize

    def put(self, item, timeout=None):
        """Put an item into the queue, perhaps waiting until there is room.

        Returns a Future, which raises `tornado.gen.TimeoutError` after a
        timeout.
        """
        try:
            self.put_nowait(item)
        except QueueFull:
            future = Future()
            self._putters.append((item, future))
            if timeout:
                def on_timeout():
                    future.set_exception(gen.TimeoutError())
                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)
            return future
        else:
            return gen._null_future

    def put_nowait(self, item):
        """Put an item into the queue without blocking.

        If no free slot is immediately available, raise `QueueFull`.
        """
        self._consume_expired()
        if self._getters:
            assert self.empty(), "queue non-empty, why are getters waiting?"
            getter = self._getters.popleft()
            self._put(item)
            getter.set_result(self._get())
        elif self.full():
            raise QueueFull
        else:
            self._put(item)

    def get(self, timeout=None):
        """Remove and return an item from the queue.

        Returns a Future which resolves once an item is available, or raises
        `tornado.gen.TimeoutError` after a timeout.
        """
        future = Future()
        try:
            future.set_result(self.get_nowait())
        except QueueEmpty:
            self._getters.append(future)
            if timeout:
                def on_timeout():
                    future.set_exception(gen.TimeoutError())
                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)
        return future

    def get_nowait(self):
        """Remove and return an item from the queue without blocking.

        Return an item if one is immediately available, else raise
        `QueueEmpty`.
        """
        self._consume_expired()
        if self._putters:
            assert self.full(), "queue not full, why are putters waiting?"
            item, putter = self._putters.popleft()
            self._put(item)
            putter.set_result(None)
            return self._get()
        elif self.qsize():
            return self._get()
        else:
            raise QueueEmpty

    def task_done(self):
        """Indicate that a formerly enqueued task is complete.

        Used by queue consumers. For each `.get` used to fetch a task, a
        subsequent call to `.task_done` tells the queue that the processing
        on the task is complete.

        If a `.join` is blocking, it resumes when all items have been
        processed; that is, when every `.put` is matched by a `.task_done`.

        Raises `ValueError` if called more times than `.put`.
        """
        if self._unfinished_tasks <= 0:
            raise ValueError('task_done() called too many times')
        self._unfinished_tasks -= 1
        if self._unfinished_tasks == 0:
            self._finished.set()

    def join(self, timeout=None):
        """Block until all items in the queue are processed. Returns a Future.

        Returns a Future, which raises `tornado.gen.TimeoutError` after a
        timeout.
        """
        return self._finished.wait(timeout)

    def _init(self):
        self._queue = collections.deque()

    def _get(self):
        return self._queue.popleft()

    def _put(self, item):
        self._unfinished_tasks += 1
        self._finished.clear()
        self._queue.append(item)

    def _consume_expired(self):
        # Remove timed-out waiters.
        while self._putters and self._putters[0][1].done():
            self._putters.popleft()

        while self._getters and self._getters[0].done():
            self._getters.popleft()

    def __repr__(self):
        return '<%s at %s %s>' % (
            type(self).__name__, hex(id(self)), self._format())

    def __str__(self):
        return '<%s %s>' % (type(self).__name__, self._format())

    def _format(self):
        result = 'maxsize=%r' % (self.maxsize, )
        if getattr(self, '_queue', None):
            result += ' queue=%r' % self._queue
        if self._getters:
            result += ' getters[%s]' % len(self._getters)
        if self._putters:
            result += ' putters[%s]' % len(self._putters)
        if self._unfinished_tasks:
            result += ' tasks=%s' % self._unfinished_tasks
        return result

```
--- 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:** Timeouts Not Properly Canceled in Queue Operations Leading to TypeError

**Description:**
When using `Queue.get` or `Queue.put` with a timeout, if the operation completes before the timeout duration, the timeout is not correctly canceled. This results in a `TypeError` because the internal callbacks are improperly handled.

**Example Buggy Code:**
```python
from tornado.queues import Queue
from tornado import gen
from datetime import timedelta

@gen.coroutine
def example():
    q = Queue()
    # Attempt to get an item with a short timeout
    get_future = q.get(timeout=timedelta(seconds=0.01))
    # Put an item into the queue before the timeout
    q.put_nowait(0)
    yield gen.sleep(0.02)
    result = yield get_future  # This line raises TypeError
```

**Expected Behavior:**
The `get` operation should successfully retrieve the item (`0`) without any errors, as the item was put into the queue before the timeout expired.

**Actual Behavior:**
A `TypeError` is raised with the message `'NoneType' object is not iterable` when the `get` operation attempts to handle the already resolved future. This occurs because the timeout callback was not properly canceled, leaving the future in an inconsistent state.

[/ISSUE]
{
    "QueueBasicTest.test_maxsize": "PASSED",
    "QueueBasicTest.test_order": "PASSED",
    "QueueBasicTest.test_repr_and_str": "PASSED",
    "QueueGetTest.test_blocking_get": "PASSED",
    "QueueGetTest.test_blocking_get_wait": "PASSED",
    "QueueGetTest.test_get_clears_timed_out_getters": "PASSED",
    "QueueGetTest.test_get_clears_timed_out_putters": "PASSED",
    "QueueGetTest.test_get_timeout": "PASSED",
    "QueueGetTest.test_get_timeout_preempted": "PASSED",
    "QueueGetTest.test_get_with_putters": "PASSED",
    "QueueGetTest.test_nonblocking_get": "PASSED",
    "QueueGetTest.test_nonblocking_get_exception": "PASSED",
    "QueueJoinTest.test_join_empty_queue": "PASSED",
    "QueueJoinTest.test_join_timeout": "PASSED",
    "QueueJoinTest.test_producer_consumer": "PASSED",
    "QueueJoinTest.test_task_done": "PASSED",
    "QueueJoinTest.test_task_done_delay": "PASSED",
    "QueueJoinTest.test_task_done_underflow": "PASSED",
    "QueuePutTest.test_blocking_put": "PASSED",
    "QueuePutTest.test_blocking_put_wait": "PASSED",
    "QueuePutTest.test_float_maxsize": "PASSED",
    "QueuePutTest.test_nonblocking_put_exception": "PASSED",
    "QueuePutTest.test_nonblocking_put_with_getters": "PASSED",
    "QueuePutTest.test_put_clears_timed_out_getters": "PASSED",
    "QueuePutTest.test_put_clears_timed_out_putters": "PASSED",
    "QueuePutTest.test_put_timeout": "PASSED",
    "QueuePutTest.test_put_timeout_preempted": "PASSED",
    "QueuePutTest.test_put_with_getters": "PASSED"
}
diff --git a/tornado/queues.py b/tornado/queues.py
index 1e4a31ab..5986ccc9 100644
--- a/tornado/queues.py
+++ b/tornado/queues.py
@@ -33,6 +33,16 @@ class QueueFull(Exception):
     pass
 
 
+def _set_timeout(future, timeout):
+    if timeout:
+        def on_timeout():
+            future.set_exception(gen.TimeoutError())
+        io_loop = ioloop.IOLoop.current()
+        timeout_handle = io_loop.add_timeout(timeout, on_timeout)
+        future.add_done_callback(
+            lambda _: io_loop.remove_timeout(timeout_handle))
+
+
 class Queue(object):
     """Coordinate producer and consumer coroutines.
 
@@ -82,10 +92,7 @@ class Queue(object):
         except QueueFull:
             future = Future()
             self._putters.append((item, future))
-            if timeout:
-                def on_timeout():
-                    future.set_exception(gen.TimeoutError())
-                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)
+            _set_timeout(future, timeout)
             return future
         else:
             return gen._null_future
@@ -117,10 +124,7 @@ class Queue(object):
             future.set_result(self.get_nowait())
         except QueueEmpty:
             self._getters.append(future)
-            if timeout:
-                def on_timeout():
-                    future.set_exception(gen.TimeoutError())
-                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)
+            _set_timeout(future, timeout)
         return future
 
     def get_nowait(self):

{"tornado/queues.py": "# Copyright 2015 The Tornado Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\n__all__ = ['Queue', 'QueueFull', 'QueueEmpty']\n\nimport collections\n\nfrom tornado import gen, ioloop\nfrom tornado.concurrent import Future\nfrom tornado.locks import Event\n\n\nclass QueueEmpty(Exception):\n    \"\"\"Raised by `.Queue.get_nowait` when the queue has no items.\"\"\"\n    pass\n\n\nclass QueueFull(Exception):\n    \"\"\"Raised by `.Queue.put_nowait` when a queue is at its maximum size.\"\"\"\n    pass\n\n\nclass Queue(object):\n    \"\"\"Coordinate producer and consumer coroutines.\n\n    If maxsize is 0 (the default) the queue size is unbounded.\n    \"\"\"\n    def __init__(self, maxsize=0):\n        if maxsize is None:\n            raise TypeError(\"maxsize can't be None\")\n\n        if maxsize < 0:\n            raise ValueError(\"maxsize can't be negative\")\n\n        self._maxsize = maxsize\n        self._init()\n        self._getters = collections.deque([])  # Futures.\n        self._putters = collections.deque([])  # Pairs of (item, Future).\n        self._unfinished_tasks = 0\n        self._finished = Event()\n        self._finished.set()\n\n    @property\n    def maxsize(self):\n        \"\"\"Number of items allowed in the queue.\"\"\"\n        return self._maxsize\n\n    def qsize(self):\n        \"\"\"Number of items in the queue.\"\"\"\n        return len(self._queue)\n\n    def empty(self):\n        return not self._queue\n\n    def full(self):\n        if self.maxsize == 0:\n            return False\n        else:\n            return self.qsize() >= self.maxsize\n\n    def put(self, item, timeout=None):\n        \"\"\"Put an item into the queue, perhaps waiting until there is room.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        try:\n            self.put_nowait(item)\n        except QueueFull:\n            future = Future()\n            self._putters.append((item, future))\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n            return future\n        else:\n            return gen._null_future\n\n    def put_nowait(self, item):\n        \"\"\"Put an item into the queue without blocking.\n\n        If no free slot is immediately available, raise `QueueFull`.\n        \"\"\"\n        self._consume_expired()\n        if self._getters:\n            assert self.empty(), \"queue non-empty, why are getters waiting?\"\n            getter = self._getters.popleft()\n            self._put(item)\n            getter.set_result(self._get())\n        elif self.full():\n            raise QueueFull\n        else:\n            self._put(item)\n\n    def get(self, timeout=None):\n        \"\"\"Remove and return an item from the queue.\n\n        Returns a Future which resolves once an item is available, or raises\n        `tornado.gen.TimeoutError` after a timeout.\n        \"\"\"\n        future = Future()\n        try:\n            future.set_result(self.get_nowait())\n        except QueueEmpty:\n            self._getters.append(future)\n            if timeout:\n                def on_timeout():\n                    future.set_exception(gen.TimeoutError())\n                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)\n        return future\n\n    def get_nowait(self):\n        \"\"\"Remove and return an item from the queue without blocking.\n\n        Return an item if one is immediately available, else raise\n        `QueueEmpty`.\n        \"\"\"\n        self._consume_expired()\n        if self._putters:\n            assert self.full(), \"queue not full, why are putters waiting?\"\n            item, putter = self._putters.popleft()\n            self._put(item)\n            putter.set_result(None)\n            return self._get()\n        elif self.qsize():\n            return self._get()\n        else:\n            raise QueueEmpty\n\n    def task_done(self):\n        \"\"\"Indicate that a formerly enqueued task is complete.\n\n        Used by queue consumers. For each `.get` used to fetch a task, a\n        subsequent call to `.task_done` tells the queue that the processing\n        on the task is complete.\n\n        If a `.join` is blocking, it resumes when all items have been\n        processed; that is, when every `.put` is matched by a `.task_done`.\n\n        Raises `ValueError` if called more times than `.put`.\n        \"\"\"\n        if self._unfinished_tasks <= 0:\n            raise ValueError('task_done() called too many times')\n        self._unfinished_tasks -= 1\n        if self._unfinished_tasks == 0:\n            self._finished.set()\n\n    def join(self, timeout=None):\n        \"\"\"Block until all items in the queue are processed. Returns a Future.\n\n        Returns a Future, which raises `tornado.gen.TimeoutError` after a\n        timeout.\n        \"\"\"\n        return self._finished.wait(timeout)\n\n    def _init(self):\n        self._queue = collections.deque()\n\n    def _get(self):\n        return self._queue.popleft()\n\n    def _put(self, item):\n        self._unfinished_tasks += 1\n        self._finished.clear()\n        self._queue.append(item)\n\n    def _consume_expired(self):\n        # Remove timed-out waiters.\n        while self._putters and self._putters[0][1].done():\n            self._putters.popleft()\n\n        while self._getters and self._getters[0].done():\n            self._getters.popleft()\n\n    def __repr__(self):\n        return '<%s at %s %s>' % (\n            type(self).__name__, hex(id(self)), self._format())\n\n    def __str__(self):\n        return '<%s %s>' % (type(self).__name__, self._format())\n\n    def _format(self):\n        result = 'maxsize=%r' % (self.maxsize, )\n        if getattr(self, '_queue', None):\n            result += ' queue=%r' % self._queue\n        if self._getters:\n            result += ' getters[%s]' % len(self._getters)\n        if self._putters:\n            result += ' putters[%s]' % len(self._putters)\n        if self._unfinished_tasks:\n            result += ' tasks=%s' % self._unfinished_tasks\n        return result\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:** Timeouts Not Properly Canceled in Queue Operations Leading to TypeError

**Description:**
When using `Queue.get` or `Queue.put` with a timeout, if the operation completes before the timeout duration, the timeout is not correctly canceled. This results in a `TypeError` because the internal callbacks are improperly handled.

**Example Buggy Code:**
```python
from tornado.queues import Queue
from tornado import gen
from datetime import timedelta

@gen.coroutine
def example():
    q = Queue()
    # Attempt to get an item with a short timeout
    get_future = q.get(timeout=timedelta(seconds=0.01))
    # Put an item into the queue before the timeout
    q.put_nowait(0)
    yield gen.sleep(0.02)
    result = yield get_future  # This line raises TypeError
```

**Expected Behavior:**
The `get` operation should successfully retrieve the item (`0`) without any errors, as the item was put into the queue before the timeout expired.

**Actual Behavior:**
A `TypeError` is raised with the message `'NoneType' object is not iterable` when the `get` operation attempts to handle the already resolved future. This occurs because the timeout callback was not properly canceled, leaving the future in an inconsistent state.

[/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/queues.py
# Copyright 2015 The Tornado Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

from __future__ import absolute_import, division, print_function, with_statement

__all__ = ['Queue', 'QueueFull', 'QueueEmpty']

import collections

from tornado import gen, ioloop
from tornado.concurrent import Future
from tornado.locks import Event


class QueueEmpty(Exception):
    """Raised by `.Queue.get_nowait` when the queue has no items."""
    pass


class QueueFull(Exception):
    """Raised by `.Queue.put_nowait` when a queue is at its maximum size."""
    pass


class Queue(object):
    """Coordinate producer and consumer coroutines.

    If maxsize is 0 (the default) the queue size is unbounded.
    """
    def __init__(self, maxsize=0):
        if maxsize is None:
            raise TypeError("maxsize can't be None")

        if maxsize < 0:
            raise ValueError("maxsize can't be negative")

        self._maxsize = maxsize
        self._init()
        self._getters = collections.deque([])  # Futures.
        self._putters = collections.deque([])  # Pairs of (item, Future).
        self._unfinished_tasks = 0
        self._finished = Event()
        self._finished.set()

    @property
    def maxsize(self):
        """Number of items allowed in the queue."""
        return self._maxsize

    def qsize(self):
        """Number of items in the queue."""
        return len(self._queue)

    def empty(self):
        return not self._queue

    def full(self):
        if self.maxsize == 0:
            return False
        else:
            return self.qsize() >= self.maxsize

    def put(self, item, timeout=None):
        """Put an item into the queue, perhaps waiting until there is room.

        Returns a Future, which raises `tornado.gen.TimeoutError` after a
        timeout.
        """
        try:
            self.put_nowait(item)
        except QueueFull:
            future = Future()
            self._putters.append((item, future))
            if timeout:
                def on_timeout():
                    future.set_exception(gen.TimeoutError())
                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)
            return future
        else:
            return gen._null_future

    def put_nowait(self, item):
        """Put an item into the queue without blocking.

        If no free slot is immediately available, raise `QueueFull`.
        """
        self._consume_expired()
        if self._getters:
            assert self.empty(), "queue non-empty, why are getters waiting?"
            getter = self._getters.popleft()
            self._put(item)
            getter.set_result(self._get())
        elif self.full():
            raise QueueFull
        else:
            self._put(item)

    def get(self, timeout=None):
        """Remove and return an item from the queue.

        Returns a Future which resolves once an item is available, or raises
        `tornado.gen.TimeoutError` after a timeout.
        """
        future = Future()
        try:
            future.set_result(self.get_nowait())
        except QueueEmpty:
            self._getters.append(future)
            if timeout:
                def on_timeout():
                    future.set_exception(gen.TimeoutError())
                ioloop.IOLoop.current().add_timeout(timeout, on_timeout)
        return future

    def get_nowait(self):
        """Remove and return an item from the queue without blocking.

        Return an item if one is immediately available, else raise
        `QueueEmpty`.
        """
        self._consume_expired()
        if self._putters:
            assert self.full(), "queue not full, why are putters waiting?"
            item, putter = self._putters.popleft()
            self._put(item)
            putter.set_result(None)
            return self._get()
        elif self.qsize():
            return self._get()
        else:
            raise QueueEmpty

    def task_done(self):
        """Indicate that a formerly enqueued task is complete.

        Used by queue consumers. For each `.get` used to fetch a task, a
        subsequent call to `.task_done` tells the queue that the processing
        on the task is complete.

        If a `.join` is blocking, it resumes when all items have been
        processed; that is, when every `.put` is matched by a `.task_done`.

        Raises `ValueError` if called more times than `.put`.
        """
        if self._unfinished_tasks <= 0:
            raise ValueError('task_done() called too many times')
        self._unfinished_tasks -= 1
        if self._unfinished_tasks == 0:
            self._finished.set()

    def join(self, timeout=None):
        """Block until all items in the queue are processed. Returns a Future.

        Returns a Future, which raises `tornado.gen.TimeoutError` after a
        timeout.
        """
        return self._finished.wait(timeout)

    def _init(self):
        self._queue = collections.deque()

    def _get(self):
        return self._queue.popleft()

    def _put(self, item):
        self._unfinished_tasks += 1
        self._finished.clear()
        self._queue.append(item)

    def _consume_expired(self):
        # Remove timed-out waiters.
        while self._putters and self._putters[0][1].done():
            self._putters.popleft()

        while self._getters and self._getters[0].done():
            self._getters.popleft()

    def __repr__(self):
        return '<%s at %s %s>' % (
            type(self).__name__, hex(id(self)), self._format())

    def __str__(self):
        return '<%s %s>' % (type(self).__name__, self._format())

    def _format(self):
        result = 'maxsize=%r' % (self.maxsize, )
        if getattr(self, '_queue', None):
            result += ' queue=%r' % self._queue
        if self._getters:
            result += ' getters[%s]' % len(self._getters)
        if self._putters:
            result += ' putters[%s]' % len(self._putters)
        if self._unfinished_tasks:
            result += ' tasks=%s' % self._unfinished_tasks
        return result

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