Added a configuration flag for enable request task handler cancelling when client connection closing.

aiohttp/web.py

@@ -306,6 +306,7 @@ async def _run_app(
     handle_signals: bool = True,
     reuse_address: Optional[bool] = None,
     reuse_port: Optional[bool] = None,
+    cancel_handler_on_connection_lost: bool = False,
 ) -> None:
     # An internal function to actually do all dirty job for application running
     if asyncio.iscoroutine(app):
@@ -320,6 +321,7 @@ async def _run_app(
         access_log_format=access_log_format,
         access_log=access_log,
         keepalive_timeout=keepalive_timeout,
+        cancel_handler_on_connection_lost=cancel_handler_on_connection_lost,
     )
 
     await runner.setup()
@@ -480,6 +482,7 @@ def run_app(
     handle_signals: bool = True,
     reuse_address: Optional[bool] = None,
     reuse_port: Optional[bool] = None,
+    cancel_handler_on_connection_lost: bool = False,
     loop: Optional[asyncio.AbstractEventLoop] = None,
 ) -> None:
     """Run an app locally"""
@@ -512,6 +515,7 @@ def run_app(
             handle_signals=handle_signals,
             reuse_address=reuse_address,
             reuse_port=reuse_port,
+            cancel_handler_on_connection_lost=cancel_handler_on_connection_lost,
         )
     )
 

aiohttp/web_protocol.py

@@ -313,6 +313,8 @@ def connection_lost(self, exc: Optional[BaseException]) -> None:
 
         super().connection_lost(exc)
 
+        should_cancel_task_handler = self._manager.cancel_handler_on_connection_lost
+
         self._manager = None
         self._force_close = True
         self._request_factory = None
@@ -330,6 +332,9 @@ def connection_lost(self, exc: Optional[BaseException]) -> None:
         if self._waiter is not None:
             self._waiter.cancel()
 
+        if should_cancel_task_handler and self._task_handler is not None:
+            self._task_handler.cancel()
+
         self._task_handler = None
 
         if self._payload_parser is not None:

aiohttp/web_server.py

@@ -19,6 +19,7 @@ def __init__(
         *,
         request_factory: Optional[_RequestFactory] = None,
         debug: Optional[bool] = None,
+        cancel_handler_on_connection_lost: bool = False,
         **kwargs: Any,
     ) -> None:
         if debug is not None:
@@ -33,6 +34,7 @@ def __init__(
         self.requests_count = 0
         self.request_handler = handler
         self.request_factory = request_factory or self._make_request
+        self.cancel_handler_on_connection_lost = cancel_handler_on_connection_lost
 
     @property
     def connections(self) -> List[RequestHandler]:

docs/web_reference.rst

@@ -2809,7 +2809,8 @@ Utilities
                       access_log=aiohttp.log.access_logger, \
                       handle_signals=True, \
                       reuse_address=None, \
-                      reuse_port=None)
+                      reuse_port=None, \
+                      cancel_handler_on_connection_lost=False)
 
    A high-level function for running an application, serving it until
    keyboard interrupt and performing a
@@ -2904,6 +2905,11 @@ Utilities
                            this flag when being created. This option is not
                            supported on Windows.
 
+   :param bool cancel_handler_on_connection_lost: tells the runner whether to
+                                                  cancel the execution of the
+                                                  handler task if the client
+                                                  connection has been closed.
+
    .. versionadded:: 3.0
 
       Support *access_log_class* parameter.

tests/test_web_server.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-from aiohttp import client, helpers, web
+from aiohttp import ClientSession, client, helpers, web
 
 
 async def test_simple_server(aiohttp_raw_server: Any, aiohttp_client: Any) -> None:
@@ -207,3 +207,53 @@ async def handler(request):
     )
 
     logger.exception.assert_called_with("Error handling request", exc_info=exc)
+
+
+async def test_cancel_handler_on_connection_lost(aiohttp_unused_port) -> None:
+    event = asyncio.Event()
+    port = aiohttp_unused_port()
+
+    async def on_request(_: web.Request) -> web.Response:
+        nonlocal event
+        try:
+            await asyncio.sleep(10)
+        except asyncio.CancelledError:
+            event.set()
+            raise
+        else:
+            raise web.HTTPInternalServerError()
+
+    app = web.Application()
+    app.router.add_route("GET", "/", on_request)
+
+    runner = web.AppRunner(app, cancel_handler_on_connection_lost=True)
+    await runner.setup()
+
+    site = web.TCPSite(runner, host="localhost", port=port)
+
+    await site.start()
+
+    async def client_request_maker():
+        async with ClientSession(base_url=f"http://localhost:{port}") as session:
+            request = session.get("/")
+
+            with pytest.raises(asyncio.TimeoutError):
+                await asyncio.wait_for(request, timeout=0.1)
+
+    try:
+        assert (
+            runner.server.cancel_handler_on_connection_lost
+        ), "Flag was not propagated"
+        await client_request_maker()
+
+        await asyncio.wait_for(event.wait(), timeout=1)
+        assert event.is_set(), "Request handler hasn't been cancelled"
+    finally:
+        await asyncio.gather(runner.shutdown(), site.stop())
+
+
+async def test_cancel_handler_on_connection_lost_flag_on_runner() -> None:
+    runner = web.AppRunner(web.Application(), cancel_handler_on_connection_lost=True)
+    await runner.setup()
+    assert runner.server.cancel_handler_on_connection_lost, "Flag was not propagated"
+    await runner.shutdown()