aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib')
-rw-r--r--src/core/lib/channel/channel_trace.cc49
-rw-r--r--src/core/lib/channel/channel_trace.h30
-rw-r--r--src/core/lib/channel/channelz.cc185
-rw-r--r--src/core/lib/channel/channelz.h85
-rw-r--r--src/core/lib/channel/connected_channel.cc9
-rw-r--r--src/core/lib/gprpp/inlined_vector.h2
-rw-r--r--src/core/lib/iomgr/call_combiner.h80
-rw-r--r--src/core/lib/iomgr/closure.h5
-rw-r--r--src/core/lib/iomgr/socket_utils.h9
-rw-r--r--src/core/lib/iomgr/socket_utils_common_posix.cc4
-rw-r--r--src/core/lib/iomgr/socket_utils_uv.cc4
-rw-r--r--src/core/lib/iomgr/socket_utils_windows.cc4
-rw-r--r--src/core/lib/security/credentials/google_default/google_default_credentials.cc3
-rw-r--r--src/core/lib/security/credentials/ssl/ssl_credentials.cc20
-rw-r--r--src/core/lib/security/security_connector/security_connector.cc35
-rw-r--r--src/core/lib/security/security_connector/security_connector.h1
-rw-r--r--src/core/lib/surface/call.cc108
-rw-r--r--src/core/lib/surface/channel.cc33
-rw-r--r--src/core/lib/surface/channel.h4
-rw-r--r--src/core/lib/transport/transport.cc29
-rw-r--r--src/core/lib/transport/transport.h22
-rw-r--r--src/core/lib/transport/transport_op_string.cc7
22 files changed, 602 insertions, 126 deletions
diff --git a/src/core/lib/channel/channel_trace.cc b/src/core/lib/channel/channel_trace.cc
index eb7214b355..0f655d8716 100644
--- a/src/core/lib/channel/channel_trace.cc
+++ b/src/core/lib/channel/channel_trace.cc
@@ -28,7 +28,6 @@
#include <stdlib.h>
#include <string.h>
-#include "src/core/lib/channel/channelz_registry.h"
#include "src/core/lib/channel/status_util.h"
#include "src/core/lib/gpr/string.h"
#include "src/core/lib/gpr/useful.h"
@@ -40,16 +39,17 @@
#include "src/core/lib/transport/error_utils.h"
namespace grpc_core {
+namespace channelz {
ChannelTrace::TraceEvent::TraceEvent(
Severity severity, grpc_slice data,
- RefCountedPtr<ChannelTrace> referenced_tracer, ReferencedType type)
+ RefCountedPtr<ChannelNode> referenced_channel, ReferencedType type)
: severity_(severity),
data_(data),
timestamp_(grpc_millis_to_timespec(grpc_core::ExecCtx::Get()->Now(),
GPR_CLOCK_REALTIME)),
next_(nullptr),
- referenced_tracer_(std::move(referenced_tracer)),
+ referenced_channel_(std::move(referenced_channel)),
referenced_type_(type) {}
ChannelTrace::TraceEvent::TraceEvent(Severity severity, grpc_slice data)
@@ -62,15 +62,13 @@ ChannelTrace::TraceEvent::TraceEvent(Severity severity, grpc_slice data)
ChannelTrace::TraceEvent::~TraceEvent() { grpc_slice_unref_internal(data_); }
ChannelTrace::ChannelTrace(size_t max_events)
- : channel_uuid_(-1),
- num_events_logged_(0),
+ : num_events_logged_(0),
list_size_(0),
max_list_size_(max_events),
head_trace_(nullptr),
tail_trace_(nullptr) {
if (max_list_size_ == 0) return; // tracing is disabled if max_events == 0
gpr_mu_init(&tracer_mu_);
- channel_uuid_ = ChannelzRegistry::Register(this);
time_created_ = grpc_millis_to_timespec(grpc_core::ExecCtx::Get()->Now(),
GPR_CLOCK_REALTIME);
}
@@ -83,12 +81,9 @@ ChannelTrace::~ChannelTrace() {
it = it->next();
Delete<TraceEvent>(to_free);
}
- ChannelzRegistry::Unregister(channel_uuid_);
gpr_mu_destroy(&tracer_mu_);
}
-intptr_t ChannelTrace::GetUuid() const { return channel_uuid_; }
-
void ChannelTrace::AddTraceEventHelper(TraceEvent* new_trace_event) {
++num_events_logged_;
// first event case
@@ -117,20 +112,21 @@ void ChannelTrace::AddTraceEvent(Severity severity, grpc_slice data) {
void ChannelTrace::AddTraceEventReferencingChannel(
Severity severity, grpc_slice data,
- RefCountedPtr<ChannelTrace> referenced_tracer) {
+ RefCountedPtr<ChannelNode> referenced_channel) {
if (max_list_size_ == 0) return; // tracing is disabled if max_events == 0
// create and fill up the new event
- AddTraceEventHelper(
- New<TraceEvent>(severity, data, std::move(referenced_tracer), Channel));
+ AddTraceEventHelper(New<TraceEvent>(
+ severity, data, std::move(referenced_channel), ReferencedType::Channel));
}
void ChannelTrace::AddTraceEventReferencingSubchannel(
Severity severity, grpc_slice data,
- RefCountedPtr<ChannelTrace> referenced_tracer) {
+ RefCountedPtr<ChannelNode> referenced_subchannel) {
if (max_list_size_ == 0) return; // tracing is disabled if max_events == 0
// create and fill up the new event
- AddTraceEventHelper(New<TraceEvent>(
- severity, data, std::move(referenced_tracer), Subchannel));
+ AddTraceEventHelper(New<TraceEvent>(severity, data,
+ std::move(referenced_subchannel),
+ ReferencedType::Subchannel));
}
namespace {
@@ -193,22 +189,24 @@ void ChannelTrace::TraceEvent::RenderTraceEvent(grpc_json* json) const {
json_iterator =
grpc_json_create_child(json_iterator, json, "timestamp",
fmt_time(timestamp_), GRPC_JSON_STRING, true);
- if (referenced_tracer_ != nullptr) {
+ if (referenced_channel_ != nullptr) {
char* uuid_str;
- gpr_asprintf(&uuid_str, "%" PRIdPTR, referenced_tracer_->channel_uuid_);
+ gpr_asprintf(&uuid_str, "%" PRIdPTR, referenced_channel_->channel_uuid());
grpc_json* child_ref = grpc_json_create_child(
json_iterator, json,
- (referenced_type_ == Channel) ? "channelRef" : "subchannelRef", nullptr,
- GRPC_JSON_OBJECT, false);
+ (referenced_type_ == ReferencedType::Channel) ? "channelRef"
+ : "subchannelRef",
+ nullptr, GRPC_JSON_OBJECT, false);
json_iterator = grpc_json_create_child(
nullptr, child_ref,
- (referenced_type_ == Channel) ? "channelId" : "subchannelId", uuid_str,
- GRPC_JSON_STRING, true);
+ (referenced_type_ == ReferencedType::Channel) ? "channelId"
+ : "subchannelId",
+ uuid_str, GRPC_JSON_STRING, true);
json_iterator = child_ref;
}
}
-char* ChannelTrace::RenderTrace() const {
+grpc_json* ChannelTrace::RenderJSON() const {
if (!max_list_size_)
return nullptr; // tracing is disabled if max_events == 0
grpc_json* json = grpc_json_create(GRPC_JSON_OBJECT);
@@ -219,7 +217,7 @@ char* ChannelTrace::RenderTrace() const {
grpc_json_create_child(json_iterator, json, "numEventsLogged",
num_events_logged_str, GRPC_JSON_STRING, true);
json_iterator =
- grpc_json_create_child(json_iterator, json, "creationTime",
+ grpc_json_create_child(json_iterator, json, "creationTimestamp",
fmt_time(time_created_), GRPC_JSON_STRING, true);
grpc_json* events = grpc_json_create_child(json_iterator, json, "events",
nullptr, GRPC_JSON_ARRAY, false);
@@ -231,9 +229,8 @@ char* ChannelTrace::RenderTrace() const {
it->RenderTraceEvent(json_iterator);
it = it->next();
}
- char* json_str = grpc_json_dump_to_string(json, 0);
- grpc_json_destroy(json);
- return json_str;
+ return json;
}
+} // namespace channelz
} // namespace grpc_core
diff --git a/src/core/lib/channel/channel_trace.h b/src/core/lib/channel/channel_trace.h
index 1df1e585f2..0dd162a777 100644
--- a/src/core/lib/channel/channel_trace.h
+++ b/src/core/lib/channel/channel_trace.h
@@ -28,18 +28,18 @@
#include "src/core/lib/json/json.h"
namespace grpc_core {
+namespace channelz {
+
+class ChannelNode;
// Object used to hold live data for a channel. This data is exposed via the
// channelz service:
// https://github.com/grpc/proposal/blob/master/A14-channelz.md
-class ChannelTrace : public RefCounted<ChannelTrace> {
+class ChannelTrace {
public:
ChannelTrace(size_t max_events);
~ChannelTrace();
- // returns the tracer's uuid
- intptr_t GetUuid() const;
-
enum Severity {
Unset = 0, // never to be used
Info, // we start at 1 to avoid using proto default values
@@ -59,34 +59,30 @@ class ChannelTrace : public RefCounted<ChannelTrace> {
// created a new subchannel, then it would record that with a TraceEvent
// referencing the new subchannel.
//
- // TODO(ncteisen): Once channelz is implemented, the events should reference
- // the overall channelz object, not just the ChannelTrace object.
// TODO(ncteisen): as this call is used more and more throughout the gRPC
// stack, determine if it makes more sense to accept a char* instead of a
// slice.
void AddTraceEventReferencingChannel(
Severity severity, grpc_slice data,
- RefCountedPtr<ChannelTrace> referenced_tracer);
+ RefCountedPtr<ChannelNode> referenced_channel);
void AddTraceEventReferencingSubchannel(
Severity severity, grpc_slice data,
- RefCountedPtr<ChannelTrace> referenced_tracer);
+ RefCountedPtr<ChannelNode> referenced_subchannel);
- // Returns the tracing data rendered as a grpc json string.
- // The string is owned by the caller and must be freed.
- char* RenderTrace() const;
+ // Creates and returns the raw grpc_json object, so a parent channelz
+ // object may incorporate the json before rendering.
+ grpc_json* RenderJSON() const;
private:
// Types of objects that can be references by trace events.
- enum ReferencedType { Channel, Subchannel };
+ enum class ReferencedType { Channel, Subchannel };
// Private class to encapsulate all the data and bookkeeping needed for a
// a trace event.
class TraceEvent {
public:
// Constructor for a TraceEvent that references a different channel.
- // TODO(ncteisen): once channelz is implemented, this should reference the
- // overall channelz object, not just the ChannelTrace object
TraceEvent(Severity severity, grpc_slice data,
- RefCountedPtr<ChannelTrace> referenced_tracer,
+ RefCountedPtr<ChannelNode> referenced_channel,
ReferencedType type);
// Constructor for a TraceEvent that does not reverence a different
@@ -109,7 +105,7 @@ class ChannelTrace : public RefCounted<ChannelTrace> {
gpr_timespec timestamp_;
TraceEvent* next_;
// the tracer object for the (sub)channel that this trace event refers to.
- RefCountedPtr<ChannelTrace> referenced_tracer_;
+ RefCountedPtr<ChannelNode> referenced_channel_;
// the type that the referenced tracer points to. Unused if this trace
// does not point to any channel or subchannel
ReferencedType referenced_type_;
@@ -119,7 +115,6 @@ class ChannelTrace : public RefCounted<ChannelTrace> {
void AddTraceEventHelper(TraceEvent* new_trace_event);
gpr_mu tracer_mu_;
- intptr_t channel_uuid_;
uint64_t num_events_logged_;
size_t list_size_;
size_t max_list_size_;
@@ -128,6 +123,7 @@ class ChannelTrace : public RefCounted<ChannelTrace> {
gpr_timespec time_created_;
};
+} // namespace channelz
} // namespace grpc_core
#endif /* GRPC_CORE_LIB_CHANNEL_CHANNEL_TRACE_H */
diff --git a/src/core/lib/channel/channelz.cc b/src/core/lib/channel/channelz.cc
new file mode 100644
index 0000000000..3550fc0551
--- /dev/null
+++ b/src/core/lib/channel/channelz.cc
@@ -0,0 +1,185 @@
+/*
+ *
+ * Copyright 2017 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.
+ *
+ */
+
+#include <grpc/impl/codegen/port_platform.h>
+
+#include "src/core/lib/channel/channelz.h"
+
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "src/core/lib/channel/channelz_registry.h"
+#include "src/core/lib/channel/status_util.h"
+#include "src/core/lib/gpr/string.h"
+#include "src/core/lib/gpr/useful.h"
+#include "src/core/lib/gprpp/memory.h"
+#include "src/core/lib/iomgr/error.h"
+#include "src/core/lib/slice/slice_internal.h"
+#include "src/core/lib/surface/channel.h"
+#include "src/core/lib/transport/connectivity_state.h"
+#include "src/core/lib/transport/error_utils.h"
+
+namespace grpc_core {
+namespace channelz {
+
+namespace {
+
+// TODO(ncteisen): move this function to a common helper location.
+//
+// returns an allocated string that represents tm according to RFC-3339, and,
+// more specifically, follows:
+// https://developers.google.com/protocol-buffers/docs/proto3#json
+//
+// "Uses RFC 3339, where generated output will always be Z-normalized and uses
+// 0, 3, 6 or 9 fractional digits."
+char* fmt_time(gpr_timespec tm) {
+ char time_buffer[35];
+ char ns_buffer[11]; // '.' + 9 digits of precision
+ struct tm* tm_info = localtime((const time_t*)&tm.tv_sec);
+ strftime(time_buffer, sizeof(time_buffer), "%Y-%m-%dT%H:%M:%S", tm_info);
+ snprintf(ns_buffer, 11, ".%09d", tm.tv_nsec);
+ // This loop trims off trailing zeros by inserting a null character that the
+ // right point. We iterate in chunks of three because we want 0, 3, 6, or 9
+ // fractional digits.
+ for (int i = 7; i >= 1; i -= 3) {
+ if (ns_buffer[i] == '0' && ns_buffer[i + 1] == '0' &&
+ ns_buffer[i + 2] == '0') {
+ ns_buffer[i] = '\0';
+ // Edge case in which all fractional digits were 0.
+ if (i == 1) {
+ ns_buffer[0] = '\0';
+ }
+ } else {
+ break;
+ }
+ }
+ char* full_time_str;
+ gpr_asprintf(&full_time_str, "%s%sZ", time_buffer, ns_buffer);
+ return full_time_str;
+}
+
+// TODO(ncteisen); move this to json library
+grpc_json* add_num_str(grpc_json* parent, grpc_json* it, const char* name,
+ int64_t num) {
+ char* num_str;
+ gpr_asprintf(&num_str, "%" PRId64, num);
+ return grpc_json_create_child(it, parent, name, num_str, GRPC_JSON_STRING,
+ true);
+}
+
+} // namespace
+
+ChannelNode::ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes)
+ : channel_(channel), target_(nullptr), channel_uuid_(-1) {
+ trace_.Init(channel_tracer_max_nodes);
+ target_ = UniquePtr<char>(grpc_channel_get_target(channel_));
+ channel_uuid_ = ChannelzRegistry::Register(this);
+ gpr_atm_no_barrier_store(&last_call_started_millis_,
+ (gpr_atm)ExecCtx::Get()->Now());
+}
+
+ChannelNode::~ChannelNode() {
+ trace_.Destroy();
+ ChannelzRegistry::Unregister(channel_uuid_);
+}
+
+void ChannelNode::RecordCallStarted() {
+ gpr_atm_no_barrier_fetch_add(&calls_started_, (gpr_atm)1);
+ gpr_atm_no_barrier_store(&last_call_started_millis_,
+ (gpr_atm)ExecCtx::Get()->Now());
+}
+
+grpc_connectivity_state ChannelNode::GetConnectivityState() {
+ if (channel_ == nullptr) {
+ return GRPC_CHANNEL_SHUTDOWN;
+ } else {
+ return grpc_channel_check_connectivity_state(channel_, false);
+ }
+}
+
+char* ChannelNode::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 = add_num_str(json, json_iterator, "channelId", channel_uuid_);
+ // reset json iterators to top level object
+ json = top_level_json;
+ json_iterator = nullptr;
+ // create and fill the data child.
+ grpc_json* data = grpc_json_create_child(json_iterator, json, "data", nullptr,
+ GRPC_JSON_OBJECT, false);
+ json = data;
+ json_iterator = nullptr;
+ // create and fill the connectivity state child.
+ grpc_connectivity_state connectivity_state = GetConnectivityState();
+ json_iterator = grpc_json_create_child(json_iterator, json, "state", nullptr,
+ GRPC_JSON_OBJECT, false);
+ json = json_iterator;
+ grpc_json_create_child(nullptr, json, "state",
+ grpc_connectivity_state_name(connectivity_state),
+ GRPC_JSON_STRING, false);
+ // reset the parent to be the data object.
+ json = data;
+ json_iterator = grpc_json_create_child(
+ json_iterator, json, "target", target_.get(), GRPC_JSON_STRING, false);
+ // fill in the channel trace if applicable
+ grpc_json* trace = trace_->RenderJSON();
+ if (trace != nullptr) {
+ // we manuall link up and fill the child since it was created for us in
+ // ChannelTrace::RenderJSON
+ json_iterator = grpc_json_link_child(json, trace, json_iterator);
+ trace->parent = json;
+ trace->value = nullptr;
+ trace->key = "trace";
+ trace->owns_value = false;
+ }
+ // reset the parent to be the data object.
+ json = data;
+ json_iterator = nullptr;
+ // We use -1 as sentinel values since proto default value for integers is
+ // zero, and the confuses the parser into thinking the value weren't present
+ json_iterator =
+ add_num_str(json, json_iterator, "callsStarted", calls_started_);
+ json_iterator =
+ add_num_str(json, json_iterator, "callsSucceeded", calls_succeeded_);
+ json_iterator =
+ add_num_str(json, json_iterator, "callsFailed", calls_failed_);
+ gpr_timespec ts =
+ grpc_millis_to_timespec(last_call_started_millis_, GPR_CLOCK_REALTIME);
+ json_iterator =
+ grpc_json_create_child(json_iterator, json, "lastCallStartedTimestamp",
+ fmt_time(ts), GRPC_JSON_STRING, true);
+ // render and return the over json object
+ char* json_str = grpc_json_dump_to_string(top_level_json, 0);
+ grpc_json_destroy(top_level_json);
+ return json_str;
+}
+
+} // namespace channelz
+} // namespace grpc_core
diff --git a/src/core/lib/channel/channelz.h b/src/core/lib/channel/channelz.h
new file mode 100644
index 0000000000..2aad1e82f4
--- /dev/null
+++ b/src/core/lib/channel/channelz.h
@@ -0,0 +1,85 @@
+/*
+ *
+ * Copyright 2018 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_CORE_LIB_CHANNEL_CHANNELZ_H
+#define GRPC_CORE_LIB_CHANNEL_CHANNELZ_H
+
+#include <grpc/impl/codegen/port_platform.h>
+
+#include <grpc/grpc.h>
+
+#include "src/core/lib/channel/channel_trace.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"
+#include "src/core/lib/iomgr/error.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/json/json.h"
+
+namespace grpc_core {
+namespace channelz {
+
+namespace testing {
+class ChannelNodePeer;
+}
+
+class ChannelNode : public RefCounted<ChannelNode> {
+ public:
+ ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes);
+ ~ChannelNode();
+
+ void RecordCallStarted();
+ void RecordCallFailed() {
+ gpr_atm_no_barrier_fetch_add(&calls_failed_, (gpr_atm(1)));
+ }
+ void RecordCallSucceeded() {
+ gpr_atm_no_barrier_fetch_add(&calls_succeeded_, (gpr_atm(1)));
+ }
+
+ char* RenderJSON();
+
+ ChannelTrace* trace() { return trace_.get(); }
+
+ void set_channel_destroyed() {
+ GPR_ASSERT(channel_ != nullptr);
+ channel_ = nullptr;
+ }
+
+ intptr_t channel_uuid() { return channel_uuid_; }
+
+ private:
+ // testing peer friend.
+ friend class testing::ChannelNodePeer;
+
+ // helper for getting connectivity state.
+ grpc_connectivity_state GetConnectivityState();
+
+ grpc_channel* channel_ = nullptr;
+ UniquePtr<char> target_;
+ gpr_atm calls_started_ = 0;
+ gpr_atm calls_succeeded_ = 0;
+ gpr_atm calls_failed_ = 0;
+ gpr_atm last_call_started_millis_ = 0;
+ intptr_t channel_uuid_;
+ ManualConstructor<ChannelTrace> trace_;
+};
+
+} // namespace channelz
+} // namespace grpc_core
+
+#endif /* GRPC_CORE_LIB_CHANNEL_CHANNELZ_H */
diff --git a/src/core/lib/channel/connected_channel.cc b/src/core/lib/channel/connected_channel.cc
index ddd3029402..e2ea334ded 100644
--- a/src/core/lib/channel/connected_channel.cc
+++ b/src/core/lib/channel/connected_channel.cc
@@ -51,6 +51,7 @@ typedef struct connected_channel_call_data {
callback_state on_complete[6]; // Max number of pending batches.
callback_state recv_initial_metadata_ready;
callback_state recv_message_ready;
+ callback_state recv_trailing_metadata_ready;
} call_data;
static void run_in_call_combiner(void* arg, grpc_error* error) {
@@ -111,6 +112,12 @@ static void con_start_transport_stream_op_batch(
intercept_callback(calld, state, false, "recv_message_ready",
&batch->payload->recv_message.recv_message_ready);
}
+ if (batch->recv_trailing_metadata) {
+ callback_state* state = &calld->recv_trailing_metadata_ready;
+ intercept_callback(
+ calld, state, false, "recv_trailing_metadata_ready",
+ &batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready);
+ }
if (batch->cancel_stream) {
// There can be more than one cancellation batch in flight at any
// given time, so we can't just pick out a fixed index into
@@ -121,7 +128,7 @@ static void con_start_transport_stream_op_batch(
static_cast<callback_state*>(gpr_malloc(sizeof(*state)));
intercept_callback(calld, state, true, "on_complete (cancel_stream)",
&batch->on_complete);
- } else {
+ } else if (batch->on_complete != nullptr) {
callback_state* state = get_state_for_batch(calld, batch);
intercept_callback(calld, state, false, "on_complete", &batch->on_complete);
}
diff --git a/src/core/lib/gprpp/inlined_vector.h b/src/core/lib/gprpp/inlined_vector.h
index f36f6cb706..0d2586e507 100644
--- a/src/core/lib/gprpp/inlined_vector.h
+++ b/src/core/lib/gprpp/inlined_vector.h
@@ -99,6 +99,8 @@ class InlinedVector {
void push_back(T&& value) { emplace_back(std::move(value)); }
size_t size() const { return size_; }
+ bool empty() const { return size_ == 0; }
+
size_t capacity() const { return capacity_; }
void clear() {
diff --git a/src/core/lib/iomgr/call_combiner.h b/src/core/lib/iomgr/call_combiner.h
index 0ccd08ea57..641fa18082 100644
--- a/src/core/lib/iomgr/call_combiner.h
+++ b/src/core/lib/iomgr/call_combiner.h
@@ -26,6 +26,7 @@
#include <grpc/support/atm.h>
#include "src/core/lib/gpr/mpscq.h"
+#include "src/core/lib/gprpp/inlined_vector.h"
#include "src/core/lib/iomgr/closure.h"
// A simple, lock-free mechanism for serializing activity related to a
@@ -109,4 +110,83 @@ void grpc_call_combiner_set_notify_on_cancel(grpc_call_combiner* call_combiner,
void grpc_call_combiner_cancel(grpc_call_combiner* call_combiner,
grpc_error* error);
+namespace grpc_core {
+
+// Helper for running a list of closures in a call combiner.
+//
+// Each callback running in the call combiner will eventually be
+// returned to the surface, at which point the surface will yield the
+// call combiner. So when we are running in the call combiner and have
+// more than one callback to return to the surface, we need to re-enter
+// the call combiner for all but one of those callbacks.
+class CallCombinerClosureList {
+ public:
+ CallCombinerClosureList() {}
+
+ // Adds a closure to the list. The closure must eventually result in
+ // the call combiner being yielded.
+ void Add(grpc_closure* closure, grpc_error* error, const char* reason) {
+ closures_.emplace_back(closure, error, reason);
+ }
+
+ // Runs all closures in the call combiner and yields the call combiner.
+ //
+ // All but one of the closures in the list will be scheduled via
+ // GRPC_CALL_COMBINER_START(), and the remaining closure will be
+ // scheduled via GRPC_CLOSURE_SCHED(), which will eventually result in
+ // yielding the call combiner. If the list is empty, then the call
+ // combiner will be yielded immediately.
+ void RunClosures(grpc_call_combiner* call_combiner) {
+ if (closures_.empty()) {
+ GRPC_CALL_COMBINER_STOP(call_combiner, "no closures to schedule");
+ return;
+ }
+ for (size_t i = 1; i < closures_.size(); ++i) {
+ auto& closure = closures_[i];
+ GRPC_CALL_COMBINER_START(call_combiner, closure.closure, closure.error,
+ closure.reason);
+ }
+ if (grpc_call_combiner_trace.enabled()) {
+ gpr_log(GPR_INFO,
+ "CallCombinerClosureList executing closure while already "
+ "holding call_combiner %p: closure=%p error=%s reason=%s",
+ call_combiner, closures_[0].closure,
+ grpc_error_string(closures_[0].error), closures_[0].reason);
+ }
+ // This will release the call combiner.
+ GRPC_CLOSURE_SCHED(closures_[0].closure, closures_[0].error);
+ closures_.clear();
+ }
+
+ // Runs all closures in the call combiner, but does NOT yield the call
+ // combiner. All closures will be scheduled via GRPC_CALL_COMBINER_START().
+ void RunClosuresWithoutYielding(grpc_call_combiner* call_combiner) {
+ for (size_t i = 0; i < closures_.size(); ++i) {
+ auto& closure = closures_[i];
+ GRPC_CALL_COMBINER_START(call_combiner, closure.closure, closure.error,
+ closure.reason);
+ }
+ closures_.clear();
+ }
+
+ size_t size() const { return closures_.size(); }
+
+ private:
+ struct CallCombinerClosure {
+ grpc_closure* closure;
+ grpc_error* error;
+ const char* reason;
+
+ CallCombinerClosure(grpc_closure* closure, grpc_error* error,
+ const char* reason)
+ : closure(closure), error(error), reason(reason) {}
+ };
+
+ // There are generally a maximum of 6 closures to run in the call
+ // combiner, one for each pending op.
+ InlinedVector<CallCombinerClosure, 6> closures_;
+};
+
+} // namespace grpc_core
+
#endif /* GRPC_CORE_LIB_IOMGR_CALL_COMBINER_H */
diff --git a/src/core/lib/iomgr/closure.h b/src/core/lib/iomgr/closure.h
index 34a494485d..f14c723844 100644
--- a/src/core/lib/iomgr/closure.h
+++ b/src/core/lib/iomgr/closure.h
@@ -283,9 +283,10 @@ inline void grpc_closure_sched(grpc_closure* c, grpc_error* error) {
if (c->scheduled) {
gpr_log(GPR_ERROR,
"Closure already scheduled. (closure: %p, created: [%s:%d], "
- "previously scheduled at: [%s: %d] run?: %s",
+ "previously scheduled at: [%s: %d], newly scheduled at [%s: %d], "
+ "run?: %s",
c, c->file_created, c->line_created, c->file_initiated,
- c->line_initiated, c->run ? "true" : "false");
+ c->line_initiated, file, line, c->run ? "true" : "false");
abort();
}
c->scheduled = true;
diff --git a/src/core/lib/iomgr/socket_utils.h b/src/core/lib/iomgr/socket_utils.h
index cf1a7be648..14bb081e93 100644
--- a/src/core/lib/iomgr/socket_utils.h
+++ b/src/core/lib/iomgr/socket_utils.h
@@ -23,12 +23,21 @@
#include <stddef.h>
+// TODO(juanlishen): The following functions might be simple enough to implement
+// ourselves, so that they don't cause any portability hassle.
+
/* A wrapper for htons on POSIX and Windows */
uint16_t grpc_htons(uint16_t hostshort);
/* A wrapper for ntohs on POSIX and WINDOWS */
uint16_t grpc_ntohs(uint16_t netshort);
+/* A wrapper for htonl on POSIX and Windows */
+uint32_t grpc_htonl(uint32_t hostlong);
+
+/* A wrapper for ntohl on POSIX and WINDOWS */
+uint32_t grpc_ntohl(uint32_t netlong);
+
/* A wrapper for inet_pton on POSIX and WINDOWS */
int grpc_inet_pton(int af, const char* src, void* dst);
diff --git a/src/core/lib/iomgr/socket_utils_common_posix.cc b/src/core/lib/iomgr/socket_utils_common_posix.cc
index caee652307..c4b991c94d 100644
--- a/src/core/lib/iomgr/socket_utils_common_posix.cc
+++ b/src/core/lib/iomgr/socket_utils_common_posix.cc
@@ -339,6 +339,10 @@ uint16_t grpc_htons(uint16_t hostshort) { return htons(hostshort); }
uint16_t grpc_ntohs(uint16_t netshort) { return ntohs(netshort); }
+uint32_t grpc_htonl(uint32_t hostlong) { return htonl(hostlong); }
+
+uint32_t grpc_ntohl(uint32_t netlong) { return ntohl(netlong); }
+
int grpc_inet_pton(int af, const char* src, void* dst) {
return inet_pton(af, src, dst);
}
diff --git a/src/core/lib/iomgr/socket_utils_uv.cc b/src/core/lib/iomgr/socket_utils_uv.cc
index 7eba40c46b..b5f96b52df 100644
--- a/src/core/lib/iomgr/socket_utils_uv.cc
+++ b/src/core/lib/iomgr/socket_utils_uv.cc
@@ -33,6 +33,10 @@ uint16_t grpc_htons(uint16_t hostshort) { return htons(hostshort); }
uint16_t grpc_ntohs(uint16_t netshort) { return ntohs(netshort); }
+uint32_t grpc_htonl(uint32_t hostlong) { return htonl(hostlong); }
+
+uint32_t grpc_ntohl(uint32_t netlong) { return ntohl(netlong); }
+
int grpc_inet_pton(int af, const char* src, void* dst) {
return inet_pton(af, src, dst);
}
diff --git a/src/core/lib/iomgr/socket_utils_windows.cc b/src/core/lib/iomgr/socket_utils_windows.cc
index 3e7b5b812d..9137ab98e6 100644
--- a/src/core/lib/iomgr/socket_utils_windows.cc
+++ b/src/core/lib/iomgr/socket_utils_windows.cc
@@ -31,6 +31,10 @@ uint16_t grpc_htons(uint16_t hostshort) { return htons(hostshort); }
uint16_t grpc_ntohs(uint16_t netshort) { return ntohs(netshort); }
+uint32_t grpc_htonl(uint32_t hostlong) { return htonl(hostlong); }
+
+uint32_t grpc_ntohl(uint32_t netlong) { return ntohl(netlong); }
+
int grpc_inet_pton(int af, const char* src, void* dst) {
return inet_pton(af, src, dst);
}
diff --git a/src/core/lib/security/credentials/google_default/google_default_credentials.cc b/src/core/lib/security/credentials/google_default/google_default_credentials.cc
index 38c9175717..c456ffaf5d 100644
--- a/src/core/lib/security/credentials/google_default/google_default_credentials.cc
+++ b/src/core/lib/security/credentials/google_default/google_default_credentials.cc
@@ -231,7 +231,8 @@ end:
creds->base.vtable = &google_default_credentials_vtable;
creds->base.type = GRPC_CHANNEL_CREDENTIALS_TYPE_GOOGLE_DEFAULT;
gpr_ref_init(&creds->base.refcount, 1);
- creds->ssl_creds = grpc_ssl_credentials_create(nullptr, nullptr, nullptr);
+ creds->ssl_creds =
+ grpc_ssl_credentials_create(nullptr, nullptr, nullptr, nullptr);
GPR_ASSERT(creds->ssl_creds != nullptr);
grpc_alts_credentials_options* options =
grpc_alts_credentials_client_options_create();
diff --git a/src/core/lib/security/credentials/ssl/ssl_credentials.cc b/src/core/lib/security/credentials/ssl/ssl_credentials.cc
index 2b6377d3ec..3d6f2f200a 100644
--- a/src/core/lib/security/credentials/ssl/ssl_credentials.cc
+++ b/src/core/lib/security/credentials/ssl/ssl_credentials.cc
@@ -48,6 +48,10 @@ static void ssl_destruct(grpc_channel_credentials* creds) {
grpc_ssl_credentials* c = reinterpret_cast<grpc_ssl_credentials*>(creds);
gpr_free(c->config.pem_root_certs);
grpc_tsi_ssl_pem_key_cert_pairs_destroy(c->config.pem_key_cert_pair, 1);
+ if (c->config.verify_options.verify_peer_destruct != nullptr) {
+ c->config.verify_options.verify_peer_destruct(
+ c->config.verify_options.verify_peer_callback_userdata);
+ }
}
static grpc_security_status ssl_create_security_connector(
@@ -87,6 +91,7 @@ static grpc_channel_credentials_vtable ssl_vtable = {
static void ssl_build_config(const char* pem_root_certs,
grpc_ssl_pem_key_cert_pair* pem_key_cert_pair,
+ const verify_peer_options* verify_options,
grpc_ssl_config* config) {
if (pem_root_certs != nullptr) {
config->pem_root_certs = gpr_strdup(pem_root_certs);
@@ -101,23 +106,32 @@ static void ssl_build_config(const char* pem_root_certs,
config->pem_key_cert_pair->private_key =
gpr_strdup(pem_key_cert_pair->private_key);
}
+ if (verify_options != nullptr) {
+ memcpy(&config->verify_options, verify_options,
+ sizeof(verify_peer_options));
+ } else {
+ // Otherwise set all options to default values
+ memset(&config->verify_options, 0, sizeof(verify_peer_options));
+ }
}
grpc_channel_credentials* grpc_ssl_credentials_create(
const char* pem_root_certs, grpc_ssl_pem_key_cert_pair* pem_key_cert_pair,
- void* reserved) {
+ const verify_peer_options* verify_options, void* reserved) {
grpc_ssl_credentials* c = static_cast<grpc_ssl_credentials*>(
gpr_zalloc(sizeof(grpc_ssl_credentials)));
GRPC_API_TRACE(
"grpc_ssl_credentials_create(pem_root_certs=%s, "
"pem_key_cert_pair=%p, "
+ "verify_options=%p, "
"reserved=%p)",
- 3, (pem_root_certs, pem_key_cert_pair, reserved));
+ 4, (pem_root_certs, pem_key_cert_pair, verify_options, reserved));
GPR_ASSERT(reserved == nullptr);
c->base.type = GRPC_CHANNEL_CREDENTIALS_TYPE_SSL;
c->base.vtable = &ssl_vtable;
gpr_ref_init(&c->base.refcount, 1);
- ssl_build_config(pem_root_certs, pem_key_cert_pair, &c->config);
+ ssl_build_config(pem_root_certs, pem_key_cert_pair, verify_options,
+ &c->config);
return &c->base;
}
diff --git a/src/core/lib/security/security_connector/security_connector.cc b/src/core/lib/security/security_connector/security_connector.cc
index b54a7643e4..cc72bb6164 100644
--- a/src/core/lib/security/security_connector/security_connector.cc
+++ b/src/core/lib/security/security_connector/security_connector.cc
@@ -620,6 +620,7 @@ typedef struct {
tsi_ssl_client_handshaker_factory* client_handshaker_factory;
char* target_name;
char* overridden_target_name;
+ const verify_peer_options* verify_options;
} grpc_ssl_channel_security_connector;
typedef struct {
@@ -878,11 +879,34 @@ static void ssl_channel_check_peer(grpc_security_connector* sc, tsi_peer peer,
grpc_closure* on_peer_checked) {
grpc_ssl_channel_security_connector* c =
reinterpret_cast<grpc_ssl_channel_security_connector*>(sc);
- grpc_error* error = ssl_check_peer(sc,
- c->overridden_target_name != nullptr
- ? c->overridden_target_name
- : c->target_name,
- &peer, auth_context);
+ const char* target_name = c->overridden_target_name != nullptr
+ ? c->overridden_target_name
+ : c->target_name;
+ grpc_error* error = ssl_check_peer(sc, target_name, &peer, auth_context);
+ if (error == GRPC_ERROR_NONE &&
+ c->verify_options->verify_peer_callback != nullptr) {
+ const tsi_peer_property* p =
+ tsi_peer_get_property_by_name(&peer, TSI_X509_PEM_CERT_PROPERTY);
+ if (p == nullptr) {
+ error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+ "Cannot check peer: missing pem cert property.");
+ } else {
+ char* peer_pem = static_cast<char*>(gpr_malloc(p->value.length + 1));
+ memcpy(peer_pem, p->value.data, p->value.length);
+ peer_pem[p->value.length] = '\0';
+ int callback_status = c->verify_options->verify_peer_callback(
+ target_name, peer_pem,
+ c->verify_options->verify_peer_callback_userdata);
+ gpr_free(peer_pem);
+ if (callback_status) {
+ char* msg;
+ gpr_asprintf(&msg, "Verify peer callback returned a failure (%d)",
+ callback_status);
+ error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg);
+ gpr_free(msg);
+ }
+ }
+ }
GRPC_CLOSURE_SCHED(on_peer_checked, error);
tsi_peer_destruct(&peer);
}
@@ -1047,6 +1071,7 @@ grpc_security_status grpc_ssl_channel_security_connector_create(
if (overridden_target_name != nullptr) {
c->overridden_target_name = gpr_strdup(overridden_target_name);
}
+ c->verify_options = &config->verify_options;
has_key_cert_pair = config->pem_key_cert_pair != nullptr &&
config->pem_key_cert_pair->private_key != nullptr &&
diff --git a/src/core/lib/security/security_connector/security_connector.h b/src/core/lib/security/security_connector/security_connector.h
index f9723166d0..67a506b576 100644
--- a/src/core/lib/security/security_connector/security_connector.h
+++ b/src/core/lib/security/security_connector/security_connector.h
@@ -193,6 +193,7 @@ grpc_server_security_connector* grpc_fake_server_security_connector_create(
typedef struct {
tsi_ssl_pem_key_cert_pair* pem_key_cert_pair;
char* pem_root_certs;
+ verify_peer_options verify_options;
} grpc_ssl_config;
/* Creates an SSL channel_security_connector.
diff --git a/src/core/lib/surface/call.cc b/src/core/lib/surface/call.cc
index 1cf8ea94e7..556eb234b4 100644
--- a/src/core/lib/surface/call.cc
+++ b/src/core/lib/surface/call.cc
@@ -233,6 +233,7 @@ struct grpc_call {
grpc_closure receiving_slice_ready;
grpc_closure receiving_stream_ready;
grpc_closure receiving_initial_metadata_ready;
+ grpc_closure receiving_trailing_metadata_ready;
uint32_t test_only_last_message_flags;
grpc_closure release_call;
@@ -270,8 +271,17 @@ struct grpc_call {
grpc_core::TraceFlag grpc_call_error_trace(false, "call_error");
grpc_core::TraceFlag grpc_compression_trace(false, "compression");
-#define CALL_STACK_FROM_CALL(call) ((grpc_call_stack*)((call) + 1))
-#define CALL_FROM_CALL_STACK(call_stack) (((grpc_call*)(call_stack)) - 1)
+/* Given a size, round up to the next multiple of sizeof(void*) */
+#define ROUND_UP_TO_ALIGNMENT_SIZE(x) \
+ (((x) + GPR_MAX_ALIGNMENT - 1u) & ~(GPR_MAX_ALIGNMENT - 1u))
+
+#define CALL_STACK_FROM_CALL(call) \
+ (grpc_call_stack*)((char*)(call) + \
+ ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call)))
+#define CALL_FROM_CALL_STACK(call_stack) \
+ (grpc_call*)(((char*)(call_stack)) - \
+ ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call)))
+
#define CALL_ELEM_FROM_CALL(call, idx) \
grpc_call_stack_element(CALL_STACK_FROM_CALL(call), idx)
#define CALL_FROM_TOP_ELEM(top_elem) \
@@ -342,8 +352,9 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,
size_t initial_size = grpc_channel_get_call_size_estimate(args->channel);
GRPC_STATS_INC_CALL_INITIAL_SIZE(initial_size);
gpr_arena* arena = gpr_arena_create(initial_size);
- call = static_cast<grpc_call*>(gpr_arena_alloc(
- arena, sizeof(grpc_call) + channel_stack->call_stack_size));
+ call = static_cast<grpc_call*>(
+ gpr_arena_alloc(arena, ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call)) +
+ channel_stack->call_stack_size));
gpr_ref_init(&call->ext_ref, 1);
call->arena = arena;
grpc_call_combiner_init(&call->call_combiner);
@@ -478,6 +489,12 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args,
&call->pollent);
}
+ grpc_core::channelz::ChannelNode* channelz_channel =
+ grpc_channel_get_channelz_node(call->channel);
+ if (channelz_channel != nullptr) {
+ channelz_channel->RecordCallStarted();
+ }
+
grpc_slice_unref_internal(path);
return error;
@@ -520,7 +537,6 @@ static void release_call(void* call, grpc_error* error) {
GRPC_CHANNEL_INTERNAL_UNREF(channel, "call");
}
-static void set_status_value_directly(grpc_status_code status, void* dest);
static void destroy_call(void* call, grpc_error* error) {
GPR_TIMER_SCOPE("destroy_call", 0);
size_t i;
@@ -1076,13 +1092,12 @@ static void recv_trailing_filter(void* args, grpc_metadata_batch* b) {
if (b->idx.named.grpc_status != nullptr) {
grpc_status_code status_code =
grpc_get_status_code_from_metadata(b->idx.named.grpc_status->md);
- grpc_error* error =
- status_code == GRPC_STATUS_OK
- ? GRPC_ERROR_NONE
- : grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
- "Error received from peer"),
- GRPC_ERROR_INT_GRPC_STATUS,
- static_cast<intptr_t>(status_code));
+ grpc_error* error = GRPC_ERROR_NONE;
+ if (status_code != GRPC_STATUS_OK) {
+ error = grpc_error_set_int(
+ GRPC_ERROR_CREATE_FROM_STATIC_STRING("Error received from peer"),
+ GRPC_ERROR_INT_GRPC_STATUS, static_cast<intptr_t>(status_code));
+ }
if (b->idx.named.grpc_message != nullptr) {
error = grpc_error_set_str(
error, GRPC_ERROR_STR_GRPC_MESSAGE,
@@ -1209,7 +1224,6 @@ static void post_batch_completion(batch_control* bctl) {
if (bctl->op.send_initial_metadata) {
grpc_metadata_batch_destroy(
-
&call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]);
}
if (bctl->op.send_message) {
@@ -1217,14 +1231,9 @@ static void post_batch_completion(batch_control* bctl) {
}
if (bctl->op.send_trailing_metadata) {
grpc_metadata_batch_destroy(
-
&call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]);
}
if (bctl->op.recv_trailing_metadata) {
- grpc_metadata_batch* md =
- &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
- recv_trailing_filter(call, md);
-
/* propagate cancellation to any interested children */
gpr_atm_rel_store(&call->received_final_op_atm, 1);
parent_call* pc = get_parent_call(call);
@@ -1246,7 +1255,6 @@ static void post_batch_completion(batch_control* bctl) {
}
gpr_mu_unlock(&pc->child_list_mu);
}
-
if (call->is_client) {
get_final_status(call, set_status_value_directly,
call->final_op.client.status,
@@ -1256,7 +1264,15 @@ static void post_batch_completion(batch_control* bctl) {
get_final_status(call, set_cancelled_value,
call->final_op.server.cancelled, nullptr, nullptr);
}
-
+ grpc_core::channelz::ChannelNode* channelz_channel =
+ grpc_channel_get_channelz_node(call->channel);
+ if (channelz_channel != nullptr) {
+ if (*call->final_op.client.status != GRPC_STATUS_OK) {
+ channelz_channel->RecordCallFailed();
+ } else {
+ channelz_channel->RecordCallSucceeded();
+ }
+ }
GRPC_ERROR_UNREF(error);
error = GRPC_ERROR_NONE;
}
@@ -1538,6 +1554,17 @@ static void receiving_initial_metadata_ready(void* bctlp, grpc_error* error) {
finish_batch_step(bctl);
}
+static void receiving_trailing_metadata_ready(void* bctlp, grpc_error* error) {
+ batch_control* bctl = static_cast<batch_control*>(bctlp);
+ grpc_call* call = bctl->call;
+ GRPC_CALL_COMBINER_STOP(&call->call_combiner, "recv_trailing_metadata_ready");
+ add_batch_error(bctl, GRPC_ERROR_REF(error), false);
+ grpc_metadata_batch* md =
+ &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
+ recv_trailing_filter(call, md);
+ finish_batch_step(bctl);
+}
+
static void finish_batch(void* bctlp, grpc_error* error) {
batch_control* bctl = static_cast<batch_control*>(bctlp);
grpc_call* call = bctl->call;
@@ -1558,7 +1585,8 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
size_t i;
const grpc_op* op;
batch_control* bctl;
- int num_completion_callbacks_needed = 1;
+ bool has_send_ops = false;
+ int num_recv_ops = 0;
grpc_call_error error = GRPC_CALL_OK;
grpc_transport_stream_op_batch* stream_op;
grpc_transport_stream_op_batch_payload* stream_op_payload;
@@ -1664,6 +1692,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
stream_op_payload->send_initial_metadata.peer_string =
&call->peer_string;
}
+ has_send_ops = true;
break;
}
case GRPC_OP_SEND_MESSAGE: {
@@ -1693,6 +1722,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
&op->data.send_message.send_message->data.raw.slice_buffer, flags);
stream_op_payload->send_message.send_message.reset(
call->sending_stream.get());
+ has_send_ops = true;
break;
}
case GRPC_OP_SEND_CLOSE_FROM_CLIENT: {
@@ -1713,6 +1743,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
call->sent_final_op = true;
stream_op_payload->send_trailing_metadata.send_trailing_metadata =
&call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */];
+ has_send_ops = true;
break;
}
case GRPC_OP_SEND_STATUS_FROM_SERVER: {
@@ -1777,6 +1808,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
}
stream_op_payload->send_trailing_metadata.send_trailing_metadata =
&call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */];
+ has_send_ops = true;
break;
}
case GRPC_OP_RECV_INITIAL_METADATA: {
@@ -1804,7 +1836,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
stream_op_payload->recv_initial_metadata.peer_string =
&call->peer_string;
}
- num_completion_callbacks_needed++;
+ ++num_recv_ops;
break;
}
case GRPC_OP_RECV_MESSAGE: {
@@ -1826,7 +1858,7 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
grpc_schedule_on_exec_ctx);
stream_op_payload->recv_message.recv_message_ready =
&call->receiving_stream_ready;
- num_completion_callbacks_needed++;
+ ++num_recv_ops;
break;
}
case GRPC_OP_RECV_STATUS_ON_CLIENT: {
@@ -1852,11 +1884,16 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
call->final_op.client.error_string =
op->data.recv_status_on_client.error_string;
stream_op->recv_trailing_metadata = true;
- stream_op->collect_stats = true;
stream_op_payload->recv_trailing_metadata.recv_trailing_metadata =
&call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
- stream_op_payload->collect_stats.collect_stats =
+ stream_op_payload->recv_trailing_metadata.collect_stats =
&call->final_info.stats.transport_stream_stats;
+ GRPC_CLOSURE_INIT(&call->receiving_trailing_metadata_ready,
+ receiving_trailing_metadata_ready, bctl,
+ grpc_schedule_on_exec_ctx);
+ stream_op_payload->recv_trailing_metadata.recv_trailing_metadata_ready =
+ &call->receiving_trailing_metadata_ready;
+ ++num_recv_ops;
break;
}
case GRPC_OP_RECV_CLOSE_ON_SERVER: {
@@ -1877,11 +1914,16 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
call->final_op.server.cancelled =
op->data.recv_close_on_server.cancelled;
stream_op->recv_trailing_metadata = true;
- stream_op->collect_stats = true;
stream_op_payload->recv_trailing_metadata.recv_trailing_metadata =
&call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
- stream_op_payload->collect_stats.collect_stats =
+ stream_op_payload->recv_trailing_metadata.collect_stats =
&call->final_info.stats.transport_stream_stats;
+ GRPC_CLOSURE_INIT(&call->receiving_trailing_metadata_ready,
+ receiving_trailing_metadata_ready, bctl,
+ grpc_schedule_on_exec_ctx);
+ stream_op_payload->recv_trailing_metadata.recv_trailing_metadata_ready =
+ &call->receiving_trailing_metadata_ready;
+ ++num_recv_ops;
break;
}
}
@@ -1891,13 +1933,15 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
if (!is_notify_tag_closure) {
GPR_ASSERT(grpc_cq_begin_op(call->cq, notify_tag));
}
- gpr_ref_init(&bctl->steps_to_complete, num_completion_callbacks_needed);
+ gpr_ref_init(&bctl->steps_to_complete, (has_send_ops ? 1 : 0) + num_recv_ops);
- GRPC_CLOSURE_INIT(&bctl->finish_batch, finish_batch, bctl,
- grpc_schedule_on_exec_ctx);
- stream_op->on_complete = &bctl->finish_batch;
- gpr_atm_rel_store(&call->any_ops_sent_atm, 1);
+ if (has_send_ops) {
+ GRPC_CLOSURE_INIT(&bctl->finish_batch, finish_batch, bctl,
+ grpc_schedule_on_exec_ctx);
+ stream_op->on_complete = &bctl->finish_batch;
+ }
+ gpr_atm_rel_store(&call->any_ops_sent_atm, 1);
execute_batch(call, stream_op, &bctl->start_batch);
done:
diff --git a/src/core/lib/surface/channel.cc b/src/core/lib/surface/channel.cc
index a466b325be..d5d75fcb2a 100644
--- a/src/core/lib/surface/channel.cc
+++ b/src/core/lib/surface/channel.cc
@@ -32,6 +32,7 @@
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/channel_trace.h"
+#include "src/core/lib/channel/channelz.h"
#include "src/core/lib/debug/stats.h"
#include "src/core/lib/gpr/string.h"
#include "src/core/lib/gprpp/manual_constructor.h"
@@ -66,7 +67,7 @@ struct grpc_channel {
gpr_mu registered_call_mu;
registered_call* registered_calls;
- grpc_core::RefCountedPtr<grpc_core::ChannelTrace> tracer;
+ grpc_core::RefCountedPtr<grpc_core::channelz::ChannelNode> channelz_channel;
char* target;
};
@@ -103,6 +104,7 @@ grpc_channel* grpc_channel_create_with_builder(
channel->target = target;
channel->is_client = grpc_channel_stack_type_is_client(channel_stack_type);
size_t channel_tracer_max_nodes = 0; // default to off
+ bool channelz_enabled = false;
gpr_mu_init(&channel->registered_call_mu);
channel->registered_calls = nullptr;
@@ -141,15 +143,20 @@ grpc_channel* grpc_channel_create_with_builder(
const grpc_integer_options options = {0, 0, INT_MAX};
channel_tracer_max_nodes =
(size_t)grpc_channel_arg_get_integer(&args->args[i], options);
+ } else if (0 == strcmp(args->args[i].key, GRPC_ARG_ENABLE_CHANNELZ)) {
+ channelz_enabled = grpc_channel_arg_get_bool(&args->args[i], false);
}
}
grpc_channel_args_destroy(args);
- channel->tracer = grpc_core::MakeRefCounted<grpc_core::ChannelTrace>(
- channel_tracer_max_nodes);
- channel->tracer->AddTraceEvent(
- grpc_core::ChannelTrace::Severity::Info,
- grpc_slice_from_static_string("Channel created"));
+ if (channelz_enabled) {
+ channel->channelz_channel =
+ grpc_core::MakeRefCounted<grpc_core::channelz::ChannelNode>(
+ channel, channel_tracer_max_nodes);
+ channel->channelz_channel->trace()->AddTraceEvent(
+ grpc_core::channelz::ChannelTrace::Severity::Info,
+ grpc_slice_from_static_string("Channel created"));
+ }
return channel;
}
@@ -184,12 +191,9 @@ static grpc_channel_args* build_channel_args(
return grpc_channel_args_copy_and_add(input_args, new_args, num_new_args);
}
-char* grpc_channel_get_trace(grpc_channel* channel) {
- return channel->tracer->RenderTrace();
-}
-
-intptr_t grpc_channel_get_uuid(grpc_channel* channel) {
- return channel->tracer->GetUuid();
+grpc_core::channelz::ChannelNode* grpc_channel_get_channelz_node(
+ grpc_channel* channel) {
+ return channel->channelz_channel.get();
}
grpc_channel* grpc_channel_create(const char* target,
@@ -395,6 +399,10 @@ void grpc_channel_internal_unref(grpc_channel* c REF_ARG) {
static void destroy_channel(void* arg, grpc_error* error) {
grpc_channel* channel = static_cast<grpc_channel*>(arg);
+ if (channel->channelz_channel != nullptr) {
+ channel->channelz_channel->set_channel_destroyed();
+ channel->channelz_channel.reset();
+ }
grpc_channel_stack_destroy(CHANNEL_STACK_FROM_CHANNEL(channel));
while (channel->registered_calls) {
registered_call* rc = channel->registered_calls;
@@ -403,7 +411,6 @@ static void destroy_channel(void* arg, grpc_error* error) {
GRPC_MDELEM_UNREF(rc->authority);
gpr_free(rc);
}
- channel->tracer.reset();
gpr_mu_destroy(&channel->registered_call_mu);
gpr_free(channel->target);
gpr_free(channel);
diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h
index 288313951e..e5ff2c3596 100644
--- a/src/core/lib/surface/channel.h
+++ b/src/core/lib/surface/channel.h
@@ -23,6 +23,7 @@
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/channel/channel_stack_builder.h"
+#include "src/core/lib/channel/channelz.h"
#include "src/core/lib/surface/channel_stack_type.h"
grpc_channel* grpc_channel_create(const char* target,
@@ -50,6 +51,9 @@ grpc_call* grpc_channel_create_pollset_set_call(
/** Get a (borrowed) pointer to this channels underlying channel stack */
grpc_channel_stack* grpc_channel_get_channel_stack(grpc_channel* channel);
+grpc_core::channelz::ChannelNode* grpc_channel_get_channelz_node(
+ grpc_channel* channel);
+
/** Get a grpc_mdelem of grpc-status: X where X is the numeric value of
status_code.
diff --git a/src/core/lib/transport/transport.cc b/src/core/lib/transport/transport.cc
index 039d603394..cbdb77c844 100644
--- a/src/core/lib/transport/transport.cc
+++ b/src/core/lib/transport/transport.cc
@@ -212,21 +212,32 @@ void grpc_transport_stream_op_batch_finish_with_failure(
if (batch->send_message) {
batch->payload->send_message.send_message.reset();
}
- if (batch->recv_message) {
- GRPC_CALL_COMBINER_START(
- call_combiner, batch->payload->recv_message.recv_message_ready,
- GRPC_ERROR_REF(error), "failing recv_message_ready");
+ if (batch->cancel_stream) {
+ GRPC_ERROR_UNREF(batch->payload->cancel_stream.cancel_error);
}
+ // Construct a list of closures to execute.
+ grpc_core::CallCombinerClosureList closures;
if (batch->recv_initial_metadata) {
- GRPC_CALL_COMBINER_START(
- call_combiner,
+ closures.Add(
batch->payload->recv_initial_metadata.recv_initial_metadata_ready,
GRPC_ERROR_REF(error), "failing recv_initial_metadata_ready");
}
- GRPC_CLOSURE_SCHED(batch->on_complete, error);
- if (batch->cancel_stream) {
- GRPC_ERROR_UNREF(batch->payload->cancel_stream.cancel_error);
+ if (batch->recv_message) {
+ closures.Add(batch->payload->recv_message.recv_message_ready,
+ GRPC_ERROR_REF(error), "failing recv_message_ready");
+ }
+ if (batch->recv_trailing_metadata) {
+ closures.Add(
+ batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready,
+ GRPC_ERROR_REF(error), "failing recv_trailing_metadata_ready");
+ }
+ if (batch->on_complete != nullptr) {
+ closures.Add(batch->on_complete, GRPC_ERROR_REF(error),
+ "failing on_complete");
}
+ // Execute closures.
+ closures.RunClosures(call_combiner);
+ GRPC_ERROR_UNREF(error);
}
typedef struct {
diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h
index b2e252d939..585b9dfae9 100644
--- a/src/core/lib/transport/transport.h
+++ b/src/core/lib/transport/transport.h
@@ -122,9 +122,15 @@ typedef struct grpc_transport_stream_op_batch_payload
/* Transport stream op: a set of operations to perform on a transport
against a single stream */
typedef struct grpc_transport_stream_op_batch {
- /** Should be enqueued when all requested operations (excluding recv_message
- and recv_initial_metadata which have their own closures) in a given batch
- have been completed. */
+ /** Should be scheduled when all of the non-recv operations in the batch
+ are complete.
+
+ The recv ops (recv_initial_metadata, recv_message, and
+ recv_trailing_metadata) each have their own callbacks. If a batch
+ contains both recv ops and non-recv ops, on_complete should be
+ scheduled as soon as the non-recv ops are complete, regardless of
+ whether or not the recv ops are complete. If a batch contains
+ only recv ops, on_complete can be null. */
grpc_closure* on_complete;
/** Values for the stream op (fields set are determined by flags above) */
@@ -149,9 +155,6 @@ typedef struct grpc_transport_stream_op_batch {
*/
bool recv_trailing_metadata : 1;
- /** Collect any stats into provided buffer, zero internal stat counters */
- bool collect_stats : 1;
-
/** Cancel this stream with the provided error */
bool cancel_stream : 1;
@@ -219,11 +222,10 @@ struct grpc_transport_stream_op_batch_payload {
struct {
grpc_metadata_batch* recv_trailing_metadata;
- } recv_trailing_metadata;
-
- struct {
grpc_transport_stream_stats* collect_stats;
- } collect_stats;
+ /** Should be enqueued when initial metadata is ready to be processed. */
+ grpc_closure* recv_trailing_metadata_ready;
+ } recv_trailing_metadata;
/** Forcefully close this stream.
The HTTP2 semantics should be:
diff --git a/src/core/lib/transport/transport_op_string.cc b/src/core/lib/transport/transport_op_string.cc
index 25ab492f3a..8c7db642a5 100644
--- a/src/core/lib/transport/transport_op_string.cc
+++ b/src/core/lib/transport/transport_op_string.cc
@@ -120,13 +120,6 @@ char* grpc_transport_stream_op_batch_string(
gpr_strvec_add(&b, tmp);
}
- if (op->collect_stats) {
- gpr_strvec_add(&b, gpr_strdup(" "));
- gpr_asprintf(&tmp, "COLLECT_STATS:%p",
- op->payload->collect_stats.collect_stats);
- gpr_strvec_add(&b, tmp);
- }
-
out = gpr_strvec_flatten(&b, nullptr);
gpr_strvec_destroy(&b);