• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import linecache
2import traceback
3
4from . import base_futures
5from . import coroutines
6
7
8def _task_repr_info(task):
9    info = base_futures._future_repr_info(task)
10
11    if task._must_cancel:
12        # replace status
13        info[0] = 'cancelling'
14
15    info.insert(1, 'name=%r' % task.get_name())
16
17    coro = coroutines._format_coroutine(task._coro)
18    info.insert(2, f'coro=<{coro}>')
19
20    if task._fut_waiter is not None:
21        info.insert(3, f'wait_for={task._fut_waiter!r}')
22    return info
23
24
25def _task_get_stack(task, limit):
26    frames = []
27    try:
28        # 'async def' coroutines
29        f = task._coro.cr_frame
30    except AttributeError:
31        f = task._coro.gi_frame
32    if f is not None:
33        while f is not None:
34            if limit is not None:
35                if limit <= 0:
36                    break
37                limit -= 1
38            frames.append(f)
39            f = f.f_back
40        frames.reverse()
41    elif task._exception is not None:
42        tb = task._exception.__traceback__
43        while tb is not None:
44            if limit is not None:
45                if limit <= 0:
46                    break
47                limit -= 1
48            frames.append(tb.tb_frame)
49            tb = tb.tb_next
50    return frames
51
52
53def _task_print_stack(task, limit, file):
54    extracted_list = []
55    checked = set()
56    for f in task.get_stack(limit=limit):
57        lineno = f.f_lineno
58        co = f.f_code
59        filename = co.co_filename
60        name = co.co_name
61        if filename not in checked:
62            checked.add(filename)
63            linecache.checkcache(filename)
64        line = linecache.getline(filename, lineno, f.f_globals)
65        extracted_list.append((filename, lineno, name, line))
66
67    exc = task._exception
68    if not extracted_list:
69        print(f'No stack for {task!r}', file=file)
70    elif exc is not None:
71        print(f'Traceback for {task!r} (most recent call last):', file=file)
72    else:
73        print(f'Stack for {task!r} (most recent call last):', file=file)
74
75    traceback.print_list(extracted_list, file=file)
76    if exc is not None:
77        for line in traceback.format_exception_only(exc.__class__, exc):
78            print(line, file=file, end='')
79