Merge heads

Lib/asyncio/base_events.py

@@ -179,6 +179,7 @@ class BaseEventLoop(events.AbstractEventLoop):
         # 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
+        self._current_handle = None
 
     def __repr__(self):
         return ('<%s running=%s closed=%s debug=%s>'
@@ -723,7 +724,13 @@ class BaseEventLoop(events.AbstractEventLoop):
                 logger.debug("Datagram endpoint remote_addr=%r created: "
                              "(%r, %r)",
                              remote_addr, transport, protocol)
-        yield from waiter
+
+        try:
+            yield from waiter
+        except:
+            transport.close()
+            raise
+
         return transport, protocol
 
     @coroutine
@@ -815,7 +822,13 @@ class BaseEventLoop(events.AbstractEventLoop):
         protocol = protocol_factory()
         waiter = futures.Future(loop=self)
         transport = self._make_read_pipe_transport(pipe, protocol, waiter)
-        yield from waiter
+
+        try:
+            yield from waiter
+        except:
+            transport.close()
+            raise
+
         if self._debug:
             logger.debug('Read pipe %r connected: (%r, %r)',
                          pipe.fileno(), transport, protocol)
@@ -826,7 +839,13 @@ class BaseEventLoop(events.AbstractEventLoop):
         protocol = protocol_factory()
         waiter = futures.Future(loop=self)
         transport = self._make_write_pipe_transport(pipe, protocol, waiter)
-        yield from waiter
+
+        try:
+            yield from waiter
+        except:
+            transport.close()
+            raise
+
         if self._debug:
             logger.debug('Write pipe %r connected: (%r, %r)',
                          pipe.fileno(), transport, protocol)
@@ -937,6 +956,10 @@ class BaseEventLoop(events.AbstractEventLoop):
         else:
             exc_info = False
 
+        if (self._current_handle is not None
+        and self._current_handle._source_traceback):
+            context['handle_traceback'] = self._current_handle._source_traceback
+
         log_lines = [message]
         for key in sorted(context):
             if key in {'message', 'exception'}:
@@ -946,6 +969,10 @@ class BaseEventLoop(events.AbstractEventLoop):
                 tb = ''.join(traceback.format_list(value))
                 value = 'Object created at (most recent call last):\n'
                 value += tb.rstrip()
+            elif key == 'handle_traceback':
+                tb = ''.join(traceback.format_list(value))
+                value = 'Handle created at (most recent call last):\n'
+                value += tb.rstrip()
             else:
                 value = repr(value)
             log_lines.append('{}: {}'.format(key, value))
@@ -1103,12 +1130,16 @@ class BaseEventLoop(events.AbstractEventLoop):
             if handle._cancelled:
                 continue
             if self._debug:
-                t0 = self.time()
-                handle._run()
-                dt = self.time() - t0
-                if dt >= self.slow_callback_duration:
-                    logger.warning('Executing %s took %.3f seconds',
-                                   _format_handle(handle), dt)
+                try:
+                    self._current_handle = handle
+                    t0 = self.time()
+                    handle._run()
+                    dt = self.time() - t0
+                    if dt >= self.slow_callback_duration:
+                        logger.warning('Executing %s took %.3f seconds',
+                                       _format_handle(handle), dt)
+                finally:
+                    self._current_handle = None
             else:
                 handle._run()
         handle = None  # Needed to break cycles when an exception occurs.

Lib/asyncio/windows_events.py

@@ -694,12 +694,7 @@ class IocpProactor:
     def close(self):
         # Cancel remaining registered operations.
         for address, (fut, ov, obj, callback) in list(self._cache.items()):
-            if obj is None:
-                # The operation was started with connect_pipe() which
-                # queues a task to Windows' thread pool.  This cannot
-                # be cancelled, so just forget it.
-                del self._cache[address]
-            elif fut.cancelled():
+            if fut.cancelled():
                 # Nothing to do with cancelled futures
                 pass
             elif isinstance(fut, _WaitCancelFuture):