RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Aditya 2020-06-14 22:40:49 +05:30
parent d06bb12e35
commit de4a34365a
2 changed files with 129 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
scrapy/core/http2/protocol.py
View File

@ -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

85
scrapy/core/http2/stream.py
View File

@ -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"],