aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib')
-rw-r--r--src/core/lib/channel/channelz.cc70
-rw-r--r--src/core/lib/channel/channelz.h21
-rw-r--r--src/core/lib/channel/channelz_registry.cc75
-rw-r--r--src/core/lib/channel/channelz_registry.h16
-rw-r--r--src/core/lib/gprpp/inlined_vector.h8
-rw-r--r--src/core/lib/surface/server.cc35
-rw-r--r--src/core/lib/surface/server.h15
-rw-r--r--src/core/lib/transport/metadata_batch.cc1
8 files changed, 226 insertions, 15 deletions
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 {
generated by cgit v1.2.3 (git 2.39.1) at 2024-08-13 19:11:06 +0000