FM / asyncio

/usr/lib64/python3.6/asyncio

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
base_events.py	57346	0644	edit dl rm
base_futures.py	2074	0644	edit dl rm
base_subprocess.py	9096	0644	edit dl rm
base_tasks.py	2186	0644	edit dl rm
compat.py	543	0644	edit dl rm
constants.py	371	0644	edit dl rm
coroutines.py	11135	0644	edit dl rm
events.py	23511	0644	edit dl rm
futures.py	15901	0644	edit dl rm
locks.py	15582	0644	edit dl rm
log.py	124	0644	edit dl rm
proactor_events.py	20403	0644	edit dl rm
protocols.py	4512	0644	edit dl rm
queues.py	7957	0644	edit dl rm
selector_events.py	41640	0644	edit dl rm
sslproto.py	26005	0644	edit dl rm
streams.py	24472	0644	edit dl rm
subprocess.py	7626	0644	edit dl rm
tasks.py	25060	0644	edit dl rm
test_utils.py	15091	0644	edit dl rm
transports.py	10066	0644	edit dl rm
unix_events.py	37268	0644	edit dl rm
windows_events.py	27831	0644	edit dl rm
windows_utils.py	6883	0644	edit dl rm
__init__.py	1436	0644	edit dl rm

Edit: /usr/lib64/python3.6/asyncio/coroutines.py (11135B)

__all__ = ['coroutine',
           'iscoroutinefunction', 'iscoroutine']

import functools
import inspect
import opcode
import os
import sys
import traceback
import types

from . import compat
from . import constants
from . import events
from . import base_futures
from .log import logger

# Opcode of "yield from" instruction
_YIELD_FROM = opcode.opmap['YIELD_FROM']

# If you set _DEBUG to true, @coroutine will wrap the resulting
# generator objects in a CoroWrapper instance (defined below).  That
# instance will log a message when the generator is never iterated
# over, which may happen when you forget to use "yield from" with a
# coroutine call.  Note that the value of the _DEBUG flag is taken
# when the decorator is used, so to be of any use it must be set
# before you define your coroutines.  A downside of using this feature
# is that tracebacks show entries for the CoroWrapper.__next__ method
# when _DEBUG is true.
_DEBUG = (not sys.flags.ignore_environment and
          bool(os.environ.get('PYTHONASYNCIODEBUG')))

try:
    _types_coroutine = types.coroutine
    _types_CoroutineType = types.CoroutineType
except AttributeError:
    # Python 3.4
    _types_coroutine = None
    _types_CoroutineType = None

try:
    _inspect_iscoroutinefunction = inspect.iscoroutinefunction
except AttributeError:
    # Python 3.4
    _inspect_iscoroutinefunction = lambda func: False

try:
    from collections.abc import Coroutine as _CoroutineABC, \
                                Awaitable as _AwaitableABC
except ImportError:
    _CoroutineABC = _AwaitableABC = None

# Check for CPython issue #21209
def has_yield_from_bug():
    class MyGen:
        def __init__(self):
            self.send_args = None
        def __iter__(self):
            return self
        def __next__(self):
            return 42
        def send(self, *what):
            self.send_args = what
            return None
    def yield_from_gen(gen):
        yield from gen
    value = (1, 2, 3)
    gen = MyGen()
    coro = yield_from_gen(gen)
    next(coro)
    coro.send(value)
    return gen.send_args != (value,)
_YIELD_FROM_BUG = has_yield_from_bug()
del has_yield_from_bug

def debug_wrapper(gen):
    # This function is called from 'sys.set_coroutine_wrapper'.
    # We only wrap here coroutines defined via 'async def' syntax.
    # Generator-based coroutines are wrapped in @coroutine
    # decorator.
    return CoroWrapper(gen, None)

class CoroWrapper:
    # Wrapper for coroutine object in _DEBUG mode.

def __init__(self, gen, func=None):
        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.__name__ = getattr(gen, '__name__', None)
        self.__qualname__ = getattr(gen, '__qualname__', None)

def __repr__(self):
        coro_repr = _format_coroutine(self)
        if self._source_traceback:
            frame = self._source_traceback[-1]
            coro_repr += ', created at %s:%s' % (frame[0], frame[1])
        return '<%s %s>' % (self.__class__.__name__, coro_repr)

def __iter__(self):
        return self

def __next__(self):
        return self.gen.send(None)

if _YIELD_FROM_BUG:
        # For for CPython issue #21209: using "yield from" and a custom
        # generator, generator.send(tuple) unpacks the tuple instead of passing
        # the tuple unchanged. Check if the caller is a generator using "yield
        # from" to decide if the parameter should be unpacked or not.
        def send(self, *value):
            frame = sys._getframe()
            caller = frame.f_back
            assert caller.f_lasti >= 0
            if caller.f_code.co_code[caller.f_lasti] != _YIELD_FROM:
                value = value[0]
            return self.gen.send(value)
    else:
        def send(self, value):
            return self.gen.send(value)

def throw(self, type, value=None, traceback=None):
        return self.gen.throw(type, value, traceback)

def close(self):
        return self.gen.close()

@property
    def gi_frame(self):
        return self.gen.gi_frame

@property
    def gi_running(self):
        return self.gen.gi_running

@property
    def gi_code(self):
        return self.gen.gi_code

if compat.PY35:

def __await__(self):
            cr_await = getattr(self.gen, 'cr_await', None)
            if cr_await is not None:
                raise RuntimeError(
                    "Cannot await on coroutine {!r} while it's "
                    "awaiting for {!r}".format(self.gen, cr_await))
            return self

@property
        def gi_yieldfrom(self):
            return self.gen.gi_yieldfrom

@property
        def cr_await(self):
            return self.gen.cr_await

@property
        def cr_running(self):
            return self.gen.cr_running

@property
        def cr_code(self):
            return self.gen.cr_code

@property
        def cr_frame(self):
            return self.gen.cr_frame

def __del__(self):
        # Be careful accessing self.gen.frame -- self.gen might not exist.
        gen = getattr(self, 'gen', None)
        frame = getattr(gen, 'gi_frame', None)
        if frame is None:
            frame = getattr(gen, 'cr_frame', None)
        if frame is not None and frame.f_lasti == -1:
            msg = '%r was never yielded from' % self
            tb = getattr(self, '_source_traceback', ())
            if tb:
                tb = ''.join(traceback.format_list(tb))
                msg += (f'\nCoroutine object created at '
                        f'(most recent call last, truncated to '
                        f'{constants.DEBUG_STACK_DEPTH} last lines):\n')
                msg += tb.rstrip()
            logger.error(msg)

def coroutine(func):
    """Decorator to mark coroutines.

If the coroutine is not yielded from before it is destroyed,
    an error message is logged.
    """
    if _inspect_iscoroutinefunction(func):
        # In Python 3.5 that's all we need to do for coroutines
        # defined with "async def".
        # Wrapping in CoroWrapper will happen via
        # 'sys.set_coroutine_wrapper' function.
        return func

if inspect.isgeneratorfunction(func):
        coro = func
    else:
        @functools.wraps(func)
        def coro(*args, **kw):
            res = func(*args, **kw)
            if (base_futures.isfuture(res) or inspect.isgenerator(res) or
                isinstance(res, CoroWrapper)):
                res = yield from res
            elif _AwaitableABC is not None:
                # If 'func' returns an Awaitable (new in 3.5) we
                # want to run it.
                try:
                    await_meth = res.__await__
                except AttributeError:
                    pass
                else:
                    if isinstance(res, _AwaitableABC):
                        res = yield from await_meth()
            return res

if not _DEBUG:
        if _types_coroutine is None:
            wrapper = coro
        else:
            wrapper = _types_coroutine(coro)
    else:
        @functools.wraps(func)
        def wrapper(*args, **kwds):
            w = CoroWrapper(coro(*args, **kwds), func=func)
            if w._source_traceback:
                del w._source_traceback[-1]
            # Python < 3.5 does not implement __qualname__
            # on generator objects, so we set it manually.
            # We use getattr as some callables (such as
            # functools.partial may lack __qualname__).
            w.__name__ = getattr(func, '__name__', None)
            w.__qualname__ = getattr(func, '__qualname__', None)
            return w

wrapper._is_coroutine = _is_coroutine  # For iscoroutinefunction().
    return wrapper

# A marker for iscoroutinefunction.
_is_coroutine = object()

def iscoroutinefunction(func):
    """Return True if func is a decorated coroutine function."""
    return (getattr(func, '_is_coroutine', None) is _is_coroutine or
            _inspect_iscoroutinefunction(func))

_COROUTINE_TYPES = (types.GeneratorType, CoroWrapper)
if _CoroutineABC is not None:
    _COROUTINE_TYPES += (_CoroutineABC,)
if _types_CoroutineType is not None:
    # Prioritize native coroutine check to speed-up
    # asyncio.iscoroutine.
    _COROUTINE_TYPES = (_types_CoroutineType,) + _COROUTINE_TYPES

def iscoroutine(obj):
    """Return True if obj is a coroutine object."""
    return isinstance(obj, _COROUTINE_TYPES)

def _format_coroutine(coro):
    assert iscoroutine(coro)

if not hasattr(coro, 'cr_code') and not hasattr(coro, 'gi_code'):
        # Most likely a built-in type or a Cython coroutine.

# Built-in types might not have __qualname__ or __name__.
        coro_name = getattr(
            coro, '__qualname__',
            getattr(coro, '__name__', type(coro).__name__))
        coro_name = '{}()'.format(coro_name)

running = False
        try:
            running = coro.cr_running
        except AttributeError:
            try:
                running = coro.gi_running
            except AttributeError:
                pass

if running:
            return '{} running'.format(coro_name)
        else:
            return coro_name

coro_name = None
    if isinstance(coro, CoroWrapper):
        func = coro.func
        coro_name = coro.__qualname__
        if coro_name is not None:
            coro_name = '{}()'.format(coro_name)
    else:
        func = coro

if coro_name is None:
        coro_name = events._format_callback(func, (), {})

coro_code = None
    if hasattr(coro, 'cr_code') and coro.cr_code:
        coro_code = coro.cr_code
    elif hasattr(coro, 'gi_code') and coro.gi_code:
        coro_code = coro.gi_code

coro_frame = None
    if hasattr(coro, 'cr_frame') and coro.cr_frame:
        coro_frame = coro.cr_frame
    elif hasattr(coro, 'gi_frame') and coro.gi_frame:
        coro_frame = coro.gi_frame

filename = '<empty co_filename>'
    if coro_code and coro_code.co_filename:
        filename = coro_code.co_filename

lineno = 0
    coro_repr = coro_name

if (isinstance(coro, CoroWrapper) and
            not inspect.isgeneratorfunction(coro.func) and
            coro.func is not None):
        source = events._get_function_source(coro.func)
        if source is not None:
            filename, lineno = source
        if coro_frame is None:
            coro_repr = ('%s done, defined at %s:%s'
                         % (coro_name, filename, lineno))
        else:
            coro_repr = ('%s running, defined at %s:%s'
                         % (coro_name, filename, lineno))
    elif coro_frame is not None:
        lineno = coro_frame.f_lineno
        coro_repr = ('%s running at %s:%s'
                     % (coro_name, filename, lineno))
    elif coro_code:
        lineno = coro_code.co_firstlineno
        coro_repr = ('%s done, defined at %s:%s'
                     % (coro_name, filename, lineno))

return coro_repr

cmd: