From 303485a9b4fd86c5123c96c81a9401b5e323a91d Mon Sep 17 00:00:00 2001 From: Aditya Date: Sun, 21 Jun 2020 00:33:34 +0530 Subject: [PATCH] fix(http2): POST request not sending large body --- scrapy/core/http2/protocol.py | 51 +++++++----- scrapy/core/http2/stream.py | 148 ++++++++++++++++++++++++---------- 2 files changed, 138 insertions(+), 61 deletions(-) diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index dbc048ffa..d6134183a 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -1,3 +1,4 @@ +import ipaddress import itertools import logging from collections import deque @@ -5,7 +6,7 @@ from collections import deque from h2.config import H2Configuration from h2.connection import H2Connection from h2.events import ( - ConnectionTerminated, DataReceived, ResponseReceived, + DataReceived, ResponseReceived, SettingsAcknowledged, StreamEnded, StreamReset, WindowUpdated ) from twisted.internet.protocol import connectionDone, Protocol @@ -49,6 +50,13 @@ class H2ClientProtocol(Protocol): self.is_connection_made = False self._pending_request_stream_pool = deque() + # Some meta data of this connection + # initialized when connection is successfully made + self._metadata = { + 'certificate': None, + 'ip_address': None + } + def _stream_close_cb(self, stream_id: int): """Called when stream is closed completely """ @@ -63,6 +71,7 @@ class H2ClientProtocol(Protocol): stream_id=stream_id, request=request, connection=self.conn, + metadata=self._metadata, write_to_transport=self._write_to_transport, cb_close=self._stream_close_cb ) @@ -73,7 +82,7 @@ class H2ClientProtocol(Protocol): def _send_pending_requests(self): # TODO: handle MAX_CONCURRENT_STREAMS # Initiate all pending requests - while len(self._pending_request_stream_pool): + while self._pending_request_stream_pool: stream = self._pending_request_stream_pool.popleft() stream.initiate_request() @@ -84,6 +93,8 @@ class H2ClientProtocol(Protocol): data = self.conn.data_to_send() self.transport.write(data) + LOGGER.debug("Sent {} bytes to {} via transport".format(len(data), self._metadata['ip_address'])) + def request(self, _request: Request): stream = self._new_stream(_request) d = stream.get_response() @@ -101,17 +112,16 @@ 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() + self.destination = self.transport.getPeer() LOGGER.info('Connection made to {}'.format(self.destination)) + self._metadata['certificate'] = self.transport.getPeerCertificate() + self._metadata['ip_address'] = ipaddress.ip_address(self.destination.host) + self.conn.initiate_connection() self._write_to_transport() self.is_connection_made = True - # Send off all the pending requests - # as now we have established a proper HTTP/2 connection - self._send_pending_requests() - def dataReceived(self, data): events = self.conn.receive_data(data) self._handle_events(events) @@ -123,7 +133,7 @@ class H2ClientProtocol(Protocol): """ # Pop all streams which were pending and were not yet started for stream_id in list(self.streams): - self.streams[stream_id].lost_connection() + self.streams[stream_id].close() self.conn.close_connection() @@ -139,9 +149,7 @@ class H2ClientProtocol(Protocol): """ for event in events: LOGGER.debug(event) - if isinstance(event, ConnectionTerminated): - self.connection_terminated(event) - elif isinstance(event, DataReceived): + if isinstance(event, DataReceived): self.data_received(event) elif isinstance(event, ResponseReceived): self.response_received(event) @@ -151,13 +159,12 @@ class H2ClientProtocol(Protocol): self.stream_reset(event) elif isinstance(event, WindowUpdated): self.window_updated(event) + elif isinstance(event, SettingsAcknowledged): + self.settings_acknowledged(event) else: LOGGER.info("Received unhandled event {}".format(event)) # Event handler functions starts here - def connection_terminated(self, event: ConnectionTerminated): - pass - def data_received(self, event: DataReceived): stream_id = event.stream_id self.streams[stream_id].receive_data(event.data, event.flow_controlled_length) @@ -166,20 +173,26 @@ class H2ClientProtocol(Protocol): stream_id = event.stream_id self.streams[stream_id].receive_headers(event.headers) + def settings_acknowledged(self, event: SettingsAcknowledged): + # Send off all the pending requests + # as now we have established a proper HTTP/2 connection + self._send_pending_requests() + def stream_ended(self, event: StreamEnded): stream_id = event.stream_id - self.streams[stream_id].end_stream() + self.streams[stream_id].close() def stream_reset(self, event: StreamReset): # TODO: event.stream_id was abruptly closed # Q. What should be the response? (Failure/Partial/???) - self.streams[event.stream_id].reset() + self.streams[event.stream_id].close(event) def window_updated(self, event: WindowUpdated): stream_id = event.stream_id if stream_id != 0: self.streams[stream_id].receive_window_update(event.delta) else: - # TODO: - # Q. What to do when StreamID=0 ? - pass + # Send leftover data for all the streams + for stream in self.streams.values(): + if stream.request_sent: + stream.send_data() diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index c2e1adce5..e8b4471d6 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -1,11 +1,15 @@ import logging +from typing import Dict from urllib.parse import urlparse from h2.connection import H2Connection +from h2.events import StreamEnded +from h2.exceptions import StreamClosedError from twisted.internet.defer import Deferred -from scrapy.http import Request, Response +from scrapy.http import Request from scrapy.http.headers import Headers +from scrapy.responsetypes import responsetypes LOGGER = logging.getLogger(__name__) @@ -22,12 +26,13 @@ class Stream: """ def __init__( - self, - stream_id: int, - request: Request, - connection: H2Connection, - write_to_transport, - cb_close + self, + stream_id: int, + request: Request, + connection: H2Connection, + metadata: Dict, + write_to_transport, + cb_close ): """ Arguments: @@ -35,6 +40,7 @@ class Stream: uniquely identified by a single integer request {Request} -- HTTP request connection {H2Connection} -- HTTP/2 connection this stream belongs to. + metadata {Dict} -- Reference to dictionary having metadata of HTTP/2 connection 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 @@ -43,12 +49,16 @@ class Stream: self.stream_id = stream_id self._request = request self._conn = connection + self._metadata = metadata 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) + # Flag to keep track whether this stream has initiated the request + self.request_sent = False + # Each time we send a data frame, we will decrease value by the amount send. self.remaining_content_length = self.content_length @@ -58,20 +68,30 @@ class Stream: # 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({}) + # Private variable used to build the response + # this response is then converted to appropriate Response class + # passed to the response deferred callback + self._response = { + # Data received frame by frame from the server is appended + # and passed to the response Deferred when completely received. + 'body': b'', + + # Headers received after sending the request + 'headers': Headers({}) + } # TODO: Add canceller for the Deferred below self._deferred_response = Deferred() + def __str__(self): + return "Stream(id={})".format(self.stream_id) + + __repr__ = __str__ + def get_response(self): """Simply return a Deferred which fires when response from the asynchronous request is available @@ -104,6 +124,8 @@ class Stream: self._conn.send_headers(self.stream_id, headers, end_stream=False) self._write_to_transport() + self.request_sent = True + self.send_data() def send_data(self): @@ -112,7 +134,18 @@ class Stream: If the content length is 0 initially then we end the stream immediately and wait for response data. + + Warning: Only call this method when stream not closed from client side + and has initiated request already by sending HEADER frame. If not then + stream will be closed from client side with 499 response. + + TODO: Q. Should we instead raise ProtocolError here with a proper message? """ + if self.stream_closed_local or self.stream_closed_server: + raise StreamClosedError(self.stream_id) + elif not self.request_sent: + self.close() + return # TODO: # 1. Add test for sending very large data @@ -132,7 +165,8 @@ class Stream: bytes_to_send_size = min(window_size, self.remaining_content_length) # We now need to send a number of data frames. - while bytes_to_send_size: + data_frames_sent = 0 + while bytes_to_send_size > 0: chunk_size = min(bytes_to_send_size, max_frame_size) data_chunk_start_id = self.content_length - self.remaining_content_length @@ -140,16 +174,24 @@ class Stream: self._conn.send_data(self.stream_id, data_chunk, end_stream=False) + data_frames_sent += 1 bytes_to_send_size = bytes_to_send_size - chunk_size self.remaining_content_length = self.remaining_content_length - chunk_size - # End the stream if no more data has to be send - if self.remaining_content_length <= 0: - self.remaining_content_length = 0 + self.remaining_content_length = max(0, self.remaining_content_length) + LOGGER.debug("{} sending {}/{} data bytes ({} frames) to {}".format( + self, + self.content_length - self.remaining_content_length, self.content_length, + data_frames_sent, + self._metadata['ip_address']) + ) + # End the stream if no more data needs to be send + if self.remaining_content_length == 0: self.stream_closed_local = True self._conn.end_stream(self.stream_id) + # Write data to transport -- Empty the outstanding data self._write_to_transport() # Q. What about the rest of the data? @@ -163,11 +205,11 @@ class Stream: Arguments: delta -- Window change delta """ - if self.stream_closed_local is False: + if self.remaining_content_length > 0 and not self.stream_closed_server: self.send_data() def receive_data(self, data: bytes, flow_controlled_length: int): - self._response_data += data + self._response['body'] += data self._response_flow_controlled_size += flow_controlled_length # Acknowledge the data received @@ -178,47 +220,69 @@ class Stream: def receive_headers(self, headers): for name, value in headers: - self._response_headers[name] = value + 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(self.stream_id) + def close(self, event=None): + """Based on the event sent we will handle each case. - def lost_connection(self): - # TODO: Same as self.reset - self.stream_closed_server = True - self._cb_close(self.stream_id) - - def end_stream(self): - """Stream is ended by the server hence no further + event: StreamEnded + Stream is ended by the server hence no further data or headers should be expected on this stream. - We will call the response deferred callback passing the response object + + event: StreamReset + Stream reset via RST_FRAME by the upstream hence forcefully close + this stream and send TODO: ? + + event: None + No event is launched -- Hence we will simply close this stream """ + # TODO: In case of abruptly stream close + # Q1. Do we need to send the request again? + # Q2. What response should we send now? assert self.stream_closed_server is False self.stream_closed_server = True + if not isinstance(event, StreamEnded): + # TODO + # Stream was abruptly ended here + # Partial - Content-Length header not provided + pass + self._fire_response_deferred() self._cb_close(self.stream_id) - def _fire_response_deferred(self): + def _fire_response_deferred(self, flags=None): + """Builds response from the self._response dict + and fires the response deferred callback with the + generated response instance""" # 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 + # 3. Update Client Side Status Codes here - # NOTE: Presently on fired with successful response - response = Response( + response_cls = responsetypes.from_args( + headers=self._response['headers'], url=self._request.url, - status=self._response_headers[":status"], - headers=self._response_headers, - body=self._response_data, - request=self._request + body=self._response['body'] ) + + # If there is :status in headers then + # HTTP Status Code: 499 - Client Closed Request + status = self._response['headers'].get(':status', '499') + + response = response_cls( + url=self._request.url, + status=status, + headers=self._response['headers'], + body=self._response['body'], + request=self._request, + flags=flags, + certificate=self._metadata['certificate'], + ip_address=self._metadata['ip_address'] + ) + self._deferred_response.callback(response)