diff options
Diffstat (limited to 'src/core/ext/transport/chttp2/transport/internal.h')
| -rw-r--r-- | src/core/ext/transport/chttp2/transport/internal.h | 206 |
1 files changed, 115 insertions, 91 deletions
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 32a13df48c..8a83f4894c 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -36,6 +36,7 @@ #include "src/core/ext/transport/chttp2/transport/hpack_parser.h" #include "src/core/ext/transport/chttp2/transport/incoming_metadata.h" #include "src/core/ext/transport/chttp2/transport/stream_map.h" +#include "src/core/lib/channel/channelz.h" #include "src/core/lib/compression/stream_compression.h" #include "src/core/lib/gprpp/manual_constructor.h" #include "src/core/lib/iomgr/combiner.h" @@ -106,8 +107,8 @@ const char* grpc_chttp2_initiate_write_reason_string( grpc_chttp2_initiate_write_reason reason); typedef struct { - grpc_closure_list lists[GRPC_CHTTP2_PCL_COUNT]; - uint64_t inflight_id; + grpc_closure_list lists[GRPC_CHTTP2_PCL_COUNT] = {}; + uint64_t inflight_id = 0; } grpc_chttp2_ping_queue; typedef struct { @@ -250,6 +251,8 @@ class Chttp2IncomingByteStream : public ByteStream { static void NextLocked(void* arg, grpc_error* error_ignored); static void OrphanLocked(void* arg, grpc_error* error_ignored); + void MaybeCreateStreamDecompressionCtx(); + grpc_chttp2_transport* transport_; // Immutable. grpc_chttp2_stream* stream_; // Immutable. @@ -281,34 +284,41 @@ typedef enum { } grpc_chttp2_keepalive_state; struct grpc_chttp2_transport { + grpc_chttp2_transport(const grpc_channel_args* channel_args, + grpc_endpoint* ep, bool is_client, + grpc_resource_user* resource_user); + ~grpc_chttp2_transport(); + grpc_transport base; /* must be first */ gpr_refcount refs; grpc_endpoint* ep; char* peer_string; + grpc_resource_user* resource_user; + grpc_combiner* combiner; - grpc_closure* notify_on_receive_settings; + grpc_closure* notify_on_receive_settings = nullptr; /** write execution state of the transport */ - grpc_chttp2_write_state write_state; + grpc_chttp2_write_state write_state = GRPC_CHTTP2_WRITE_STATE_IDLE; /** is this the first write in a series of writes? set when we initiate writing from idle, cleared when we initiate writing from writing+more */ - bool is_first_write_in_batch; + bool is_first_write_in_batch = false; /** is the transport destroying itself? */ - uint8_t destroying; + uint8_t destroying = false; /** has the upper layer closed the transport? */ - grpc_error* closed_with_error; + grpc_error* closed_with_error = GRPC_ERROR_NONE; /** is there a read request to the endpoint outstanding? */ - uint8_t endpoint_reading; + uint8_t endpoint_reading = 1; - grpc_chttp2_optimization_target opt_target; + grpc_chttp2_optimization_target opt_target = GRPC_CHTTP2_OPTIMIZE_FOR_LATENCY; /** various lists of streams */ - grpc_chttp2_stream_list lists[STREAM_LIST_COUNT]; + grpc_chttp2_stream_list lists[STREAM_LIST_COUNT] = {}; /** maps stream id to grpc_chttp2_stream objects */ grpc_chttp2_stream_map stream_map; @@ -325,7 +335,7 @@ struct grpc_chttp2_transport { /** address to place a newly accepted stream - set and unset by grpc_chttp2_parsing_accept_stream; used by init_stream to publish the accepted server stream */ - grpc_chttp2_stream** accepting_stream; + grpc_chttp2_stream** accepting_stream = nullptr; struct { /* accept stream callback */ @@ -349,41 +359,43 @@ struct grpc_chttp2_transport { /** how much data are we willing to buffer when the WRITE_BUFFER_HINT is set? */ - uint32_t write_buffer_size; + uint32_t write_buffer_size = grpc_core::chttp2::kDefaultWindow; /** Set to a grpc_error object if a goaway frame is received. By default, set * to GRPC_ERROR_NONE */ - grpc_error* goaway_error; + grpc_error* goaway_error = GRPC_ERROR_NONE; - grpc_chttp2_sent_goaway_state sent_goaway_state; + grpc_chttp2_sent_goaway_state sent_goaway_state = GRPC_CHTTP2_NO_GOAWAY_SEND; /** are the local settings dirty and need to be sent? */ - bool dirtied_local_settings; + bool dirtied_local_settings = true; /** have local settings been sent? */ - bool sent_local_settings; - /** bitmask of setting indexes to send out */ - uint32_t force_send_settings; + bool sent_local_settings = false; + /** bitmask of setting indexes to send out + Hack: it's common for implementations to assume 65536 bytes initial send + window -- this should by rights be 0 */ + uint32_t force_send_settings = 1 << GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE; /** settings values */ uint32_t settings[GRPC_NUM_SETTING_SETS][GRPC_CHTTP2_NUM_SETTINGS]; /** what is the next stream id to be allocated by this peer? copied to next_stream_id in parsing when parsing commences */ - uint32_t next_stream_id; + uint32_t next_stream_id = 0; /** last new stream id */ - uint32_t last_new_stream_id; + uint32_t last_new_stream_id = 0; /** ping queues for various ping insertion points */ - grpc_chttp2_ping_queue ping_queue; + grpc_chttp2_ping_queue ping_queue = grpc_chttp2_ping_queue(); grpc_chttp2_repeated_ping_policy ping_policy; grpc_chttp2_repeated_ping_state ping_state; - uint64_t ping_ctr; /* unique id for pings */ + uint64_t ping_ctr = 0; /* unique id for pings */ grpc_closure retry_initiate_ping_locked; /** ping acks */ - size_t ping_ack_count; - size_t ping_ack_capacity; - uint64_t* ping_acks; + size_t ping_ack_count = 0; + size_t ping_ack_capacity = 0; + uint64_t* ping_acks = nullptr; grpc_chttp2_server_ping_recv_state ping_recv_state; /** parser for headers */ @@ -409,22 +421,22 @@ struct grpc_chttp2_transport { int64_t initial_window_update = 0; /* deframing */ - grpc_chttp2_deframe_transport_state deframe_state; - uint8_t incoming_frame_type; - uint8_t incoming_frame_flags; - uint8_t header_eof; - bool is_first_frame; - uint32_t expect_continuation_stream_id; - uint32_t incoming_frame_size; - uint32_t incoming_stream_id; + grpc_chttp2_deframe_transport_state deframe_state = GRPC_DTS_CLIENT_PREFIX_0; + uint8_t incoming_frame_type = 0; + uint8_t incoming_frame_flags = 0; + uint8_t header_eof = 0; + bool is_first_frame = true; + uint32_t expect_continuation_stream_id = 0; + uint32_t incoming_frame_size = 0; + uint32_t incoming_stream_id = 0; /* active parser */ - void* parser_data; - grpc_chttp2_stream* incoming_stream; + void* parser_data = nullptr; + grpc_chttp2_stream* incoming_stream = nullptr; grpc_error* (*parser)(void* parser_user_data, grpc_chttp2_transport* t, grpc_chttp2_stream* s, grpc_slice slice, int is_last); - grpc_chttp2_write_cb* write_cb_pool; + grpc_chttp2_write_cb* write_cb_pool = nullptr; /* bdp estimator */ grpc_closure next_bdp_ping_timer_expired_locked; @@ -433,23 +445,23 @@ struct grpc_chttp2_transport { /* if non-NULL, close the transport with this error when writes are finished */ - grpc_error* close_transport_on_writes_finished; + grpc_error* close_transport_on_writes_finished = GRPC_ERROR_NONE; /* a list of closures to run after writes are finished */ - grpc_closure_list run_after_write; + grpc_closure_list run_after_write = GRPC_CLOSURE_LIST_INIT; /* buffer pool state */ /** have we scheduled a benign cleanup? */ - bool benign_reclaimer_registered; + bool benign_reclaimer_registered = false; /** have we scheduled a destructive cleanup? */ - bool destructive_reclaimer_registered; + bool destructive_reclaimer_registered = false; /** benign cleanup closure */ grpc_closure benign_reclaimer_locked; /** destructive cleanup closure */ grpc_closure destructive_reclaimer_locked; /* next bdp ping timer */ - bool have_next_bdp_ping_timer; + bool have_next_bdp_ping_timer = false; grpc_timer next_bdp_ping_timer; /* keep-alive ping support */ @@ -470,10 +482,12 @@ struct grpc_chttp2_transport { /** grace period for a ping to complete before watchdog kicks in */ grpc_millis keepalive_timeout; /** if keepalive pings are allowed when there's no outstanding streams */ - bool keepalive_permit_without_calls; + bool keepalive_permit_without_calls = false; /** keep-alive state machine state */ grpc_chttp2_keepalive_state keepalive_state; grpc_core::ContextList* cl; + grpc_core::RefCountedPtr<grpc_core::channelz::SocketNode> channelz_socket; + uint32_t num_messages_in_next_write = 0; }; typedef enum { @@ -484,6 +498,10 @@ typedef enum { } grpc_published_metadata_method; struct grpc_chttp2_stream { + grpc_chttp2_stream(grpc_chttp2_transport* t, grpc_stream_refcount* refcount, + const void* server_data, gpr_arena* arena); + ~grpc_chttp2_stream(); + void* context; grpc_chttp2_transport* t; grpc_stream_refcount* refcount; @@ -492,59 +510,63 @@ struct grpc_chttp2_stream { grpc_closure* destroy_stream_arg; grpc_chttp2_stream_link links[STREAM_LIST_COUNT]; - uint8_t included[STREAM_LIST_COUNT]; + uint8_t included[STREAM_LIST_COUNT] = {}; /** HTTP2 stream id for this stream, or zero if one has not been assigned */ - uint32_t id; + uint32_t id = 0; /** things the upper layers would like to send */ - grpc_metadata_batch* send_initial_metadata; - grpc_closure* send_initial_metadata_finished; - grpc_metadata_batch* send_trailing_metadata; - grpc_closure* send_trailing_metadata_finished; + grpc_metadata_batch* send_initial_metadata = nullptr; + grpc_closure* send_initial_metadata_finished = nullptr; + grpc_metadata_batch* send_trailing_metadata = nullptr; + grpc_closure* send_trailing_metadata_finished = nullptr; grpc_core::OrphanablePtr<grpc_core::ByteStream> fetching_send_message; - uint32_t fetched_send_message_length; - grpc_slice fetching_slice; + uint32_t fetched_send_message_length = 0; + grpc_slice fetching_slice = grpc_empty_slice(); int64_t next_message_end_offset; - int64_t flow_controlled_bytes_written; - int64_t flow_controlled_bytes_flowed; + int64_t flow_controlled_bytes_written = 0; + int64_t flow_controlled_bytes_flowed = 0; grpc_closure complete_fetch_locked; - grpc_closure* fetching_send_message_finished; + grpc_closure* fetching_send_message_finished = nullptr; grpc_metadata_batch* recv_initial_metadata; - grpc_closure* recv_initial_metadata_ready; - bool* trailing_metadata_available; + grpc_closure* recv_initial_metadata_ready = nullptr; + bool* trailing_metadata_available = nullptr; grpc_core::OrphanablePtr<grpc_core::ByteStream>* recv_message; - grpc_closure* recv_message_ready; + grpc_closure* recv_message_ready = nullptr; grpc_metadata_batch* recv_trailing_metadata; - grpc_closure* recv_trailing_metadata_finished; + grpc_closure* recv_trailing_metadata_finished = nullptr; - grpc_transport_stream_stats* collecting_stats; - grpc_transport_stream_stats stats; + grpc_transport_stream_stats* collecting_stats = nullptr; + grpc_transport_stream_stats stats = grpc_transport_stream_stats(); /** Is this stream closed for writing. */ - bool write_closed; + bool write_closed = false; /** Is this stream reading half-closed. */ - bool read_closed; + bool read_closed = false; /** Are all published incoming byte streams closed. */ - bool all_incoming_byte_streams_finished; + bool all_incoming_byte_streams_finished = false; /** Has this stream seen an error. If true, then pending incoming frames can be thrown away. */ - bool seen_error; + bool seen_error = false; /** Are we buffering writes on this stream? If yes, we won't become writable until there's enough queued up in the flow_controlled_buffer */ - bool write_buffering; + bool write_buffering = false; /** Has trailing metadata been received. */ - bool received_trailing_metadata; + bool received_trailing_metadata = false; + + /* have we sent or received the EOS bit? */ + bool eos_received = false; + bool eos_sent = false; /** the error that resulted in this stream being read-closed */ - grpc_error* read_closed_error; + grpc_error* read_closed_error = GRPC_ERROR_NONE; /** the error that resulted in this stream being write-closed */ - grpc_error* write_closed_error; + grpc_error* write_closed_error = GRPC_ERROR_NONE; - grpc_published_metadata_method published_metadata[2]; - bool final_metadata_requested; + grpc_published_metadata_method published_metadata[2] = {}; + bool final_metadata_requested = false; grpc_chttp2_incoming_metadata_buffer metadata_buffer[2]; @@ -554,33 +576,33 @@ struct grpc_chttp2_stream { * Accessed only by application thread when stream->pending_byte_stream == * true */ grpc_slice_buffer unprocessed_incoming_frames_buffer; - grpc_closure* on_next; /* protected by t combiner */ - bool pending_byte_stream; /* protected by t combiner */ + grpc_closure* on_next = nullptr; /* protected by t combiner */ + bool pending_byte_stream = false; /* protected by t combiner */ // cached length of buffer to be used by the transport thread in cases where // stream->pending_byte_stream == true. The value is saved before // application threads are allowed to modify // unprocessed_incoming_frames_buffer - size_t unprocessed_incoming_frames_buffer_cached_length; + size_t unprocessed_incoming_frames_buffer_cached_length = 0; grpc_closure reset_byte_stream; - grpc_error* byte_stream_error; /* protected by t combiner */ - bool received_last_frame; /* protected by t combiner */ + grpc_error* byte_stream_error = GRPC_ERROR_NONE; /* protected by t combiner */ + bool received_last_frame = false; /* protected by t combiner */ - grpc_millis deadline; + grpc_millis deadline = GRPC_MILLIS_INF_FUTURE; /** saw some stream level error */ - grpc_error* forced_close_error; + grpc_error* forced_close_error = GRPC_ERROR_NONE; /** how many header frames have we received? */ - uint8_t header_frames_received; + uint8_t header_frames_received = 0; /** parsing state for data frames */ /* Accessed only by transport thread when stream->pending_byte_stream == false * Accessed only by application thread when stream->pending_byte_stream == * true */ grpc_chttp2_data_parser data_parser; /** number of bytes received - reset at end of parse thread execution */ - int64_t received_bytes; + int64_t received_bytes = 0; - bool sent_initial_metadata; - bool sent_trailing_metadata; + bool sent_initial_metadata = false; + bool sent_trailing_metadata = false; grpc_core::PolymorphicManualConstructor< grpc_core::chttp2::StreamFlowControlBase, @@ -590,32 +612,34 @@ struct grpc_chttp2_stream { grpc_slice_buffer flow_controlled_buffer; - grpc_chttp2_write_cb* on_flow_controlled_cbs; - grpc_chttp2_write_cb* on_write_finished_cbs; - grpc_chttp2_write_cb* finish_after_write; - size_t sending_bytes; + grpc_chttp2_write_cb* on_flow_controlled_cbs = nullptr; + grpc_chttp2_write_cb* on_write_finished_cbs = nullptr; + grpc_chttp2_write_cb* finish_after_write = nullptr; + size_t sending_bytes = 0; /* Stream compression method to be used. */ - grpc_stream_compression_method stream_compression_method; + grpc_stream_compression_method stream_compression_method = + GRPC_STREAM_COMPRESSION_IDENTITY_COMPRESS; /* Stream decompression method to be used. */ - grpc_stream_compression_method stream_decompression_method; + grpc_stream_compression_method stream_decompression_method = + GRPC_STREAM_COMPRESSION_IDENTITY_COMPRESS; /** Stream compression decompress context */ - grpc_stream_compression_context* stream_decompression_ctx; + grpc_stream_compression_context* stream_decompression_ctx = nullptr; /** Stream compression compress context */ - grpc_stream_compression_context* stream_compression_ctx; + grpc_stream_compression_context* stream_compression_ctx = nullptr; /** Buffer storing data that is compressed but not sent */ grpc_slice_buffer compressed_data_buffer; /** Amount of uncompressed bytes sent out when compressed_data_buffer is * emptied */ - size_t uncompressed_data_size; + size_t uncompressed_data_size = 0; /** Temporary buffer storing decompressed data */ grpc_slice_buffer decompressed_data_buffer; /** Whether bytes stored in unprocessed_incoming_byte_stream is decompressed */ - bool unprocessed_incoming_frames_decompressed; + bool unprocessed_incoming_frames_decompressed = false; /** gRPC header bytes that are already decompressed */ - size_t decompressed_header_bytes; + size_t decompressed_header_bytes = 0; }; /** Transport writing call flow: |