asyncio: Log an error if a Task is destroyed while it is still pending

Lib/asyncio/futures.py

@@ -169,6 +169,9 @@ def __repr__(self):
             res += '<{}>'.format(self._state)
         return res
 
+    # On Python 3.3 or older, objects with a destructor part of a reference
+    # cycle are never destroyed. It's not more the case on Python 3.4 thanks to
+    # the PEP 442.
     if _PY34:
         def __del__(self):
             if not self._log_traceback:

Lib/asyncio/tasks.py

@@ -32,6 +32,7 @@
 _DEBUG = (not sys.flags.ignore_environment
           and bool(os.environ.get('PYTHONASYNCIODEBUG')))
 
+_PY34 = (sys.version_info >= (3, 4))
 _PY35 = (sys.version_info >= (3, 5))
 
 
@@ -181,6 +182,18 @@ def __init__(self, coro, *, loop=None):
         self._loop.call_soon(self._step)
         self.__class__._all_tasks.add(self)
 
+    # On Python 3.3 or older, objects with a destructor part of a reference
+    # cycle are never destroyed. It's not more the case on Python 3.4 thanks to
+    # the PEP 442.
+    if _PY34:
+        def __del__(self):
+            if self._state == futures._PENDING:
+                self._loop.call_exception_handler({
+                    'task': self,
+                    'message': 'Task was destroyed but it is pending!',
+                })
+            futures.Future.__del__(self)
+
     def __repr__(self):
         res = super().__repr__()
         if (self._must_cancel and

Lib/test/test_asyncio/test_base_events.py

@@ -244,7 +244,8 @@ def test_set_debug(self):
     @mock.patch('asyncio.base_events.logger')
     def test__run_once_logging(self, m_logger):
         def slow_select(timeout):
-            time.sleep(1.0)
+            # Sleep a bit longer than a second to avoid timer resolution issues.
+            time.sleep(1.1)
             return []
 
         # logging needs debug flag

Lib/test/test_asyncio/test_tasks.py

@@ -5,13 +5,16 @@
 import types
 import unittest
 import weakref
+from test import support
 from test.script_helper import assert_python_ok
+from unittest import mock
 
 import asyncio
 from asyncio import tasks
 from asyncio import test_utils
 
 
+PY34 = (sys.version_info >= (3, 4))
 PY35 = (sys.version_info >= (3, 5))
 
 
@@ -1501,9 +1504,45 @@ def call(arg):
     def test_corowrapper_weakref(self):
         wd = weakref.WeakValueDictionary()
         def foo(): yield from []
-        cw = asyncio.tasks.CoroWrapper(foo(), foo)
-        wd['cw'] = cw  # Would fail without __weakref__ slot.
-        cw.gen = None  # Suppress warning from __del__.
+
+    @unittest.skipUnless(PY34,
+                         'need python 3.4 or later')
+    def test_log_destroyed_pending_task(self):
+        @asyncio.coroutine
+        def kill_me(loop):
+            future = asyncio.Future(loop=loop)
+            yield from future
+            # at this point, the only reference to kill_me() task is
+            # the Task._wakeup() method in future._callbacks
+            raise Exception("code never reached")
+
+        mock_handler = mock.Mock()
+        self.loop.set_exception_handler(mock_handler)
+
+        # schedule the task
+        coro = kill_me(self.loop)
+        task = asyncio.async(coro, loop=self.loop)
+        self.assertEqual(asyncio.Task.all_tasks(loop=self.loop), {task})
+
+        # execute the task so it waits for future
+        self.loop._run_once()
+        self.assertEqual(len(self.loop._ready), 0)
+
+        # remove the future used in kill_me(), and references to the task
+        del coro.gi_frame.f_locals['future']
+        coro = None
+        task = None
+
+        # no more reference to kill_me() task: the task is destroyed by the GC
+        support.gc_collect()
+
+        self.assertEqual(asyncio.Task.all_tasks(loop=self.loop), set())
+
+        mock_handler.assert_called_with(self.loop, {
+            'message': 'Task was destroyed but it is pending!',
+            'task': mock.ANY,
+        })
+        mock_handler.reset_mock()
 
 
 class GatherTestsBase: