merge

20 files changed, 390 insertions, 162 deletions

diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index e40d3ad..7108f25 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -26,6 +26,7 @@ import threading
 import time
 import traceback
 import sys
+import warnings
 
 from . import coroutines
 from . import events
@@ -333,6 +334,16 @@ class BaseEventLoop(events.AbstractEventLoop):
         """Returns True if the event loop was closed."""
         return self._closed
 
+    # On Python 3.3 and 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 sys.version_info >= (3, 4):
+        def __del__(self):
+            if not self.is_closed():
+                warnings.warn("unclosed event loop %r" % self, ResourceWarning)
+                if not self.is_running():
+                    self.close()
+
     def is_running(self):
         """Returns True if the event loop is running."""
         return (self._owner is not None)
diff --git a/Lib/asyncio/base_subprocess.py b/Lib/asyncio/base_subprocess.py
index 81c6f1a..02b9e89 100644
--- a/Lib/asyncio/base_subprocess.py
+++ b/Lib/asyncio/base_subprocess.py
@@ -1,6 +1,9 @@
 import collections
 import subprocess
+import sys
+import warnings
 
+from . import futures
 from . import protocols
 from . import transports
 from .coroutines import coroutine
@@ -11,26 +14,32 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
 
     def __init__(self, loop, protocol, args, shell,
                  stdin, stdout, stderr, bufsize,
-                 extra=None, **kwargs):
+                 waiter=None, extra=None, **kwargs):
         super().__init__(extra)
+        self._closed = False
         self._protocol = protocol
         self._loop = loop
+        self._proc = None
         self._pid = None
-
+        self._returncode = None
+        self._exit_waiters = []
+        self._pending_calls = collections.deque()
         self._pipes = {}
+        self._finished = False
+
         if stdin == subprocess.PIPE:
             self._pipes[0] = None
         if stdout == subprocess.PIPE:
             self._pipes[1] = None
         if stderr == subprocess.PIPE:
             self._pipes[2] = None
-        self._pending_calls = collections.deque()
-        self._finished = False
-        self._returncode = None
+
+        # Create the child process: set the _proc attribute
         self._start(args=args, shell=shell, stdin=stdin, stdout=stdout,
                     stderr=stderr, bufsize=bufsize, **kwargs)
         self._pid = self._proc.pid
         self._extra['subprocess'] = self._proc
+
         if self._loop.get_debug():
             if isinstance(args, (bytes, str)):
                 program = args
@@ -39,8 +48,13 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
             logger.debug('process %r created: pid %s',
                          program, self._pid)
 
+        self._loop.create_task(self._connect_pipes(waiter))
+
     def __repr__(self):
-        info = [self.__class__.__name__, 'pid=%s' % self._pid]
+        info = [self.__class__.__name__]
+        if self._closed:
+            info.append('closed')
+        info.append('pid=%s' % self._pid)
         if self._returncode is not None:
             info.append('returncode=%s' % self._returncode)
 
@@ -70,12 +84,34 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
         raise NotImplementedError
 
     def close(self):
+        if self._closed:
+            return
+        self._closed = True
+
         for proto in self._pipes.values():
             if proto is None:
                 continue
             proto.pipe.close()
-        if self._returncode is None:
-            self.terminate()
+
+        if self._proc is not None and self._returncode is None:
+            if self._loop.get_debug():
+                logger.warning('Close running child process: kill %r', self)
+
+            try:
+                self._proc.kill()
+            except ProcessLookupError:
+                pass
+
+            # Don't clear the _proc reference yet: _post_init() may still run
+
+    # On Python 3.3 and 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 sys.version_info >= (3, 4):
+        def __del__(self):
+            if not self._closed:
+                warnings.warn("unclosed transport %r" % self, ResourceWarning)
+                self.close()
 
     def get_pid(self):
         return self._pid
@@ -89,58 +125,40 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
         else:
             return None
 
+    def _check_proc(self):
+        if self._proc is None:
+            raise ProcessLookupError()
+
     def send_signal(self, signal):
+        self._check_proc()
         self._proc.send_signal(signal)
 
     def terminate(self):
+        self._check_proc()
         self._proc.terminate()
 
     def kill(self):
+        self._check_proc()
         self._proc.kill()
 
-    def _kill_wait(self):
-        """Close pipes, kill the subprocess and read its return status.
-
-        Function called when an exception is raised during the creation
-        of a subprocess.
-        """
-        if self._loop.get_debug():
-            logger.warning('Exception during subprocess creation, '
-                           'kill the subprocess %r',
-                           self,
-                           exc_info=True)
-
-        proc = self._proc
-        if proc.stdout:
-            proc.stdout.close()
-        if proc.stderr:
-            proc.stderr.close()
-        if proc.stdin:
-            proc.stdin.close()
-
-        try:
-            proc.kill()
-        except ProcessLookupError:
-            pass
-        self._returncode = proc.wait()
-
-        self.close()
-
     @coroutine
-    def _post_init(self):
+    def _connect_pipes(self, waiter):
         try:
             proc = self._proc
             loop = self._loop
+
             if proc.stdin is not None:
                 _, pipe = yield from loop.connect_write_pipe(
                     lambda: WriteSubprocessPipeProto(self, 0),
                     proc.stdin)
                 self._pipes[0] = pipe
+
             if proc.stdout is not None:
                 _, pipe = yield from loop.connect_read_pipe(
                     lambda: ReadSubprocessPipeProto(self, 1),
                     proc.stdout)
                 self._pipes[1] = pipe
+
             if proc.stderr is not None:
                 _, pipe = yield from loop.connect_read_pipe(
                     lambda: ReadSubprocessPipeProto(self, 2),
@@ -149,13 +167,16 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
 
             assert self._pending_calls is not None
 
-            self._loop.call_soon(self._protocol.connection_made, self)
+            loop.call_soon(self._protocol.connection_made, self)
             for callback, data in self._pending_calls:
-                self._loop.call_soon(callback, *data)
+                loop.call_soon(callback, *data)
             self._pending_calls = None
-        except:
-            self._kill_wait()
-            raise
+        except Exception as exc:
+            if waiter is not None and not waiter.cancelled():
+                waiter.set_exception(exc)
+        else:
+            if waiter is not None and not waiter.cancelled():
+                waiter.set_result(None)
 
     def _call(self, cb, *data):
         if self._pending_calls is not None:
@@ -180,6 +201,23 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
         self._call(self._protocol.process_exited)
         self._try_finish()
 
+        # wake up futures waiting for wait()
+        for waiter in self._exit_waiters:
+            if not waiter.cancelled():
+                waiter.set_result(returncode)
+        self._exit_waiters = None
+
+    def _wait(self):
+        """Wait until the process exit and return the process return code.
+
+        This method is a coroutine."""
+        if self._returncode is not None:
+            return self._returncode
+
+        waiter = futures.Future(loop=self._loop)
+        self._exit_waiters.append(waiter)
+        return (yield from waiter)
+
     def _try_finish(self):
         assert not self._finished
         if self._returncode is None:
@@ -193,9 +231,9 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
         try:
             self._protocol.connection_lost(exc)
         finally:
+            self._loop = None
             self._proc = None
             self._protocol = None
-            self._loop = None
 
 
 class WriteSubprocessPipeProto(protocols.BaseProtocol):
diff --git a/Lib/asyncio/futures.py b/Lib/asyncio/futures.py
index 19212a9..2c741fd 100644
--- a/Lib/asyncio/futures.py
+++ b/Lib/asyncio/futures.py
@@ -195,9 +195,9 @@ class Future:
         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
-    # cycle are never destroyed. It's not more the case on Python 3.4 thanks to
-    # the PEP 442.
+    # On Python 3.3 and 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:
diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
index 0f533a5..65de926 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -7,6 +7,8 @@ proactor is only implemented on Windows with IOCP.
 __all__ = ['BaseProactorEventLoop']
 
 import socket
+import sys
+import warnings
 
 from . import base_events
 from . import constants
@@ -74,6 +76,15 @@ class _ProactorBasePipeTransport(transports._FlowControlMixin,
             self._read_fut.cancel()
             self._read_fut = None
 
+    # On Python 3.3 and 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 sys.version_info >= (3, 4):
+        def __del__(self):
+            if self._sock is not None:
+                warnings.warn("unclosed transport %r" % self, ResourceWarning)
+                self.close()
+
     def _fatal_error(self, exc, message='Fatal error on pipe transport'):
         if isinstance(exc, (BrokenPipeError, ConnectionResetError)):
             if self._loop.get_debug():
diff --git a/Lib/asyncio/protocols.py b/Lib/asyncio/protocols.py
index 52fc25c..80fcac9 100644
--- a/Lib/asyncio/protocols.py
+++ b/Lib/asyncio/protocols.py
@@ -78,6 +78,11 @@ class Protocol(BaseProtocol):
     State machine of calls:
 
       start -> CM [-> DR*] [-> ER?] -> CL -> end
+
+    * CM: connection_made()
+    * DR: data_received()
+    * ER: eof_received()
+    * CL: connection_lost()
     """
 
     def data_received(self, data):
diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
index f499629..4bd6dc8 100644
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -10,6 +10,8 @@ import collections
 import errno
 import functools
 import socket
+import sys
+import warnings
 try:
     import ssl
 except ImportError:  # pragma: no cover
@@ -22,6 +24,7 @@ from . import futures
 from . import selectors
 from . import transports
 from . import sslproto
+from .coroutines import coroutine
 from .log import logger
 
 
@@ -181,16 +184,47 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
             else:
                 raise  # The event loop will catch, log and ignore it.
         else:
+            extra = {'peername': addr}
+            accept = self._accept_connection2(protocol_factory, conn, extra,
+                                              sslcontext, server)
+            self.create_task(accept)
+
+    @coroutine
+    def _accept_connection2(self, protocol_factory, conn, extra,
+                            sslcontext=None, server=None):
+        protocol = None
+        transport = None
+        try:
             protocol = protocol_factory()
+            waiter = futures.Future(loop=self)
             if sslcontext:
-                self._make_ssl_transport(
-                    conn, protocol, sslcontext,
-                    server_side=True, extra={'peername': addr}, server=server)
+                transport = self._make_ssl_transport(
+                    conn, protocol, sslcontext, waiter=waiter,
+                    server_side=True, extra=extra, server=server)
             else:
-                self._make_socket_transport(
-                    conn, protocol , extra={'peername': addr},
+                transport = self._make_socket_transport(
+                    conn, protocol, waiter=waiter, extra=extra,
                     server=server)
-        # It's now up to the protocol to handle the connection.
+
+            try:
+                yield from waiter
+            except:
+                transport.close()
+                raise
+
+            # It's now up to the protocol to handle the connection.
+        except Exception as exc:
+            if self.get_debug():
+                context = {
+                    'message': ('Error on transport creation '
+                                'for incoming connection'),
+                    'exception': exc,
+                }
+                if protocol is not None:
+                    context['protocol'] = protocol
+                if transport is not None:
+                    context['transport'] = transport
+                self.call_exception_handler(context)
 
     def add_reader(self, fd, callback, *args):
         """Add a reader callback."""
@@ -467,7 +501,12 @@ class _SelectorTransport(transports._FlowControlMixin,
 
     _buffer_factory = bytearray  # Constructs initial value for self._buffer.
 
-    def __init__(self, loop, sock, protocol, extra, server=None):
+    # Attribute used in the destructor: it must be set even if the constructor
+    # is not called (see _SelectorSslTransport which may start by raising an
+    # exception)
+    _sock = None
+
+    def __init__(self, loop, sock, protocol, extra=None, server=None):
         super().__init__(extra, loop)
         self._extra['socket'] = sock
         self._extra['sockname'] = sock.getsockname()
@@ -479,6 +518,7 @@ class _SelectorTransport(transports._FlowControlMixin,
         self._sock = sock
         self._sock_fd = sock.fileno()
         self._protocol = protocol
+        self._protocol_connected = True
         self._server = server
         self._buffer = self._buffer_factory()
         self._conn_lost = 0  # Set when call to connection_lost scheduled.
@@ -526,6 +566,15 @@ class _SelectorTransport(transports._FlowControlMixin,
             self._conn_lost += 1
             self._loop.call_soon(self._call_connection_lost, None)
 
+    # On Python 3.3 and 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 sys.version_info >= (3, 4):
+        def __del__(self):
+            if self._sock is not None:
+                warnings.warn("unclosed transport %r" % self, ResourceWarning)
+                self._sock.close()
+
     def _fatal_error(self, exc, message='Fatal error on transport'):
         # Should be called from exception handler only.
         if isinstance(exc, (BrokenPipeError,
@@ -555,7 +604,8 @@ class _SelectorTransport(transports._FlowControlMixin,
 
     def _call_connection_lost(self, exc):
         try:
-            self._protocol.connection_lost(exc)
+            if self._protocol_connected:
+                self._protocol.connection_lost(exc)
         finally:
             self._sock.close()
             self._sock = None
@@ -718,6 +768,8 @@ class _SelectorSslTransport(_SelectorTransport):
         sslsock = sslcontext.wrap_socket(rawsock, **wrap_kwargs)
 
         super().__init__(loop, sslsock, protocol, extra, server)
+        # the protocol connection is only made after the SSL handshake
+        self._protocol_connected = False
 
         self._server_hostname = server_hostname
         self._waiter = waiter
@@ -797,6 +849,7 @@ class _SelectorSslTransport(_SelectorTransport):
         self._read_wants_write = False
         self._write_wants_read = False
         self._loop.add_reader(self._sock_fd, self._read_ready)
+        self._protocol_connected = True
         self._loop.call_soon(self._protocol.connection_made, self)
         # only wake up the waiter when connection_made() has been called
         self._loop.call_soon(self._wakeup_waiter)
@@ -928,8 +981,10 @@ class _SelectorDatagramTransport(_SelectorTransport):
                  waiter=None, extra=None):
         super().__init__(loop, sock, protocol, extra)
         self._address = address
-        self._loop.add_reader(self._sock_fd, self._read_ready)
         self._loop.call_soon(self._protocol.connection_made, self)
+        # only start reading when connection_made() has been called
+        self._loop.call_soon(self._loop.add_reader,
+                             self._sock_fd, self._read_ready)
         if waiter is not None:
             # only wake up the waiter when connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
diff --git a/Lib/asyncio/sslproto.py b/Lib/asyncio/sslproto.py
index fc809b9..235855e 100644
--- a/Lib/asyncio/sslproto.py
+++ b/Lib/asyncio/sslproto.py
@@ -1,4 +1,6 @@
 import collections
+import sys
+import warnings
 try:
     import ssl
 except ImportError:  # pragma: no cover
@@ -295,6 +297,7 @@ class _SSLProtocolTransport(transports._FlowControlMixin,
         self._loop = loop
         self._ssl_protocol = ssl_protocol
         self._app_protocol = app_protocol
+        self._closed = False
 
     def get_extra_info(self, name, default=None):
         """Get optional transport information."""
@@ -308,8 +311,18 @@ class _SSLProtocolTransport(transports._FlowControlMixin,
         protocol's connection_lost() method will (eventually) called
         with None as its argument.
         """
+        self._closed = True
         self._ssl_protocol._start_shutdown()
 
+    # On Python 3.3 and 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 sys.version_info >= (3, 4):
+        def __del__(self):
+            if not self._closed:
+                warnings.warn("unclosed transport %r" % self, ResourceWarning)
+                self.close()
+
     def pause_reading(self):
         """Pause the receiving end.
 
diff --git a/Lib/asyncio/subprocess.py b/Lib/asyncio/subprocess.py
index c848a21..4600a9f 100644
--- a/Lib/asyncio/subprocess.py
+++ b/Lib/asyncio/subprocess.py
@@ -25,8 +25,6 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
         super().__init__(loop=loop)
         self._limit = limit
         self.stdin = self.stdout = self.stderr = None
-        self.waiter = futures.Future(loop=loop)
-        self._waiters = collections.deque()
         self._transport = None
 
     def __repr__(self):
@@ -61,9 +59,6 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
                                               reader=None,
                                               loop=self._loop)
 
-        if not self.waiter.cancelled():
-            self.waiter.set_result(None)
-
     def pipe_data_received(self, fd, data):
         if fd == 1:
             reader = self.stdout
@@ -94,16 +89,9 @@ class SubprocessStreamProtocol(streams.FlowControlMixin,
                 reader.set_exception(exc)
 
     def process_exited(self):
-        returncode = self._transport.get_returncode()
         self._transport.close()
         self._transport = None
 
-        # wake up futures waiting for wait()
-        while self._waiters:
-            waiter = self._waiters.popleft()
-            if not waiter.cancelled():
-                waiter.set_result(returncode)
-
 
 class Process:
     def __init__(self, transport, protocol, loop):
@@ -124,30 +112,18 @@ class Process:
 
     @coroutine
     def wait(self):
-        """Wait until the process exit and return the process return code."""
-        returncode = self._transport.get_returncode()
-        if returncode is not None:
-            return returncode
-
-        waiter = futures.Future(loop=self._loop)
-        self._protocol._waiters.append(waiter)
-        yield from waiter
-        return waiter.result()
+        """Wait until the process exit and return the process return code.
 
-    def _check_alive(self):
-        if self._transport.get_returncode() is not None:
-            raise ProcessLookupError()
+        This method is a coroutine."""
+        return (yield from self._transport._wait())
 
     def send_signal(self, signal):
-        self._check_alive()
         self._transport.send_signal(signal)
 
     def terminate(self):
-        self._check_alive()
         self._transport.terminate()
 
     def kill(self):
-        self._check_alive()
         self._transport.kill()
 
     @coroutine
@@ -221,11 +197,6 @@ def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
                                             protocol_factory,
                                             cmd, stdin=stdin, stdout=stdout,
                                             stderr=stderr, **kwds)
-    try:
-        yield from protocol.waiter
-    except:
-        transport._kill_wait()
-        raise
     return Process(transport, protocol, loop)
 
 @coroutine
@@ -241,9 +212,4 @@ def create_subprocess_exec(program, *args, stdin=None, stdout=None,
                                             program, *args,
                                             stdin=stdin, stdout=stdout,
                                             stderr=stderr, **kwds)
-    try:
-        yield from protocol.waiter
-    except:
-        transport._kill_wait()
-        raise
     return Process(transport, protocol, loop)
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index 63412a9..4f19a25 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -592,7 +592,7 @@ def gather(*coros_or_futures, loop=None, return_exceptions=False):
                 fut.exception()
             return
 
-        if fut._state == futures._CANCELLED:
+        if fut.cancelled():
             res = futures.CancelledError()
             if not return_exceptions:
                 outer.set_exception(res)
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index 67973f1..1fc39ab 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -8,6 +8,7 @@ import stat
 import subprocess
 import sys
 import threading
+import warnings
 
 
 from . import base_events
@@ -15,6 +16,7 @@ from . import base_subprocess
 from . import constants
 from . import coroutines
 from . import events
+from . import futures
 from . import selector_events
 from . import selectors
 from . import transports
@@ -174,16 +176,20 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
                                    stdin, stdout, stderr, bufsize,
                                    extra=None, **kwargs):
         with events.get_child_watcher() as watcher:
+            waiter = futures.Future(loop=self)
             transp = _UnixSubprocessTransport(self, protocol, args, shell,
                                               stdin, stdout, stderr, bufsize,
-                                              extra=extra, **kwargs)
+                                              waiter=waiter, extra=extra,
+                                              **kwargs)
+
+            watcher.add_child_handler(transp.get_pid(),
+                                      self._child_watcher_callback, transp)
             try:
-                yield from transp._post_init()
+                yield from waiter
             except:
                 transp.close()
+                yield from transp._wait()
                 raise
-            watcher.add_child_handler(transp.get_pid(),
-                                      self._child_watcher_callback, transp)
 
         return transp
 
@@ -298,8 +304,10 @@ class _UnixReadPipeTransport(transports.ReadTransport):
         _set_nonblocking(self._fileno)
         self._protocol = protocol
         self._closing = False
-        self._loop.add_reader(self._fileno, self._read_ready)
         self._loop.call_soon(self._protocol.connection_made, self)
+        # only start reading when connection_made() has been called
+        self._loop.call_soon(self._loop.add_reader,
+                             self._fileno, self._read_ready)
         if waiter is not None:
             # only wake up the waiter when connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
@@ -351,6 +359,15 @@ class _UnixReadPipeTransport(transports.ReadTransport):
         if not self._closing:
             self._close(None)
 
+    # On Python 3.3 and 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 sys.version_info >= (3, 4):
+        def __del__(self):
+            if self._pipe is not None:
+                warnings.warn("unclosed transport %r" % self, ResourceWarning)
+                self._pipe.close()
+
     def _fatal_error(self, exc, message='Fatal error on pipe transport'):
         # should be called by exception handler only
         if (isinstance(exc, OSError) and exc.errno == errno.EIO):
@@ -401,13 +418,16 @@ class _UnixWritePipeTransport(transports._FlowControlMixin,
         self._conn_lost = 0
         self._closing = False  # Set when close() or write_eof() called.
 
-        # On AIX, the reader trick only works for sockets.
-        # On other platforms it works for pipes and sockets.
-        # (Exception: OS X 10.4?  Issue #19294.)
+        self._loop.call_soon(self._protocol.connection_made, self)
+
+        # On AIX, the reader trick (to be notified when the read end of the
+        # socket is closed) only works for sockets. On other platforms it
+        # works for pipes and sockets. (Exception: OS X 10.4?  Issue #19294.)
         if is_socket or not sys.platform.startswith("aix"):
-            self._loop.add_reader(self._fileno, self._read_ready)
+            # only start reading when connection_made() has been called
+            self._loop.call_soon(self._loop.add_reader,
+                                 self._fileno, self._read_ready)
 
-        self._loop.call_soon(self._protocol.connection_made, self)
         if waiter is not None:
             # only wake up the waiter when connection_made() has been called
             self._loop.call_soon(waiter._set_result_unless_cancelled, None)
@@ -524,6 +544,15 @@ class _UnixWritePipeTransport(transports._FlowControlMixin,
             # write_eof is all what we needed to close the write pipe
             self.write_eof()
 
+    # On Python 3.3 and 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 sys.version_info >= (3, 4):
+        def __del__(self):
+            if self._pipe is not None:
+                warnings.warn("unclosed transport %r" % self, ResourceWarning)
+                self._pipe.close()
+
     def abort(self):
         self._close(None)
 
@@ -750,7 +779,7 @@ class SafeChildWatcher(BaseChildWatcher):
         pass
 
     def add_child_handler(self, pid, callback, *args):
-        self._callbacks[pid] = callback, args
+        self._callbacks[pid] = (callback, args)
 
         # Prevent a race condition in case the child is already terminated.
         self._do_waitpid(pid)
diff --git a/Lib/asyncio/windows_events.py b/Lib/asyncio/windows_events.py
index 94aafb6..c4bffc4 100644
--- a/Lib/asyncio/windows_events.py
+++ b/Lib/asyncio/windows_events.py
@@ -366,13 +366,16 @@ class ProactorEventLoop(proactor_events.BaseProactorEventLoop):
     def _make_subprocess_transport(self, protocol, args, shell,
                                    stdin, stdout, stderr, bufsize,
                                    extra=None, **kwargs):
+        waiter = futures.Future(loop=self)
         transp = _WindowsSubprocessTransport(self, protocol, args, shell,
                                              stdin, stdout, stderr, bufsize,
-                                             extra=extra, **kwargs)
+                                             waiter=waiter, extra=extra,
+                                             **kwargs)
         try:
-            yield from transp._post_init()
+            yield from waiter
         except:
             transp.close()
+            yield from transp._wait()
             raise
 
         return transp
diff --git a/Lib/asyncio/windows_utils.py b/Lib/asyncio/windows_utils.py
index 5f8327e..870cd13 100644
--- a/Lib/asyncio/windows_utils.py
+++ b/Lib/asyncio/windows_utils.py
@@ -14,6 +14,7 @@ import os
 import socket
 import subprocess
 import tempfile
+import warnings
 
 
 __all__ = ['socketpair', 'pipe', 'Popen', 'PIPE', 'PipeHandle']
@@ -156,7 +157,10 @@ class PipeHandle:
             CloseHandle(self._handle)
             self._handle = None
 
-    __del__ = close
+    def __del__(self):
+        if self._handle is not None:
+            warnings.warn("unclosed %r" % self, ResourceWarning)
+            self.close()
 
     def __enter__(self):
         return self
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index a38c90e..4b957d8 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -886,13 +886,18 @@ class EventLoopTestsMixin:
         if hasattr(sslcontext_client, 'check_hostname'):
             sslcontext_client.check_hostname = True
 
+
         # no CA loaded
         f_c = self.loop.create_connection(MyProto, host, port,
                                           ssl=sslcontext_client)
-        with test_utils.disable_logger():
-            with self.assertRaisesRegex(ssl.SSLError,
-                                        'certificate verify failed '):
-                self.loop.run_until_complete(f_c)
+        with mock.patch.object(self.loop, 'call_exception_handler'):
+            with test_utils.disable_logger():
+                with self.assertRaisesRegex(ssl.SSLError,
+                                            'certificate verify failed '):
+                    self.loop.run_until_complete(f_c)
+
+            # execute the loop to log the connection error
+            test_utils.run_briefly(self.loop)
 
         # close connection
         self.assertIsNone(proto.transport)
@@ -919,15 +924,20 @@ class EventLoopTestsMixin:
         f_c = self.loop.create_unix_connection(MyProto, path,
                                                ssl=sslcontext_client,
                                                server_hostname='invalid')
-        with test_utils.disable_logger():
-            with self.assertRaisesRegex(ssl.SSLError,
-                                        'certificate verify failed '):
-                self.loop.run_until_complete(f_c)
+        with mock.patch.object(self.loop, 'call_exception_handler'):
+            with test_utils.disable_logger():
+                with self.assertRaisesRegex(ssl.SSLError,
+                                            'certificate verify failed '):
+                    self.loop.run_until_complete(f_c)
+
+            # execute the loop to log the connection error
+            test_utils.run_briefly(self.loop)
 
         # close connection
         self.assertIsNone(proto.transport)
         server.close()
 
+
     def test_legacy_create_unix_server_ssl_verify_failed(self):
         with test_utils.force_legacy_ssl_support():
             self.test_create_unix_server_ssl_verify_failed()
@@ -949,11 +959,12 @@ class EventLoopTestsMixin:
         # incorrect server_hostname
         f_c = self.loop.create_connection(MyProto, host, port,
                                           ssl=sslcontext_client)
-        with test_utils.disable_logger():
-            with self.assertRaisesRegex(
-                    ssl.CertificateError,
-                    "hostname '127.0.0.1' doesn't match 'localhost'"):
-                self.loop.run_until_complete(f_c)
+        with mock.patch.object(self.loop, 'call_exception_handler'):
+            with test_utils.disable_logger():
+                with self.assertRaisesRegex(
+                        ssl.CertificateError,
+                        "hostname '127.0.0.1' doesn't match 'localhost'"):
+                    self.loop.run_until_complete(f_c)
 
         # close connection
         proto.transport.close()
@@ -1540,9 +1551,10 @@ class SubprocessTestsMixin:
         stdin = transp.get_pipe_transport(0)
         stdin.write(b'Python The Winner')
         self.loop.run_until_complete(proto.got_data[1].wait())
-        transp.close()
+        with test_utils.disable_logger():
+            transp.close()
         self.loop.run_until_complete(proto.completed)
-        self.check_terminated(proto.returncode)
+        self.check_killed(proto.returncode)
         self.assertEqual(b'Python The Winner', proto.data[1])
 
     def test_subprocess_interactive(self):
@@ -1556,21 +1568,20 @@ class SubprocessTestsMixin:
         self.loop.run_until_complete(proto.connected)
         self.assertEqual('CONNECTED', proto.state)
 
-        try:
-            stdin = transp.get_pipe_transport(0)
-            stdin.write(b'Python ')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            proto.got_data[1].clear()
-            self.assertEqual(b'Python ', proto.data[1])
-
-            stdin.write(b'The Winner')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            self.assertEqual(b'Python The Winner', proto.data[1])
-        finally:
-            transp.close()
+        stdin = transp.get_pipe_transport(0)
+        stdin.write(b'Python ')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        proto.got_data[1].clear()
+        self.assertEqual(b'Python ', proto.data[1])
 
+        stdin.write(b'The Winner')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        self.assertEqual(b'Python The Winner', proto.data[1])
+
+        with test_utils.disable_logger():
+            transp.close()
         self.loop.run_until_complete(proto.completed)
-        self.check_terminated(proto.returncode)
+        self.check_killed(proto.returncode)
 
     def test_subprocess_shell(self):
         connect = self.loop.subprocess_shell(
@@ -1728,9 +1739,10 @@ class SubprocessTestsMixin:
             # GetLastError()==ERROR_INVALID_NAME on Windows!?!  (Using
             # WriteFile() we get ERROR_BROKEN_PIPE as expected.)
             self.assertEqual(b'ERR:OSError', proto.data[2])
-        transp.close()
+        with test_utils.disable_logger():
+            transp.close()
         self.loop.run_until_complete(proto.completed)
-        self.check_terminated(proto.returncode)
+        self.check_killed(proto.returncode)
 
     def test_subprocess_wait_no_same_group(self):
         # start the new process in a new session
diff --git a/Lib/test/test_asyncio/test_proactor_events.py b/Lib/test/test_asyncio/test_proactor_events.py
index 33a8a67..fcd9ab1 100644
--- a/Lib/test/test_asyncio/test_proactor_events.py
+++ b/Lib/test/test_asyncio/test_proactor_events.py
@@ -499,8 +499,12 @@ class BaseProactorEventLoopTests(test_utils.TestCase):
         self.proactor.accept.assert_called_with(self.sock)
 
     def test_socketpair(self):
+        class EventLoop(BaseProactorEventLoop):
+            # override the destructor to not log a ResourceWarning
+            def __del__(self):
+                pass
         self.assertRaises(
-            NotImplementedError, BaseProactorEventLoop, self.proactor)
+            NotImplementedError, EventLoop, self.proactor)
 
     def test_make_socket_transport(self):
         tr = self.loop._make_socket_transport(self.sock, asyncio.Protocol())
diff --git a/Lib/test/test_asyncio/test_selector_events.py b/Lib/test/test_asyncio/test_selector_events.py
index 5152616..f64e40d 100644
--- a/Lib/test/test_asyncio/test_selector_events.py
+++ b/Lib/test/test_asyncio/test_selector_events.py
@@ -1427,7 +1427,7 @@ class SelectorSslTransportTests(test_utils.TestCase):
         self.assertFalse(tr.can_write_eof())
         self.assertRaises(NotImplementedError, tr.write_eof)
 
-    def test_close(self):
+    def check_close(self):
         tr = self._make_one()
         tr.close()
 
@@ -1439,6 +1439,19 @@ class SelectorSslTransportTests(test_utils.TestCase):
         self.assertEqual(tr._conn_lost, 1)
         self.assertEqual(1, self.loop.remove_reader_count[1])
 
+        test_utils.run_briefly(self.loop)
+
+    def test_close(self):
+        self.check_close()
+        self.assertTrue(self.protocol.connection_made.called)
+        self.assertTrue(self.protocol.connection_lost.called)
+
+    def test_close_not_connected(self):
+        self.sslsock.do_handshake.side_effect = ssl.SSLWantReadError
+        self.check_close()
+        self.assertFalse(self.protocol.connection_made.called)
+        self.assertFalse(self.protocol.connection_lost.called)
+
     @unittest.skipIf(ssl is None, 'No SSL support')
     def test_server_hostname(self):
         self.ssl_transport(server_hostname='localhost')
diff --git a/Lib/test/test_asyncio/test_sslproto.py b/Lib/test/test_asyncio/test_sslproto.py
index 148e30d..a72967e 100644
--- a/Lib/test/test_asyncio/test_sslproto.py
+++ b/Lib/test/test_asyncio/test_sslproto.py
@@ -22,7 +22,9 @@ class SslProtoHandshakeTests(test_utils.TestCase):
     def ssl_protocol(self, waiter=None):
         sslcontext = test_utils.dummy_ssl_context()
         app_proto = asyncio.Protocol()
-        return sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter)
+        proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter)
+        self.addCleanup(proto._app_transport.close)
+        return proto
 
     def connection_made(self, ssl_proto, do_handshake=None):
         transport = mock.Mock()
@@ -56,9 +58,6 @@ class SslProtoHandshakeTests(test_utils.TestCase):
         with test_utils.disable_logger():
             self.loop.run_until_complete(handshake_fut)
 
-        # Close the transport
-        ssl_proto._app_transport.close()
-
     def test_eof_received_waiter(self):
         waiter = asyncio.Future(loop=self.loop)
         ssl_proto = self.ssl_protocol(waiter)
diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py
index ecc2c9d..b467b04 100644
--- a/Lib/test/test_asyncio/test_subprocess.py
+++ b/Lib/test/test_asyncio/test_subprocess.py
@@ -4,6 +4,7 @@ import unittest
 from unittest import mock
 
 import asyncio
+from asyncio import base_subprocess
 from asyncio import subprocess
 from asyncio import test_utils
 try:
@@ -23,6 +24,56 @@ PROGRAM_CAT = [
               'data = sys.stdin.buffer.read()',
               'sys.stdout.buffer.write(data)'))]
 
+class TestSubprocessTransport(base_subprocess.BaseSubprocessTransport):
+    def _start(self, *args, **kwargs):
+        self._proc = mock.Mock()
+        self._proc.stdin = None
+        self._proc.stdout = None
+        self._proc.stderr = None
+
+
+class SubprocessTransportTests(test_utils.TestCase):
+    def setUp(self):
+        self.loop = self.new_test_loop()
+        self.set_event_loop(self.loop)
+
+
+    def create_transport(self, waiter=None):
+        protocol = mock.Mock()
+        protocol.connection_made._is_coroutine = False
+        protocol.process_exited._is_coroutine = False
+        transport = TestSubprocessTransport(
+                        self.loop, protocol, ['test'], False,
+                        None, None, None, 0, waiter=waiter)
+        return (transport, protocol)
+
+    def test_proc_exited(self):
+        waiter = asyncio.Future(loop=self.loop)
+        transport, protocol = self.create_transport(waiter)
+        transport._process_exited(6)
+        self.loop.run_until_complete(waiter)
+
+        self.assertEqual(transport.get_returncode(), 6)
+
+        self.assertTrue(protocol.connection_made.called)
+        self.assertTrue(protocol.process_exited.called)
+        self.assertTrue(protocol.connection_lost.called)
+        self.assertEqual(protocol.connection_lost.call_args[0], (None,))
+
+        self.assertFalse(transport._closed)
+        self.assertIsNone(transport._loop)
+        self.assertIsNone(transport._proc)
+        self.assertIsNone(transport._protocol)
+
+        # methods must raise ProcessLookupError if the process exited
+        self.assertRaises(ProcessLookupError,
+                          transport.send_signal, signal.SIGTERM)
+        self.assertRaises(ProcessLookupError, transport.terminate)
+        self.assertRaises(ProcessLookupError, transport.kill)
+
+        transport.close()
+
+
 class SubprocessMixin:
 
     def test_stdin_stdout(self):
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
index 126196d..41249ff 100644
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/Lib/test/test_asyncio/test_unix_events.py
@@ -350,16 +350,13 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
         return transport
 
     def test_ctor(self):
-        tr = self.read_pipe_transport()
-        self.loop.assert_reader(5, tr._read_ready)
-        test_utils.run_briefly(self.loop)
-        self.protocol.connection_made.assert_called_with(tr)
+        waiter = asyncio.Future(loop=self.loop)
+        tr = self.read_pipe_transport(waiter=waiter)
+        self.loop.run_until_complete(waiter)
 
-    def test_ctor_with_waiter(self):
-        fut = asyncio.Future(loop=self.loop)
-        tr = self.read_pipe_transport(waiter=fut)
-        test_utils.run_briefly(self.loop)
-        self.assertIsNone(fut.result())
+        self.protocol.connection_made.assert_called_with(tr)
+        self.loop.assert_reader(5, tr._read_ready)
+        self.assertIsNone(waiter.result())
 
     @mock.patch('os.read')
     def test__read_ready(self, m_read):
@@ -502,17 +499,13 @@ class UnixWritePipeTransportTests(test_utils.TestCase):
         return transport
 
     def test_ctor(self):
-        tr = self.write_pipe_transport()
-        self.loop.assert_reader(5, tr._read_ready)
-        test_utils.run_briefly(self.loop)
-        self.protocol.connection_made.assert_called_with(tr)
+        waiter = asyncio.Future(loop=self.loop)
+        tr = self.write_pipe_transport(waiter=waiter)
+        self.loop.run_until_complete(waiter)
 
-    def test_ctor_with_waiter(self):
-        fut = asyncio.Future(loop=self.loop)
-        tr = self.write_pipe_transport(waiter=fut)
+        self.protocol.connection_made.assert_called_with(tr)
         self.loop.assert_reader(5, tr._read_ready)
-        test_utils.run_briefly(self.loop)
-        self.assertEqual(None, fut.result())
+        self.assertEqual(None, waiter.result())
 
     def test_can_write_eof(self):
         tr = self.write_pipe_transport()
diff --git a/Lib/test/test_memoryview.py b/Lib/test/test_memoryview.py
index e7df8a7..e9bd9fd 100644
--- a/Lib/test/test_memoryview.py
+++ b/Lib/test/test_memoryview.py
@@ -360,6 +360,27 @@ class AbstractMemoryTests:
             self.assertEqual(list(reversed(m)), aslist)
             self.assertEqual(list(reversed(m)), list(m[::-1]))
 
+    def test_issue22668(self):
+        a = array.array('H', [256, 256, 256, 256])
+        x = memoryview(a)
+        m = x.cast('B')
+        b = m.cast('H')
+        c = b[0:2]
+        d = memoryview(b)
+
+        del b
+
+        self.assertEqual(c[0], 256)
+        self.assertEqual(d[0], 256)
+        self.assertEqual(c.format, "H")
+        self.assertEqual(d.format, "H")
+
+        _ = m.cast('I')
+        self.assertEqual(c[0], 256)
+        self.assertEqual(d[0], 256)
+        self.assertEqual(c.format, "H")
+        self.assertEqual(d.format, "H")
+
 
 # Variations on source objects for the buffer: bytes-like objects, then arrays
 # with itemsize > 1.
diff --git a/Lib/test/test_sys.py b/Lib/test/test_sys.py
index bd2d204..edb9c10 100644
--- a/Lib/test/test_sys.py
+++ b/Lib/test/test_sys.py
@@ -965,7 +965,7 @@ class SizeofTest(unittest.TestCase):
         check(int(PyLong_BASE**2-1), vsize('') + 2*self.longdigit)
         check(int(PyLong_BASE**2), vsize('') + 3*self.longdigit)
         # memoryview
-        check(memoryview(b''), size('Pnin 2P2n2i5P 3cPn'))
+        check(memoryview(b''), size('Pnin 2P2n2i5P Pn'))
         # module
         check(unittest, size('PnPPP'))
         # None