We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
3.3 socket read timeouts not working correctly
@asvetlov looks like the new `sock_read` timeout implementation is not working, see testcase:

```python
import multiprocessing
import asyncio
import aiohttp
import socket

host = "localhost"


def get_free_tcp_port():
    sckt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sckt.bind((host, 0))
    addr, port = sckt.getsockname()
    sckt.close()
    return port


class AIOServer(multiprocessing.Process):
    """
    This is a mock AWS service which will 5 seconds before returning
    a response to test socket timeouts.
    """
    def __init__(self):
        super().__init__(target=self._run)
        self._loop = None
        self._port = get_free_tcp_port()
        self.endpoint_url = 'http://{}:{}'.format(host, self._port)
        self.daemon = True  # die when parent dies

    def _run(self):
        asyncio.set_event_loop(asyncio.new_event_loop())
        app = aiohttp.web.Application()
        app.router.add_route('*', '/ok', self.ok)
        app.router.add_route('*', '/{anything:.*}', self.stream_handler)

        try:
            aiohttp.web.run_app(app, host=host, port=self._port,
                                handle_signals=False)
        except BaseException:
            pytest.fail('unable to start and connect to aiohttp server')
            raise

    async def __aenter__(self):
        self.start()
        await self._wait_until_up()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        try:
            self.terminate()
        except:
            pytest.fail("Unable to shut down server")
            raise

    async def ok(self, request):
        return aiohttp.web.Response()

    async def stream_handler(self, request):
        # Without the Content-Type, most (all?) browsers will not render
        # partially downloaded content. Note, the response type is
        # StreamResponse not Response.
        resp = aiohttp.web.StreamResponse(status=200, reason='OK',
                              headers={'Content-Type': 'text/html'})

        await resp.prepare(request)
        await asyncio.sleep(5, loop=self._loop)
        await resp.drain()
        return resp

    async def _wait_until_up(self):
        async with aiohttp.ClientSession() as session:
            for i in range(0, 30):
                if self.exitcode is not None:
                    pytest.fail('unable to start and connect to aiohttp server')
                    return

                try:
                    # we need to bypass the proxies due to monkey patches
                    await session.get(self.endpoint_url + '/ok', timeout=0.5)
                    return
                except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
                    await asyncio.sleep(0.5)
                except BaseException:
                    pytest.fail('unable to start and connect to aiohttp server')
                    raise

        pytest.fail('unable to start and connect to aiohttp server')


async def main():
    async with AIOServer() as server, \
            aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(sock_connect=1, sock_read=1)) as session:

        try:
            response = await session.get(server.endpoint_url)
            await response.read()
        except asyncio.TimeoutError:
            print("Success")

        print("failure")


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `aiohttp/client_proto.py`
Content:
```
1 from contextlib import suppress
2 
3 from .base_protocol import BaseProtocol
4 from .client_exceptions import (ClientOSError, ClientPayloadError,
5                                 ServerDisconnectedError, ServerTimeoutError)
6 from .http import HttpResponseParser
7 from .streams import EMPTY_PAYLOAD, DataQueue
8 
9 
10 class ResponseHandler(BaseProtocol, DataQueue):
11     """Helper class to adapt between Protocol and StreamReader."""
12 
13     def __init__(self, *, loop=None):
14         BaseProtocol.__init__(self, loop=loop)
15         DataQueue.__init__(self, loop=loop)
16 
17         self._should_close = False
18 
19         self._payload = None
20         self._skip_payload = False
21         self._payload_parser = None
22         self._reading_paused = False
23 
24         self._timer = None
25 
26         self._tail = b''
27         self._upgraded = False
28         self._parser = None
29 
30         self._read_timeout = None
31         self._read_timeout_handle = None
32 
33     @property
34     def upgraded(self):
35         return self._upgraded
36 
37     @property
38     def should_close(self):
39         if (self._payload is not None and
40                 not self._payload.is_eof() or self._upgraded):
41             return True
42 
43         return (self._should_close or self._upgraded or
44                 self.exception() is not None or
45                 self._payload_parser is not None or
46                 len(self) or self._tail)
47 
48     def close(self):
49         transport = self.transport
50         if transport is not None:
51             transport.close()
52             self.transport = None
53             self._payload = None
54         return transport
55 
56     def is_connected(self):
57         return self.transport is not None
58 
59     def connection_lost(self, exc):
60         self._drop_timeout()
61 
62         if self._payload_parser is not None:
63             with suppress(Exception):
64                 self._payload_parser.feed_eof()
65 
66         try:
67             uncompleted = self._parser.feed_eof()
68         except Exception:
69             uncompleted = None
70             if self._payload is not None:
71                 self._payload.set_exception(
72                     ClientPayloadError('Response payload is not completed'))
73 
74         if not self.is_eof():
75             if isinstance(exc, OSError):
76                 exc = ClientOSError(*exc.args)
77             if exc is None:
78                 exc = ServerDisconnectedError(uncompleted)
79             # assigns self._should_close to True as side effect,
80             # we do it anyway below
81             self.set_exception(exc)
82 
83         self._should_close = True
84         self._parser = None
85         self._payload = None
86         self._payload_parser = None
87         self._reading_paused = False
88 
89         super().connection_lost(exc)
90 
91     def eof_received(self):
92         # should call parser.feed_eof() most likely
93         self._drop_timeout()
94 
95     def pause_reading(self):
96         if not self._reading_paused:
97             try:
98                 self.transport.pause_reading()
99             except (AttributeError, NotImplementedError, RuntimeError):
100                 pass
101             self._reading_paused = True
102             self._drop_timeout()
103 
104     def resume_reading(self):
105         if self._reading_paused:
106             try:
107                 self.transport.resume_reading()
108             except (AttributeError, NotImplementedError, RuntimeError):
109                 pass
110             self._reading_paused = False
111             self._reschedule_timeout()
112 
113     def set_exception(self, exc):
114         self._should_close = True
115         self._drop_timeout()
116         super().set_exception(exc)
117 
118     def set_parser(self, parser, payload):
119         self._payload = payload
120         self._payload_parser = parser
121 
122         self._drop_timeout()
123 
124         if self._tail:
125             data, self._tail = self._tail, b''
126             self.data_received(data)
127 
128     def set_response_params(self, *, timer=None,
129                             skip_payload=False,
130                             read_until_eof=False,
131                             auto_decompress=True,
132                             read_timeout=None):
133         self._skip_payload = skip_payload
134 
135         self._read_timeout = read_timeout
136         self._reschedule_timeout()
137 
138         self._parser = HttpResponseParser(
139             self, self._loop, timer=timer,
140             payload_exception=ClientPayloadError,
141             read_until_eof=read_until_eof,
142             auto_decompress=auto_decompress)
143 
144         if self._tail:
145             data, self._tail = self._tail, b''
146             self.data_received(data)
147 
148     def _drop_timeout(self):
149         if self._read_timeout_handle is not None:
150             self._read_timeout_handle.cancel()
151             self._read_timeout_handle = None
152 
153     def _reschedule_timeout(self):
154         timeout = self._read_timeout
155         if self._read_timeout_handle is not None:
156             self._read_timeout_handle.cancel()
157 
158         if timeout:
159             self._read_timeout_handle = self._loop.call_later(
160                 timeout, self._on_read_timeout)
161         else:
162             self._read_timeout_handle = None
163 
164     def _on_read_timeout(self):
165         self.set_exception(
166             ServerTimeoutError("Timeout on reading data from socket"))
167 
168     def data_received(self, data):
169         if not data:
170             return
171 
172         # custom payload parser
173         if self._payload_parser is not None:
174             eof, tail = self._payload_parser.feed_data(data)
175             if eof:
176                 self._payload = None
177                 self._payload_parser = None
178 
179                 if tail:
180                     self.data_received(tail)
181             return
182         else:
183             if self._upgraded or self._parser is None:
184                 # i.e. websocket connection, websocket parser is not set yet
185                 self._tail += data
186             else:
187                 # parse http messages
188                 try:
189                     messages, upgraded, tail = self._parser.feed_data(data)
190                 except BaseException as exc:
191                     self.transport.close()
192                     # should_close is True after the call
193                     self.set_exception(exc)
194                     return
195 
196                 self._upgraded = upgraded
197 
198                 payload = None
199                 for message, payload in messages:
200                     if message.should_close:
201                         self._should_close = True
202 
203                     self._payload = payload
204 
205                     if self._skip_payload or message.code in (204, 304):
206                         self.feed_data((message, EMPTY_PAYLOAD), 0)
207                     else:
208                         self.feed_data((message, payload), 0)
209                 if payload is not None:
210                     # new message(s) was processed
211                     # register timeout handler unsubscribing
212                     # either on end-of-stream or immediatelly for
213                     # EMPTY_PAYLOAD
214                     if payload is not EMPTY_PAYLOAD:
215                         payload.on_eof(self._drop_timeout)
216                     else:
217                         self._drop_timeout()
218 
219                 if tail:
220                     if upgraded:
221                         self.data_received(tail)
222                     else:
223                         self._tail = tail
224
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/aiohttp/client_proto.py b/aiohttp/client_proto.py
--- a/aiohttp/client_proto.py
+++ b/aiohttp/client_proto.py
@@ -51,6 +51,7 @@
             transport.close()
             self.transport = None
             self._payload = None
+            self._drop_timeout()
         return transport
 
     def is_connected(self):
@@ -162,8 +163,10 @@
             self._read_timeout_handle = None
 
     def _on_read_timeout(self):
-        self.set_exception(
-            ServerTimeoutError("Timeout on reading data from socket"))
+        exc = ServerTimeoutError("Timeout on reading data from socket")
+        self.set_exception(exc)
+        if self._payload is not None:
+            self._payload.set_exception(exc)
 
     def data_received(self, data):
         if not data:

{"golden_diff": "diff --git a/aiohttp/client_proto.py b/aiohttp/client_proto.py\n--- a/aiohttp/client_proto.py\n+++ b/aiohttp/client_proto.py\n@@ -51,6 +51,7 @@\n             transport.close()\n             self.transport = None\n             self._payload = None\n+            self._drop_timeout()\n         return transport\n \n     def is_connected(self):\n@@ -162,8 +163,10 @@\n             self._read_timeout_handle = None\n \n     def _on_read_timeout(self):\n-        self.set_exception(\n-            ServerTimeoutError(\"Timeout on reading data from socket\"))\n+        exc = ServerTimeoutError(\"Timeout on reading data from socket\")\n+        self.set_exception(exc)\n+        if self._payload is not None:\n+            self._payload.set_exception(exc)\n \n     def data_received(self, data):\n         if not data:\n", "issue": "3.3 socket read timeouts not working correctly\n@asvetlov looks like the new `sock_read` timeout implementation is not working, see testcase:\r\n\r\n```python\r\nimport multiprocessing\r\nimport asyncio\r\nimport aiohttp\r\nimport socket\r\n\r\nhost = \"localhost\"\r\n\r\n\r\ndef get_free_tcp_port():\r\n    sckt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\r\n    sckt.bind((host, 0))\r\n    addr, port = sckt.getsockname()\r\n    sckt.close()\r\n    return port\r\n\r\n\r\nclass AIOServer(multiprocessing.Process):\r\n    \"\"\"\r\n    This is a mock AWS service which will 5 seconds before returning\r\n    a response to test socket timeouts.\r\n    \"\"\"\r\n    def __init__(self):\r\n        super().__init__(target=self._run)\r\n        self._loop = None\r\n        self._port = get_free_tcp_port()\r\n        self.endpoint_url = 'http://{}:{}'.format(host, self._port)\r\n        self.daemon = True  # die when parent dies\r\n\r\n    def _run(self):\r\n        asyncio.set_event_loop(asyncio.new_event_loop())\r\n        app = aiohttp.web.Application()\r\n        app.router.add_route('*', '/ok', self.ok)\r\n        app.router.add_route('*', '/{anything:.*}', self.stream_handler)\r\n\r\n        try:\r\n            aiohttp.web.run_app(app, host=host, port=self._port,\r\n                                handle_signals=False)\r\n        except BaseException:\r\n            pytest.fail('unable to start and connect to aiohttp server')\r\n            raise\r\n\r\n    async def __aenter__(self):\r\n        self.start()\r\n        await self._wait_until_up()\r\n        return self\r\n\r\n    async def __aexit__(self, exc_type, exc_val, exc_tb):\r\n        try:\r\n            self.terminate()\r\n        except:\r\n            pytest.fail(\"Unable to shut down server\")\r\n            raise\r\n\r\n    async def ok(self, request):\r\n        return aiohttp.web.Response()\r\n\r\n    async def stream_handler(self, request):\r\n        # Without the Content-Type, most (all?) browsers will not render\r\n        # partially downloaded content. Note, the response type is\r\n        # StreamResponse not Response.\r\n        resp = aiohttp.web.StreamResponse(status=200, reason='OK',\r\n                              headers={'Content-Type': 'text/html'})\r\n\r\n        await resp.prepare(request)\r\n        await asyncio.sleep(5, loop=self._loop)\r\n        await resp.drain()\r\n        return resp\r\n\r\n    async def _wait_until_up(self):\r\n        async with aiohttp.ClientSession() as session:\r\n            for i in range(0, 30):\r\n                if self.exitcode is not None:\r\n                    pytest.fail('unable to start and connect to aiohttp server')\r\n                    return\r\n\r\n                try:\r\n                    # we need to bypass the proxies due to monkey patches\r\n                    await session.get(self.endpoint_url + '/ok', timeout=0.5)\r\n                    return\r\n                except (aiohttp.ClientConnectionError, asyncio.TimeoutError):\r\n                    await asyncio.sleep(0.5)\r\n                except BaseException:\r\n                    pytest.fail('unable to start and connect to aiohttp server')\r\n                    raise\r\n\r\n        pytest.fail('unable to start and connect to aiohttp server')\r\n\r\n\r\nasync def main():\r\n    async with AIOServer() as server, \\\r\n            aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(sock_connect=1, sock_read=1)) as session:\r\n\r\n        try:\r\n            response = await session.get(server.endpoint_url)\r\n            await response.read()\r\n        except asyncio.TimeoutError:\r\n            print(\"Success\")\r\n\r\n        print(\"failure\")\r\n\r\n\r\nif __name__ == '__main__':\r\n    asyncio.get_event_loop().run_until_complete(main())\r\n```\n", "before_files": [{"content": "from contextlib import suppress\n\nfrom .base_protocol import BaseProtocol\nfrom .client_exceptions import (ClientOSError, ClientPayloadError,\n                                ServerDisconnectedError, ServerTimeoutError)\nfrom .http import HttpResponseParser\nfrom .streams import EMPTY_PAYLOAD, DataQueue\n\n\nclass ResponseHandler(BaseProtocol, DataQueue):\n    \"\"\"Helper class to adapt between Protocol and StreamReader.\"\"\"\n\n    def __init__(self, *, loop=None):\n        BaseProtocol.__init__(self, loop=loop)\n        DataQueue.__init__(self, loop=loop)\n\n        self._should_close = False\n\n        self._payload = None\n        self._skip_payload = False\n        self._payload_parser = None\n        self._reading_paused = False\n\n        self._timer = None\n\n        self._tail = b''\n        self._upgraded = False\n        self._parser = None\n\n        self._read_timeout = None\n        self._read_timeout_handle = None\n\n    @property\n    def upgraded(self):\n        return self._upgraded\n\n    @property\n    def should_close(self):\n        if (self._payload is not None and\n                not self._payload.is_eof() or self._upgraded):\n            return True\n\n        return (self._should_close or self._upgraded or\n                self.exception() is not None or\n                self._payload_parser is not None or\n                len(self) or self._tail)\n\n    def close(self):\n        transport = self.transport\n        if transport is not None:\n            transport.close()\n            self.transport = None\n            self._payload = None\n        return transport\n\n    def is_connected(self):\n        return self.transport is not None\n\n    def connection_lost(self, exc):\n        self._drop_timeout()\n\n        if self._payload_parser is not None:\n            with suppress(Exception):\n                self._payload_parser.feed_eof()\n\n        try:\n            uncompleted = self._parser.feed_eof()\n        except Exception:\n            uncompleted = None\n            if self._payload is not None:\n                self._payload.set_exception(\n                    ClientPayloadError('Response payload is not completed'))\n\n        if not self.is_eof():\n            if isinstance(exc, OSError):\n                exc = ClientOSError(*exc.args)\n            if exc is None:\n                exc = ServerDisconnectedError(uncompleted)\n            # assigns self._should_close to True as side effect,\n            # we do it anyway below\n            self.set_exception(exc)\n\n        self._should_close = True\n        self._parser = None\n        self._payload = None\n        self._payload_parser = None\n        self._reading_paused = False\n\n        super().connection_lost(exc)\n\n    def eof_received(self):\n        # should call parser.feed_eof() most likely\n        self._drop_timeout()\n\n    def pause_reading(self):\n        if not self._reading_paused:\n            try:\n                self.transport.pause_reading()\n            except (AttributeError, NotImplementedError, RuntimeError):\n                pass\n            self._reading_paused = True\n            self._drop_timeout()\n\n    def resume_reading(self):\n        if self._reading_paused:\n            try:\n                self.transport.resume_reading()\n            except (AttributeError, NotImplementedError, RuntimeError):\n                pass\n            self._reading_paused = False\n            self._reschedule_timeout()\n\n    def set_exception(self, exc):\n        self._should_close = True\n        self._drop_timeout()\n        super().set_exception(exc)\n\n    def set_parser(self, parser, payload):\n        self._payload = payload\n        self._payload_parser = parser\n\n        self._drop_timeout()\n\n        if self._tail:\n            data, self._tail = self._tail, b''\n            self.data_received(data)\n\n    def set_response_params(self, *, timer=None,\n                            skip_payload=False,\n                            read_until_eof=False,\n                            auto_decompress=True,\n                            read_timeout=None):\n        self._skip_payload = skip_payload\n\n        self._read_timeout = read_timeout\n        self._reschedule_timeout()\n\n        self._parser = HttpResponseParser(\n            self, self._loop, timer=timer,\n            payload_exception=ClientPayloadError,\n            read_until_eof=read_until_eof,\n            auto_decompress=auto_decompress)\n\n        if self._tail:\n            data, self._tail = self._tail, b''\n            self.data_received(data)\n\n    def _drop_timeout(self):\n        if self._read_timeout_handle is not None:\n            self._read_timeout_handle.cancel()\n            self._read_timeout_handle = None\n\n    def _reschedule_timeout(self):\n        timeout = self._read_timeout\n        if self._read_timeout_handle is not None:\n            self._read_timeout_handle.cancel()\n\n        if timeout:\n            self._read_timeout_handle = self._loop.call_later(\n                timeout, self._on_read_timeout)\n        else:\n            self._read_timeout_handle = None\n\n    def _on_read_timeout(self):\n        self.set_exception(\n            ServerTimeoutError(\"Timeout on reading data from socket\"))\n\n    def data_received(self, data):\n        if not data:\n            return\n\n        # custom payload parser\n        if self._payload_parser is not None:\n            eof, tail = self._payload_parser.feed_data(data)\n            if eof:\n                self._payload = None\n                self._payload_parser = None\n\n                if tail:\n                    self.data_received(tail)\n            return\n        else:\n            if self._upgraded or self._parser is None:\n                # i.e. websocket connection, websocket parser is not set yet\n                self._tail += data\n            else:\n                # parse http messages\n                try:\n                    messages, upgraded, tail = self._parser.feed_data(data)\n                except BaseException as exc:\n                    self.transport.close()\n                    # should_close is True after the call\n                    self.set_exception(exc)\n                    return\n\n                self._upgraded = upgraded\n\n                payload = None\n                for message, payload in messages:\n                    if message.should_close:\n                        self._should_close = True\n\n                    self._payload = payload\n\n                    if self._skip_payload or message.code in (204, 304):\n                        self.feed_data((message, EMPTY_PAYLOAD), 0)\n                    else:\n                        self.feed_data((message, payload), 0)\n                if payload is not None:\n                    # new message(s) was processed\n                    # register timeout handler unsubscribing\n                    # either on end-of-stream or immediatelly for\n                    # EMPTY_PAYLOAD\n                    if payload is not EMPTY_PAYLOAD:\n                        payload.on_eof(self._drop_timeout)\n                    else:\n                        self._drop_timeout()\n\n                if tail:\n                    if upgraded:\n                        self.data_received(tail)\n                    else:\n                        self._tail = tail\n", "path": "aiohttp/client_proto.py"}], "after_files": [{"content": "from contextlib import suppress\n\nfrom .base_protocol import BaseProtocol\nfrom .client_exceptions import (ClientOSError, ClientPayloadError,\n                                ServerDisconnectedError, ServerTimeoutError)\nfrom .http import HttpResponseParser\nfrom .streams import EMPTY_PAYLOAD, DataQueue\n\n\nclass ResponseHandler(BaseProtocol, DataQueue):\n    \"\"\"Helper class to adapt between Protocol and StreamReader.\"\"\"\n\n    def __init__(self, *, loop=None):\n        BaseProtocol.__init__(self, loop=loop)\n        DataQueue.__init__(self, loop=loop)\n\n        self._should_close = False\n\n        self._payload = None\n        self._skip_payload = False\n        self._payload_parser = None\n        self._reading_paused = False\n\n        self._timer = None\n\n        self._tail = b''\n        self._upgraded = False\n        self._parser = None\n\n        self._read_timeout = None\n        self._read_timeout_handle = None\n\n    @property\n    def upgraded(self):\n        return self._upgraded\n\n    @property\n    def should_close(self):\n        if (self._payload is not None and\n                not self._payload.is_eof() or self._upgraded):\n            return True\n\n        return (self._should_close or self._upgraded or\n                self.exception() is not None or\n                self._payload_parser is not None or\n                len(self) or self._tail)\n\n    def close(self):\n        transport = self.transport\n        if transport is not None:\n            transport.close()\n            self.transport = None\n            self._payload = None\n            self._drop_timeout()\n        return transport\n\n    def is_connected(self):\n        return self.transport is not None\n\n    def connection_lost(self, exc):\n        self._drop_timeout()\n\n        if self._payload_parser is not None:\n            with suppress(Exception):\n                self._payload_parser.feed_eof()\n\n        try:\n            uncompleted = self._parser.feed_eof()\n        except Exception:\n            uncompleted = None\n            if self._payload is not None:\n                self._payload.set_exception(\n                    ClientPayloadError('Response payload is not completed'))\n\n        if not self.is_eof():\n            if isinstance(exc, OSError):\n                exc = ClientOSError(*exc.args)\n            if exc is None:\n                exc = ServerDisconnectedError(uncompleted)\n            # assigns self._should_close to True as side effect,\n            # we do it anyway below\n            self.set_exception(exc)\n\n        self._should_close = True\n        self._parser = None\n        self._payload = None\n        self._payload_parser = None\n        self._reading_paused = False\n\n        super().connection_lost(exc)\n\n    def eof_received(self):\n        # should call parser.feed_eof() most likely\n        self._drop_timeout()\n\n    def pause_reading(self):\n        if not self._reading_paused:\n            try:\n                self.transport.pause_reading()\n            except (AttributeError, NotImplementedError, RuntimeError):\n                pass\n            self._reading_paused = True\n            self._drop_timeout()\n\n    def resume_reading(self):\n        if self._reading_paused:\n            try:\n                self.transport.resume_reading()\n            except (AttributeError, NotImplementedError, RuntimeError):\n                pass\n            self._reading_paused = False\n            self._reschedule_timeout()\n\n    def set_exception(self, exc):\n        self._should_close = True\n        self._drop_timeout()\n        super().set_exception(exc)\n\n    def set_parser(self, parser, payload):\n        self._payload = payload\n        self._payload_parser = parser\n\n        self._drop_timeout()\n\n        if self._tail:\n            data, self._tail = self._tail, b''\n            self.data_received(data)\n\n    def set_response_params(self, *, timer=None,\n                            skip_payload=False,\n                            read_until_eof=False,\n                            auto_decompress=True,\n                            read_timeout=None):\n        self._skip_payload = skip_payload\n\n        self._read_timeout = read_timeout\n        self._reschedule_timeout()\n\n        self._parser = HttpResponseParser(\n            self, self._loop, timer=timer,\n            payload_exception=ClientPayloadError,\n            read_until_eof=read_until_eof,\n            auto_decompress=auto_decompress)\n\n        if self._tail:\n            data, self._tail = self._tail, b''\n            self.data_received(data)\n\n    def _drop_timeout(self):\n        if self._read_timeout_handle is not None:\n            self._read_timeout_handle.cancel()\n            self._read_timeout_handle = None\n\n    def _reschedule_timeout(self):\n        timeout = self._read_timeout\n        if self._read_timeout_handle is not None:\n            self._read_timeout_handle.cancel()\n\n        if timeout:\n            self._read_timeout_handle = self._loop.call_later(\n                timeout, self._on_read_timeout)\n        else:\n            self._read_timeout_handle = None\n\n    def _on_read_timeout(self):\n        exc = ServerTimeoutError(\"Timeout on reading data from socket\")\n        self.set_exception(exc)\n        if self._payload is not None:\n            self._payload.set_exception(exc)\n\n    def data_received(self, data):\n        if not data:\n            return\n\n        # custom payload parser\n        if self._payload_parser is not None:\n            eof, tail = self._payload_parser.feed_data(data)\n            if eof:\n                self._payload = None\n                self._payload_parser = None\n\n                if tail:\n                    self.data_received(tail)\n            return\n        else:\n            if self._upgraded or self._parser is None:\n                # i.e. websocket connection, websocket parser is not set yet\n                self._tail += data\n            else:\n                # parse http messages\n                try:\n                    messages, upgraded, tail = self._parser.feed_data(data)\n                except BaseException as exc:\n                    self.transport.close()\n                    # should_close is True after the call\n                    self.set_exception(exc)\n                    return\n\n                self._upgraded = upgraded\n\n                payload = None\n                for message, payload in messages:\n                    if message.should_close:\n                        self._should_close = True\n\n                    self._payload = payload\n\n                    if self._skip_payload or message.code in (204, 304):\n                        self.feed_data((message, EMPTY_PAYLOAD), 0)\n                    else:\n                        self.feed_data((message, payload), 0)\n                if payload is not None:\n                    # new message(s) was processed\n                    # register timeout handler unsubscribing\n                    # either on end-of-stream or immediatelly for\n                    # EMPTY_PAYLOAD\n                    if payload is not EMPTY_PAYLOAD:\n                        payload.on_eof(self._drop_timeout)\n                    else:\n                        self._drop_timeout()\n\n                if tail:\n                    if upgraded:\n                        self.data_received(tail)\n                    else:\n                        self._tail = tail\n", "path": "aiohttp/client_proto.py"}]}