diff options
Diffstat (limited to 'src/core')
19 files changed, 277 insertions, 52 deletions
diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index 388736b60a..35c8bb27ee 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -3289,8 +3289,9 @@ static void try_to_connect_locked(void* arg, grpc_error* error_ignored) { } void grpc_client_channel_populate_child_refs( - grpc_channel_element* elem, grpc_core::ChildRefsList* child_subchannels, - grpc_core::ChildRefsList* child_channels) { + grpc_channel_element* elem, + grpc_core::channelz::ChildRefsList* child_subchannels, + grpc_core::channelz::ChildRefsList* child_channels) { channel_data* chand = static_cast<channel_data*>(elem->channel_data); if (chand->lb_policy != nullptr) { chand->lb_policy->FillChildRefsForChannelz(child_subchannels, diff --git a/src/core/ext/filters/client_channel/client_channel.h b/src/core/ext/filters/client_channel/client_channel.h index 0b44a17562..d64faaabd2 100644 --- a/src/core/ext/filters/client_channel/client_channel.h +++ b/src/core/ext/filters/client_channel/client_channel.h @@ -41,8 +41,9 @@ extern grpc_core::TraceFlag grpc_client_channel_trace; extern const grpc_channel_filter grpc_client_channel_filter; void grpc_client_channel_populate_child_refs( - grpc_channel_element* elem, grpc_core::ChildRefsList* child_subchannels, - grpc_core::ChildRefsList* child_channels); + grpc_channel_element* elem, + grpc_core::channelz::ChildRefsList* child_subchannels, + grpc_core::channelz::ChildRefsList* child_channels); grpc_connectivity_state grpc_client_channel_check_connectivity_state( grpc_channel_element* elem, int try_to_connect); diff --git a/src/core/ext/filters/client_channel/client_channel_channelz.h b/src/core/ext/filters/client_channel/client_channel_channelz.h index 8ce331e529..8a5c3e7e5e 100644 --- a/src/core/ext/filters/client_channel/client_channel_channelz.h +++ b/src/core/ext/filters/client_channel/client_channel_channelz.h @@ -25,17 +25,10 @@ #include "src/core/lib/channel/channel_stack.h" #include "src/core/lib/channel/channel_trace.h" #include "src/core/lib/channel/channelz.h" -#include "src/core/lib/gprpp/inlined_vector.h" typedef struct grpc_subchannel grpc_subchannel; namespace grpc_core { - -// TODO(ncteisen), this only contains the uuids of the children for now, -// since that is all that is strictly needed. In a future enhancement we will -// add human readable names as in the channelz.proto -typedef InlinedVector<intptr_t, 10> ChildRefsList; - namespace channelz { // Subtype of ChannelNode that overrides and provides client_channel specific diff --git a/src/core/ext/filters/client_channel/lb_policy.h b/src/core/ext/filters/client_channel/lb_policy.h index 3c0a9c1118..21f80b7b94 100644 --- a/src/core/ext/filters/client_channel/lb_policy.h +++ b/src/core/ext/filters/client_channel/lb_policy.h @@ -151,9 +151,9 @@ class LoadBalancingPolicy /// LB policy's referenced children. This is not invoked from the /// client_channel's combiner. The implementation is responsible for /// providing its own synchronization. - virtual void FillChildRefsForChannelz(ChildRefsList* child_subchannels, - ChildRefsList* child_channels) - GRPC_ABSTRACT; + virtual void FillChildRefsForChannelz( + channelz::ChildRefsList* child_subchannels, + channelz::ChildRefsList* child_channels) GRPC_ABSTRACT; void Orphan() override { // Invoke ShutdownAndUnrefLocked() inside of the combiner. @@ -212,8 +212,8 @@ class LoadBalancingPolicy // Dummy classes needed for alignment issues. // See https://github.com/grpc/grpc/issues/16032 for context. // TODO(ncteisen): remove this as soon as the issue is resolved. - ChildRefsList dummy_list_foo; - ChildRefsList dummy_list_bar; + channelz::ChildRefsList dummy_list_foo; + channelz::ChildRefsList dummy_list_bar; }; } // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc index 0061190016..384e7931b5 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc @@ -136,8 +136,9 @@ class GrpcLb : public LoadBalancingPolicy { void HandOffPendingPicksLocked(LoadBalancingPolicy* new_policy) override; void ExitIdleLocked() override; void ResetBackoffLocked() override; - void FillChildRefsForChannelz(ChildRefsList* child_subchannels, - ChildRefsList* child_channels) override; + void FillChildRefsForChannelz( + channelz::ChildRefsList* child_subchannels, + channelz::ChildRefsList* child_channels) override; private: /// Linked list of pending pick requests. It stores all information needed to @@ -1258,8 +1259,9 @@ bool GrpcLb::PickLocked(PickState* pick, grpc_error** error) { return pick_done; } -void GrpcLb::FillChildRefsForChannelz(ChildRefsList* child_subchannels, - ChildRefsList* child_channels) { +void GrpcLb::FillChildRefsForChannelz( + channelz::ChildRefsList* child_subchannels, + channelz::ChildRefsList* child_channels) { // delegate to the RoundRobin to fill the children subchannels. rr_policy_->FillChildRefsForChannelz(child_subchannels, child_channels); MutexLock lock(&lb_channel_mu_); diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc index ed8cc60ea1..f4dca146f7 100644 --- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc +++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc @@ -59,8 +59,8 @@ class PickFirst : public LoadBalancingPolicy { void HandOffPendingPicksLocked(LoadBalancingPolicy* new_policy) override; void ExitIdleLocked() override; void ResetBackoffLocked() override; - void FillChildRefsForChannelz(ChildRefsList* child_subchannels, - ChildRefsList* ignored) override; + void FillChildRefsForChannelz(channelz::ChildRefsList* child_subchannels, + channelz::ChildRefsList* ignored) override; private: ~PickFirst(); @@ -147,8 +147,8 @@ class PickFirst : public LoadBalancingPolicy { /// Lock and data used to capture snapshots of this channels child /// channels and subchannels. This data is consumed by channelz. gpr_mu child_refs_mu_; - ChildRefsList child_subchannels_; - ChildRefsList child_channels_; + channelz::ChildRefsList child_subchannels_; + channelz::ChildRefsList child_channels_; }; PickFirst::PickFirst(const Args& args) : LoadBalancingPolicy(args) { @@ -300,7 +300,8 @@ void PickFirst::NotifyOnStateChangeLocked(grpc_connectivity_state* current, } void PickFirst::FillChildRefsForChannelz( - ChildRefsList* child_subchannels_to_fill, ChildRefsList* ignored) { + channelz::ChildRefsList* child_subchannels_to_fill, + channelz::ChildRefsList* ignored) { MutexLock lock(&child_refs_mu_); for (size_t i = 0; i < child_subchannels_.size(); ++i) { // TODO(ncteisen): implement a de dup loop that is not O(n^2). Might @@ -320,7 +321,7 @@ void PickFirst::FillChildRefsForChannelz( } void PickFirst::UpdateChildRefsLocked() { - ChildRefsList cs; + channelz::ChildRefsList cs; if (subchannel_list_ != nullptr) { subchannel_list_->PopulateChildRefsList(&cs); } diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc index 8dd5820bae..e9ed85cf66 100644 --- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc +++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc @@ -70,8 +70,8 @@ class RoundRobin : public LoadBalancingPolicy { void HandOffPendingPicksLocked(LoadBalancingPolicy* new_policy) override; void ExitIdleLocked() override; void ResetBackoffLocked() override; - void FillChildRefsForChannelz(ChildRefsList* child_subchannels, - ChildRefsList* ignored) override; + void FillChildRefsForChannelz(channelz::ChildRefsList* child_subchannels, + channelz::ChildRefsList* ignored) override; private: ~RoundRobin(); @@ -223,8 +223,8 @@ class RoundRobin : public LoadBalancingPolicy { /// Lock and data used to capture snapshots of this channel's child /// channels and subchannels. This data is consumed by channelz. gpr_mu child_refs_mu_; - ChildRefsList child_subchannels_; - ChildRefsList child_channels_; + channelz::ChildRefsList child_subchannels_; + channelz::ChildRefsList child_channels_; }; RoundRobin::RoundRobin(const Args& args) : LoadBalancingPolicy(args) { @@ -402,7 +402,8 @@ bool RoundRobin::PickLocked(PickState* pick, grpc_error** error) { } void RoundRobin::FillChildRefsForChannelz( - ChildRefsList* child_subchannels_to_fill, ChildRefsList* ignored) { + channelz::ChildRefsList* child_subchannels_to_fill, + channelz::ChildRefsList* ignored) { MutexLock lock(&child_refs_mu_); for (size_t i = 0; i < child_subchannels_.size(); ++i) { // TODO(ncteisen): implement a de dup loop that is not O(n^2). Might @@ -422,7 +423,7 @@ void RoundRobin::FillChildRefsForChannelz( } void RoundRobin::UpdateChildRefsLocked() { - ChildRefsList cs; + channelz::ChildRefsList cs; if (subchannel_list_ != nullptr) { subchannel_list_->PopulateChildRefsList(&cs); } diff --git a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h index 5e8682e056..e0e0e1e638 100644 --- a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h +++ b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h @@ -201,7 +201,7 @@ class SubchannelList bool shutting_down() const { return shutting_down_; } // Populates refs_list with the uuids of this SubchannelLists's subchannels. - void PopulateChildRefsList(ChildRefsList* refs_list) { + void PopulateChildRefsList(channelz::ChildRefsList* refs_list) { for (size_t i = 0; i < subchannels_.size(); ++i) { if (subchannels_[i].subchannel() != nullptr) { grpc_core::channelz::SubchannelNode* subchannel_node = diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.cc b/src/core/ext/transport/chttp2/server/chttp2_server.cc index 3f8a26ae32..a5fc6062aa 100644 --- a/src/core/ext/transport/chttp2/server/chttp2_server.cc +++ b/src/core/ext/transport/chttp2/server/chttp2_server.cc @@ -53,6 +53,8 @@ typedef struct { grpc_closure tcp_server_shutdown_complete; grpc_closure* server_destroy_listener_done; grpc_handshake_manager* pending_handshake_mgrs; + grpc_core::RefCountedPtr<grpc_core::channelz::ListenSocketNode> + channelz_listen_socket; } server_state; typedef struct { @@ -133,7 +135,8 @@ static void on_handshake_done(void* arg, grpc_error* error) { grpc_create_chttp2_transport(args->args, args->endpoint, false); grpc_server_setup_transport( connection_state->svr_state->server, transport, - connection_state->accepting_pollset, args->args); + connection_state->accepting_pollset, args->args, + grpc_chttp2_transport_get_socket_uuid(transport)); // Use notify_on_receive_settings callback to enforce the // handshake deadline. connection_state->transport = @@ -223,6 +226,7 @@ static void tcp_server_shutdown_complete(void* arg, grpc_error* error) { GPR_ASSERT(state->shutdown); grpc_handshake_manager_pending_list_shutdown_all( state->pending_handshake_mgrs, GRPC_ERROR_REF(error)); + state->channelz_listen_socket.reset(); gpr_mu_unlock(&state->mu); // Flush queued work before destroying handshaker factory, since that // may do a synchronous unref. @@ -262,6 +266,8 @@ grpc_error* grpc_chttp2_server_add_port(grpc_server* server, const char* addr, server_state* state = nullptr; grpc_error** errors = nullptr; size_t naddrs = 0; + const grpc_arg* arg = nullptr; + intptr_t socket_uuid = 0; *port_num = -1; @@ -323,9 +329,16 @@ grpc_error* grpc_chttp2_server_add_port(grpc_server* server, const char* addr, } grpc_resolved_addresses_destroy(resolved); + arg = grpc_channel_args_find(args, GRPC_ARG_ENABLE_CHANNELZ); + if (grpc_channel_arg_get_bool(arg, false)) { + state->channelz_listen_socket = + grpc_core::MakeRefCounted<grpc_core::channelz::ListenSocketNode>(); + socket_uuid = state->channelz_listen_socket->uuid(); + } + /* Register with the server only upon success */ grpc_server_add_listener(server, state, server_start_listener, - server_destroy_listener); + server_destroy_listener, socket_uuid); goto done; /* Error path: cleanup and return */ diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc index e4bd91d07b..b9024a87e2 100644 --- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc +++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc @@ -61,7 +61,7 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server* server, grpc_endpoint_add_to_pollset(server_endpoint, pollsets[i]); } - grpc_server_setup_transport(server, transport, nullptr, server_args); + grpc_server_setup_transport(server, transport, nullptr, server_args, 0); grpc_chttp2_transport_start_reading(transport, nullptr, nullptr); } diff --git a/src/core/ext/transport/inproc/inproc_transport.cc b/src/core/ext/transport/inproc/inproc_transport.cc index b0ca7f8207..9dbd095843 100644 --- a/src/core/ext/transport/inproc/inproc_transport.cc +++ b/src/core/ext/transport/inproc/inproc_transport.cc @@ -1256,7 +1256,9 @@ grpc_channel* grpc_inproc_channel_create(grpc_server* server, inproc_transports_create(&server_transport, server_args, &client_transport, client_args); - grpc_server_setup_transport(server, server_transport, nullptr, server_args); + // TODO(ncteisen): design and support channelz GetSocket for inproc. + grpc_server_setup_transport(server, server_transport, nullptr, server_args, + 0); grpc_channel* channel = grpc_channel_create( "inproc", client_args, GRPC_CLIENT_DIRECT_CHANNEL, client_transport); diff --git a/src/core/lib/channel/channelz.cc b/src/core/lib/channel/channelz.cc index 573292fba1..f3d0c03715 100644 --- a/src/core/lib/channel/channelz.cc +++ b/src/core/lib/channel/channelz.cc @@ -37,6 +37,7 @@ #include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/surface/channel.h" +#include "src/core/lib/surface/server.h" #include "src/core/lib/transport/error_utils.h" namespace grpc_core { @@ -190,11 +191,45 @@ RefCountedPtr<ChannelNode> ChannelNode::MakeChannelNode( channel, channel_tracer_max_nodes, is_top_level_channel); } -ServerNode::ServerNode(size_t channel_tracer_max_nodes) - : BaseNode(EntityType::kServer), trace_(channel_tracer_max_nodes) {} +ServerNode::ServerNode(grpc_server* server, size_t channel_tracer_max_nodes) + : BaseNode(EntityType::kServer), + server_(server), + trace_(channel_tracer_max_nodes) {} ServerNode::~ServerNode() {} +char* ServerNode::RenderServerSockets(intptr_t start_socket_id) { + grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); + grpc_json* json = top_level_json; + grpc_json* json_iterator = nullptr; + ChildRefsList socket_refs; + // uuids index into entities one-off (idx 0 is really uuid 1, since 0 is + // reserved). However, we want to support requests coming in with + // start_server_id=0, which signifies "give me everything." + size_t start_idx = start_socket_id == 0 ? 0 : start_socket_id - 1; + grpc_server_populate_server_sockets(server_, &socket_refs, start_idx); + if (!socket_refs.empty()) { + // create list of socket refs + grpc_json* array_parent = grpc_json_create_child( + nullptr, json, "socketRef", nullptr, GRPC_JSON_ARRAY, false); + for (size_t i = 0; i < socket_refs.size(); ++i) { + json_iterator = + grpc_json_create_child(json_iterator, array_parent, nullptr, nullptr, + GRPC_JSON_OBJECT, false); + grpc_json_add_number_string_child(json_iterator, nullptr, "socketId", + socket_refs[i]); + } + } + // For now we do not have any pagination rules. In the future we could + // pick a constant for max_channels_sent for a GetServers request. + // Tracking: https://github.com/grpc/grpc/issues/16019. + json_iterator = grpc_json_create_child(nullptr, json, "end", nullptr, + GRPC_JSON_TRUE, false); + char* json_str = grpc_json_dump_to_string(top_level_json, 0); + grpc_json_destroy(top_level_json); + return json_str; +} + grpc_json* ServerNode::RenderJson() { // We need to track these three json objects to build our object grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); @@ -223,6 +258,20 @@ grpc_json* ServerNode::RenderJson() { } // ask CallCountingHelper to populate trace and call count data. call_counter_.PopulateCallCounts(json); + json = top_level_json; + ChildRefsList listen_sockets; + grpc_server_populate_listen_sockets(server_, &listen_sockets); + if (!listen_sockets.empty()) { + grpc_json* array_parent = grpc_json_create_child( + nullptr, json, "listenSocket", nullptr, GRPC_JSON_ARRAY, false); + for (size_t i = 0; i < listen_sockets.size(); ++i) { + json_iterator = + grpc_json_create_child(json_iterator, array_parent, nullptr, nullptr, + GRPC_JSON_OBJECT, false); + grpc_json_add_number_string_child(json_iterator, nullptr, "socketId", + listen_sockets[i]); + } + } return top_level_json; } @@ -323,5 +372,22 @@ grpc_json* SocketNode::RenderJson() { return top_level_json; } +ListenSocketNode::ListenSocketNode() : BaseNode(EntityType::kSocket) {} + +grpc_json* ListenSocketNode::RenderJson() { + // We need to track these three json objects to build our object + grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT); + grpc_json* json = top_level_json; + grpc_json* json_iterator = nullptr; + // create and fill the ref child + json_iterator = grpc_json_create_child(json_iterator, json, "ref", nullptr, + GRPC_JSON_OBJECT, false); + json = json_iterator; + json_iterator = nullptr; + json_iterator = grpc_json_add_number_string_child(json, json_iterator, + "socketId", uuid()); + return top_level_json; +} + } // namespace channelz } // namespace grpc_core diff --git a/src/core/lib/channel/channelz.h b/src/core/lib/channel/channelz.h index 3f728cda6d..d8113585c2 100644 --- a/src/core/lib/channel/channelz.h +++ b/src/core/lib/channel/channelz.h @@ -24,6 +24,7 @@ #include <grpc/grpc.h> #include "src/core/lib/channel/channel_trace.h" +#include "src/core/lib/gprpp/inlined_vector.h" #include "src/core/lib/gprpp/manual_constructor.h" #include "src/core/lib/gprpp/ref_counted.h" #include "src/core/lib/gprpp/ref_counted_ptr.h" @@ -50,8 +51,14 @@ #define GRPC_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE_DEFAULT 0 namespace grpc_core { + namespace channelz { +// TODO(ncteisen), this only contains the uuids of the children for now, +// since that is all that is strictly needed. In a future enhancement we will +// add human readable names as in the channelz.proto +typedef InlinedVector<intptr_t, 10> ChildRefsList; + namespace testing { class CallCountingHelperPeer; class ChannelNodePeer; @@ -193,11 +200,13 @@ class ChannelNode : public BaseNode { // Handles channelz bookkeeping for servers class ServerNode : public BaseNode { public: - explicit ServerNode(size_t channel_tracer_max_nodes); + ServerNode(grpc_server* server, size_t channel_tracer_max_nodes); ~ServerNode() override; grpc_json* RenderJson() override; + char* RenderServerSockets(intptr_t start_socket_id); + // proxy methods to composed classes. void AddTraceEvent(ChannelTrace::Severity severity, grpc_slice data) { trace_.AddTraceEvent(severity, data); @@ -213,6 +222,7 @@ class ServerNode : public BaseNode { void RecordCallSucceeded() { call_counter_.RecordCallSucceeded(); } private: + grpc_server* server_; CallCountingHelper call_counter_; ChannelTrace trace_; }; @@ -253,6 +263,15 @@ class SocketNode : public BaseNode { UniquePtr<char> peer_string_; }; +// Handles channelz bookkeeping for listen sockets +class ListenSocketNode : public BaseNode { + public: + ListenSocketNode(); + ~ListenSocketNode() override {} + + grpc_json* RenderJson() override; +}; + // Creation functions typedef RefCountedPtr<ChannelNode> (*ChannelNodeCreationFunc)(grpc_channel*, diff --git a/src/core/lib/channel/channelz_registry.cc b/src/core/lib/channel/channelz_registry.cc index 841f1c6104..1b54b19be3 100644 --- a/src/core/lib/channel/channelz_registry.cc +++ b/src/core/lib/channel/channelz_registry.cc @@ -56,23 +56,71 @@ ChannelzRegistry::~ChannelzRegistry() { gpr_mu_destroy(&mu_); } intptr_t ChannelzRegistry::InternalRegister(BaseNode* node) { MutexLock lock(&mu_); entities_.push_back(node); - intptr_t uuid = entities_.size(); - return uuid; + return ++uuid_generator_; +} + +void ChannelzRegistry::MaybePerformCompactionLocked() { + constexpr double kEmptinessTheshold = 1 / 3; + double emptiness_ratio = + double(num_empty_slots_) / double(entities_.capacity()); + if (emptiness_ratio > kEmptinessTheshold) { + int front = 0; + for (size_t i = 0; i < entities_.size(); ++i) { + if (entities_[i] != nullptr) { + entities_[front++] = entities_[i]; + } + } + for (int i = 0; i < num_empty_slots_; ++i) { + entities_.pop_back(); + } + num_empty_slots_ = 0; + } +} + +int ChannelzRegistry::FindByUuidLocked(intptr_t target_uuid) { + size_t left = 0; + size_t right = entities_.size() - 1; + while (left <= right) { + size_t true_middle = left + (right - left) / 2; + size_t first_non_null = true_middle; + while (first_non_null < right && entities_[first_non_null] == nullptr) { + first_non_null++; + } + if (entities_[first_non_null] == nullptr) { + right = true_middle - 1; + continue; + } + intptr_t uuid = entities_[first_non_null]->uuid(); + if (uuid == target_uuid) { + return int(first_non_null); + } + if (uuid < target_uuid) { + left = first_non_null + 1; + } else { + right = true_middle - 1; + } + } + return -1; } void ChannelzRegistry::InternalUnregister(intptr_t uuid) { GPR_ASSERT(uuid >= 1); MutexLock lock(&mu_); - GPR_ASSERT(static_cast<size_t>(uuid) <= entities_.size()); - entities_[uuid - 1] = nullptr; + GPR_ASSERT(uuid <= uuid_generator_); + int idx = FindByUuidLocked(uuid); + GPR_ASSERT(idx >= 0); + entities_[idx] = nullptr; + num_empty_slots_++; + MaybePerformCompactionLocked(); } BaseNode* ChannelzRegistry::InternalGet(intptr_t uuid) { MutexLock lock(&mu_); - if (uuid < 1 || uuid > static_cast<intptr_t>(entities_.size())) { + if (uuid < 1 || uuid > uuid_generator_) { return nullptr; } - return entities_[uuid - 1]; + int idx = FindByUuidLocked(uuid); + return idx < 0 ? nullptr : entities_[idx]; } char* ChannelzRegistry::InternalGetTopChannels(intptr_t start_channel_id) { @@ -160,6 +208,21 @@ char* grpc_channelz_get_servers(intptr_t start_server_id) { return grpc_core::channelz::ChannelzRegistry::GetServers(start_server_id); } +char* grpc_channelz_get_server_sockets(intptr_t server_id, + intptr_t start_socket_id) { + grpc_core::channelz::BaseNode* base_node = + grpc_core::channelz::ChannelzRegistry::Get(server_id); + if (base_node == nullptr || + base_node->type() != grpc_core::channelz::BaseNode::EntityType::kServer) { + return nullptr; + } + // This cast is ok since we have just checked to make sure base_node is + // actually a server node + grpc_core::channelz::ServerNode* server_node = + static_cast<grpc_core::channelz::ServerNode*>(base_node); + return server_node->RenderServerSockets(start_socket_id); +} + char* grpc_channelz_get_channel(intptr_t channel_id) { grpc_core::channelz::BaseNode* channel_node = grpc_core::channelz::ChannelzRegistry::Get(channel_id); diff --git a/src/core/lib/channel/channelz_registry.h b/src/core/lib/channel/channelz_registry.h index d0d660600d..9c43d960d3 100644 --- a/src/core/lib/channel/channelz_registry.h +++ b/src/core/lib/channel/channelz_registry.h @@ -30,6 +30,10 @@ namespace grpc_core { namespace channelz { +namespace testing { +class ChannelzRegistryPeer; +} + // singleton registry object to track all objects that are needed to support // channelz bookkeeping. All objects share globally distributed uuids. class ChannelzRegistry { @@ -61,6 +65,7 @@ class ChannelzRegistry { private: GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_NEW GPRC_ALLOW_CLASS_TO_USE_NON_PUBLIC_DELETE + friend class testing::ChannelzRegistryPeer; ChannelzRegistry(); ~ChannelzRegistry(); @@ -82,9 +87,18 @@ class ChannelzRegistry { char* InternalGetTopChannels(intptr_t start_channel_id); char* InternalGetServers(intptr_t start_server_id); - // protects entities_ and uuid_ + // If entities_ has over a certain threshold of empty slots, it will + // compact the vector and move all used slots to the front. + void MaybePerformCompactionLocked(); + + // Performs binary search on entities_ to find the index with that uuid. + int FindByUuidLocked(intptr_t uuid); + + // protects members gpr_mu mu_; InlinedVector<BaseNode*, 20> entities_; + intptr_t uuid_generator_ = 0; + int num_empty_slots_ = 0; }; } // namespace channelz diff --git a/src/core/lib/gprpp/inlined_vector.h b/src/core/lib/gprpp/inlined_vector.h index 76e2f0a785..65c2b9634f 100644 --- a/src/core/lib/gprpp/inlined_vector.h +++ b/src/core/lib/gprpp/inlined_vector.h @@ -123,6 +123,14 @@ class InlinedVector { void push_back(T&& value) { emplace_back(std::move(value)); } + void pop_back() { + assert(!empty()); + size_t s = size(); + T& value = data()[s - 1]; + value.~T(); + size_--; + } + void copy_from(const InlinedVector& v) { // if v is allocated, copy over the buffer. if (v.dynamic_ != nullptr) { diff --git a/src/core/lib/surface/server.cc b/src/core/lib/surface/server.cc index 449aef768e..35ab2c3bce 100644 --- a/src/core/lib/surface/server.cc +++ b/src/core/lib/surface/server.cc @@ -54,6 +54,7 @@ struct listener { size_t pollset_count); void (*destroy)(grpc_server* server, void* arg, grpc_closure* closure); struct listener* next; + intptr_t socket_uuid; grpc_closure destroy_done; }; @@ -104,6 +105,7 @@ struct channel_data { uint32_t registered_method_max_probes; grpc_closure finish_destroy_channel_closure; grpc_closure channel_connectivity_changed; + intptr_t socket_uuid; }; typedef struct shutdown_tag { @@ -1016,7 +1018,7 @@ grpc_server* grpc_server_create(const grpc_channel_args* args, void* reserved) { {GRPC_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE_DEFAULT, 0, INT_MAX}); server->channelz_server = grpc_core::MakeRefCounted<grpc_core::channelz::ServerNode>( - channel_tracer_max_memory); + server, channel_tracer_max_memory); server->channelz_server->AddTraceEvent( grpc_core::channelz::ChannelTrace::Severity::Info, grpc_slice_from_static_string("Server created")); @@ -1119,7 +1121,8 @@ void grpc_server_get_pollsets(grpc_server* server, grpc_pollset*** pollsets, void grpc_server_setup_transport(grpc_server* s, grpc_transport* transport, grpc_pollset* accepting_pollset, - const grpc_channel_args* args) { + const grpc_channel_args* args, + intptr_t socket_uuid) { size_t num_registered_methods; size_t alloc; registered_method* rm; @@ -1139,6 +1142,7 @@ void grpc_server_setup_transport(grpc_server* s, grpc_transport* transport, chand->server = s; server_ref(s); chand->channel = channel; + chand->socket_uuid = socket_uuid; size_t cq_idx; for (cq_idx = 0; cq_idx < s->cq_count; cq_idx++) { @@ -1213,6 +1217,29 @@ void grpc_server_setup_transport(grpc_server* s, grpc_transport* transport, grpc_transport_perform_op(transport, op); } +void grpc_server_populate_server_sockets( + grpc_server* s, grpc_core::channelz::ChildRefsList* server_sockets, + intptr_t start_idx) { + gpr_mu_lock(&s->mu_global); + channel_data* c = nullptr; + for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) { + intptr_t socket_uuid = c->socket_uuid; + if (socket_uuid >= start_idx) { + server_sockets->push_back(socket_uuid); + } + } + gpr_mu_unlock(&s->mu_global); +} + +void grpc_server_populate_listen_sockets( + grpc_server* server, grpc_core::channelz::ChildRefsList* listen_sockets) { + gpr_mu_lock(&server->mu_global); + for (listener* l = server->listeners; l != nullptr; l = l->next) { + listen_sockets->push_back(l->socket_uuid); + } + gpr_mu_unlock(&server->mu_global); +} + void done_published_shutdown(void* done_arg, grpc_cq_completion* storage) { (void)done_arg; gpr_free(storage); @@ -1346,11 +1373,13 @@ void grpc_server_add_listener(grpc_server* server, void* arg, grpc_pollset** pollsets, size_t pollset_count), void (*destroy)(grpc_server* server, void* arg, - grpc_closure* on_done)) { + grpc_closure* on_done), + intptr_t socket_uuid) { listener* l = static_cast<listener*>(gpr_malloc(sizeof(listener))); l->arg = arg; l->start = start; l->destroy = destroy; + l->socket_uuid = socket_uuid; l->next = server->listeners; server->listeners = l; } diff --git a/src/core/lib/surface/server.h b/src/core/lib/surface/server.h index 0196743ff9..33c205417e 100644 --- a/src/core/lib/surface/server.h +++ b/src/core/lib/surface/server.h @@ -39,13 +39,24 @@ void grpc_server_add_listener(grpc_server* server, void* listener, grpc_pollset** pollsets, size_t npollsets), void (*destroy)(grpc_server* server, void* arg, - grpc_closure* on_done)); + grpc_closure* on_done), + intptr_t socket_uuid); /* Setup a transport - creates a channel stack, binds the transport to the server */ void grpc_server_setup_transport(grpc_server* server, grpc_transport* transport, grpc_pollset* accepting_pollset, - const grpc_channel_args* args); + const grpc_channel_args* args, + intptr_t socket_uuid); + +/* fills in the uuids of all sockets used for connections on this server */ +void grpc_server_populate_server_sockets( + grpc_server* server, grpc_core::channelz::ChildRefsList* server_sockets, + intptr_t start_idx); + +/* fills in the uuids of all listen sockets on this server */ +void grpc_server_populate_listen_sockets( + grpc_server* server, grpc_core::channelz::ChildRefsList* listen_sockets); grpc_core::channelz::ServerNode* grpc_server_get_channelz_node( grpc_server* server); diff --git a/src/core/lib/transport/metadata_batch.cc b/src/core/lib/transport/metadata_batch.cc index 761aad19c4..928ed73cda 100644 --- a/src/core/lib/transport/metadata_batch.cc +++ b/src/core/lib/transport/metadata_batch.cc @@ -139,6 +139,7 @@ static void link_head(grpc_mdelem_list* list, grpc_linked_mdelem* storage) { GPR_ASSERT(!GRPC_MDISNULL(storage->md)); storage->prev = nullptr; storage->next = list->head; + storage->reserved = nullptr; if (list->head != nullptr) { list->head->prev = storage; } else { |