Merge Tulip into Trollius

7 files changed, 112 insertions, 73 deletions

diff --git a/tests/test_futures.py b/tests/test_futures.py
index 2086515..fc5c043 100644
--- a/tests/test_futures.py
+++ b/tests/test_futures.py
@@ -89,6 +89,15 @@ class FutureTests(test_utils.TestCase):
         self.assertIsInstance(f.exception(), RuntimeError)
 
     def test_future_repr(self):
+        self.loop.set_debug(True)
+        f_pending_debug = asyncio.Future(loop=self.loop)
+        frame = f_pending_debug._source_traceback[-1]
+        self.assertEqual(repr(f_pending_debug),
+                         '<Future pending created at %s:%s>'
+                         % (frame[0], frame[1]))
+        f_pending_debug.cancel()
+
+        self.loop.set_debug(False)
         f_pending = asyncio.Future(loop=self.loop)
         self.assertEqual(repr(f_pending), '<Future pending>')
         f_pending.cancel()
@@ -274,12 +283,6 @@ class FutureTests(test_utils.TestCase):
 
     @mock.patch('trollius.base_events.logger')
     def test_future_exception_never_retrieved(self, m_log):
-        # FIXME: Python issue #21163, other tests may "leak" pending task which
-        # emit a warning when they are destroyed by the GC
-        support.gc_collect()
-        m_log.error.reset_mock()
-        # ---
-
         self.loop.set_debug(True)
 
         def memory_error():
@@ -299,7 +302,7 @@ class FutureTests(test_utils.TestCase):
         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'future: <Future finished exception=MemoryError\(\) created at {filename}:{lineno}>\n'
                      r'source_traceback: Object created at \(most recent call last\):\n'
                      r'  File'
                      r'.*\n'
diff --git a/tests/test_tasks.py b/tests/test_tasks.py
index 0b9fdf2..b2c9d1e 100644
--- a/tests/test_tasks.py
+++ b/tests/test_tasks.py
@@ -134,6 +134,8 @@ class TaskTests(test_utils.TestCase):
             asyncio.async('ok')
 
     def test_task_repr(self):
+        self.loop.set_debug(False)
+
         @asyncio.coroutine
         def noop():
             yield From(None)
@@ -196,6 +198,8 @@ class TaskTests(test_utils.TestCase):
                          "<Task finished %s result='abc'>" % coro)
 
     def test_task_repr_coro_decorator(self):
+        self.loop.set_debug(False)
+
         @asyncio.coroutine
         def notmuch():
             # notmuch() function doesn't use yield: it will be wrapped by
@@ -262,6 +266,8 @@ class TaskTests(test_utils.TestCase):
         self.loop.run_until_complete(t)
 
     def test_task_repr_wait_for(self):
+        self.loop.set_debug(False)
+
         @asyncio.coroutine
         def wait_for(fut):
             res = yield From(fut)
diff --git a/trollius/events.py b/trollius/events.py
index 131399b..2498094 100644
--- a/trollius/events.py
+++ b/trollius/events.py
@@ -11,11 +11,16 @@ __all__ = ['AbstractEventLoopPolicy',
 
 import functools
 import inspect
-import subprocess
-import traceback
-import threading
 import socket
+import subprocess
 import sys
+import threading
+import traceback
+try:
+    import reprlib   # Python 3
+except ImportError:
+    import repr as reprlib   # Python 2
+
 from trollius import compat
 try:
     import asyncio
@@ -60,8 +65,12 @@ def _get_function_source(func):
 
 
 def _format_args(args):
-    # function formatting ('hello',) as ('hello')
-    args_repr = repr(args)
+    """Format function arguments.
+
+    Special case for a single parameter: ('hello',) is formatted as ('hello').
+    """
+    # use reprlib to limit the length of the output
+    args_repr = reprlib.repr(args)
     if len(args) == 1 and args_repr.endswith(',)'):
         args_repr = args_repr[:-2] + ')'
     return args_repr
diff --git a/trollius/futures.py b/trollius/futures.py
index fa0a7e7..780c73d 100644
--- a/trollius/futures.py
+++ b/trollius/futures.py
@@ -8,6 +8,10 @@ __all__ = ['CancelledError', 'TimeoutError',
 import logging
 import sys
 import traceback
+try:
+    import reprlib   # Python 3
+except ImportError:
+    import repr as reprlib   # Python 2
 
 from . import events
 from . import executor
@@ -172,20 +176,25 @@ class Future(object):
                                                format_cb(cb[-1]))
         return 'cb=[%s]' % cb
 
-    def _format_result(self):
-        if self._state != _FINISHED:
-            return None
-        elif self._exception is not None:
-            return 'exception={0!r}'.format(self._exception)
-        else:
-            return 'result={0!r}'.format(self._result)
-
-    def __repr__(self):
+    def _repr_info(self):
         info = [self._state.lower()]
         if self._state == _FINISHED:
-            info.append(self._format_result())
+            if self._exception is not None:
+                info.append('exception={0!r}'.format(self._exception))
+            else:
+                # use reprlib to limit the length of the output, especially
+                # for very long strings
+                result = reprlib.repr(self._result)
+                info.append('result={0}'.format(result))
         if self._callbacks:
             info.append(self._format_callbacks())
+        if self._source_traceback:
+            frame = self._source_traceback[-1]
+            info.append('created at %s:%s' % (frame[0], frame[1]))
+        return info
+
+    def __repr__(self):
+        info = self._repr_info()
         return '<%s %s>' % (self.__class__.__name__, ' '.join(info))
 
     # On Python 3.3 or older, objects with a destructor part of a reference
diff --git a/trollius/selectors.py b/trollius/selectors.py
index ffd7555..5d0a188 100644
--- a/trollius/selectors.py
+++ b/trollius/selectors.py
@@ -452,14 +452,14 @@ if hasattr(select, 'devpoll'):
         """Solaris /dev/poll selector."""
 
         def __init__(self):
-            super().__init__()
+            super(DevpollSelector, self).__init__()
             self._devpoll = select.devpoll()
 
         def fileno(self):
             return self._devpoll.fileno()
 
         def register(self, fileobj, events, data=None):
-            key = super().register(fileobj, events, data)
+            key = super(DevpollSelector, self).register(fileobj, events, data)
             poll_events = 0
             if events & EVENT_READ:
                 poll_events |= select.POLLIN
@@ -469,7 +469,7 @@ if hasattr(select, 'devpoll'):
             return key
 
         def unregister(self, fileobj):
-            key = super().unregister(fileobj)
+            key = super(DevpollSelector, self).unregister(fileobj)
             self._devpoll.unregister(key.fd)
             return key
 
@@ -501,7 +501,7 @@ if hasattr(select, 'devpoll'):
 
         def close(self):
             self._devpoll.close()
-            super().close()
+            super(DevpollSelector, self).close()
 
 
 if hasattr(select, 'kqueue'):
diff --git a/trollius/tasks.py b/trollius/tasks.py
index 93ad6c8..a14175e 100644
--- a/trollius/tasks.py
+++ b/trollius/tasks.py
@@ -105,30 +105,19 @@ class Task(futures.Future):
                 self._loop.call_exception_handler(context)
             futures.Future.__del__(self)
 
-    def __repr__(self):
-        info = []
+    def _repr_info(self):
+        info = super(Task, self)._repr_info()
+
         if self._must_cancel:
-            info.append('cancelling')
-        else:
-            info.append(self._state.lower())
+            # replace status
+            info[0] = 'cancelling'
 
         coro = coroutines._format_coroutine(self._coro)
-        info.append('coro=<%s>' % coro)
-
-        if self._source_traceback:
-            frame = self._source_traceback[-1]
-            info.append('created at %s:%s' % (frame[0], frame[1]))
-
-        if self._state == futures._FINISHED:
-            info.append(self._format_result())
-
-        if self._callbacks:
-            info.append(self._format_callbacks())
+        info.insert(1, 'coro=<%s>' % coro)
 
         if self._fut_waiter is not None:
-            info.append('wait_for=%r' % self._fut_waiter)
-
-        return '<%s %s>' % (self.__class__.__name__, ' '.join(info))
+            info.insert(2, 'wait_for=%r' % self._fut_waiter)
+        return info
 
     def get_stack(self, limit=None):
         """Return the list of stack frames for this task's coroutine.
diff --git a/trollius/windows_events.py b/trollius/windows_events.py
index aef840e..a4468ea 100644
--- a/trollius/windows_events.py
+++ b/trollius/windows_events.py
@@ -42,16 +42,12 @@ class _OverlappedFuture(futures.Future):
             del self._source_traceback[-1]
         self._ov = ov
 
-    def __repr__(self):
-        info = [self._state.lower()]
+    def _repr_info(self):
+        info = super(_OverlappedFuture, self)._repr_info()
         if self._ov is not None:
             state = 'pending' if self._ov.pending else 'completed'
-            info.append('overlapped=<%s, %#x>' % (state, self._ov.address))
-        if self._state == futures._FINISHED:
-            info.append(self._format_result())
-        if self._callbacks:
-            info.append(self._format_callbacks())
-        return '<%s %s>' % (self.__class__.__name__, ' '.join(info))
+            info.insert(1, 'overlapped=<%s, %#x>' % (state, self._ov.address))
+        return info
 
     def _cancel_overlapped(self):
         if self._ov is None:
@@ -85,8 +81,14 @@ class _OverlappedFuture(futures.Future):
 class _WaitHandleFuture(futures.Future):
     """Subclass of Future which represents a wait handle."""
 
-    def __init__(self, handle, wait_handle, loop=None):
+    def __init__(self, iocp, ov, handle, wait_handle, loop=None):
         super(_WaitHandleFuture, self).__init__(loop=loop)
+        if self._source_traceback:
+            del self._source_traceback[-1]
+        # iocp and ov are only used by cancel() to notify IocpProactor
+        # that the wait was cancelled
+        self._iocp = iocp
+        self._ov = ov
         self._handle = handle
         self._wait_handle = wait_handle
 
@@ -95,19 +97,16 @@ class _WaitHandleFuture(futures.Future):
         return (_winapi.WaitForSingleObject(self._handle, 0) ==
                 _winapi.WAIT_OBJECT_0)
 
-    def __repr__(self):
-        info = [self._state.lower()]
+    def _repr_info(self):
+        info = super(_WaitHandleFuture, self)._repr_info()
+        info.insert(1, 'handle=%#x' % self._handle)
         if self._wait_handle:
-            state = 'pending' if self._poll() else 'completed'
-            info.append('wait_handle=<%s, %#x>' % (state, self._wait_handle))
-        info.append('handle=<%#x>' % self._handle)
-        if self._state == futures._FINISHED:
-            info.append(self._format_result())
-        if self._callbacks:
-            info.append(self._format_callbacks())
-        return '<%s %s>' % (self.__class__.__name__, ' '.join(info))
-
-    def _unregister(self):
+            state = 'signaled' if self._poll() else 'waiting'
+            info.insert(1, 'wait_handle=<%s, %#x>'
+                           % (state, self._wait_handle))
+        return info
+
+    def _unregister_wait(self):
         if self._wait_handle is None:
             return
         try:
@@ -117,10 +116,25 @@ class _WaitHandleFuture(futures.Future):
                 raise
             # ERROR_IO_PENDING is not an error, the wait was unregistered
         self._wait_handle = None
+        self._iocp = None
+        self._ov = None
 
     def cancel(self):
-        self._unregister()
-        return super(_WaitHandleFuture, self).cancel()
+        result = super(_WaitHandleFuture, self).cancel()
+        if self._ov is not None:
+            # signal the cancellation to the overlapped object
+            _overlapped.PostQueuedCompletionStatus(self._iocp, True,
+                                                   0, self._ov.address)
+        self._unregister_wait()
+        return result
+
+    def set_exception(self, exception):
+        super(_WaitHandleFuture, self).set_exception(exception)
+        self._unregister_wait()
+
+    def set_result(self, result):
+        super(_WaitHandleFuture, self).set_result(result)
+        self._unregister_wait()
 
 
 class PipeServer(object):
@@ -397,7 +411,9 @@ class IocpProactor(object):
         ov = _overlapped.Overlapped(NULL)
         wh = _overlapped.RegisterWaitWithQueue(
             handle, self._iocp, ov.address, ms)
-        f = _WaitHandleFuture(handle, wh, loop=self._loop)
+        f = _WaitHandleFuture(self._iocp, ov, handle, wh, loop=self._loop)
+        if f._source_traceback:
+            del f._source_traceback[-1]
 
         def finish_wait_for_handle(trans, key, ov):
             # Note that this second wait means that we should only use
@@ -406,12 +422,17 @@ class IocpProactor(object):
             # or semaphores are not.  Also note if the handle is
             # signalled and then quickly reset, then we may return
             # False even though we have not timed out.
+            return f._poll()
+
+        if f._poll():
             try:
-                return f._poll()
-            finally:
-                f._unregister()
+                result = f._poll()
+            except OSError as exc:
+                f.set_exception(exc)
+            else:
+                f.set_result(result)
 
-        self._cache[ov.address] = (f, ov, None, finish_wait_for_handle)
+        self._cache[ov.address] = (f, ov, 0, finish_wait_for_handle)
         return f
 
     def _register_with_iocp(self, obj):
@@ -430,6 +451,8 @@ class IocpProactor(object):
         # operation when it completes.  The future's value is actually
         # the value returned by callback().
         f = _OverlappedFuture(ov, loop=self._loop)
+        if f._source_traceback:
+            del f._source_traceback[-1]
         if not ov.pending and not wait_for_post:
             # The operation has completed, so no need to postpone the
             # work.  We cannot take this short cut if we need the