mirror of https://github.com/python/cpython
(Merge 3.4) asyncio, Tulip issue 137: In debug mode, save traceback where
Future, Task and Handle objects are created. Pass the traceback to call_exception_handler() in the 'source_traceback' key. The traceback is truncated to hide internal calls in asyncio, show only the traceback from user code. Add tests for the new source_traceback, and a test for the 'Future/Task exception was never retrieved' log.
This commit is contained in:
commit
eb39199f3d
|
@ -21,6 +21,7 @@ import inspect
|
|||
import logging
|
||||
import socket
|
||||
import subprocess
|
||||
import traceback
|
||||
import time
|
||||
import os
|
||||
import sys
|
||||
|
@ -290,7 +291,10 @@ class BaseEventLoop(events.AbstractEventLoop):
|
|||
Any positional arguments after the callback will be passed to
|
||||
the callback when it is called.
|
||||
"""
|
||||
return self.call_at(self.time() + delay, callback, *args)
|
||||
timer = self.call_at(self.time() + delay, callback, *args)
|
||||
if timer._source_traceback:
|
||||
del timer._source_traceback[-1]
|
||||
return timer
|
||||
|
||||
def call_at(self, when, callback, *args):
|
||||
"""Like call_later(), but uses an absolute time."""
|
||||
|
@ -299,6 +303,8 @@ class BaseEventLoop(events.AbstractEventLoop):
|
|||
if self._debug:
|
||||
self._assert_is_current_event_loop()
|
||||
timer = events.TimerHandle(when, callback, args, self)
|
||||
if timer._source_traceback:
|
||||
del timer._source_traceback[-1]
|
||||
heapq.heappush(self._scheduled, timer)
|
||||
return timer
|
||||
|
||||
|
@ -312,7 +318,10 @@ class BaseEventLoop(events.AbstractEventLoop):
|
|||
Any positional arguments after the callback will be passed to
|
||||
the callback when it is called.
|
||||
"""
|
||||
return self._call_soon(callback, args, check_loop=True)
|
||||
handle = self._call_soon(callback, args, check_loop=True)
|
||||
if handle._source_traceback:
|
||||
del handle._source_traceback[-1]
|
||||
return handle
|
||||
|
||||
def _call_soon(self, callback, args, check_loop):
|
||||
if tasks.iscoroutinefunction(callback):
|
||||
|
@ -320,6 +329,8 @@ class BaseEventLoop(events.AbstractEventLoop):
|
|||
if self._debug and check_loop:
|
||||
self._assert_is_current_event_loop()
|
||||
handle = events.Handle(callback, args, self)
|
||||
if handle._source_traceback:
|
||||
del handle._source_traceback[-1]
|
||||
self._ready.append(handle)
|
||||
return handle
|
||||
|
||||
|
@ -344,6 +355,8 @@ class BaseEventLoop(events.AbstractEventLoop):
|
|||
def call_soon_threadsafe(self, callback, *args):
|
||||
"""Like call_soon(), but thread safe."""
|
||||
handle = self._call_soon(callback, args, check_loop=False)
|
||||
if handle._source_traceback:
|
||||
del handle._source_traceback[-1]
|
||||
self._write_to_self()
|
||||
return handle
|
||||
|
||||
|
@ -757,7 +770,14 @@ class BaseEventLoop(events.AbstractEventLoop):
|
|||
for key in sorted(context):
|
||||
if key in {'message', 'exception'}:
|
||||
continue
|
||||
log_lines.append('{}: {!r}'.format(key, context[key]))
|
||||
value = context[key]
|
||||
if key == 'source_traceback':
|
||||
tb = ''.join(traceback.format_list(value))
|
||||
value = 'Object created at (most recent call last):\n'
|
||||
value += tb.rstrip()
|
||||
else:
|
||||
value = repr(value)
|
||||
log_lines.append('{}: {}'.format(key, value))
|
||||
|
||||
logger.error('\n'.join(log_lines), exc_info=exc_info)
|
||||
|
||||
|
|
|
@ -11,6 +11,7 @@ __all__ = ['AbstractEventLoopPolicy',
|
|||
import functools
|
||||
import inspect
|
||||
import subprocess
|
||||
import traceback
|
||||
import threading
|
||||
import socket
|
||||
import sys
|
||||
|
@ -66,7 +67,8 @@ def _format_callback(func, args, suffix=''):
|
|||
class Handle:
|
||||
"""Object returned by callback registration methods."""
|
||||
|
||||
__slots__ = ['_callback', '_args', '_cancelled', '_loop', '__weakref__']
|
||||
__slots__ = ('_callback', '_args', '_cancelled', '_loop',
|
||||
'_source_traceback', '__weakref__')
|
||||
|
||||
def __init__(self, callback, args, loop):
|
||||
assert not isinstance(callback, Handle), 'A Handle is not a callback'
|
||||
|
@ -74,6 +76,10 @@ class Handle:
|
|||
self._callback = callback
|
||||
self._args = args
|
||||
self._cancelled = False
|
||||
if self._loop.get_debug():
|
||||
self._source_traceback = traceback.extract_stack(sys._getframe(1))
|
||||
else:
|
||||
self._source_traceback = None
|
||||
|
||||
def __repr__(self):
|
||||
info = []
|
||||
|
@ -91,11 +97,14 @@ class Handle:
|
|||
except Exception as exc:
|
||||
cb = _format_callback(self._callback, self._args)
|
||||
msg = 'Exception in callback {}'.format(cb)
|
||||
self._loop.call_exception_handler({
|
||||
context = {
|
||||
'message': msg,
|
||||
'exception': exc,
|
||||
'handle': self,
|
||||
})
|
||||
}
|
||||
if self._source_traceback:
|
||||
context['source_traceback'] = self._source_traceback
|
||||
self._loop.call_exception_handler(context)
|
||||
self = None # Needed to break cycles when an exception occurs.
|
||||
|
||||
|
||||
|
@ -107,7 +116,8 @@ class TimerHandle(Handle):
|
|||
def __init__(self, when, callback, args, loop):
|
||||
assert when is not None
|
||||
super().__init__(callback, args, loop)
|
||||
|
||||
if self._source_traceback:
|
||||
del self._source_traceback[-1]
|
||||
self._when = when
|
||||
|
||||
def __repr__(self):
|
||||
|
|
|
@ -82,10 +82,11 @@ class _TracebackLogger:
|
|||
in a discussion about closing files when they are collected.
|
||||
"""
|
||||
|
||||
__slots__ = ['exc', 'tb', 'loop']
|
||||
__slots__ = ('loop', 'source_traceback', 'exc', 'tb')
|
||||
|
||||
def __init__(self, exc, loop):
|
||||
self.loop = loop
|
||||
def __init__(self, future, exc):
|
||||
self.loop = future._loop
|
||||
self.source_traceback = future._source_traceback
|
||||
self.exc = exc
|
||||
self.tb = None
|
||||
|
||||
|
@ -102,11 +103,12 @@ class _TracebackLogger:
|
|||
|
||||
def __del__(self):
|
||||
if self.tb:
|
||||
msg = 'Future/Task exception was never retrieved:\n{tb}'
|
||||
context = {
|
||||
'message': msg.format(tb=''.join(self.tb)),
|
||||
}
|
||||
self.loop.call_exception_handler(context)
|
||||
msg = 'Future/Task exception was never retrieved'
|
||||
if self.source_traceback:
|
||||
msg += '\nFuture/Task created at (most recent call last):\n'
|
||||
msg += ''.join(traceback.format_list(self.source_traceback))
|
||||
msg += ''.join(self.tb).rstrip()
|
||||
self.loop.call_exception_handler({'message': msg})
|
||||
|
||||
|
||||
class Future:
|
||||
|
@ -149,6 +151,10 @@ class Future:
|
|||
else:
|
||||
self._loop = loop
|
||||
self._callbacks = []
|
||||
if self._loop.get_debug():
|
||||
self._source_traceback = traceback.extract_stack(sys._getframe(1))
|
||||
else:
|
||||
self._source_traceback = None
|
||||
|
||||
def _format_callbacks(self):
|
||||
cb = self._callbacks
|
||||
|
@ -196,10 +202,13 @@ class Future:
|
|||
return
|
||||
exc = self._exception
|
||||
context = {
|
||||
'message': 'Future/Task exception was never retrieved',
|
||||
'message': ('%s exception was never retrieved'
|
||||
% self.__class__.__name__),
|
||||
'exception': exc,
|
||||
'future': self,
|
||||
}
|
||||
if self._source_traceback:
|
||||
context['source_traceback'] = self._source_traceback
|
||||
self._loop.call_exception_handler(context)
|
||||
|
||||
def cancel(self):
|
||||
|
@ -335,7 +344,7 @@ class Future:
|
|||
if _PY34:
|
||||
self._log_traceback = True
|
||||
else:
|
||||
self._tb_logger = _TracebackLogger(exception, self._loop)
|
||||
self._tb_logger = _TracebackLogger(self, exception)
|
||||
# Arrange for the logger to be activated after all callbacks
|
||||
# have had a chance to call result() or exception().
|
||||
self._loop.call_soon(self._tb_logger.activate)
|
||||
|
|
|
@ -195,6 +195,8 @@ class Task(futures.Future):
|
|||
def __init__(self, coro, *, loop=None):
|
||||
assert iscoroutine(coro), repr(coro) # Not a coroutine function!
|
||||
super().__init__(loop=loop)
|
||||
if self._source_traceback:
|
||||
del self._source_traceback[-1]
|
||||
self._coro = iter(coro) # Use the iterator just in case.
|
||||
self._fut_waiter = None
|
||||
self._must_cancel = False
|
||||
|
@ -207,10 +209,13 @@ class Task(futures.Future):
|
|||
if _PY34:
|
||||
def __del__(self):
|
||||
if self._state == futures._PENDING:
|
||||
self._loop.call_exception_handler({
|
||||
context = {
|
||||
'task': self,
|
||||
'message': 'Task was destroyed but it is pending!',
|
||||
})
|
||||
}
|
||||
if self._source_traceback:
|
||||
context['source_traceback'] = self._source_traceback
|
||||
self._loop.call_exception_handler(context)
|
||||
futures.Future.__del__(self)
|
||||
|
||||
def __repr__(self):
|
||||
|
@ -620,7 +625,10 @@ def async(coro_or_future, *, loop=None):
|
|||
raise ValueError('loop argument must agree with Future')
|
||||
return coro_or_future
|
||||
elif iscoroutine(coro_or_future):
|
||||
return Task(coro_or_future, loop=loop)
|
||||
task = Task(coro_or_future, loop=loop)
|
||||
if task._source_traceback:
|
||||
del task._source_traceback[-1]
|
||||
return task
|
||||
else:
|
||||
raise TypeError('A Future or coroutine is required')
|
||||
|
||||
|
|
|
@ -406,19 +406,22 @@ class BaseEventLoopTests(test_utils.TestCase):
|
|||
1/0
|
||||
|
||||
def run_loop():
|
||||
self.loop.call_soon(zero_error)
|
||||
handle = self.loop.call_soon(zero_error)
|
||||
self.loop._run_once()
|
||||
return handle
|
||||
|
||||
self.loop.set_debug(True)
|
||||
self.loop._process_events = mock.Mock()
|
||||
|
||||
mock_handler = mock.Mock()
|
||||
self.loop.set_exception_handler(mock_handler)
|
||||
run_loop()
|
||||
handle = run_loop()
|
||||
mock_handler.assert_called_with(self.loop, {
|
||||
'exception': MOCK_ANY,
|
||||
'message': test_utils.MockPattern(
|
||||
'Exception in callback.*zero_error'),
|
||||
'handle': MOCK_ANY,
|
||||
'handle': handle,
|
||||
'source_traceback': handle._source_traceback,
|
||||
})
|
||||
mock_handler.reset_mock()
|
||||
|
||||
|
|
|
@ -1751,10 +1751,11 @@ def noop(*args):
|
|||
pass
|
||||
|
||||
|
||||
class HandleTests(unittest.TestCase):
|
||||
class HandleTests(test_utils.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.loop = None
|
||||
self.loop = mock.Mock()
|
||||
self.loop.get_debug.return_value = True
|
||||
|
||||
def test_handle(self):
|
||||
def callback(*args):
|
||||
|
@ -1789,7 +1790,8 @@ class HandleTests(unittest.TestCase):
|
|||
self.loop.call_exception_handler.assert_called_with({
|
||||
'message': test_utils.MockPattern('Exception in callback.*'),
|
||||
'exception': mock.ANY,
|
||||
'handle': h
|
||||
'handle': h,
|
||||
'source_traceback': h._source_traceback,
|
||||
})
|
||||
|
||||
def test_handle_weakref(self):
|
||||
|
@ -1837,6 +1839,35 @@ class HandleTests(unittest.TestCase):
|
|||
% (cb_regex, re.escape(filename), lineno))
|
||||
self.assertRegex(repr(h), regex)
|
||||
|
||||
def test_handle_source_traceback(self):
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
loop.set_debug(True)
|
||||
self.set_event_loop(loop)
|
||||
|
||||
def check_source_traceback(h):
|
||||
lineno = sys._getframe(1).f_lineno - 1
|
||||
self.assertIsInstance(h._source_traceback, list)
|
||||
self.assertEqual(h._source_traceback[-1][:3],
|
||||
(__file__,
|
||||
lineno,
|
||||
'test_handle_source_traceback'))
|
||||
|
||||
# call_soon
|
||||
h = loop.call_soon(noop)
|
||||
check_source_traceback(h)
|
||||
|
||||
# call_soon_threadsafe
|
||||
h = loop.call_soon_threadsafe(noop)
|
||||
check_source_traceback(h)
|
||||
|
||||
# call_later
|
||||
h = loop.call_later(0, noop)
|
||||
check_source_traceback(h)
|
||||
|
||||
# call_at
|
||||
h = loop.call_later(0, noop)
|
||||
check_source_traceback(h)
|
||||
|
||||
|
||||
class TimerTests(unittest.TestCase):
|
||||
|
||||
|
|
|
@ -2,8 +2,10 @@
|
|||
|
||||
import concurrent.futures
|
||||
import re
|
||||
import sys
|
||||
import threading
|
||||
import unittest
|
||||
from test import support
|
||||
from unittest import mock
|
||||
|
||||
import asyncio
|
||||
|
@ -284,6 +286,63 @@ class FutureTests(test_utils.TestCase):
|
|||
self.assertEqual(f1.result(), 42)
|
||||
self.assertTrue(f2.cancelled())
|
||||
|
||||
def test_future_source_traceback(self):
|
||||
self.loop.set_debug(True)
|
||||
|
||||
future = asyncio.Future(loop=self.loop)
|
||||
lineno = sys._getframe().f_lineno - 1
|
||||
self.assertIsInstance(future._source_traceback, list)
|
||||
self.assertEqual(future._source_traceback[-1][:3],
|
||||
(__file__,
|
||||
lineno,
|
||||
'test_future_source_traceback'))
|
||||
|
||||
@mock.patch('asyncio.base_events.logger')
|
||||
def test_future_exception_never_retrieved(self, m_log):
|
||||
self.loop.set_debug(True)
|
||||
|
||||
def memroy_error():
|
||||
try:
|
||||
raise MemoryError()
|
||||
except BaseException as exc:
|
||||
return exc
|
||||
exc = memroy_error()
|
||||
|
||||
future = asyncio.Future(loop=self.loop)
|
||||
source_traceback = future._source_traceback
|
||||
future.set_exception(exc)
|
||||
future = None
|
||||
test_utils.run_briefly(self.loop)
|
||||
support.gc_collect()
|
||||
|
||||
if sys.version_info >= (3, 4):
|
||||
frame = source_traceback[-1]
|
||||
regex = (r'^Future exception was never retrieved\n'
|
||||
r'future: <Future finished exception=MemoryError\(\)>\n'
|
||||
r'source_traceback: Object created at \(most recent call last\):\n'
|
||||
r' File'
|
||||
r'.*\n'
|
||||
r' File "%s", line %s, in test_future_exception_never_retrieved\n'
|
||||
r' future = asyncio\.Future\(loop=self\.loop\)$'
|
||||
% (frame[0], frame[1]))
|
||||
exc_info = (type(exc), exc, exc.__traceback__)
|
||||
m_log.error.assert_called_once_with(mock.ANY, exc_info=exc_info)
|
||||
else:
|
||||
frame = source_traceback[-1]
|
||||
regex = (r'^Future/Task exception was never retrieved\n'
|
||||
r'Future/Task created at \(most recent call last\):\n'
|
||||
r' File'
|
||||
r'.*\n'
|
||||
r' File "%s", line %s, in test_future_exception_never_retrieved\n'
|
||||
r' future = asyncio\.Future\(loop=self\.loop\)\n'
|
||||
r'Traceback \(most recent call last\):\n'
|
||||
r'.*\n'
|
||||
r'MemoryError$'
|
||||
% (frame[0], frame[1]))
|
||||
m_log.error.assert_called_once_with(mock.ANY, exc_info=False)
|
||||
message = m_log.error.call_args[0][0]
|
||||
self.assertRegex(message, re.compile(regex, re.DOTALL))
|
||||
|
||||
|
||||
class FutureDoneCallbackTests(test_utils.TestCase):
|
||||
|
||||
|
|
|
@ -1546,6 +1546,7 @@ class TaskTests(test_utils.TestCase):
|
|||
raise Exception("code never reached")
|
||||
|
||||
mock_handler = mock.Mock()
|
||||
self.loop.set_debug(True)
|
||||
self.loop.set_exception_handler(mock_handler)
|
||||
|
||||
# schedule the task
|
||||
|
@ -1560,6 +1561,7 @@ class TaskTests(test_utils.TestCase):
|
|||
# remove the future used in kill_me(), and references to the task
|
||||
del coro.gi_frame.f_locals['future']
|
||||
coro = None
|
||||
source_traceback = task._source_traceback
|
||||
task = None
|
||||
|
||||
# no more reference to kill_me() task: the task is destroyed by the GC
|
||||
|
@ -1570,6 +1572,7 @@ class TaskTests(test_utils.TestCase):
|
|||
mock_handler.assert_called_with(self.loop, {
|
||||
'message': 'Task was destroyed but it is pending!',
|
||||
'task': mock.ANY,
|
||||
'source_traceback': source_traceback,
|
||||
})
|
||||
mock_handler.reset_mock()
|
||||
|
||||
|
@ -1604,6 +1607,17 @@ class TaskTests(test_utils.TestCase):
|
|||
|
||||
self.assertRegex(message, re.compile(regex, re.DOTALL))
|
||||
|
||||
def test_task_source_traceback(self):
|
||||
self.loop.set_debug(True)
|
||||
|
||||
task = asyncio.Task(coroutine_function(), loop=self.loop)
|
||||
lineno = sys._getframe().f_lineno - 1
|
||||
self.assertIsInstance(task._source_traceback, list)
|
||||
self.assertEqual(task._source_traceback[-1][:3],
|
||||
(__file__,
|
||||
lineno,
|
||||
'test_task_source_traceback'))
|
||||
|
||||
|
||||
class GatherTestsBase:
|
||||
|
||||
|
|
Loading…
Reference in New Issue