-
Notifications
You must be signed in to change notification settings - Fork 89
/
http2.py
475 lines (405 loc) · 18.7 KB
/
http2.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
import enum
import time
import types
import typing
import h2.config
import h2.connection
import h2.events
import h2.exceptions
import h2.settings
from .._exceptions import (
ConnectionNotAvailable,
LocalProtocolError,
RemoteProtocolError,
)
from .._models import Origin, Request, Response
from .._synchronization import Lock, Semaphore
from .._trace import Trace
from ..backends.base import NetworkStream
from .interfaces import ConnectionInterface
def has_body_headers(request: Request) -> bool:
return any(
[
k.lower() == b"content-length" or k.lower() == b"transfer-encoding"
for k, v in request.headers
]
)
class HTTPConnectionState(enum.IntEnum):
ACTIVE = 1
IDLE = 2
CLOSED = 3
class HTTP2Connection(ConnectionInterface):
READ_NUM_BYTES = 64 * 1024
CONFIG = h2.config.H2Configuration(validate_inbound_headers=False)
def __init__(
self,
origin: Origin,
stream: NetworkStream,
keepalive_expiry: typing.Optional[float] = None,
):
self._origin = origin
self._network_stream = stream
self._keepalive_expiry: typing.Optional[float] = keepalive_expiry
self._h2_state = h2.connection.H2Connection(config=self.CONFIG)
self._state = HTTPConnectionState.IDLE
self._expire_at: typing.Optional[float] = None
self._request_count = 0
self._init_lock = Lock()
self._state_lock = Lock()
self._read_lock = Lock()
self._write_lock = Lock()
self._sent_connection_init = False
self._used_all_stream_ids = False
self._connection_error = False
self._events: typing.Dict[int, h2.events.Event] = {}
self._read_exception: typing.Optional[Exception] = None
self._write_exception: typing.Optional[Exception] = None
self._connection_error_event: typing.Optional[h2.events.Event] = None
def handle_request(self, request: Request) -> Response:
if not self.can_handle_request(request.url.origin):
# This cannot occur in normal operation, since the connection pool
# will only send requests on connections that handle them.
# It's in place simply for resilience as a guard against incorrect
# usage, for anyone working directly with httpcore connections.
raise RuntimeError(
f"Attempted to send request to {request.url.origin} on connection "
f"to {self._origin}"
)
with self._state_lock:
if self._state in (HTTPConnectionState.ACTIVE, HTTPConnectionState.IDLE):
self._request_count += 1
self._expire_at = None
self._state = HTTPConnectionState.ACTIVE
else:
raise ConnectionNotAvailable()
with self._init_lock:
if not self._sent_connection_init:
kwargs = {"request": request}
with Trace("http2.send_connection_init", request, kwargs):
self._send_connection_init(**kwargs)
self._sent_connection_init = True
max_streams = self._h2_state.local_settings.max_concurrent_streams
self._max_streams_semaphore = Semaphore(max_streams)
self._max_streams_semaphore.acquire()
try:
stream_id = self._h2_state.get_next_available_stream_id()
self._events[stream_id] = []
except h2.exceptions.NoAvailableStreamIDError: # pragma: nocover
self._used_all_stream_ids = True
raise ConnectionNotAvailable()
try:
kwargs = {"request": request, "stream_id": stream_id}
with Trace("http2.send_request_headers", request, kwargs):
self._send_request_headers(request=request, stream_id=stream_id)
with Trace("http2.send_request_body", request, kwargs):
self._send_request_body(request=request, stream_id=stream_id)
with Trace(
"http2.receive_response_headers", request, kwargs
) as trace:
status, headers = self._receive_response(
request=request, stream_id=stream_id
)
trace.return_value = (status, headers)
return Response(
status=status,
headers=headers,
content=HTTP2ConnectionByteStream(self, request, stream_id=stream_id),
extensions={"stream_id": stream_id, "http_version": b"HTTP/2"},
)
except Exception as exc: # noqa: PIE786
kwargs = {"stream_id": stream_id}
with Trace("http2.response_closed", request, kwargs):
self._response_closed(stream_id=stream_id)
if isinstance(exc, h2.exceptions.ProtocolError):
# One case where h2 can raise a protocol error is when a
# closed frame has been seen by the state machine.
#
# This happens when one stream is reading, and encounters
# a GOAWAY event. Other flows of control may then raise
# a protocol error at any point they interact with the 'h2_state'.
#
# In this case we'll have stored the event, and should raise
# it as a RemoteProtocolError.
if self._connection_error_event:
raise RemoteProtocolError(self._connection_error_event)
# If h2 raises a protocol error in some other state then we
# must somehow have made a protocol violation.
raise LocalProtocolError(exc) # pragma: nocover
raise exc
def _send_connection_init(self, request: Request) -> None:
"""
The HTTP/2 connection requires some initial setup before we can start
using individual request/response streams on it.
"""
# Need to set these manually here instead of manipulating via
# __setitem__() otherwise the H2Connection will emit SettingsUpdate
# frames in addition to sending the undesired defaults.
self._h2_state.local_settings = h2.settings.Settings(
client=True,
initial_values={
# Disable PUSH_PROMISE frames from the server since we don't do anything
# with them for now. Maybe when we support caching?
h2.settings.SettingCodes.ENABLE_PUSH: 0,
# These two are taken from h2 for safe defaults
h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 100,
h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: 65536,
},
)
# Some websites (*cough* Yahoo *cough*) balk at this setting being
# present in the initial handshake since it's not defined in the original
# RFC despite the RFC mandating ignoring settings you don't know about.
del self._h2_state.local_settings[
h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL
]
self._h2_state.initiate_connection()
self._h2_state.increment_flow_control_window(2**24)
self._write_outgoing_data(request)
# Sending the request...
def _send_request_headers(self, request: Request, stream_id: int) -> None:
end_stream = not has_body_headers(request)
# In HTTP/2 the ':authority' pseudo-header is used instead of 'Host'.
# In order to gracefully handle HTTP/1.1 and HTTP/2 we always require
# HTTP/1.1 style headers, and map them appropriately if we end up on
# an HTTP/2 connection.
authority = [v for k, v in request.headers if k.lower() == b"host"][0]
headers = [
(b":method", request.method),
(b":authority", authority),
(b":scheme", request.url.scheme),
(b":path", request.url.target),
] + [
(k.lower(), v)
for k, v in request.headers
if k.lower()
not in (
b"host",
b"transfer-encoding",
)
]
self._h2_state.send_headers(stream_id, headers, end_stream=end_stream)
self._h2_state.increment_flow_control_window(2**24, stream_id=stream_id)
self._write_outgoing_data(request)
def _send_request_body(self, request: Request, stream_id: int) -> None:
if not has_body_headers(request):
return
assert isinstance(request.stream, typing.Iterable)
for data in request.stream:
while data:
max_flow = self._wait_for_outgoing_flow(request, stream_id)
chunk_size = min(len(data), max_flow)
chunk, data = data[:chunk_size], data[chunk_size:]
self._h2_state.send_data(stream_id, chunk)
self._write_outgoing_data(request)
self._h2_state.end_stream(stream_id)
self._write_outgoing_data(request)
# Receiving the response...
def _receive_response(
self, request: Request, stream_id: int
) -> typing.Tuple[int, typing.List[typing.Tuple[bytes, bytes]]]:
while True:
event = self._receive_stream_event(request, stream_id)
if isinstance(event, h2.events.ResponseReceived):
break
status_code = 200
headers = []
for k, v in event.headers:
if k == b":status":
status_code = int(v.decode("ascii", errors="ignore"))
elif not k.startswith(b":"):
headers.append((k, v))
return (status_code, headers)
def _receive_response_body(
self, request: Request, stream_id: int
) -> typing.Iterator[bytes]:
while True:
event = self._receive_stream_event(request, stream_id)
if isinstance(event, h2.events.DataReceived):
amount = event.flow_controlled_length
self._h2_state.acknowledge_received_data(amount, stream_id)
self._write_outgoing_data(request)
yield event.data
elif isinstance(event, (h2.events.StreamEnded, h2.events.StreamReset)):
break
def _receive_stream_event(
self, request: Request, stream_id: int
) -> h2.events.Event:
while not self._events.get(stream_id):
self._receive_events(request, stream_id)
event = self._events[stream_id].pop(0)
# The StreamReset event applies to a single stream.
if hasattr(event, "error_code"):
raise RemoteProtocolError(event)
return event
def _receive_events(
self, request: Request, stream_id: typing.Optional[int] = None
) -> None:
with self._read_lock:
if self._connection_error_event is not None: # pragma: nocover
raise RemoteProtocolError(self._connection_error_event)
# This conditional is a bit icky. We don't want to block reading if we've
# actually got an event to return for a given stream. We need to do that
# check *within* the atomic read lock. Though it also need to be optional,
# because when we call it from `_wait_for_outgoing_flow` we *do* want to
# block until we've available flow control, event when we have events
# pending for the stream ID we're attempting to send on.
if stream_id is None or not self._events.get(stream_id):
events = self._read_incoming_data(request)
for event in events:
event_stream_id = getattr(event, "stream_id", 0)
# The ConnectionTerminatedEvent applies to the entire connection,
# and should be saved so it can be raised on all streams.
if hasattr(event, "error_code") and event_stream_id == 0:
self._connection_error_event = event
raise RemoteProtocolError(event)
if event_stream_id in self._events:
self._events[event_stream_id].append(event)
self._write_outgoing_data(request)
def _response_closed(self, stream_id: int) -> None:
self._max_streams_semaphore.release()
del self._events[stream_id]
with self._state_lock:
if self._state == HTTPConnectionState.ACTIVE and not self._events:
self._state = HTTPConnectionState.IDLE
if self._keepalive_expiry is not None:
now = time.monotonic()
self._expire_at = now + self._keepalive_expiry
if self._used_all_stream_ids: # pragma: nocover
self.close()
def close(self) -> None:
# Note that this method unilaterally closes the connection, and does
# not have any kind of locking in place around it.
self._h2_state.close_connection()
self._state = HTTPConnectionState.CLOSED
self._network_stream.close()
# Wrappers around network read/write operations...
def _read_incoming_data(
self, request: Request
) -> typing.List[h2.events.Event]:
timeouts = request.extensions.get("timeout", {})
timeout = timeouts.get("read", None)
if self._read_exception is not None:
raise self._read_exception # pragma: nocover
try:
data = self._network_stream.read(self.READ_NUM_BYTES, timeout)
if data == b"":
raise RemoteProtocolError("Server disconnected")
except Exception as exc:
# If we get a network error we should:
#
# 1. Save the exception and just raise it immediately on any future reads.
# (For example, this means that a single read timeout or disconnect will
# immediately close all pending streams. Without requiring multiple
# sequential timeouts.)
# 2. Mark the connection as errored, so that we don't accept any other
# incoming requests.
self._read_exception = exc
self._connection_error = True
raise exc
events: typing.List[h2.events.Event] = self._h2_state.receive_data(data)
return events
def _write_outgoing_data(self, request: Request) -> None:
timeouts = request.extensions.get("timeout", {})
timeout = timeouts.get("write", None)
with self._write_lock:
data_to_send = self._h2_state.data_to_send()
if self._write_exception is not None:
raise self._write_exception # pragma: nocover
try:
self._network_stream.write(data_to_send, timeout)
except Exception as exc: # pragma: nocover
# If we get a network error we should:
#
# 1. Save the exception and just raise it immediately on any future write.
# (For example, this means that a single write timeout or disconnect will
# immediately close all pending streams. Without requiring multiple
# sequential timeouts.)
# 2. Mark the connection as errored, so that we don't accept any other
# incoming requests.
self._write_exception = exc
self._connection_error = True
raise exc
# Flow control...
def _wait_for_outgoing_flow(self, request: Request, stream_id: int) -> int:
"""
Returns the maximum allowable outgoing flow for a given stream.
If the allowable flow is zero, then waits on the network until
WindowUpdated frames have increased the flow rate.
https://tools.ietf.org/html/rfc7540#section-6.9
"""
local_flow: int = self._h2_state.local_flow_control_window(stream_id)
max_frame_size: int = self._h2_state.max_outbound_frame_size
flow = min(local_flow, max_frame_size)
while flow == 0:
self._receive_events(request)
local_flow = self._h2_state.local_flow_control_window(stream_id)
max_frame_size = self._h2_state.max_outbound_frame_size
flow = min(local_flow, max_frame_size)
return flow
# Interface for connection pooling...
def can_handle_request(self, origin: Origin) -> bool:
return origin == self._origin
def is_available(self) -> bool:
return (
self._state != HTTPConnectionState.CLOSED
and not self._connection_error
and not self._used_all_stream_ids
)
def has_expired(self) -> bool:
now = time.monotonic()
return self._expire_at is not None and now > self._expire_at
def is_idle(self) -> bool:
return self._state == HTTPConnectionState.IDLE
def is_closed(self) -> bool:
return self._state == HTTPConnectionState.CLOSED
def info(self) -> str:
origin = str(self._origin)
return (
f"{origin!r}, HTTP/2, {self._state.name}, "
f"Request Count: {self._request_count}"
)
def __repr__(self) -> str:
class_name = self.__class__.__name__
origin = str(self._origin)
return (
f"<{class_name} [{origin!r}, {self._state.name}, "
f"Request Count: {self._request_count}]>"
)
# These context managers are not used in the standard flow, but are
# useful for testing or working with connection instances directly.
def __enter__(self) -> "HTTP2Connection":
return self
def __exit__(
self,
exc_type: typing.Optional[typing.Type[BaseException]] = None,
exc_value: typing.Optional[BaseException] = None,
traceback: typing.Optional[types.TracebackType] = None,
) -> None:
self.close()
class HTTP2ConnectionByteStream:
def __init__(
self, connection: HTTP2Connection, request: Request, stream_id: int
) -> None:
self._connection = connection
self._request = request
self._stream_id = stream_id
self._closed = False
def __iter__(self) -> typing.Iterator[bytes]:
kwargs = {"request": self._request, "stream_id": self._stream_id}
try:
with Trace("http2.receive_response_body", self._request, kwargs):
for chunk in self._connection._receive_response_body(
request=self._request, stream_id=self._stream_id
):
yield chunk
except BaseException as exc:
# If we get an exception while streaming the response,
# we want to close the response (and possibly the connection)
# before raising that exception.
self.close()
raise exc
def close(self) -> None:
if not self._closed:
self._closed = True
kwargs = {"stream_id": self._stream_id}
with Trace("http2.response_closed", self._request, kwargs):
self._connection._response_closed(stream_id=self._stream_id)