// // // Copyright 2015 gRPC authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. // // #ifndef GRPC_SRC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H #define GRPC_SRC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H #include <grpc/support/port_platform.h> #include <stddef.h> #include <stdint.h> #include <memory> #include "absl/strings/string_view.h" #include "absl/types/optional.h" #include <grpc/event_engine/event_engine.h> #include <grpc/event_engine/memory_allocator.h> #include <grpc/grpc.h> #include <grpc/slice.h> #include <grpc/support/time.h> #include "src/core/ext/transport/chttp2/transport/flow_control.h" #include "src/core/ext/transport/chttp2/transport/frame.h" #include "src/core/ext/transport/chttp2/transport/frame_goaway.h" #include "src/core/ext/transport/chttp2/transport/frame_ping.h" #include "src/core/ext/transport/chttp2/transport/frame_rst_stream.h" #include "src/core/ext/transport/chttp2/transport/frame_settings.h" #include "src/core/ext/transport/chttp2/transport/frame_window_update.h" #include "src/core/ext/transport/chttp2/transport/hpack_encoder.h" #include "src/core/ext/transport/chttp2/transport/hpack_parser.h" #include "src/core/ext/transport/chttp2/transport/http2_settings.h" #include "src/core/ext/transport/chttp2/transport/stream_map.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/channelz.h" #include "src/core/lib/debug/trace.h" #include "src/core/lib/gprpp/bitset.h" #include "src/core/lib/gprpp/debug_location.h" #include "src/core/lib/gprpp/ref_counted.h" #include "src/core/lib/gprpp/ref_counted_ptr.h" #include "src/core/lib/gprpp/time.h" #include "src/core/lib/iomgr/closure.h" #include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/iomgr/endpoint.h" #include "src/core/lib/iomgr/error.h" #include "src/core/lib/resource_quota/arena.h" #include "src/core/lib/resource_quota/memory_quota.h" #include "src/core/lib/slice/slice.h" #include "src/core/lib/slice/slice_buffer.h" #include "src/core/lib/surface/init_internally.h" #include "src/core/lib/transport/connectivity_state.h" #include "src/core/lib/transport/metadata_batch.h" #include "src/core/lib/transport/transport.h" #include "src/core/lib/transport/transport_fwd.h" #include "src/core/lib/transport/transport_impl.h" // Flag that this closure barrier may be covering a write in a pollset, and so // we should not complete this closure until we can prove that the write got // scheduled #define CLOSURE_BARRIER_MAY_COVER_WRITE … // First bit of the reference count, stored in the high order bits (with the low // bits being used for flags defined above) #define CLOSURE_BARRIER_FIRST_REF_BIT … namespace grpc_core { class ContextList; } // streams are kept in various linked lists depending on what things need to // happen to them... this enum labels each list grpc_chttp2_stream_list_id; grpc_chttp2_write_state; grpc_chttp2_optimization_target; grpc_chttp2_ping_closure_list; grpc_chttp2_initiate_write_reason; const char* grpc_chttp2_initiate_write_reason_string( grpc_chttp2_initiate_write_reason reason); struct grpc_chttp2_ping_queue { … }; struct grpc_chttp2_repeated_ping_policy { … }; struct grpc_chttp2_repeated_ping_state { … }; struct grpc_chttp2_server_ping_recv_state { … }; // deframer state for the overall http2 stream of bytes grpc_chttp2_deframe_transport_state; struct grpc_chttp2_stream_list { … }; struct grpc_chttp2_stream_link { … }; // We keep several sets of connection wide parameters grpc_chttp2_setting_set; grpc_chttp2_sent_goaway_state; grpc_chttp2_write_cb; grpc_chttp2_keepalive_state; struct grpc_chttp2_transport // TODO(ctiller): #31319 fixed a crash on Linux & Mac whereby iomgr was // accessed after shutdown by chttp2. We've not seen similar behavior on // Windows afaik, but this fix has exposed another refcounting bug whereby // transports leak on Windows and prevent test shutdown. // This hack attempts to compromise between two things that are blocking our CI // from giving us a good quality signal, but are unlikely to be problems for // most customers. We should continue tracking down what's causing the failure, // but this gives us some runway to do so - and given that we're actively // working on removing the problematic code paths, it may be that effort brings // the result we need. #ifndef GPR_WINDOWS : public grpc_core::KeepsGrpcInitialized #endif { … }; grpc_published_metadata_method; struct grpc_chttp2_stream { … }; /// Transport writing call flow: /// grpc_chttp2_initiate_write() is called anywhere that we know bytes need to /// go out on the wire. /// If no other write has been started, a task is enqueued onto our workqueue. /// When that task executes, it obtains the global lock, and gathers the data /// to write. /// The global lock is dropped and we do the syscall to write. /// After writing, a follow-up check is made to see if another round of writing /// should be performed. /// The actual call chain is documented in the implementation of this function. /// void grpc_chttp2_initiate_write(grpc_chttp2_transport* t, grpc_chttp2_initiate_write_reason reason); struct grpc_chttp2_begin_write_result { … }; grpc_chttp2_begin_write_result grpc_chttp2_begin_write( grpc_chttp2_transport* t); void grpc_chttp2_end_write(grpc_chttp2_transport* t, grpc_error_handle error); /// Process one slice of incoming data; return 1 if the connection is still /// viable after reading, or 0 if the connection should be torn down grpc_error_handle grpc_chttp2_perform_read(grpc_chttp2_transport* t, const grpc_slice& slice); bool grpc_chttp2_list_add_writable_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s); /// Get a writable stream /// returns non-zero if there was a stream available bool grpc_chttp2_list_pop_writable_stream(grpc_chttp2_transport* t, grpc_chttp2_stream** s); bool grpc_chttp2_list_remove_writable_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s); bool grpc_chttp2_list_add_writing_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s); bool grpc_chttp2_list_have_writing_streams(grpc_chttp2_transport* t); bool grpc_chttp2_list_pop_writing_stream(grpc_chttp2_transport* t, grpc_chttp2_stream** s); void grpc_chttp2_list_add_written_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s); bool grpc_chttp2_list_pop_written_stream(grpc_chttp2_transport* t, grpc_chttp2_stream** s); void grpc_chttp2_list_add_waiting_for_concurrency(grpc_chttp2_transport* t, grpc_chttp2_stream* s); bool grpc_chttp2_list_pop_waiting_for_concurrency(grpc_chttp2_transport* t, grpc_chttp2_stream** s); void grpc_chttp2_list_remove_waiting_for_concurrency(grpc_chttp2_transport* t, grpc_chttp2_stream* s); void grpc_chttp2_list_add_stalled_by_transport(grpc_chttp2_transport* t, grpc_chttp2_stream* s); bool grpc_chttp2_list_pop_stalled_by_transport(grpc_chttp2_transport* t, grpc_chttp2_stream** s); void grpc_chttp2_list_remove_stalled_by_transport(grpc_chttp2_transport* t, grpc_chttp2_stream* s); void grpc_chttp2_list_add_stalled_by_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s); bool grpc_chttp2_list_pop_stalled_by_stream(grpc_chttp2_transport* t, grpc_chttp2_stream** s); bool grpc_chttp2_list_remove_stalled_by_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s); //******** Flow Control ************** // Takes in a flow control action and performs all the needed operations. void grpc_chttp2_act_on_flowctl_action( const grpc_core::chttp2::FlowControlAction& action, grpc_chttp2_transport* t, grpc_chttp2_stream* s); //******** End of Flow Control ************** inline grpc_chttp2_stream* grpc_chttp2_parsing_lookup_stream( grpc_chttp2_transport* t, uint32_t id) { … } grpc_chttp2_stream* grpc_chttp2_parsing_accept_stream(grpc_chttp2_transport* t, uint32_t id); void grpc_chttp2_add_incoming_goaway(grpc_chttp2_transport* t, uint32_t goaway_error, uint32_t last_stream_id, absl::string_view goaway_text); void grpc_chttp2_parsing_become_skip_parser(grpc_chttp2_transport* t); void grpc_chttp2_complete_closure_step(grpc_chttp2_transport* t, grpc_chttp2_stream* s, grpc_closure** pclosure, grpc_error_handle error, const char* desc); #define GRPC_HEADER_SIZE_IN_BYTES … #define MAX_SIZE_T … #define GRPC_CHTTP2_CLIENT_CONNECT_STRING … #define GRPC_CHTTP2_CLIENT_CONNECT_STRLEN … // extern grpc_core::TraceFlag grpc_flowctl_trace; #define GRPC_CHTTP2_IF_TRACING(stmt) … void grpc_chttp2_fake_status(grpc_chttp2_transport* t, grpc_chttp2_stream* stream, grpc_error_handle error); void grpc_chttp2_mark_stream_closed(grpc_chttp2_transport* t, grpc_chttp2_stream* s, int close_reads, int close_writes, grpc_error_handle error); void grpc_chttp2_start_writing(grpc_chttp2_transport* t); #ifndef NDEBUG #define GRPC_CHTTP2_STREAM_REF(stream, reason) … #define GRPC_CHTTP2_STREAM_UNREF(stream, reason) … void grpc_chttp2_stream_ref(grpc_chttp2_stream* s, const char* reason); void grpc_chttp2_stream_unref(grpc_chttp2_stream* s, const char* reason); #else #define GRPC_CHTTP2_STREAM_REF … #define GRPC_CHTTP2_STREAM_UNREF … void grpc_chttp2_stream_ref(grpc_chttp2_stream* s); void grpc_chttp2_stream_unref(grpc_chttp2_stream* s); #endif #ifndef NDEBUG #define GRPC_CHTTP2_REF_TRANSPORT(t, r) … #define GRPC_CHTTP2_UNREF_TRANSPORT(t, r) … inline void grpc_chttp2_unref_transport(grpc_chttp2_transport* t, const char* reason, const char* file, int line) { … } inline void grpc_chttp2_ref_transport(grpc_chttp2_transport* t, const char* reason, const char* file, int line) { … } #else #define GRPC_CHTTP2_REF_TRANSPORT … #define GRPC_CHTTP2_UNREF_TRANSPORT … inline void grpc_chttp2_unref_transport(grpc_chttp2_transport* t) { if (t->refs.Unref()) { delete t; } } inline void grpc_chttp2_ref_transport(grpc_chttp2_transport* t) { t->refs.Ref(); } #endif void grpc_chttp2_ack_ping(grpc_chttp2_transport* t, uint64_t id); /// Add a new ping strike to ping_recv_state.ping_strikes. If /// ping_recv_state.ping_strikes > ping_policy.max_ping_strikes, it sends GOAWAY /// with error code ENHANCE_YOUR_CALM and additional debug data resembling /// "too_many_pings" followed by immediately closing the connection. void grpc_chttp2_add_ping_strike(grpc_chttp2_transport* t); /// Resets ping clock. Should be called when flushing window updates, /// initial/trailing metadata or data frames. For a server, it resets the number /// of ping strikes and the last_ping_recv_time. For a ping sender, it resets /// pings_before_data_required. void grpc_chttp2_reset_ping_clock(grpc_chttp2_transport* t); /// add a ref to the stream and add it to the writable list; /// ref will be dropped in writing.c void grpc_chttp2_mark_stream_writable(grpc_chttp2_transport* t, grpc_chttp2_stream* s); void grpc_chttp2_cancel_stream(grpc_chttp2_transport* t, grpc_chttp2_stream* s, grpc_error_handle due_to_error); void grpc_chttp2_maybe_complete_recv_initial_metadata(grpc_chttp2_transport* t, grpc_chttp2_stream* s); void grpc_chttp2_maybe_complete_recv_message(grpc_chttp2_transport* t, grpc_chttp2_stream* s); void grpc_chttp2_maybe_complete_recv_trailing_metadata(grpc_chttp2_transport* t, grpc_chttp2_stream* s); void grpc_chttp2_fail_pending_writes(grpc_chttp2_transport* t, grpc_chttp2_stream* s, grpc_error_handle error); /// Set the default keepalive configurations, must only be called at /// initialization void grpc_chttp2_config_default_keepalive_args(grpc_channel_args* args, bool is_client); void grpc_chttp2_retry_initiate_ping(grpc_chttp2_transport* t); void schedule_bdp_ping_locked(grpc_chttp2_transport* t); uint32_t grpc_chttp2_min_read_progress_size(grpc_chttp2_transport* t); #endif // GRPC_SRC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H
Codebase Browser
chromium