FM / asyncio

/opt/alt/python37/lib64/python3.7/asyncio/

/opt/alt/python37/lib64/python3.7/asyncio

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
base_events.py	69830	0644	edit dl rm
base_futures.py	2081	0644	edit dl rm
base_subprocess.py	8786	0644	edit dl rm
base_tasks.py	2158	0644	edit dl rm
constants.py	888	0644	edit dl rm
coroutines.py	8611	0644	edit dl rm
events.py	26220	0644	edit dl rm
format_helpers.py	2404	0644	edit dl rm
futures.py	12823	0644	edit dl rm
locks.py	15913	0644	edit dl rm
log.py	124	0644	edit dl rm
proactor_events.py	25653	0644	edit dl rm
protocols.py	7036	0644	edit dl rm
queues.py	8003	0644	edit dl rm
runners.py	2021	0644	edit dl rm
selector_events.py	37668	0644	edit dl rm
sslproto.py	26927	0644	edit dl rm
streams.py	24587	0644	edit dl rm
subprocess.py	7254	0644	edit dl rm
tasks.py	30704	0644	edit dl rm
transports.py	10122	0644	edit dl rm
unix_events.py	40920	0644	edit dl rm
windows_events.py	30216	0644	edit dl rm
windows_utils.py	5073	0644	edit dl rm
__init__.py	1169	0644	edit dl rm

Edit: /opt/alt/python37/lib64/python3.7/asyncio/base_tasks.py (2158B)

import linecache
import traceback

from . import base_futures
from . import coroutines

def _task_repr_info(task):
    info = base_futures._future_repr_info(task)

if task._must_cancel:
        # replace status
        info[0] = 'cancelling'

coro = coroutines._format_coroutine(task._coro)
    info.insert(1, f'coro=<{coro}>')

if task._fut_waiter is not None:
        info.insert(2, f'wait_for={task._fut_waiter!r}')
    return info

def _task_get_stack(task, limit):
    frames = []
    try:
        # 'async def' coroutines
        f = task._coro.cr_frame
    except AttributeError:
        f = task._coro.gi_frame
    if f is not None:
        while f is not None:
            if limit is not None:
                if limit <= 0:
                    break
                limit -= 1
            frames.append(f)
            f = f.f_back
        frames.reverse()
    elif task._exception is not None:
        tb = task._exception.__traceback__
        while tb is not None:
            if limit is not None:
                if limit <= 0:
                    break
                limit -= 1
            frames.append(tb.tb_frame)
            tb = tb.tb_next
    return frames

def _task_print_stack(task, limit, file):
    extracted_list = []
    checked = set()
    for f in task.get_stack(limit=limit):
        lineno = f.f_lineno
        co = f.f_code
        filename = co.co_filename
        name = co.co_name
        if filename not in checked:
            checked.add(filename)
            linecache.checkcache(filename)
        line = linecache.getline(filename, lineno, f.f_globals)
        extracted_list.append((filename, lineno, name, line))

exc = task._exception
    if not extracted_list:
        print(f'No stack for {task!r}', file=file)
    elif exc is not None:
        print(f'Traceback for {task!r} (most recent call last):', file=file)
    else:
        print(f'Stack for {task!r} (most recent call last):', file=file)

traceback.print_list(extracted_list, file=file)
    if exc is not None:
        for line in traceback.format_exception_only(exc.__class__, exc):
            print(line, file=file, end='')

import linecache import traceback from . import base_futures from . import coroutines def _task_repr_info(task): info = base_futures._future_repr_info(task) if task._must_cancel: # replace status info[0] = 'cancelling' coro = coroutines._format_coroutine(task._coro) info.insert(1, f'coro=<{coro}>') if task._fut_waiter is not None: info.insert(2, f'wait_for={task._fut_waiter!r}') return info def _task_get_stack(task, limit): frames = [] try: # 'async def' coroutines f = task._coro.cr_frame except AttributeError: f = task._coro.gi_frame if f is not None: while f is not None: if limit is not None: if limit <= 0: break limit -= 1 frames.append(f) f = f.f_back frames.reverse() elif task._exception is not None: tb = task._exception.__traceback__ while tb is not None: if limit is not None: if limit <= 0: break limit -= 1 frames.append(tb.tb_frame) tb = tb.tb_next return frames def _task_print_stack(task, limit, file): extracted_list = [] checked = set() for f in task.get_stack(limit=limit): lineno = f.f_lineno co = f.f_code filename = co.co_filename name = co.co_name if filename not in checked: checked.add(filename) linecache.checkcache(filename) line = linecache.getline(filename, lineno, f.f_globals) extracted_list.append((filename, lineno, name, line)) exc = task._exception if not extracted_list: print(f'No stack for {task!r}', file=file) elif exc is not None: print(f'Traceback for {task!r} (most recent call last):', file=file) else: print(f'Stack for {task!r} (most recent call last):', file=file) traceback.print_list(extracted_list, file=file) if exc is not None: for line in traceback.format_exception_only(exc.__class__, exc): print(line, file=file, end='')