diff options
Diffstat (limited to 'src/core/lib')
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); |