From de4a34365a2d872eb69ab2a2edfb723658e28530 Mon Sep 17 00:00:00 2001 From: Aditya Date: Sun, 14 Jun 2020 22:40:49 +0530 Subject: [PATCH] fix: large data chunk not received Every data chunk received needs to be acknowledged to - update the flow control window size - get furthur data chunks from the server --- scrapy/core/http2/protocol.py | 74 +++++++++++++++++++++--------- scrapy/core/http2/stream.py | 85 ++++++++++++++++++++++++++++++----- 2 files changed, 129 insertions(+), 30 deletions(-) diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index 4036dfb3e..167596340 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -1,10 +1,10 @@ import logging -from typing import Dict, List from h2.config import H2Configuration from h2.connection import H2Connection from h2.events import ( - ConnectionTerminated, DataReceived, ResponseReceived, StreamEnded, StreamReset, WindowUpdated + ConnectionTerminated, DataReceived, ResponseReceived, + StreamEnded, StreamReset, WindowUpdated ) from twisted.internet.protocol import connectionDone, Protocol @@ -18,33 +18,57 @@ class H2ClientProtocol(Protocol): # TODO: # 1. Check for user-agent while testing # 2. Add support for cookies - # 3. Handle priority updates + # 3. Handle priority updates (Not required) + # 4. Handle case when received events have StreamID = 0 (applied to H2Connection) + # 1 & 2: + # - Automatically handled by the Request middleware + # - request.headers will have 'Set-Cookie' value def __init__(self): config = H2Configuration(client_side=True, header_encoding='utf-8') self.conn = H2Connection(config=config) + # Address of the server we are connected to + # these are updated when connection is successfully made + self.destination = None + # ID of the next request stream - # Assuming each request stream creates a new response stream - # we increment by 2 for each new request stream created + # Following the convention made by hyper-h2 each client ID + # will be odd. self.next_stream_id = 1 # Streams are stored in a dictionary keyed off their stream IDs - self.streams: Dict[int, Stream] = {} + self.streams = {} # Boolean to keep track the connection is made # If requests are received before connection is made # we keep all requests in a pool and send them as the connection # is made self.is_connection_made = False - self._pending_request_stream_pool: List[Stream] = [] + self._pending_request_stream_pool = [] + + def _stream_close_cb(self, stream_id: int): + """Called when stream is closed completely + """ + try: + del self.streams[stream_id] + except KeyError: + pass def _new_stream(self, request: Request): """Instantiates a new Stream object """ - stream = Stream(self.next_stream_id, request, self.conn) + stream_id = self.next_stream_id self.next_stream_id += 2 + stream = Stream( + stream_id=stream_id, + request=request, + connection=self.conn, + write_to_transport=self._write_to_transport, + cb_close=lambda: self._stream_close_cb(stream_id) + ) + self.streams[stream.stream_id] = stream return stream @@ -53,7 +77,6 @@ class H2ClientProtocol(Protocol): # Initiate all pending requests for stream in self._pending_request_stream_pool: stream.initiate_request() - self._write_to_transport() self._pending_request_stream_pool.clear() @@ -81,13 +104,15 @@ class H2ClientProtocol(Protocol): """Called by Twisted when the connection is established. We can start sending some data now: we should open with the connection preamble. """ + self.destination = self.transport.connector.getDestination() + LOGGER.info('Connection made to {}'.format(self.destination)) + self.conn.initiate_connection() self._write_to_transport() + self.is_connection_made = True self._send_pending_requests() - self.is_connection_made = True - def dataReceived(self, data): events = self.conn.receive_data(data) self._handle_events(events) @@ -95,15 +120,18 @@ class H2ClientProtocol(Protocol): def connectionLost(self, reason=connectionDone): """Called by Twisted when the transport connection is lost. + No need to write anything to transport here. """ - stream_ids = list(self.streams.keys()) + # Pop all streams which were pending and were not yet started + for stream_id in list(self.streams): + try: + self.streams[stream_id].lost_connection() + except KeyError: + pass - for stream in self._pending_request_stream_pool: - stream_ids.remove(stream.stream_id) + self.conn.close_connection() - for stream_id in stream_ids: - # TODO: Close each Stream instance in a clean manner - self.conn.end_stream(stream_id) + LOGGER.info("Connection lost with reason " + str(reason)) def _handle_events(self, events): """Private method which acts as a bridge between the events @@ -136,7 +164,7 @@ class H2ClientProtocol(Protocol): def data_received(self, event: DataReceived): stream_id = event.stream_id - self.streams[stream_id].receive_data(event.data) + self.streams[stream_id].receive_data(event.data, event.flow_controlled_length) def response_received(self, event: ResponseReceived): stream_id = event.stream_id @@ -147,9 +175,15 @@ class H2ClientProtocol(Protocol): self.streams[stream_id].end_stream() def stream_reset(self, event: StreamReset): - pass + # TODO: event.stream_id was abruptly closed + # Q. What should be the response? (Failure/Partial/???) + self.streams[event.stream_id].reset() def window_updated(self, event: WindowUpdated): stream_id = event.stream_id if stream_id != 0: - self.streams[stream_id].window_updated() + self.streams[stream_id].receive_window_update(event.delta) + else: + # TODO: + # Q. What to do when StreamID=0 ? + pass diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index b37755042..7c9cf7cf5 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -1,3 +1,4 @@ +import logging from urllib.parse import urlparse from h2.connection import H2Connection @@ -6,6 +7,8 @@ from twisted.internet.defer import Deferred from scrapy.http import Request, Response from scrapy.http.headers import Headers +LOGGER = logging.getLogger(__name__) + class Stream: """Represents a single HTTP/2 Stream. @@ -18,17 +21,30 @@ class Stream: 1. Combine all the data frames """ - def __init__(self, stream_id: int, request: Request, connection: H2Connection): + def __init__( + self, + stream_id: int, + request: Request, + connection: H2Connection, + write_to_transport, + cb_close + ): """ Arguments: stream_id {int} -- For one HTTP/2 connection each stream is uniquely identified by a single integer request {Request} -- HTTP request connection {H2Connection} -- HTTP/2 connection this stream belongs to. + write_to_transport {callable} -- Method used to write & send data to the server + This method should be used whenever some frame is to be sent to the server. + cb_close {callable} -- Method called when this stream is closed + to notify the TCP connection instance. """ self.stream_id = stream_id self._request = request self._conn = connection + self._write_to_transport = write_to_transport + self._cb_close = cb_close self._request_body = self._request.body self.content_length = 0 if self._request_body is None else len(self._request_body) @@ -36,13 +52,20 @@ class Stream: # Each time we send a data frame, we will decrease value by the amount send. self.remaining_content_length = self.content_length - # Flag to keep track whether we have ended this stream - self.stream_ended = True + # Flag to keep track whether we have closed this stream + self.stream_closed_local = False + + # Flag to keep track whether the server has closed the stream + self.stream_closed_server = False # Data received frame by frame from the server is appended # and passed to the response Deferred when completely received. self._response_data = b"" + # The amount of data received that counts against the flow control + # window + self._response_flow_controlled_size = 0 + # Headers received after sending the request self._response_headers = Headers({}) @@ -77,6 +100,8 @@ class Stream: ] self._conn.send_headers(self.stream_id, http2_request_headers, end_stream=False) + self._write_to_transport() + self.send_data() def send_data(self): @@ -112,32 +137,59 @@ class Stream: data_chunk = self._request_body[data_chunk_start:data_chunk_start + chunk_size] self._conn.send_data(self.stream_id, data_chunk, end_stream=False) + self._write_to_transport() bytes_to_send = max(0, bytes_to_send - chunk_size) self.remaining_content_length = max(0, self.remaining_content_length - chunk_size) # End the stream if no more data has to be send if self.remaining_content_length == 0: + self.stream_closed_local = True self._conn.end_stream(self.stream_id) - else: - # TODO: Continue from here :) - pass - def window_updated(self): + self._write_to_transport() + + # Q. What about the rest of the data? + # Ans: Remaining Data frames will be sent when we get a WindowUpdate frame + + def receive_window_update(self, delta): """Flow control window size was changed. Send data that earlier could not be sent as we were blocked behind the flow control. + + Arguments: + delta -- Window change delta """ - if self.remaining_content_length > 0 and not self.stream_ended: + if self.remaining_content_length > 0 and not self.stream_closed_local: self.send_data() - def receive_data(self, data: bytes): + def receive_data(self, data: bytes, flow_controlled_length: int): self._response_data += data + self._response_flow_controlled_size += flow_controlled_length + + # Acknowledge the data received + self._conn.acknowledge_received_data( + self._response_flow_controlled_size, + self.stream_id + ) def receive_headers(self, headers): for name, value in headers: self._response_headers[name] = value + def reset(self): + """Received a RST_STREAM -- forcefully reset""" + # TODO: + # Q1. Do we need to send the request again? + # Q2. What response should we send now? + self.stream_closed_server = True + self._cb_close() + + def lost_connection(self): + # TODO: Same as self.reset + self.stream_closed_server = True + self._cb_close() + def end_stream(self): """Stream is ended by the server hence no further data or headers should be expected on this stream. @@ -145,7 +197,20 @@ class Stream: We will call the response deferred callback passing the response object """ - # TODO: Set flags, certificate, ip_address + assert self.stream_closed_server is False + self.stream_closed_server = True + + self._fire_response_deferred() + self._cb_close() + + def _fire_response_deferred(self): + # TODO: + # 1. Set flags, certificate, ip_address in response + # 2. Should we fire this in case of + # 2.1 StreamReset in between when data is received partially + # 2.2 Forcefully closed the stream + + # NOTE: Presently on fired with successful response response = Response( url=self._request.url, status=self._response_headers[":status"],