bpo-32311: Implement asyncio.create_task() shortcut (#4848)

* Implement functionality
* Add documentation
This commit is contained in:
Andrew Svetlov 2017-12-15 07:04:38 +02:00 committed by GitHub
parent 19a44f63c7
commit f74ef458ab
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 201 additions and 95 deletions

View file

@ -3,7 +3,7 @@ __all__ = ()
import concurrent.futures._base
import reprlib
from . import events
from . import format_helpers
Error = concurrent.futures._base.Error
CancelledError = concurrent.futures.CancelledError
@ -38,7 +38,7 @@ def _format_callbacks(cb):
cb = ''
def format_cb(callback):
return events._format_callback_source(callback, ())
return format_helpers._format_callback_source(callback, ())
if size == 1:
cb = format_cb(cb[0])

View file

@ -6,5 +6,5 @@ ACCEPT_RETRY_DELAY = 1
# Number of stack entries to capture in debug mode.
# The larger the number, the slower the operation in debug mode
# (see extract_stack() in events.py).
# (see extract_stack() in format_helpers.py).
DEBUG_STACK_DEPTH = 10

View file

@ -9,9 +9,9 @@ import types
from collections.abc import Awaitable, Coroutine
from . import constants
from . import events
from . import base_futures
from . import constants
from . import format_helpers
from .log import logger
@ -48,7 +48,7 @@ class CoroWrapper:
assert inspect.isgenerator(gen) or inspect.iscoroutine(gen), gen
self.gen = gen
self.func = func # Used to unwrap @coroutine decorator
self._source_traceback = events.extract_stack(sys._getframe(1))
self._source_traceback = format_helpers.extract_stack(sys._getframe(1))
self.__name__ = getattr(gen, '__name__', None)
self.__qualname__ = getattr(gen, '__qualname__', None)
@ -243,7 +243,7 @@ def _format_coroutine(coro):
func = coro
if coro_name is None:
coro_name = events._format_callback(func, (), {})
coro_name = format_helpers._format_callback(func, (), {})
try:
coro_code = coro.gi_code
@ -260,7 +260,7 @@ def _format_coroutine(coro):
if (isinstance(coro, CoroWrapper) and
not inspect.isgeneratorfunction(coro.func) and
coro.func is not None):
source = events._get_function_source(coro.func)
source = format_helpers._get_function_source(coro.func)
if source is not None:
filename, lineno = source
if coro_frame is None:

View file

@ -11,86 +11,14 @@ __all__ = (
'_get_running_loop',
)
import functools
import inspect
import os
import reprlib
import socket
import subprocess
import sys
import threading
import traceback
from . import constants
def _get_function_source(func):
func = inspect.unwrap(func)
if inspect.isfunction(func):
code = func.__code__
return (code.co_filename, code.co_firstlineno)
if isinstance(func, functools.partial):
return _get_function_source(func.func)
if isinstance(func, functools.partialmethod):
return _get_function_source(func.func)
return None
def _format_args_and_kwargs(args, kwargs):
"""Format function arguments and keyword arguments.
Special case for a single parameter: ('hello',) is formatted as ('hello').
"""
# use reprlib to limit the length of the output
items = []
if args:
items.extend(reprlib.repr(arg) for arg in args)
if kwargs:
items.extend(f'{k}={reprlib.repr(v)}' for k, v in kwargs.items())
return '({})'.format(', '.join(items))
def _format_callback(func, args, kwargs, suffix=''):
if isinstance(func, functools.partial):
suffix = _format_args_and_kwargs(args, kwargs) + suffix
return _format_callback(func.func, func.args, func.keywords, suffix)
if hasattr(func, '__qualname__'):
func_repr = getattr(func, '__qualname__')
elif hasattr(func, '__name__'):
func_repr = getattr(func, '__name__')
else:
func_repr = repr(func)
func_repr += _format_args_and_kwargs(args, kwargs)
if suffix:
func_repr += suffix
return func_repr
def _format_callback_source(func, args):
func_repr = _format_callback(func, args, None)
source = _get_function_source(func)
if source:
func_repr += f' at {source[0]}:{source[1]}'
return func_repr
def extract_stack(f=None, limit=None):
"""Replacement for traceback.extract_stack() that only does the
necessary work for asyncio debug mode.
"""
if f is None:
f = sys._getframe().f_back
if limit is None:
# Limit the amount of work to a reasonable amount, as extract_stack()
# can be called for each coroutine and future in debug mode.
limit = constants.DEBUG_STACK_DEPTH
stack = traceback.StackSummary.extract(traceback.walk_stack(f),
limit=limit,
lookup_lines=False)
stack.reverse()
return stack
from . import format_helpers
class Handle:
@ -106,7 +34,8 @@ class Handle:
self._cancelled = False
self._repr = None
if self._loop.get_debug():
self._source_traceback = extract_stack(sys._getframe(1))
self._source_traceback = format_helpers.extract_stack(
sys._getframe(1))
else:
self._source_traceback = None
@ -115,7 +44,8 @@ class Handle:
if self._cancelled:
info.append('cancelled')
if self._callback is not None:
info.append(_format_callback_source(self._callback, self._args))
info.append(format_helpers._format_callback_source(
self._callback, self._args))
if self._source_traceback:
frame = self._source_traceback[-1]
info.append(f'created at {frame[0]}:{frame[1]}')
@ -145,7 +75,8 @@ class Handle:
try:
self._callback(*self._args)
except Exception as exc:
cb = _format_callback_source(self._callback, self._args)
cb = format_helpers._format_callback_source(
self._callback, self._args)
msg = f'Exception in callback {cb}'
context = {
'message': msg,

View file

@ -0,0 +1,75 @@
import functools
import inspect
import reprlib
import traceback
from . import constants
def _get_function_source(func):
func = inspect.unwrap(func)
if inspect.isfunction(func):
code = func.__code__
return (code.co_filename, code.co_firstlineno)
if isinstance(func, functools.partial):
return _get_function_source(func.func)
if isinstance(func, functools.partialmethod):
return _get_function_source(func.func)
return None
def _format_callback_source(func, args):
func_repr = _format_callback(func, args, None)
source = _get_function_source(func)
if source:
func_repr += f' at {source[0]}:{source[1]}'
return func_repr
def _format_args_and_kwargs(args, kwargs):
"""Format function arguments and keyword arguments.
Special case for a single parameter: ('hello',) is formatted as ('hello').
"""
# use reprlib to limit the length of the output
items = []
if args:
items.extend(reprlib.repr(arg) for arg in args)
if kwargs:
items.extend(f'{k}={reprlib.repr(v)}' for k, v in kwargs.items())
return '({})'.format(', '.join(items))
def _format_callback(func, args, kwargs, suffix=''):
if isinstance(func, functools.partial):
suffix = _format_args_and_kwargs(args, kwargs) + suffix
return _format_callback(func.func, func.args, func.keywords, suffix)
if hasattr(func, '__qualname__'):
func_repr = getattr(func, '__qualname__')
elif hasattr(func, '__name__'):
func_repr = getattr(func, '__name__')
else:
func_repr = repr(func)
func_repr += _format_args_and_kwargs(args, kwargs)
if suffix:
func_repr += suffix
return func_repr
def extract_stack(f=None, limit=None):
"""Replacement for traceback.extract_stack() that only does the
necessary work for asyncio debug mode.
"""
if f is None:
f = sys._getframe().f_back
if limit is None:
# Limit the amount of work to a reasonable amount, as extract_stack()
# can be called for each coroutine and future in debug mode.
limit = constants.DEBUG_STACK_DEPTH
stack = traceback.StackSummary.extract(traceback.walk_stack(f),
limit=limit,
lookup_lines=False)
stack.reverse()
return stack

View file

@ -11,6 +11,7 @@ import sys
from . import base_futures
from . import events
from . import format_helpers
CancelledError = base_futures.CancelledError
@ -79,7 +80,8 @@ class Future:
self._loop = loop
self._callbacks = []
if self._loop.get_debug():
self._source_traceback = events.extract_stack(sys._getframe(1))
self._source_traceback = format_helpers.extract_stack(
sys._getframe(1))
_repr_info = base_futures._future_repr_info

View file

@ -1,7 +1,7 @@
"""Support for tasks, coroutines and the scheduler."""
__all__ = (
'Task',
'Task', 'create_task',
'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED',
'wait', 'wait_for', 'as_completed', 'sleep',
'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe',
@ -67,13 +67,19 @@ class Task(futures.Future):
return {t for t in cls._all_tasks if t._loop is loop}
def __init__(self, coro, *, loop=None):
assert coroutines.iscoroutine(coro), repr(coro)
super().__init__(loop=loop)
if self._source_traceback:
del self._source_traceback[-1]
self._coro = coro
self._fut_waiter = None
if not coroutines.iscoroutine(coro):
# raise after Future.__init__(), attrs are required for __del__
# prevent logging for pending task in __del__
self._log_destroy_pending = False
raise TypeError(f"a coroutine was expected, got {coro!r}")
self._must_cancel = False
self._fut_waiter = None
self._coro = coro
self._loop.call_soon(self._step)
self.__class__._all_tasks.add(self)
@ -263,6 +269,15 @@ else:
Task = _CTask = _asyncio.Task
def create_task(coro):
"""Schedule the execution of a coroutine object in a spawn task.
Return a Task object.
"""
loop = events.get_running_loop()
return loop.create_task(coro)
# wait() and as_completed() similar to those in PEP 3148.
FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED