| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| Expand Up | @@ -296,6 +296,10 @@ asyncio | |
| with the tasks being completed. | ||
| (Contributed by Justin Arthur in :gh:`77714`.) | ||
|
|
||
| * Add :meth:`asyncio.Queue.shutdown` (along with | ||
| :exc:`asyncio.QueueShutDown`) for queue termination. | ||
| (Contributed by Laurie Opperman in :gh:`104228`.) | ||
|
|
||
|
Comment thread
Copy link
Copy Markdown
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityWe've worked well together, I'd be nice if you'd mention me.
Sorry, something went wrong.
All reactions
|
||
| base64 | ||
| ------ | ||
|
|
||
| Expand Down | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,11 @@ | ||
| __all__ = ('Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty') | ||
| __all__ = ( | ||
| 'Queue', | ||
| 'PriorityQueue', | ||
| 'LifoQueue', | ||
| 'QueueFull', | ||
| 'QueueEmpty', | ||
| 'QueueShutDown', | ||
| ) | ||
|
|
||
| import collections | ||
| import heapq | ||
| Expand All | @@ -18,6 +25,11 @@ class QueueFull(Exception): | |
| pass | ||
|
|
||
|
|
||
| class QueueShutDown(Exception): | ||
| """Raised when putting on to or getting from a shut-down Queue.""" | ||
| pass | ||
|
|
||
|
|
||
| class Queue(mixins._LoopBoundMixin): | ||
| """A queue, useful for coordinating producer and consumer coroutines. | ||
|
|
||
| Expand All | @@ -41,6 +53,7 @@ def __init__(self, maxsize=0): | |
| self._finished = locks.Event() | ||
| self._finished.set() | ||
| self._init(maxsize) | ||
| self._is_shutdown = False | ||
|
|
||
| # These three are overridable in subclasses. | ||
|
|
||
| Expand Down Expand Up | @@ -81,6 +94,8 @@ def _format(self): | |
| result += f' _putters[{len(self._putters)}]' | ||
| if self._unfinished_tasks: | ||
| result += f' tasks={self._unfinished_tasks}' | ||
| if self._is_shutdown: | ||
| result += ' shutdown' | ||
| return result | ||
|
|
||
| def qsize(self): | ||
| Expand Down Expand Up | @@ -112,8 +127,12 @@ async def put(self, item): | |
|
|
||
| Put an item into the queue. If the queue is full, wait until a free | ||
| slot is available before adding item. | ||
|
|
||
| Raises QueueShutDown if the queue has been shut down. | ||
| """ | ||
| while self.full(): | ||
| if self._is_shutdown: | ||
| raise QueueShutDown | ||
| putter = self._get_loop().create_future() | ||
| self._putters.append(putter) | ||
| try: | ||
| Expand All | @@ -125,7 +144,7 @@ async def put(self, item): | |
| self._putters.remove(putter) | ||
| except ValueError: | ||
| # The putter could be removed from self._putters by a | ||
| # previous get_nowait call. | ||
| # previous get_nowait call or a shutdown call. | ||
| pass | ||
| if not self.full() and not putter.cancelled(): | ||
| # We were woken up by get_nowait(), but can't take | ||
| Expand All | @@ -138,7 +157,11 @@ def put_nowait(self, item): | |
| """Put an item into the queue without blocking. | ||
|
|
||
| If no free slot is immediately available, raise QueueFull. | ||
|
|
||
| Raises QueueShutDown if the queue has been shut down. | ||
| """ | ||
| if self._is_shutdown: | ||
| raise QueueShutDown | ||
| if self.full(): | ||
| raise QueueFull | ||
| self._put(item) | ||
| Expand All | @@ -150,8 +173,13 @@ async def get(self): | |
| """Remove and return an item from the queue. | ||
|
|
||
| If queue is empty, wait until an item is available. | ||
|
|
||
| Raises QueueShutDown if the queue has been shut down and is empty, or | ||
| if the queue has been shut down immediately. | ||
| """ | ||
| while self.empty(): | ||
| if self._is_shutdown and self.empty(): | ||
| raise QueueShutDown | ||
| getter = self._get_loop().create_future() | ||
| self._getters.append(getter) | ||
| try: | ||
| Expand All | @@ -163,7 +191,7 @@ async def get(self): | |
| self._getters.remove(getter) | ||
| except ValueError: | ||
| # The getter could be removed from self._getters by a | ||
| # previous put_nowait call. | ||
| # previous put_nowait call, or a shutdown call. | ||
| pass | ||
| if not self.empty() and not getter.cancelled(): | ||
| # We were woken up by put_nowait(), but can't take | ||
| Expand All | @@ -176,8 +204,13 @@ def get_nowait(self): | |
| """Remove and return an item from the queue. | ||
|
|
||
| Return an item if one is immediately available, else raise QueueEmpty. | ||
|
|
||
| Raises QueueShutDown if the queue has been shut down and is empty, or | ||
| if the queue has been shut down immediately. | ||
| """ | ||
| if self.empty(): | ||
|
Comment thread
EpicWink marked this conversation as resolved.
|
||
| if self._is_shutdown: | ||
| raise QueueShutDown | ||
| raise QueueEmpty | ||
| item = self._get() | ||
| self._wakeup_next(self._putters) | ||
| Expand All | @@ -194,6 +227,9 @@ def task_done(self): | |
| been processed (meaning that a task_done() call was received for every | ||
| item that had been put() into the queue). | ||
|
|
||
| shutdown(immediate=True) calls task_done() for each remaining item in | ||
| the queue. | ||
|
|
||
| Raises ValueError if called more times than there were items placed in | ||
| the queue. | ||
| """ | ||
| Expand All | @@ -214,6 +250,32 @@ async def join(self): | |
| if self._unfinished_tasks > 0: | ||
| await self._finished.wait() | ||
|
|
||
| def shutdown(self, immediate=False): | ||
| """Shut-down the queue, making queue gets and puts raise QueueShutDown. | ||
|
|
||
| By default, gets will only raise once the queue is empty. Set | ||
| 'immediate' to True to make gets raise immediately instead. | ||
|
|
||
| All blocked callers of put() will be unblocked, and also get() | ||
| and join() if 'immediate'. | ||
| """ | ||
|
Comment thread
Copy link
Copy Markdown
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityDocstring to modify depending of agree/disagree on my first remark about `blocked callers'. All blocked callers of put() and get() will be unblocked, and also join() if 'immediate
Sorry, something went wrong.
All reactions
|
||
| self._is_shutdown = True | ||
| if immediate: | ||
| while not self.empty(): | ||
| self._get() | ||
| if self._unfinished_tasks > 0: | ||
| self._unfinished_tasks -= 1 | ||
| if self._unfinished_tasks == 0: | ||
| self._finished.set() | ||
| while self._getters: | ||
| getter = self._getters.popleft() | ||
| if not getter.done(): | ||
| getter.set_result(None) | ||
| while self._putters: | ||
| putter = self._putters.popleft() | ||
| if not putter.done(): | ||
| putter.set_result(None) | ||
|
|
||
|
|
||
| class PriorityQueue(Queue): | ||
| """A subclass of Queue; retrieves entries in priority order (lowest first). | ||
| Expand Down | ||
| Back | FazBrowse Home | New Git URL |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualitySorry but I have a doubt, shouldn't this documentation block be rather:
All blocked callers of :meth:`~Queue.put` and :meth:`~Queue.get` will be unblocked. If *immediate* is true, also unblock callers of :meth:`~Queue.join`.In event of change, the docstring of the shutdown method must be updated.
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Qualityjoin callers aren't necessarily even unblocked anyway, if consumers are processing any items. I should probably say that a task is marked as done for each item in the queue if immediate shutdown.
Also, I think the threading queue docs are the same.
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityIt's very precise, better.
Yes, I commented here so as not to forget (see #117532 (comment)).
English is your native language, I think it'is best if you update documentations and docstrings.
Update: but I can create the follow-up PR.
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityI've made a PR: #117621
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.