Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork 35.2k
bpo-34037: Fix test_asyncio failure and add loop.shutdown_default_executor()#15735
New issue
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
Uh oh!
There was an error while loading. Please reload this page.
Changes from all commits
6ed94658608e977f39c4a09a9458c12cf4fafa09a53303753b877e761a29d836bde87c1e4a8588a4a285f6280f769c7244760892093ddce2895cad094e7af47e4980c626117d5340a5c260caced2bc6d088205847fe0fb1dbc08aefbae3ecFile filter
Filter by extension
Conversations
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -406,6 +406,8 @@ def __init__(self): | ||
| self._asyncgens = weakref.WeakSet() | ||
| # Set to True when `loop.shutdown_asyncgens` is called. | ||
| self._asyncgens_shutdown_called = False | ||
| # Set to True when `loop.shutdown_default_executor` is called. | ||
| self._executor_shutdown_called = False | ||
asvetlov marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| def __repr__(self): | ||
| return ( | ||
| @@ -503,6 +505,10 @@ def _check_closed(self): | ||
| if self._closed: | ||
| raise RuntimeError('Event loop is closed') | ||
| def _check_default_executor(self): | ||
| if self._executor_shutdown_called: | ||
| raise RuntimeError('Executor shutdown has been called') | ||
| def _asyncgen_finalizer_hook(self, agen): | ||
| self._asyncgens.discard(agen) | ||
| if not self.is_closed(): | ||
| @@ -543,6 +549,26 @@ async def shutdown_asyncgens(self): | ||
| 'asyncgen': agen | ||
| }) | ||
| async def shutdown_default_executor(self): | ||
| """Schedule the shutdown of the default executor.""" | ||
| self._executor_shutdown_called = True | ||
| if self._default_executor is None: | ||
asvetlov marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| return | ||
| future = self.create_future() | ||
| thread = threading.Thread(target=self._do_shutdown, args=(future,)) | ||
| thread.start() | ||
| try: | ||
| await future | ||
| finally: | ||
| thread.join() | ||
| def _do_shutdown(self, future): | ||
| try: | ||
| self._default_executor.shutdown(wait=True) | ||
| self.call_soon_threadsafe(future.set_result, None) | ||
| except Exception as ex: | ||
| self.call_soon_threadsafe(future.set_exception, ex) | ||
| def run_forever(self): | ||
| """Run until stop() is called.""" | ||
| self._check_closed() | ||
| @@ -632,6 +658,7 @@ def close(self): | ||
| self._closed = True | ||
| self._ready.clear() | ||
| self._scheduled.clear() | ||
| self._executor_shutdown_called = True | ||
| executor = self._default_executor | ||
| if executor is not None: | ||
| self._default_executor = None | ||
| @@ -768,6 +795,8 @@ def run_in_executor(self, executor, func, *args): | ||
| self._check_callback(func, 'run_in_executor') | ||
| if executor is None: | ||
| executor = self._default_executor | ||
| # Only check when the default executor is being used | ||
| self._check_default_executor() | ||
asvetlov marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| if executor is None: | ||
| executor = concurrent.futures.ThreadPoolExecutor() | ||
| self._default_executor = executor | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -249,6 +249,10 @@ async def shutdown_asyncgens(self): | ||
| """Shutdown all active asynchronous generators.""" | ||
| raise NotImplementedError | ||
| async def shutdown_default_executor(self): | ||
| """Schedule the shutdown of the default executor.""" | ||
aeros marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| raise NotImplementedError | ||
| # Methods scheduling callbacks. All these return Handles. | ||
| def _timer_handle_cancelled(self, handle): | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -45,6 +45,7 @@ async def main(): | ||
| try: | ||
| _cancel_all_tasks(loop) | ||
| loop.run_until_complete(loop.shutdown_asyncgens()) | ||
| loop.run_until_complete(loop.shutdown_default_executor()) | ||
asvetlov marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| finally: | ||
| events.set_event_loop(None) | ||
| loop.close() | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| For :mod:`asyncio`, add a new coroutine :meth:`loop.shutdown_default_executor`. | ||
| The new coroutine provides an API to schedule an executor shutdown that waits | ||
| on the threadpool to finish closing. Also, :func:`asyncio.run` has been updated | ||
| to utilize the new coroutine. Patch by Kyle Stanley. |
Uh oh!
There was an error while loading. Please reload this page.