Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

(Merge 3.4) asyncio: Enable the debug mode of event loops when the 

PYTHONASYNCIODEBUG environment variable is set
This commit is contained in:
Victor Stinner 2014-06-23 00:14:45 +02:00
parent 33d2e40d88 7b7120e159
commit d42b49256f
6 changed files with 11 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Doc/library/asyncio-eventloop.rst
View File

@ -585,7 +585,11 @@ Debug mode
.. method:: BaseEventLoop.get_debug()
Get the debug mode (:class:`bool`) of the event loop, ``False`` by default.
Get the debug mode (:class:`bool`) of the event loop.
The default value is ``True`` if the environment variable
:envvar:`PYTHONASYNCIODEBUG` is set to a non-empty string, ``False``
otherwise.
.. versionadded:: 3.4.2

3
Lib/asyncio/base_events.py
View File

@ -137,7 +137,8 @@ class BaseEventLoop(events.AbstractEventLoop):
self._running = False
self._clock_resolution = time.get_clock_info('monotonic').resolution
self._exception_handler = None
self._debug = False
self._debug = (not sys.flags.ignore_environment
and bool(os.environ.get('PYTHONASYNCIODEBUG')))
# In debug mode, if the execution of a callback or a step of a task
# exceed this duration in seconds, the slow callback/task is logged.
self.slow_callback_duration = 0.1

2
Lib/test/test_asyncio/test_selector_events.py
View File

@ -682,8 +682,6 @@ class SelectorTransportTests(test_utils.TestCase):
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
self.assertEqual(3, sys.getrefcount(self.loop),
pprint.pformat(gc.get_referrers(self.loop)))
class SelectorSocketTransportTests(test_utils.TestCase):

4
Lib/test/test_asyncio/test_subprocess.py
View File

@ -141,7 +141,7 @@ if sys.platform != 'win32':
policy = asyncio.get_event_loop_policy()
self.loop = policy.new_event_loop()
# ensure that the event loop is passed explicitly in the code
# ensure that the event loop is passed explicitly in asyncio
policy.set_event_loop(None)
watcher = self.Watcher()
@ -172,7 +172,7 @@ else:
policy = asyncio.get_event_loop_policy()
self.loop = asyncio.ProactorEventLoop()
# ensure that the event loop is passed explicitly in the code
# ensure that the event loop is passed explicitly in asyncio
policy.set_event_loop(None)
def tearDown(self):

2
Lib/test/test_asyncio/test_tasks.py
View File

@ -1712,6 +1712,8 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
self.assertIs(fut._loop, self.one_loop)
gen1.close()
gen2.close()
self.set_event_loop(self.other_loop, cleanup=False)
gen3 = coro()
gen4 = coro()
fut = asyncio.gather(gen3, gen4, loop=self.other_loop)

8
Lib/test/test_asyncio/test_unix_events.py
View File

@ -445,8 +445,6 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
self.assertEqual(5, sys.getrefcount(self.loop),
pprint.pformat(gc.get_referrers(self.loop)))
def test__call_connection_lost_with_err(self):
tr = unix_events._UnixReadPipeTransport(
@ -462,8 +460,6 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
self.assertEqual(5, sys.getrefcount(self.loop),
pprint.pformat(gc.get_referrers(self.loop)))
class UnixWritePipeTransportTests(test_utils.TestCase):
@ -731,8 +727,6 @@ class UnixWritePipeTransportTests(test_utils.TestCase):
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
self.assertEqual(5, sys.getrefcount(self.loop),
pprint.pformat(gc.get_referrers(self.loop)))
def test__call_connection_lost_with_err(self):
tr = unix_events._UnixWritePipeTransport(
@ -747,8 +741,6 @@ class UnixWritePipeTransportTests(test_utils.TestCase):
self.assertEqual(2, sys.getrefcount(self.protocol),
pprint.pformat(gc.get_referrers(self.protocol)))
self.assertIsNone(tr._loop)
self.assertEqual(5, sys.getrefcount(self.loop),
pprint.pformat(gc.get_referrers(self.loop)))
def test_close(self):
tr = unix_events._UnixWritePipeTransport(