Merge branch '2.3'

CHANGES.rst

@@ -14,6 +14,17 @@ Changelog
 
 .. towncrier release notes start
 
+2.3.4 (2017-11-29)
+==================
+
+- Make `request.app` point to proper application instance when using nested
+  applications (with middlewares). (#2550)
+- Change base class of ClientConnectorSSLError to ClientSSLError from
+  ClientConnectorError. (#2563)
+- Return client connection back to free pool on error in `connector.connect()`.
+  (#2567)
+
+
 2.3.3 (2017-11-17)
 ==================
 

aiohttp/connector.py

@@ -282,6 +282,15 @@ def _cleanup(self):
             self._cleanup_handle = helpers.weakref_handle(
                 self, '_cleanup', timeout, self._loop)
 
+    def _drop_acquired_per_host(self, key, val):
+        acquired_per_host = self._acquired_per_host
+        if key not in acquired_per_host:
+            return
+        conns = acquired_per_host[key]
+        conns.remove(val)
+        if not conns:
+            del self._acquired_per_host[key]
+
     def _cleanup_closed(self):
         """Double confirmation for transport close.
         Some broken ssl servers may leave socket open without proper close.
@@ -352,7 +361,7 @@ async def connect(self, req, traces=None):
 
         if self._limit:
             # total calc available connections
-            available = self._limit - len(self._waiters) - len(self._acquired)
+            available = self._limit - len(self._acquired)
 
             # check limit per host
             if (self._limit_per_host and available > 0 and
@@ -411,15 +420,16 @@ async def connect(self, req, traces=None):
                     raise ClientConnectionError("Connector is closed.")
             except Exception:
                 # signal to waiter
-                for waiter in self._waiters[key]:
-                    if not waiter.done():
-                        waiter.set_result(None)
-                        break
+                if key in self._waiters:
+                    for waiter in self._waiters[key]:
+                        if not waiter.done():
+                            waiter.set_result(None)
+                            break
                 raise
             finally:
                 if not self._closed:
                     self._acquired.remove(placeholder)
-                    self._acquired_per_host[key].remove(placeholder)
+                    self._drop_acquired_per_host(key, placeholder)
 
             if traces:
                 for trace in traces:
@@ -486,9 +496,7 @@ def _release_acquired(self, key, proto):
 
         try:
             self._acquired.remove(proto)
-            self._acquired_per_host[key].remove(proto)
-            if not self._acquired_per_host[key]:
-                del self._acquired_per_host[key]
+            self._drop_acquired_per_host(key, proto)
         except KeyError:  # pragma: no cover
             # this may be result of undetermenistic order of objects
             # finalization due garbage collection.

tests/test_client_functional.py

@@ -2481,3 +2481,57 @@ def connection_lost(self, exc):
     connector.close()
     server.close()
     await server.wait_closed()
+
+
+async def test_error_in_performing_request(loop, ssl_ctx,
+                                           test_client, test_server):
+    async def handler(request):
+        return web.Response()
+
+    app = web.Application()
+    app.router.add_route('GET', '/', handler)
+
+    server = await test_server(app, ssl=ssl_ctx)
+
+    conn = aiohttp.TCPConnector(limit=1, loop=loop)
+    client = await test_client(server, connector=conn)
+
+    with pytest.raises(aiohttp.ClientConnectionError):
+        await client.get('/')
+
+    # second try should not hang
+    with pytest.raises(aiohttp.ClientConnectionError):
+        await client.get('/')
+
+
+async def test_await_after_cancelling(loop, test_client):
+
+    async def handler(request):
+        return web.Response()
+
+    app = web.Application()
+    app.router.add_route('GET', '/', handler)
+
+    client = await test_client(app)
+
+    fut1 = loop.create_future()
+    fut2 = loop.create_future()
+
+    async def fetch1():
+        resp = await client.get('/')
+        assert resp.status == 200
+        fut1.set_result(None)
+        with pytest.raises(asyncio.CancelledError):
+            await fut2
+        resp.release()
+
+    async def fetch2():
+        await fut1
+        resp = await client.get('/')
+        assert resp.status == 200
+
+    async def canceller():
+        await fut1
+        fut2.cancel()
+
+    await asyncio.gather(fetch1(), fetch2(), canceller(), loop=loop)

tests/test_connector.py

@@ -342,6 +342,28 @@ def test_release_close(loop):
     assert proto.close.called
 
 
+def test__drop_acquire_per_host1(loop):
+    conn = aiohttp.BaseConnector(loop=loop)
+    conn._drop_acquired_per_host(123, 456)
+    assert len(conn._acquired_per_host) == 0
+
+
+def test__drop_acquire_per_host2(loop):
+    conn = aiohttp.BaseConnector(loop=loop)
+    conn._acquired_per_host[123].add(456)
+    conn._drop_acquired_per_host(123, 456)
+    assert len(conn._acquired_per_host) == 0
+
+
+def test__drop_acquire_per_host3(loop):
+    conn = aiohttp.BaseConnector(loop=loop)
+    conn._acquired_per_host[123].add(456)
+    conn._acquired_per_host[123].add(789)
+    conn._drop_acquired_per_host(123, 456)
+    assert len(conn._acquired_per_host) == 1
+    assert conn._acquired_per_host[123] == {789}
+
+
 async def test_tcp_connector_certificate_error(loop):
     req = ClientRequest('GET', URL('https://127.0.0.1:443'), loop=loop)