1 /*
2 *
3 * Copyright 2015 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 #ifndef GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H
20 #define GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H
21
22 #include <grpc/support/port_platform.h>
23
24 #include <assert.h>
25 #include <stdbool.h>
26
27 #include "src/core/ext/transport/chttp2/transport/flow_control.h"
28 #include "src/core/ext/transport/chttp2/transport/frame.h"
29 #include "src/core/ext/transport/chttp2/transport/frame_data.h"
30 #include "src/core/ext/transport/chttp2/transport/frame_goaway.h"
31 #include "src/core/ext/transport/chttp2/transport/frame_ping.h"
32 #include "src/core/ext/transport/chttp2/transport/frame_rst_stream.h"
33 #include "src/core/ext/transport/chttp2/transport/frame_settings.h"
34 #include "src/core/ext/transport/chttp2/transport/frame_window_update.h"
35 #include "src/core/ext/transport/chttp2/transport/hpack_encoder.h"
36 #include "src/core/ext/transport/chttp2/transport/hpack_parser.h"
37 #include "src/core/ext/transport/chttp2/transport/incoming_metadata.h"
38 #include "src/core/ext/transport/chttp2/transport/stream_map.h"
39 #include "src/core/lib/channel/channelz.h"
40 #include "src/core/lib/compression/stream_compression.h"
41 #include "src/core/lib/gprpp/manual_constructor.h"
42 #include "src/core/lib/iomgr/combiner.h"
43 #include "src/core/lib/iomgr/endpoint.h"
44 #include "src/core/lib/iomgr/timer.h"
45 #include "src/core/lib/transport/connectivity_state.h"
46 #include "src/core/lib/transport/transport_impl.h"
47
48 namespace grpc_core {
49 class ContextList;
50 }
51
52 /* streams are kept in various linked lists depending on what things need to
53 happen to them... this enum labels each list */
54 typedef enum {
55 /* If a stream is in the following two lists, an explicit ref is associated
56 with the stream */
57 GRPC_CHTTP2_LIST_WRITABLE,
58 GRPC_CHTTP2_LIST_WRITING,
59 /* No additional ref is taken for the following refs. Make sure to remove the
60 stream from these lists when the stream is removed. */
61 GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT,
62 GRPC_CHTTP2_LIST_STALLED_BY_STREAM,
63 /** streams that are waiting to start because there are too many concurrent
64 streams on the connection */
65 GRPC_CHTTP2_LIST_WAITING_FOR_CONCURRENCY,
66 STREAM_LIST_COUNT /* must be last */
67 } grpc_chttp2_stream_list_id;
68
69 typedef enum {
70 GRPC_CHTTP2_WRITE_STATE_IDLE,
71 GRPC_CHTTP2_WRITE_STATE_WRITING,
72 GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE,
73 } grpc_chttp2_write_state;
74
75 typedef enum {
76 GRPC_CHTTP2_OPTIMIZE_FOR_LATENCY,
77 GRPC_CHTTP2_OPTIMIZE_FOR_THROUGHPUT,
78 } grpc_chttp2_optimization_target;
79
80 typedef enum {
81 GRPC_CHTTP2_PCL_INITIATE = 0,
82 GRPC_CHTTP2_PCL_NEXT,
83 GRPC_CHTTP2_PCL_INFLIGHT,
84 GRPC_CHTTP2_PCL_COUNT /* must be last */
85 } grpc_chttp2_ping_closure_list;
86
87 typedef enum {
88 GRPC_CHTTP2_INITIATE_WRITE_INITIAL_WRITE,
89 GRPC_CHTTP2_INITIATE_WRITE_START_NEW_STREAM,
90 GRPC_CHTTP2_INITIATE_WRITE_SEND_MESSAGE,
91 GRPC_CHTTP2_INITIATE_WRITE_SEND_INITIAL_METADATA,
92 GRPC_CHTTP2_INITIATE_WRITE_SEND_TRAILING_METADATA,
93 GRPC_CHTTP2_INITIATE_WRITE_RETRY_SEND_PING,
94 GRPC_CHTTP2_INITIATE_WRITE_CONTINUE_PINGS,
95 GRPC_CHTTP2_INITIATE_WRITE_GOAWAY_SENT,
96 GRPC_CHTTP2_INITIATE_WRITE_RST_STREAM,
97 GRPC_CHTTP2_INITIATE_WRITE_CLOSE_FROM_API,
98 GRPC_CHTTP2_INITIATE_WRITE_STREAM_FLOW_CONTROL,
99 GRPC_CHTTP2_INITIATE_WRITE_TRANSPORT_FLOW_CONTROL,
100 GRPC_CHTTP2_INITIATE_WRITE_SEND_SETTINGS,
101 GRPC_CHTTP2_INITIATE_WRITE_FLOW_CONTROL_UNSTALLED_BY_SETTING,
102 GRPC_CHTTP2_INITIATE_WRITE_FLOW_CONTROL_UNSTALLED_BY_UPDATE,
103 GRPC_CHTTP2_INITIATE_WRITE_APPLICATION_PING,
104 GRPC_CHTTP2_INITIATE_WRITE_BDP_PING,
105 GRPC_CHTTP2_INITIATE_WRITE_KEEPALIVE_PING,
106 GRPC_CHTTP2_INITIATE_WRITE_TRANSPORT_FLOW_CONTROL_UNSTALLED,
107 GRPC_CHTTP2_INITIATE_WRITE_PING_RESPONSE,
108 GRPC_CHTTP2_INITIATE_WRITE_FORCE_RST_STREAM,
109 } grpc_chttp2_initiate_write_reason;
110
111 const char* grpc_chttp2_initiate_write_reason_string(
112 grpc_chttp2_initiate_write_reason reason);
113
114 struct grpc_chttp2_ping_queue {
115 grpc_closure_list lists[GRPC_CHTTP2_PCL_COUNT] = {};
116 uint64_t inflight_id = 0;
117 };
118 struct grpc_chttp2_repeated_ping_policy {
119 int max_pings_without_data;
120 int max_ping_strikes;
121 grpc_millis min_recv_ping_interval_without_data;
122 };
123 struct grpc_chttp2_repeated_ping_state {
124 grpc_millis last_ping_sent_time;
125 int pings_before_data_required;
126 grpc_timer delayed_ping_timer;
127 bool is_delayed_ping_timer_set;
128 };
129 struct grpc_chttp2_server_ping_recv_state {
130 grpc_millis last_ping_recv_time;
131 int ping_strikes;
132 };
133 /* deframer state for the overall http2 stream of bytes */
134 typedef enum {
135 /* prefix: one entry per http2 connection prefix byte */
136 GRPC_DTS_CLIENT_PREFIX_0 = 0,
137 GRPC_DTS_CLIENT_PREFIX_1,
138 GRPC_DTS_CLIENT_PREFIX_2,
139 GRPC_DTS_CLIENT_PREFIX_3,
140 GRPC_DTS_CLIENT_PREFIX_4,
141 GRPC_DTS_CLIENT_PREFIX_5,
142 GRPC_DTS_CLIENT_PREFIX_6,
143 GRPC_DTS_CLIENT_PREFIX_7,
144 GRPC_DTS_CLIENT_PREFIX_8,
145 GRPC_DTS_CLIENT_PREFIX_9,
146 GRPC_DTS_CLIENT_PREFIX_10,
147 GRPC_DTS_CLIENT_PREFIX_11,
148 GRPC_DTS_CLIENT_PREFIX_12,
149 GRPC_DTS_CLIENT_PREFIX_13,
150 GRPC_DTS_CLIENT_PREFIX_14,
151 GRPC_DTS_CLIENT_PREFIX_15,
152 GRPC_DTS_CLIENT_PREFIX_16,
153 GRPC_DTS_CLIENT_PREFIX_17,
154 GRPC_DTS_CLIENT_PREFIX_18,
155 GRPC_DTS_CLIENT_PREFIX_19,
156 GRPC_DTS_CLIENT_PREFIX_20,
157 GRPC_DTS_CLIENT_PREFIX_21,
158 GRPC_DTS_CLIENT_PREFIX_22,
159 GRPC_DTS_CLIENT_PREFIX_23,
160 /* frame header byte 0... */
161 /* must follow from the prefix states */
162 GRPC_DTS_FH_0,
163 GRPC_DTS_FH_1,
164 GRPC_DTS_FH_2,
165 GRPC_DTS_FH_3,
166 GRPC_DTS_FH_4,
167 GRPC_DTS_FH_5,
168 GRPC_DTS_FH_6,
169 GRPC_DTS_FH_7,
170 /* ... frame header byte 8 */
171 GRPC_DTS_FH_8,
172 /* inside a http2 frame */
173 GRPC_DTS_FRAME
174 } grpc_chttp2_deframe_transport_state;
175
176 struct grpc_chttp2_stream_list {
177 grpc_chttp2_stream* head;
178 grpc_chttp2_stream* tail;
179 };
180 struct grpc_chttp2_stream_link {
181 grpc_chttp2_stream* next;
182 grpc_chttp2_stream* prev;
183 };
184 /* We keep several sets of connection wide parameters */
185 typedef enum {
186 /* The settings our peer has asked for (and we have acked) */
187 GRPC_PEER_SETTINGS = 0,
188 /* The settings we'd like to have */
189 GRPC_LOCAL_SETTINGS,
190 /* The settings we've published to our peer */
191 GRPC_SENT_SETTINGS,
192 /* The settings the peer has acked */
193 GRPC_ACKED_SETTINGS,
194 GRPC_NUM_SETTING_SETS
195 } grpc_chttp2_setting_set;
196
197 typedef enum {
198 GRPC_CHTTP2_NO_GOAWAY_SEND,
199 GRPC_CHTTP2_GOAWAY_SEND_SCHEDULED,
200 GRPC_CHTTP2_GOAWAY_SENT,
201 } grpc_chttp2_sent_goaway_state;
202
203 typedef struct grpc_chttp2_write_cb {
204 int64_t call_at_byte;
205 grpc_closure* closure;
206 struct grpc_chttp2_write_cb* next;
207 } grpc_chttp2_write_cb;
208
209 namespace grpc_core {
210
211 class Chttp2IncomingByteStream : public ByteStream {
212 public:
213 Chttp2IncomingByteStream(grpc_chttp2_transport* transport,
214 grpc_chttp2_stream* stream, uint32_t frame_size,
215 uint32_t flags);
216
217 void Orphan() override;
218
219 bool Next(size_t max_size_hint, grpc_closure* on_complete) override;
220 grpc_error* Pull(grpc_slice* slice) override;
221 void Shutdown(grpc_error* error) override;
222
223 // TODO(roth): When I converted this class to C++, I wanted to make it
224 // inherit from RefCounted or InternallyRefCounted instead of continuing
225 // to use its own custom ref-counting code. However, that would require
226 // using multiple inheritance, which sucks in general. And to make matters
227 // worse, it causes problems with our New<> and Delete<> wrappers.
228 // Specifically, unless RefCounted is first in the list of parent classes,
229 // it will see a different value of the address of the object than the one
230 // we actually allocated, in which case gpr_free() will be called on a
231 // different address than the one we got from gpr_malloc(), thus causing a
232 // crash. Given the fragility of depending on that, as well as a desire to
233 // avoid multiple inheritance in general, I've decided to leave this
234 // alone for now. We can revisit this once we're able to link against
235 // libc++, at which point we can eliminate New<> and Delete<> and
236 // switch to std::shared_ptr<>.
Ref()237 void Ref() { refs_.Ref(); }
Unref()238 void Unref() {
239 if (GPR_UNLIKELY(refs_.Unref())) {
240 delete this;
241 }
242 }
243
244 void PublishError(grpc_error* error);
245
246 grpc_error* Push(const grpc_slice& slice, grpc_slice* slice_out);
247
248 grpc_error* Finished(grpc_error* error, bool reset_on_error);
249
remaining_bytes()250 uint32_t remaining_bytes() const { return remaining_bytes_; }
251
252 private:
253 static void NextLocked(void* arg, grpc_error* error_ignored);
254 static void OrphanLocked(void* arg, grpc_error* error_ignored);
255
256 void MaybeCreateStreamDecompressionCtx();
257
258 grpc_chttp2_transport* transport_; // Immutable.
259 grpc_chttp2_stream* stream_; // Immutable.
260
261 grpc_core::RefCount refs_;
262
263 /* Accessed only by transport thread when stream->pending_byte_stream == false
264 * Accessed only by application thread when stream->pending_byte_stream ==
265 * true */
266 uint32_t remaining_bytes_;
267
268 /* Accessed only by transport thread when stream->pending_byte_stream == false
269 * Accessed only by application thread when stream->pending_byte_stream ==
270 * true */
271 struct {
272 grpc_closure closure;
273 size_t max_size_hint;
274 grpc_closure* on_complete;
275 } next_action_;
276 grpc_closure destroy_action_;
277 };
278
279 } // namespace grpc_core
280
281 typedef enum {
282 GRPC_CHTTP2_KEEPALIVE_STATE_WAITING,
283 GRPC_CHTTP2_KEEPALIVE_STATE_PINGING,
284 GRPC_CHTTP2_KEEPALIVE_STATE_DYING,
285 GRPC_CHTTP2_KEEPALIVE_STATE_DISABLED,
286 } grpc_chttp2_keepalive_state;
287
288 struct grpc_chttp2_transport {
289 grpc_chttp2_transport(const grpc_channel_args* channel_args,
290 grpc_endpoint* ep, bool is_client,
291 grpc_resource_user* resource_user);
292 ~grpc_chttp2_transport();
293
294 grpc_transport base; /* must be first */
295 grpc_core::RefCount refs;
296 grpc_endpoint* ep;
297 std::string peer_string;
298
299 grpc_resource_user* resource_user;
300
301 grpc_core::Combiner* combiner;
302
303 grpc_closure* notify_on_receive_settings = nullptr;
304
305 /** write execution state of the transport */
306 grpc_chttp2_write_state write_state = GRPC_CHTTP2_WRITE_STATE_IDLE;
307
308 /** is the transport destroying itself? */
309 uint8_t destroying = false;
310 /** has the upper layer closed the transport? */
311 grpc_error* closed_with_error = GRPC_ERROR_NONE;
312
313 /** is there a read request to the endpoint outstanding? */
314 uint8_t endpoint_reading = 1;
315
316 /** various lists of streams */
317 grpc_chttp2_stream_list lists[STREAM_LIST_COUNT] = {};
318
319 /** maps stream id to grpc_chttp2_stream objects */
320 grpc_chttp2_stream_map stream_map;
321
322 grpc_closure write_action_begin_locked;
323 grpc_closure write_action;
324 grpc_closure write_action_end_locked;
325
326 grpc_closure read_action_locked;
327
328 /** incoming read bytes */
329 grpc_slice_buffer read_buffer;
330
331 /** address to place a newly accepted stream - set and unset by
332 grpc_chttp2_parsing_accept_stream; used by init_stream to
333 publish the accepted server stream */
334 grpc_chttp2_stream** accepting_stream = nullptr;
335
336 /* accept stream callback */
337 void (*accept_stream_cb)(void* user_data, grpc_transport* transport,
338 const void* server_data);
339 void* accept_stream_cb_user_data;
340
341 /** connectivity tracking */
342 grpc_core::ConnectivityStateTracker state_tracker;
343
344 /** data to write now */
345 grpc_slice_buffer outbuf;
346 /** hpack encoding */
347 grpc_chttp2_hpack_compressor hpack_compressor;
348 /** is this a client? */
349 bool is_client;
350
351 /** data to write next write */
352 grpc_slice_buffer qbuf;
353
354 /** how much data are we willing to buffer when the WRITE_BUFFER_HINT is set?
355 */
356 uint32_t write_buffer_size = grpc_core::chttp2::kDefaultWindow;
357
358 /** Set to a grpc_error object if a goaway frame is received. By default, set
359 * to GRPC_ERROR_NONE */
360 grpc_error* goaway_error = GRPC_ERROR_NONE;
361
362 grpc_chttp2_sent_goaway_state sent_goaway_state = GRPC_CHTTP2_NO_GOAWAY_SEND;
363
364 /** are the local settings dirty and need to be sent? */
365 bool dirtied_local_settings = true;
366 /** have local settings been sent? */
367 bool sent_local_settings = false;
368 /** bitmask of setting indexes to send out
369 Hack: it's common for implementations to assume 65536 bytes initial send
370 window -- this should by rights be 0 */
371 uint32_t force_send_settings = 1 << GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;
372 /** settings values */
373 uint32_t settings[GRPC_NUM_SETTING_SETS][GRPC_CHTTP2_NUM_SETTINGS];
374
375 /** what is the next stream id to be allocated by this peer?
376 copied to next_stream_id in parsing when parsing commences */
377 uint32_t next_stream_id = 0;
378
379 /** last new stream id */
380 uint32_t last_new_stream_id = 0;
381
382 /** ping queues for various ping insertion points */
383 grpc_chttp2_ping_queue ping_queue = grpc_chttp2_ping_queue();
384 grpc_chttp2_repeated_ping_policy ping_policy;
385 grpc_chttp2_repeated_ping_state ping_state;
386 uint64_t ping_ctr = 0; /* unique id for pings */
387 grpc_closure retry_initiate_ping_locked;
388
389 /** ping acks */
390 size_t ping_ack_count = 0;
391 size_t ping_ack_capacity = 0;
392 uint64_t* ping_acks = nullptr;
393 grpc_chttp2_server_ping_recv_state ping_recv_state;
394
395 /** parser for headers */
396 grpc_chttp2_hpack_parser hpack_parser;
397 /** simple one shot parsers */
398 union {
399 grpc_chttp2_window_update_parser window_update;
400 grpc_chttp2_settings_parser settings;
401 grpc_chttp2_ping_parser ping;
402 grpc_chttp2_rst_stream_parser rst_stream;
403 } simple;
404 /** parser for goaway frames */
405 grpc_chttp2_goaway_parser goaway_parser;
406
407 grpc_core::PolymorphicManualConstructor<
408 grpc_core::chttp2::TransportFlowControlBase,
409 grpc_core::chttp2::TransportFlowControl,
410 grpc_core::chttp2::TransportFlowControlDisabled>
411 flow_control;
412 /** initial window change. This is tracked as we parse settings frames from
413 * the remote peer. If there is a positive delta, then we will make all
414 * streams readable since they may have become unstalled */
415 int64_t initial_window_update = 0;
416
417 /* deframing */
418 grpc_chttp2_deframe_transport_state deframe_state = GRPC_DTS_CLIENT_PREFIX_0;
419 uint8_t incoming_frame_type = 0;
420 uint8_t incoming_frame_flags = 0;
421 uint8_t header_eof = 0;
422 bool is_first_frame = true;
423 uint32_t expect_continuation_stream_id = 0;
424 uint32_t incoming_frame_size = 0;
425 uint32_t incoming_stream_id = 0;
426
427 /* active parser */
428 void* parser_data = nullptr;
429 grpc_chttp2_stream* incoming_stream = nullptr;
430 grpc_error* (*parser)(void* parser_user_data, grpc_chttp2_transport* t,
431 grpc_chttp2_stream* s, const grpc_slice& slice,
432 int is_last);
433
434 grpc_chttp2_write_cb* write_cb_pool = nullptr;
435
436 /* bdp estimator */
437 bool bdp_ping_blocked =
438 false; /* Is the BDP blocked due to not receiving any data? */
439 grpc_closure next_bdp_ping_timer_expired_locked;
440 grpc_closure start_bdp_ping_locked;
441 grpc_closure finish_bdp_ping_locked;
442
443 /* if non-NULL, close the transport with this error when writes are finished
444 */
445 grpc_error* close_transport_on_writes_finished = GRPC_ERROR_NONE;
446
447 /* a list of closures to run after writes are finished */
448 grpc_closure_list run_after_write = GRPC_CLOSURE_LIST_INIT;
449
450 /* buffer pool state */
451 /** have we scheduled a benign cleanup? */
452 bool benign_reclaimer_registered = false;
453 /** have we scheduled a destructive cleanup? */
454 bool destructive_reclaimer_registered = false;
455 /** benign cleanup closure */
456 grpc_closure benign_reclaimer_locked;
457 /** destructive cleanup closure */
458 grpc_closure destructive_reclaimer_locked;
459
460 /* next bdp ping timer */
461 bool have_next_bdp_ping_timer = false;
462 /** If start_bdp_ping_locked has been called */
463 bool bdp_ping_started = false;
464 grpc_timer next_bdp_ping_timer;
465
466 /* keep-alive ping support */
467 /** Closure to initialize a keepalive ping */
468 grpc_closure init_keepalive_ping_locked;
469 /** Closure to run when the keepalive ping is sent */
470 grpc_closure start_keepalive_ping_locked;
471 /** Cousure to run when the keepalive ping ack is received */
472 grpc_closure finish_keepalive_ping_locked;
473 /** Closrue to run when the keepalive ping timeouts */
474 grpc_closure keepalive_watchdog_fired_locked;
475 /** timer to initiate ping events */
476 grpc_timer keepalive_ping_timer;
477 /** watchdog to kill the transport when waiting for the keepalive ping */
478 grpc_timer keepalive_watchdog_timer;
479 /** time duration in between pings */
480 grpc_millis keepalive_time;
481 /** grace period for a ping to complete before watchdog kicks in */
482 grpc_millis keepalive_timeout;
483 /** if keepalive pings are allowed when there's no outstanding streams */
484 bool keepalive_permit_without_calls = false;
485 /** If start_keepalive_ping_locked has been called */
486 bool keepalive_ping_started = false;
487 /** keep-alive state machine state */
488 grpc_chttp2_keepalive_state keepalive_state;
489 grpc_core::ContextList* cl = nullptr;
490 grpc_core::RefCountedPtr<grpc_core::channelz::SocketNode> channelz_socket;
491 uint32_t num_messages_in_next_write = 0;
492 /** The number of pending induced frames (SETTINGS_ACK, PINGS_ACK and
493 * RST_STREAM) in the outgoing buffer (t->qbuf). If this number goes beyond
494 * DEFAULT_MAX_PENDING_INDUCED_FRAMES, we pause reading new frames. We would
495 * only continue reading when we are able to write to the socket again,
496 * thereby reducing the number of induced frames. */
497 uint32_t num_pending_induced_frames = 0;
498 bool reading_paused_on_pending_induced_frames = false;
499 };
500
501 typedef enum {
502 GRPC_METADATA_NOT_PUBLISHED,
503 GRPC_METADATA_SYNTHESIZED_FROM_FAKE,
504 GRPC_METADATA_PUBLISHED_FROM_WIRE,
505 GRPC_METADATA_PUBLISHED_AT_CLOSE
506 } grpc_published_metadata_method;
507
508 struct grpc_chttp2_stream {
509 grpc_chttp2_stream(grpc_chttp2_transport* t, grpc_stream_refcount* refcount,
510 const void* server_data, grpc_core::Arena* arena);
511 ~grpc_chttp2_stream();
512
513 void* context;
514 grpc_chttp2_transport* t;
515 grpc_stream_refcount* refcount;
516 // Reffer is a 0-len structure, simply reffing `t` and `refcount` in its ctor
517 // before initializing the rest of the stream, to avoid cache misses. This
518 // field MUST be right after `t` and `refcount`.
519 struct Reffer {
520 explicit Reffer(grpc_chttp2_stream* s);
521 } reffer;
522
523 grpc_closure destroy_stream;
524 grpc_closure* destroy_stream_arg;
525
526 grpc_chttp2_stream_link links[STREAM_LIST_COUNT];
527 uint8_t included[STREAM_LIST_COUNT] = {};
528
529 /** HTTP2 stream id for this stream, or zero if one has not been assigned */
530 uint32_t id = 0;
531
532 /** things the upper layers would like to send */
533 grpc_metadata_batch* send_initial_metadata = nullptr;
534 grpc_closure* send_initial_metadata_finished = nullptr;
535 grpc_metadata_batch* send_trailing_metadata = nullptr;
536 // TODO(yashykt): Find a better name for the below field and others in this
537 // struct to betteer distinguish inputs, return values, and
538 // internal state.
539 // sent_trailing_metadata_op allows the transport to fill in to the upper
540 // layer whether this stream was able to send its trailing metadata (used for
541 // detecting cancellation on the server-side)..
542 bool* sent_trailing_metadata_op = nullptr;
543 grpc_closure* send_trailing_metadata_finished = nullptr;
544
545 grpc_core::OrphanablePtr<grpc_core::ByteStream> fetching_send_message;
546 uint32_t fetched_send_message_length = 0;
547 grpc_slice fetching_slice = grpc_empty_slice();
548 int64_t next_message_end_offset;
549 int64_t flow_controlled_bytes_written = 0;
550 int64_t flow_controlled_bytes_flowed = 0;
551 grpc_closure complete_fetch_locked;
552 grpc_closure* fetching_send_message_finished = nullptr;
553
554 grpc_metadata_batch* recv_initial_metadata;
555 grpc_closure* recv_initial_metadata_ready = nullptr;
556 bool* trailing_metadata_available = nullptr;
557 grpc_core::OrphanablePtr<grpc_core::ByteStream>* recv_message;
558 grpc_closure* recv_message_ready = nullptr;
559 grpc_metadata_batch* recv_trailing_metadata;
560 grpc_closure* recv_trailing_metadata_finished = nullptr;
561
562 grpc_transport_stream_stats* collecting_stats = nullptr;
563 grpc_transport_stream_stats stats = grpc_transport_stream_stats();
564
565 /** Is this stream closed for writing. */
566 bool write_closed = false;
567 /** Is this stream reading half-closed. */
568 bool read_closed = false;
569 /** Are all published incoming byte streams closed. */
570 bool all_incoming_byte_streams_finished = false;
571 /** Has this stream seen an error.
572 If true, then pending incoming frames can be thrown away. */
573 bool seen_error = false;
574 /** Are we buffering writes on this stream? If yes, we won't become writable
575 until there's enough queued up in the flow_controlled_buffer */
576 bool write_buffering = false;
577
578 /* have we sent or received the EOS bit? */
579 bool eos_received = false;
580 bool eos_sent = false;
581
582 /** the error that resulted in this stream being read-closed */
583 grpc_error* read_closed_error = GRPC_ERROR_NONE;
584 /** the error that resulted in this stream being write-closed */
585 grpc_error* write_closed_error = GRPC_ERROR_NONE;
586
587 grpc_published_metadata_method published_metadata[2] = {};
588 bool final_metadata_requested = false;
589
590 grpc_chttp2_incoming_metadata_buffer metadata_buffer[2];
591
592 grpc_slice_buffer frame_storage; /* protected by t combiner */
593
594 grpc_closure* on_next = nullptr; /* protected by t combiner */
595 bool pending_byte_stream = false; /* protected by t combiner */
596 // cached length of buffer to be used by the transport thread in cases where
597 // stream->pending_byte_stream == true. The value is saved before
598 // application threads are allowed to modify
599 // unprocessed_incoming_frames_buffer
600 size_t unprocessed_incoming_frames_buffer_cached_length = 0;
601 /* Accessed only by transport thread when stream->pending_byte_stream == false
602 * Accessed only by application thread when stream->pending_byte_stream ==
603 * true */
604 grpc_slice_buffer unprocessed_incoming_frames_buffer;
605 grpc_closure reset_byte_stream;
606 grpc_error* byte_stream_error = GRPC_ERROR_NONE; /* protected by t combiner */
607 bool received_last_frame = false; /* protected by t combiner */
608
609 grpc_millis deadline = GRPC_MILLIS_INF_FUTURE;
610
611 /** saw some stream level error */
612 grpc_error* forced_close_error = GRPC_ERROR_NONE;
613 /** how many header frames have we received? */
614 uint8_t header_frames_received = 0;
615 /** parsing state for data frames */
616 /* Accessed only by transport thread when stream->pending_byte_stream == false
617 * Accessed only by application thread when stream->pending_byte_stream ==
618 * true */
619 grpc_chttp2_data_parser data_parser;
620 /** number of bytes received - reset at end of parse thread execution */
621 int64_t received_bytes = 0;
622
623 bool sent_initial_metadata = false;
624 bool sent_trailing_metadata = false;
625
626 grpc_core::PolymorphicManualConstructor<
627 grpc_core::chttp2::StreamFlowControlBase,
628 grpc_core::chttp2::StreamFlowControl,
629 grpc_core::chttp2::StreamFlowControlDisabled>
630 flow_control;
631
632 grpc_slice_buffer flow_controlled_buffer;
633
634 grpc_chttp2_write_cb* on_flow_controlled_cbs = nullptr;
635 grpc_chttp2_write_cb* on_write_finished_cbs = nullptr;
636 grpc_chttp2_write_cb* finish_after_write = nullptr;
637 size_t sending_bytes = 0;
638
639 /* Stream compression method to be used. */
640 grpc_stream_compression_method stream_compression_method =
641 GRPC_STREAM_COMPRESSION_IDENTITY_COMPRESS;
642 /* Stream decompression method to be used. */
643 grpc_stream_compression_method stream_decompression_method =
644 GRPC_STREAM_COMPRESSION_IDENTITY_DECOMPRESS;
645
646 /** Whether bytes stored in unprocessed_incoming_byte_stream is decompressed
647 */
648 bool unprocessed_incoming_frames_decompressed = false;
649 /** Whether the bytes needs to be traced using Fathom */
650 bool traced = false;
651 /** gRPC header bytes that are already decompressed */
652 size_t decompressed_header_bytes = 0;
653 /** Byte counter for number of bytes written */
654 size_t byte_counter = 0;
655
656 /** Amount of uncompressed bytes sent out when compressed_data_buffer is
657 * emptied */
658 size_t uncompressed_data_size;
659 /** Stream compression compress context */
660 grpc_stream_compression_context* stream_compression_ctx;
661 /** Buffer storing data that is compressed but not sent */
662 grpc_slice_buffer compressed_data_buffer;
663
664 /** Stream compression decompress context */
665 grpc_stream_compression_context* stream_decompression_ctx;
666 /** Temporary buffer storing decompressed data.
667 * Initialized, used, and destroyed only when stream uses (non-identity)
668 * compression.
669 */
670 grpc_slice_buffer decompressed_data_buffer;
671 };
672
673 /** Transport writing call flow:
674 grpc_chttp2_initiate_write() is called anywhere that we know bytes need to
675 go out on the wire.
676 If no other write has been started, a task is enqueued onto our workqueue.
677 When that task executes, it obtains the global lock, and gathers the data
678 to write.
679 The global lock is dropped and we do the syscall to write.
680 After writing, a follow-up check is made to see if another round of writing
681 should be performed.
682
683 The actual call chain is documented in the implementation of this function.
684 */
685 void grpc_chttp2_initiate_write(grpc_chttp2_transport* t,
686 grpc_chttp2_initiate_write_reason reason);
687
688 struct grpc_chttp2_begin_write_result {
689 /** are we writing? */
690 bool writing;
691 /** if writing: was it a complete flush (false) or a partial flush (true) */
692 bool partial;
693 /** did we queue any completions as part of beginning the write */
694 bool early_results_scheduled;
695 };
696 grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
697 grpc_chttp2_transport* t);
698 void grpc_chttp2_end_write(grpc_chttp2_transport* t, grpc_error* error);
699
700 /** Process one slice of incoming data; return 1 if the connection is still
701 viable after reading, or 0 if the connection should be torn down */
702 grpc_error* grpc_chttp2_perform_read(grpc_chttp2_transport* t,
703 const grpc_slice& slice);
704
705 bool grpc_chttp2_list_add_writable_stream(grpc_chttp2_transport* t,
706 grpc_chttp2_stream* s);
707 /** Get a writable stream
708 returns non-zero if there was a stream available */
709 bool grpc_chttp2_list_pop_writable_stream(grpc_chttp2_transport* t,
710 grpc_chttp2_stream** s);
711 bool grpc_chttp2_list_remove_writable_stream(grpc_chttp2_transport* t,
712 grpc_chttp2_stream* s);
713
714 bool grpc_chttp2_list_add_writing_stream(grpc_chttp2_transport* t,
715 grpc_chttp2_stream* s);
716 bool grpc_chttp2_list_have_writing_streams(grpc_chttp2_transport* t);
717 bool grpc_chttp2_list_pop_writing_stream(grpc_chttp2_transport* t,
718 grpc_chttp2_stream** s);
719
720 void grpc_chttp2_list_add_written_stream(grpc_chttp2_transport* t,
721 grpc_chttp2_stream* s);
722 bool grpc_chttp2_list_pop_written_stream(grpc_chttp2_transport* t,
723 grpc_chttp2_stream** s);
724
725 void grpc_chttp2_list_add_waiting_for_concurrency(grpc_chttp2_transport* t,
726 grpc_chttp2_stream* s);
727 bool grpc_chttp2_list_pop_waiting_for_concurrency(grpc_chttp2_transport* t,
728 grpc_chttp2_stream** s);
729 void grpc_chttp2_list_remove_waiting_for_concurrency(grpc_chttp2_transport* t,
730 grpc_chttp2_stream* s);
731
732 void grpc_chttp2_list_add_stalled_by_transport(grpc_chttp2_transport* t,
733 grpc_chttp2_stream* s);
734 bool grpc_chttp2_list_pop_stalled_by_transport(grpc_chttp2_transport* t,
735 grpc_chttp2_stream** s);
736 void grpc_chttp2_list_remove_stalled_by_transport(grpc_chttp2_transport* t,
737 grpc_chttp2_stream* s);
738
739 void grpc_chttp2_list_add_stalled_by_stream(grpc_chttp2_transport* t,
740 grpc_chttp2_stream* s);
741 bool grpc_chttp2_list_pop_stalled_by_stream(grpc_chttp2_transport* t,
742 grpc_chttp2_stream** s);
743 bool grpc_chttp2_list_remove_stalled_by_stream(grpc_chttp2_transport* t,
744 grpc_chttp2_stream* s);
745
746 /********* Flow Control ***************/
747
748 // Takes in a flow control action and performs all the needed operations.
749 void grpc_chttp2_act_on_flowctl_action(
750 const grpc_core::chttp2::FlowControlAction& action,
751 grpc_chttp2_transport* t, grpc_chttp2_stream* s);
752
753 /********* End of Flow Control ***************/
754
grpc_chttp2_parsing_lookup_stream(grpc_chttp2_transport * t,uint32_t id)755 inline grpc_chttp2_stream* grpc_chttp2_parsing_lookup_stream(
756 grpc_chttp2_transport* t, uint32_t id) {
757 return static_cast<grpc_chttp2_stream*>(
758 grpc_chttp2_stream_map_find(&t->stream_map, id));
759 }
760 grpc_chttp2_stream* grpc_chttp2_parsing_accept_stream(grpc_chttp2_transport* t,
761 uint32_t id);
762
763 void grpc_chttp2_add_incoming_goaway(grpc_chttp2_transport* t,
764 uint32_t goaway_error,
765 uint32_t last_stream_id,
766 const grpc_slice& goaway_text);
767
768 void grpc_chttp2_parsing_become_skip_parser(grpc_chttp2_transport* t);
769
770 void grpc_chttp2_complete_closure_step(grpc_chttp2_transport* t,
771 grpc_chttp2_stream* s,
772 grpc_closure** pclosure,
773 grpc_error* error, const char* desc);
774
775 #define GRPC_HEADER_SIZE_IN_BYTES 5
776 #define MAX_SIZE_T (~(size_t)0)
777
778 #define GRPC_CHTTP2_CLIENT_CONNECT_STRING "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
779 #define GRPC_CHTTP2_CLIENT_CONNECT_STRLEN \
780 (sizeof(GRPC_CHTTP2_CLIENT_CONNECT_STRING) - 1)
781
782 // extern grpc_core::TraceFlag grpc_http_trace;
783 // extern grpc_core::TraceFlag grpc_flowctl_trace;
784
785 #define GRPC_CHTTP2_IF_TRACING(stmt) \
786 do { \
787 if (GRPC_TRACE_FLAG_ENABLED(grpc_http_trace)) { \
788 (stmt); \
789 } \
790 } while (0)
791
792 void grpc_chttp2_fake_status(grpc_chttp2_transport* t,
793 grpc_chttp2_stream* stream, grpc_error* error);
794 void grpc_chttp2_mark_stream_closed(grpc_chttp2_transport* t,
795 grpc_chttp2_stream* s, int close_reads,
796 int close_writes, grpc_error* error);
797 void grpc_chttp2_start_writing(grpc_chttp2_transport* t);
798
799 #ifndef NDEBUG
800 #define GRPC_CHTTP2_STREAM_REF(stream, reason) \
801 grpc_chttp2_stream_ref(stream, reason)
802 #define GRPC_CHTTP2_STREAM_UNREF(stream, reason) \
803 grpc_chttp2_stream_unref(stream, reason)
804 void grpc_chttp2_stream_ref(grpc_chttp2_stream* s, const char* reason);
805 void grpc_chttp2_stream_unref(grpc_chttp2_stream* s, const char* reason);
806 #else
807 #define GRPC_CHTTP2_STREAM_REF(stream, reason) grpc_chttp2_stream_ref(stream)
808 #define GRPC_CHTTP2_STREAM_UNREF(stream, reason) \
809 grpc_chttp2_stream_unref(stream)
810 void grpc_chttp2_stream_ref(grpc_chttp2_stream* s);
811 void grpc_chttp2_stream_unref(grpc_chttp2_stream* s);
812 #endif
813
814 #ifndef NDEBUG
815 #define GRPC_CHTTP2_REF_TRANSPORT(t, r) \
816 grpc_chttp2_ref_transport(t, r, __FILE__, __LINE__)
817 #define GRPC_CHTTP2_UNREF_TRANSPORT(t, r) \
818 grpc_chttp2_unref_transport(t, r, __FILE__, __LINE__)
grpc_chttp2_unref_transport(grpc_chttp2_transport * t,const char * reason,const char * file,int line)819 inline void grpc_chttp2_unref_transport(grpc_chttp2_transport* t,
820 const char* reason, const char* file,
821 int line) {
822 if (t->refs.Unref(grpc_core::DebugLocation(file, line), reason)) {
823 delete t;
824 }
825 }
grpc_chttp2_ref_transport(grpc_chttp2_transport * t,const char * reason,const char * file,int line)826 inline void grpc_chttp2_ref_transport(grpc_chttp2_transport* t,
827 const char* reason, const char* file,
828 int line) {
829 t->refs.Ref(grpc_core::DebugLocation(file, line), reason);
830 }
831 #else
832 #define GRPC_CHTTP2_REF_TRANSPORT(t, r) grpc_chttp2_ref_transport(t)
833 #define GRPC_CHTTP2_UNREF_TRANSPORT(t, r) grpc_chttp2_unref_transport(t)
grpc_chttp2_unref_transport(grpc_chttp2_transport * t)834 inline void grpc_chttp2_unref_transport(grpc_chttp2_transport* t) {
835 if (t->refs.Unref()) {
836 delete t;
837 }
838 }
grpc_chttp2_ref_transport(grpc_chttp2_transport * t)839 inline void grpc_chttp2_ref_transport(grpc_chttp2_transport* t) {
840 t->refs.Ref();
841 }
842 #endif
843
844 void grpc_chttp2_ack_ping(grpc_chttp2_transport* t, uint64_t id);
845
846 /** Add a new ping strike to ping_recv_state.ping_strikes. If
847 ping_recv_state.ping_strikes > ping_policy.max_ping_strikes, it sends GOAWAY
848 with error code ENHANCE_YOUR_CALM and additional debug data resembling
849 "too_many_pings" followed by immediately closing the connection. */
850 void grpc_chttp2_add_ping_strike(grpc_chttp2_transport* t);
851
852 /** Resets ping clock. Should be called when flushing window updates,
853 * initial/trailing metadata or data frames. For a server, it resets the number
854 * of ping strikes and the last_ping_recv_time. For a ping sender, it resets
855 * pings_before_data_required. */
856 void grpc_chttp2_reset_ping_clock(grpc_chttp2_transport* t);
857
858 /** add a ref to the stream and add it to the writable list;
859 ref will be dropped in writing.c */
860 void grpc_chttp2_mark_stream_writable(grpc_chttp2_transport* t,
861 grpc_chttp2_stream* s);
862
863 void grpc_chttp2_cancel_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s,
864 grpc_error* due_to_error);
865
866 void grpc_chttp2_maybe_complete_recv_initial_metadata(grpc_chttp2_transport* t,
867 grpc_chttp2_stream* s);
868 void grpc_chttp2_maybe_complete_recv_message(grpc_chttp2_transport* t,
869 grpc_chttp2_stream* s);
870 void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_chttp2_transport* t,
871 grpc_chttp2_stream* s);
872
873 void grpc_chttp2_fail_pending_writes(grpc_chttp2_transport* t,
874 grpc_chttp2_stream* s, grpc_error* error);
875
876 /** Set the default keepalive configurations, must only be called at
877 initialization */
878 void grpc_chttp2_config_default_keepalive_args(grpc_channel_args* args,
879 bool is_client);
880
881 void grpc_chttp2_retry_initiate_ping(void* tp, grpc_error* error);
882
883 void schedule_bdp_ping_locked(grpc_chttp2_transport* t);
884
885 #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */
886