|
|
@@ -44,7 +44,6 @@
|
|
|
#include "src/core/lib/gprpp/mutex_lock.h"
|
|
|
#include "src/core/lib/gprpp/ref_counted_ptr.h"
|
|
|
#include "src/core/lib/iomgr/sockaddr_utils.h"
|
|
|
-#include "src/core/lib/iomgr/timer.h"
|
|
|
#include "src/core/lib/profiling/timers.h"
|
|
|
#include "src/core/lib/slice/slice_internal.h"
|
|
|
#include "src/core/lib/surface/channel.h"
|
|
|
@@ -55,153 +54,256 @@
|
|
|
#include "src/core/lib/transport/status_metadata.h"
|
|
|
#include "src/core/lib/uri/uri_parser.h"
|
|
|
|
|
|
+// Strong and weak refs.
|
|
|
#define INTERNAL_REF_BITS 16
|
|
|
#define STRONG_REF_MASK (~(gpr_atm)((1 << INTERNAL_REF_BITS) - 1))
|
|
|
|
|
|
+// Backoff parameters.
|
|
|
#define GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS 1
|
|
|
#define GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER 1.6
|
|
|
#define GRPC_SUBCHANNEL_RECONNECT_MIN_TIMEOUT_SECONDS 20
|
|
|
#define GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS 120
|
|
|
#define GRPC_SUBCHANNEL_RECONNECT_JITTER 0.2
|
|
|
|
|
|
-typedef struct external_state_watcher {
|
|
|
- grpc_subchannel* subchannel;
|
|
|
- grpc_pollset_set* pollset_set;
|
|
|
- grpc_closure* notify;
|
|
|
- grpc_closure closure;
|
|
|
- struct external_state_watcher* next;
|
|
|
- struct external_state_watcher* prev;
|
|
|
-} external_state_watcher;
|
|
|
+// Conversion between subchannel call and call stack.
|
|
|
+#define SUBCHANNEL_CALL_TO_CALL_STACK(call) \
|
|
|
+ (grpc_call_stack*)((char*)(call) + \
|
|
|
+ GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(SubchannelCall)))
|
|
|
+#define CALL_STACK_TO_SUBCHANNEL_CALL(callstack) \
|
|
|
+ (SubchannelCall*)(((char*)(call_stack)) - \
|
|
|
+ GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(SubchannelCall)))
|
|
|
|
|
|
namespace grpc_core {
|
|
|
|
|
|
-class ConnectedSubchannelStateWatcher;
|
|
|
-
|
|
|
-} // namespace grpc_core
|
|
|
+//
|
|
|
+// ConnectedSubchannel
|
|
|
+//
|
|
|
|
|
|
-struct grpc_subchannel {
|
|
|
- /** The subchannel pool this subchannel is in */
|
|
|
- grpc_core::RefCountedPtr<grpc_core::SubchannelPoolInterface> subchannel_pool;
|
|
|
+ConnectedSubchannel::ConnectedSubchannel(
|
|
|
+ grpc_channel_stack* channel_stack, const grpc_channel_args* args,
|
|
|
+ RefCountedPtr<channelz::SubchannelNode> channelz_subchannel,
|
|
|
+ intptr_t socket_uuid)
|
|
|
+ : RefCounted<ConnectedSubchannel>(&grpc_trace_stream_refcount),
|
|
|
+ channel_stack_(channel_stack),
|
|
|
+ args_(grpc_channel_args_copy(args)),
|
|
|
+ channelz_subchannel_(std::move(channelz_subchannel)),
|
|
|
+ socket_uuid_(socket_uuid) {}
|
|
|
|
|
|
- grpc_connector* connector;
|
|
|
+ConnectedSubchannel::~ConnectedSubchannel() {
|
|
|
+ grpc_channel_args_destroy(args_);
|
|
|
+ GRPC_CHANNEL_STACK_UNREF(channel_stack_, "connected_subchannel_dtor");
|
|
|
+}
|
|
|
|
|
|
- /** refcount
|
|
|
- - lower INTERNAL_REF_BITS bits are for internal references:
|
|
|
- these do not keep the subchannel open.
|
|
|
- - upper remaining bits are for public references: these do
|
|
|
- keep the subchannel open */
|
|
|
- gpr_atm ref_pair;
|
|
|
+void ConnectedSubchannel::NotifyOnStateChange(
|
|
|
+ grpc_pollset_set* interested_parties, grpc_connectivity_state* state,
|
|
|
+ grpc_closure* closure) {
|
|
|
+ grpc_transport_op* op = grpc_make_transport_op(nullptr);
|
|
|
+ grpc_channel_element* elem;
|
|
|
+ op->connectivity_state = state;
|
|
|
+ op->on_connectivity_state_change = closure;
|
|
|
+ op->bind_pollset_set = interested_parties;
|
|
|
+ elem = grpc_channel_stack_element(channel_stack_, 0);
|
|
|
+ elem->filter->start_transport_op(elem, op);
|
|
|
+}
|
|
|
|
|
|
- /** channel arguments */
|
|
|
- grpc_channel_args* args;
|
|
|
+void ConnectedSubchannel::Ping(grpc_closure* on_initiate,
|
|
|
+ grpc_closure* on_ack) {
|
|
|
+ grpc_transport_op* op = grpc_make_transport_op(nullptr);
|
|
|
+ grpc_channel_element* elem;
|
|
|
+ op->send_ping.on_initiate = on_initiate;
|
|
|
+ op->send_ping.on_ack = on_ack;
|
|
|
+ elem = grpc_channel_stack_element(channel_stack_, 0);
|
|
|
+ elem->filter->start_transport_op(elem, op);
|
|
|
+}
|
|
|
|
|
|
- grpc_core::SubchannelKey* key;
|
|
|
+namespace {
|
|
|
|
|
|
- /** set during connection */
|
|
|
- grpc_connect_out_args connecting_result;
|
|
|
+void SubchannelCallDestroy(void* arg, grpc_error* error) {
|
|
|
+ GPR_TIMER_SCOPE("subchannel_call_destroy", 0);
|
|
|
+ SubchannelCall* call = static_cast<SubchannelCall*>(arg);
|
|
|
+ grpc_closure* after_call_stack_destroy = call->after_call_stack_destroy();
|
|
|
+ call->~SubchannelCall();
|
|
|
+ // This should be the last step to destroy the subchannel call, because
|
|
|
+ // call->after_call_stack_destroy(), if not null, will free the call arena.
|
|
|
+ grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(call), nullptr,
|
|
|
+ after_call_stack_destroy);
|
|
|
+}
|
|
|
|
|
|
- /** callback for connection finishing */
|
|
|
- grpc_closure on_connected;
|
|
|
+} // namespace
|
|
|
|
|
|
- /** callback for our alarm */
|
|
|
- grpc_closure on_alarm;
|
|
|
+RefCountedPtr<SubchannelCall> ConnectedSubchannel::CreateCall(
|
|
|
+ const CallArgs& args, grpc_error** error) {
|
|
|
+ const size_t allocation_size =
|
|
|
+ GetInitialCallSizeEstimate(args.parent_data_size);
|
|
|
+ RefCountedPtr<SubchannelCall> call(
|
|
|
+ new (gpr_arena_alloc(args.arena, allocation_size))
|
|
|
+ SubchannelCall(Ref(DEBUG_LOCATION, "subchannel_call"), args));
|
|
|
+ grpc_call_stack* callstk = SUBCHANNEL_CALL_TO_CALL_STACK(call.get());
|
|
|
+ const grpc_call_element_args call_args = {
|
|
|
+ callstk, /* call_stack */
|
|
|
+ nullptr, /* server_transport_data */
|
|
|
+ args.context, /* context */
|
|
|
+ args.path, /* path */
|
|
|
+ args.start_time, /* start_time */
|
|
|
+ args.deadline, /* deadline */
|
|
|
+ args.arena, /* arena */
|
|
|
+ args.call_combiner /* call_combiner */
|
|
|
+ };
|
|
|
+ *error = grpc_call_stack_init(channel_stack_, 1, SubchannelCallDestroy,
|
|
|
+ call.get(), &call_args);
|
|
|
+ if (GPR_UNLIKELY(*error != GRPC_ERROR_NONE)) {
|
|
|
+ const char* error_string = grpc_error_string(*error);
|
|
|
+ gpr_log(GPR_ERROR, "error: %s", error_string);
|
|
|
+ return call;
|
|
|
+ }
|
|
|
+ grpc_call_stack_set_pollset_or_pollset_set(callstk, args.pollent);
|
|
|
+ if (channelz_subchannel_ != nullptr) {
|
|
|
+ channelz_subchannel_->RecordCallStarted();
|
|
|
+ }
|
|
|
+ return call;
|
|
|
+}
|
|
|
|
|
|
- /** pollset_set tracking who's interested in a connection
|
|
|
- being setup */
|
|
|
- grpc_pollset_set* pollset_set;
|
|
|
+size_t ConnectedSubchannel::GetInitialCallSizeEstimate(
|
|
|
+ size_t parent_data_size) const {
|
|
|
+ size_t allocation_size =
|
|
|
+ GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(SubchannelCall));
|
|
|
+ if (parent_data_size > 0) {
|
|
|
+ allocation_size +=
|
|
|
+ GPR_ROUND_UP_TO_ALIGNMENT_SIZE(channel_stack_->call_stack_size) +
|
|
|
+ parent_data_size;
|
|
|
+ } else {
|
|
|
+ allocation_size += channel_stack_->call_stack_size;
|
|
|
+ }
|
|
|
+ return allocation_size;
|
|
|
+}
|
|
|
|
|
|
- grpc_core::UniquePtr<char> health_check_service_name;
|
|
|
+//
|
|
|
+// SubchannelCall
|
|
|
+//
|
|
|
|
|
|
- /** mutex protecting remaining elements */
|
|
|
- gpr_mu mu;
|
|
|
+void SubchannelCall::StartTransportStreamOpBatch(
|
|
|
+ grpc_transport_stream_op_batch* batch) {
|
|
|
+ GPR_TIMER_SCOPE("subchannel_call_process_op", 0);
|
|
|
+ MaybeInterceptRecvTrailingMetadata(batch);
|
|
|
+ grpc_call_stack* call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(this);
|
|
|
+ grpc_call_element* top_elem = grpc_call_stack_element(call_stack, 0);
|
|
|
+ GRPC_CALL_LOG_OP(GPR_INFO, top_elem, batch);
|
|
|
+ top_elem->filter->start_transport_stream_op_batch(top_elem, batch);
|
|
|
+}
|
|
|
|
|
|
- /** active connection, or null */
|
|
|
- grpc_core::RefCountedPtr<grpc_core::ConnectedSubchannel> connected_subchannel;
|
|
|
- grpc_core::OrphanablePtr<grpc_core::ConnectedSubchannelStateWatcher>
|
|
|
- connected_subchannel_watcher;
|
|
|
+void* SubchannelCall::GetParentData() {
|
|
|
+ grpc_channel_stack* chanstk = connected_subchannel_->channel_stack();
|
|
|
+ return (char*)this + GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(SubchannelCall)) +
|
|
|
+ GPR_ROUND_UP_TO_ALIGNMENT_SIZE(chanstk->call_stack_size);
|
|
|
+}
|
|
|
|
|
|
- /** have we seen a disconnection? */
|
|
|
- bool disconnected;
|
|
|
- /** are we connecting */
|
|
|
- bool connecting;
|
|
|
+grpc_call_stack* SubchannelCall::GetCallStack() {
|
|
|
+ return SUBCHANNEL_CALL_TO_CALL_STACK(this);
|
|
|
+}
|
|
|
|
|
|
- /** connectivity state tracking */
|
|
|
- grpc_connectivity_state_tracker state_tracker;
|
|
|
- grpc_connectivity_state_tracker state_and_health_tracker;
|
|
|
+void SubchannelCall::SetAfterCallStackDestroy(grpc_closure* closure) {
|
|
|
+ GPR_ASSERT(after_call_stack_destroy_ == nullptr);
|
|
|
+ GPR_ASSERT(closure != nullptr);
|
|
|
+ after_call_stack_destroy_ = closure;
|
|
|
+}
|
|
|
|
|
|
- external_state_watcher root_external_state_watcher;
|
|
|
+RefCountedPtr<SubchannelCall> SubchannelCall::Ref() {
|
|
|
+ IncrementRefCount();
|
|
|
+ return RefCountedPtr<SubchannelCall>(this);
|
|
|
+}
|
|
|
|
|
|
- /** backoff state */
|
|
|
- grpc_core::ManualConstructor<grpc_core::BackOff> backoff;
|
|
|
- grpc_millis next_attempt_deadline;
|
|
|
- grpc_millis min_connect_timeout_ms;
|
|
|
+RefCountedPtr<SubchannelCall> SubchannelCall::Ref(
|
|
|
+ const grpc_core::DebugLocation& location, const char* reason) {
|
|
|
+ IncrementRefCount(location, reason);
|
|
|
+ return RefCountedPtr<SubchannelCall>(this);
|
|
|
+}
|
|
|
|
|
|
- /** do we have an active alarm? */
|
|
|
- bool have_alarm;
|
|
|
- /** have we started the backoff loop */
|
|
|
- bool backoff_begun;
|
|
|
- // reset_backoff() was called while alarm was pending
|
|
|
- bool retry_immediately;
|
|
|
- /** our alarm */
|
|
|
- grpc_timer alarm;
|
|
|
+void SubchannelCall::Unref() {
|
|
|
+ GRPC_CALL_STACK_UNREF(SUBCHANNEL_CALL_TO_CALL_STACK(this), "");
|
|
|
+}
|
|
|
|
|
|
- grpc_core::RefCountedPtr<grpc_core::channelz::SubchannelNode>
|
|
|
- channelz_subchannel;
|
|
|
-};
|
|
|
+void SubchannelCall::Unref(const DebugLocation& location, const char* reason) {
|
|
|
+ GRPC_CALL_STACK_UNREF(SUBCHANNEL_CALL_TO_CALL_STACK(this), reason);
|
|
|
+}
|
|
|
|
|
|
-struct grpc_subchannel_call {
|
|
|
- grpc_subchannel_call(grpc_core::ConnectedSubchannel* connection,
|
|
|
- const grpc_core::ConnectedSubchannel::CallArgs& args)
|
|
|
- : connection(connection), deadline(args.deadline) {}
|
|
|
-
|
|
|
- grpc_core::ConnectedSubchannel* connection;
|
|
|
- grpc_closure* schedule_closure_after_destroy = nullptr;
|
|
|
- // state needed to support channelz interception of recv trailing metadata.
|
|
|
- grpc_closure recv_trailing_metadata_ready;
|
|
|
- grpc_closure* original_recv_trailing_metadata;
|
|
|
- grpc_metadata_batch* recv_trailing_metadata = nullptr;
|
|
|
- grpc_millis deadline;
|
|
|
-};
|
|
|
+void SubchannelCall::MaybeInterceptRecvTrailingMetadata(
|
|
|
+ grpc_transport_stream_op_batch* batch) {
|
|
|
+ // only intercept payloads with recv trailing.
|
|
|
+ if (!batch->recv_trailing_metadata) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ // only add interceptor is channelz is enabled.
|
|
|
+ if (connected_subchannel_->channelz_subchannel() == nullptr) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ GRPC_CLOSURE_INIT(&recv_trailing_metadata_ready_, RecvTrailingMetadataReady,
|
|
|
+ this, grpc_schedule_on_exec_ctx);
|
|
|
+ // save some state needed for the interception callback.
|
|
|
+ GPR_ASSERT(recv_trailing_metadata_ == nullptr);
|
|
|
+ recv_trailing_metadata_ =
|
|
|
+ batch->payload->recv_trailing_metadata.recv_trailing_metadata;
|
|
|
+ original_recv_trailing_metadata_ =
|
|
|
+ batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
|
|
|
+ batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
|
|
|
+ &recv_trailing_metadata_ready_;
|
|
|
+}
|
|
|
|
|
|
-static void maybe_start_connecting_locked(grpc_subchannel* c);
|
|
|
+namespace {
|
|
|
|
|
|
-static const char* subchannel_connectivity_state_change_string(
|
|
|
- grpc_connectivity_state state) {
|
|
|
- switch (state) {
|
|
|
- case GRPC_CHANNEL_IDLE:
|
|
|
- return "Subchannel state change to IDLE";
|
|
|
- case GRPC_CHANNEL_CONNECTING:
|
|
|
- return "Subchannel state change to CONNECTING";
|
|
|
- case GRPC_CHANNEL_READY:
|
|
|
- return "Subchannel state change to READY";
|
|
|
- case GRPC_CHANNEL_TRANSIENT_FAILURE:
|
|
|
- return "Subchannel state change to TRANSIENT_FAILURE";
|
|
|
- case GRPC_CHANNEL_SHUTDOWN:
|
|
|
- return "Subchannel state change to SHUTDOWN";
|
|
|
+// Sets *status based on the rest of the parameters.
|
|
|
+void GetCallStatus(grpc_status_code* status, grpc_millis deadline,
|
|
|
+ grpc_metadata_batch* md_batch, grpc_error* error) {
|
|
|
+ if (error != GRPC_ERROR_NONE) {
|
|
|
+ grpc_error_get_status(error, deadline, status, nullptr, nullptr, nullptr);
|
|
|
+ } else {
|
|
|
+ if (md_batch->idx.named.grpc_status != nullptr) {
|
|
|
+ *status = grpc_get_status_code_from_metadata(
|
|
|
+ md_batch->idx.named.grpc_status->md);
|
|
|
+ } else {
|
|
|
+ *status = GRPC_STATUS_UNKNOWN;
|
|
|
+ }
|
|
|
}
|
|
|
- GPR_UNREACHABLE_CODE(return "UNKNOWN");
|
|
|
+ GRPC_ERROR_UNREF(error);
|
|
|
}
|
|
|
|
|
|
-static void set_subchannel_connectivity_state_locked(
|
|
|
- grpc_subchannel* c, grpc_connectivity_state state, grpc_error* error,
|
|
|
- const char* reason) {
|
|
|
- if (c->channelz_subchannel != nullptr) {
|
|
|
- c->channelz_subchannel->AddTraceEvent(
|
|
|
- grpc_core::channelz::ChannelTrace::Severity::Info,
|
|
|
- grpc_slice_from_static_string(
|
|
|
- subchannel_connectivity_state_change_string(state)));
|
|
|
+} // namespace
|
|
|
+
|
|
|
+void SubchannelCall::RecvTrailingMetadataReady(void* arg, grpc_error* error) {
|
|
|
+ SubchannelCall* call = static_cast<SubchannelCall*>(arg);
|
|
|
+ GPR_ASSERT(call->recv_trailing_metadata_ != nullptr);
|
|
|
+ grpc_status_code status = GRPC_STATUS_OK;
|
|
|
+ GetCallStatus(&status, call->deadline_, call->recv_trailing_metadata_,
|
|
|
+ GRPC_ERROR_REF(error));
|
|
|
+ channelz::SubchannelNode* channelz_subchannel =
|
|
|
+ call->connected_subchannel_->channelz_subchannel();
|
|
|
+ GPR_ASSERT(channelz_subchannel != nullptr);
|
|
|
+ if (status == GRPC_STATUS_OK) {
|
|
|
+ channelz_subchannel->RecordCallSucceeded();
|
|
|
+ } else {
|
|
|
+ channelz_subchannel->RecordCallFailed();
|
|
|
}
|
|
|
- grpc_connectivity_state_set(&c->state_tracker, state, error, reason);
|
|
|
+ GRPC_CLOSURE_RUN(call->original_recv_trailing_metadata_,
|
|
|
+ GRPC_ERROR_REF(error));
|
|
|
}
|
|
|
|
|
|
-namespace grpc_core {
|
|
|
+void SubchannelCall::IncrementRefCount() {
|
|
|
+ GRPC_CALL_STACK_REF(SUBCHANNEL_CALL_TO_CALL_STACK(this), "");
|
|
|
+}
|
|
|
|
|
|
-class ConnectedSubchannelStateWatcher
|
|
|
+void SubchannelCall::IncrementRefCount(const grpc_core::DebugLocation& location,
|
|
|
+ const char* reason) {
|
|
|
+ GRPC_CALL_STACK_REF(SUBCHANNEL_CALL_TO_CALL_STACK(this), reason);
|
|
|
+}
|
|
|
+
|
|
|
+//
|
|
|
+// Subchannel::ConnectedSubchannelStateWatcher
|
|
|
+//
|
|
|
+
|
|
|
+class Subchannel::ConnectedSubchannelStateWatcher
|
|
|
: public InternallyRefCounted<ConnectedSubchannelStateWatcher> {
|
|
|
public:
|
|
|
// Must be instantiated while holding c->mu.
|
|
|
- explicit ConnectedSubchannelStateWatcher(grpc_subchannel* c)
|
|
|
- : subchannel_(c) {
|
|
|
+ explicit ConnectedSubchannelStateWatcher(Subchannel* c) : subchannel_(c) {
|
|
|
// Steal subchannel ref for connecting.
|
|
|
GRPC_SUBCHANNEL_WEAK_REF(subchannel_, "state_watcher");
|
|
|
GRPC_SUBCHANNEL_WEAK_UNREF(subchannel_, "connecting");
|
|
|
@@ -209,15 +311,15 @@ class ConnectedSubchannelStateWatcher
|
|
|
// Callback uses initial ref to this.
|
|
|
GRPC_CLOSURE_INIT(&on_connectivity_changed_, OnConnectivityChanged, this,
|
|
|
grpc_schedule_on_exec_ctx);
|
|
|
- c->connected_subchannel->NotifyOnStateChange(c->pollset_set,
|
|
|
- &pending_connectivity_state_,
|
|
|
- &on_connectivity_changed_);
|
|
|
+ c->connected_subchannel_->NotifyOnStateChange(c->pollset_set_,
|
|
|
+ &pending_connectivity_state_,
|
|
|
+ &on_connectivity_changed_);
|
|
|
// Start health check if needed.
|
|
|
grpc_connectivity_state health_state = GRPC_CHANNEL_READY;
|
|
|
- if (c->health_check_service_name != nullptr) {
|
|
|
- health_check_client_ = grpc_core::MakeOrphanable<HealthCheckClient>(
|
|
|
- c->health_check_service_name.get(), c->connected_subchannel,
|
|
|
- c->pollset_set, c->channelz_subchannel);
|
|
|
+ if (c->health_check_service_name_ != nullptr) {
|
|
|
+ health_check_client_ = MakeOrphanable<HealthCheckClient>(
|
|
|
+ c->health_check_service_name_.get(), c->connected_subchannel_,
|
|
|
+ c->pollset_set_, c->channelz_node_);
|
|
|
GRPC_CLOSURE_INIT(&on_health_changed_, OnHealthChanged, this,
|
|
|
grpc_schedule_on_exec_ctx);
|
|
|
Ref().release(); // Ref for health callback tracked manually.
|
|
|
@@ -226,9 +328,9 @@ class ConnectedSubchannelStateWatcher
|
|
|
health_state = GRPC_CHANNEL_CONNECTING;
|
|
|
}
|
|
|
// Report initial state.
|
|
|
- set_subchannel_connectivity_state_locked(
|
|
|
- c, GRPC_CHANNEL_READY, GRPC_ERROR_NONE, "subchannel_connected");
|
|
|
- grpc_connectivity_state_set(&c->state_and_health_tracker, health_state,
|
|
|
+ c->SetConnectivityStateLocked(GRPC_CHANNEL_READY, GRPC_ERROR_NONE,
|
|
|
+ "subchannel_connected");
|
|
|
+ grpc_connectivity_state_set(&c->state_and_health_tracker_, health_state,
|
|
|
GRPC_ERROR_NONE, "subchannel_connected");
|
|
|
}
|
|
|
|
|
|
@@ -236,38 +338,39 @@ class ConnectedSubchannelStateWatcher
|
|
|
GRPC_SUBCHANNEL_WEAK_UNREF(subchannel_, "state_watcher");
|
|
|
}
|
|
|
|
|
|
+ // Must be called while holding subchannel_->mu.
|
|
|
void Orphan() override { health_check_client_.reset(); }
|
|
|
|
|
|
private:
|
|
|
static void OnConnectivityChanged(void* arg, grpc_error* error) {
|
|
|
auto* self = static_cast<ConnectedSubchannelStateWatcher*>(arg);
|
|
|
- grpc_subchannel* c = self->subchannel_;
|
|
|
+ Subchannel* c = self->subchannel_;
|
|
|
{
|
|
|
- MutexLock lock(&c->mu);
|
|
|
+ MutexLock lock(&c->mu_);
|
|
|
switch (self->pending_connectivity_state_) {
|
|
|
case GRPC_CHANNEL_TRANSIENT_FAILURE:
|
|
|
case GRPC_CHANNEL_SHUTDOWN: {
|
|
|
- if (!c->disconnected && c->connected_subchannel != nullptr) {
|
|
|
+ if (!c->disconnected_ && c->connected_subchannel_ != nullptr) {
|
|
|
if (grpc_trace_stream_refcount.enabled()) {
|
|
|
gpr_log(GPR_INFO,
|
|
|
"Connected subchannel %p of subchannel %p has gone into "
|
|
|
"%s. Attempting to reconnect.",
|
|
|
- c->connected_subchannel.get(), c,
|
|
|
+ c->connected_subchannel_.get(), c,
|
|
|
grpc_connectivity_state_name(
|
|
|
self->pending_connectivity_state_));
|
|
|
}
|
|
|
- c->connected_subchannel.reset();
|
|
|
- c->connected_subchannel_watcher.reset();
|
|
|
+ c->connected_subchannel_.reset();
|
|
|
+ c->connected_subchannel_watcher_.reset();
|
|
|
self->last_connectivity_state_ = GRPC_CHANNEL_TRANSIENT_FAILURE;
|
|
|
- set_subchannel_connectivity_state_locked(
|
|
|
- c, GRPC_CHANNEL_TRANSIENT_FAILURE, GRPC_ERROR_REF(error),
|
|
|
- "reflect_child");
|
|
|
- grpc_connectivity_state_set(&c->state_and_health_tracker,
|
|
|
+ c->SetConnectivityStateLocked(GRPC_CHANNEL_TRANSIENT_FAILURE,
|
|
|
+ GRPC_ERROR_REF(error),
|
|
|
+ "reflect_child");
|
|
|
+ grpc_connectivity_state_set(&c->state_and_health_tracker_,
|
|
|
GRPC_CHANNEL_TRANSIENT_FAILURE,
|
|
|
GRPC_ERROR_REF(error), "reflect_child");
|
|
|
- c->backoff_begun = false;
|
|
|
- c->backoff->Reset();
|
|
|
- maybe_start_connecting_locked(c);
|
|
|
+ c->backoff_begun_ = false;
|
|
|
+ c->backoff_.Reset();
|
|
|
+ c->MaybeStartConnectingLocked();
|
|
|
} else {
|
|
|
self->last_connectivity_state_ = GRPC_CHANNEL_SHUTDOWN;
|
|
|
}
|
|
|
@@ -280,15 +383,14 @@ class ConnectedSubchannelStateWatcher
|
|
|
// this watch from. And a connected subchannel should never go
|
|
|
// from READY to CONNECTING or IDLE.
|
|
|
self->last_connectivity_state_ = self->pending_connectivity_state_;
|
|
|
- set_subchannel_connectivity_state_locked(
|
|
|
- c, self->pending_connectivity_state_, GRPC_ERROR_REF(error),
|
|
|
- "reflect_child");
|
|
|
+ c->SetConnectivityStateLocked(self->pending_connectivity_state_,
|
|
|
+ GRPC_ERROR_REF(error), "reflect_child");
|
|
|
if (self->pending_connectivity_state_ != GRPC_CHANNEL_READY) {
|
|
|
- grpc_connectivity_state_set(&c->state_and_health_tracker,
|
|
|
+ grpc_connectivity_state_set(&c->state_and_health_tracker_,
|
|
|
self->pending_connectivity_state_,
|
|
|
GRPC_ERROR_REF(error), "reflect_child");
|
|
|
}
|
|
|
- c->connected_subchannel->NotifyOnStateChange(
|
|
|
+ c->connected_subchannel_->NotifyOnStateChange(
|
|
|
nullptr, &self->pending_connectivity_state_,
|
|
|
&self->on_connectivity_changed_);
|
|
|
self = nullptr; // So we don't unref below.
|
|
|
@@ -302,178 +404,82 @@ class ConnectedSubchannelStateWatcher
|
|
|
|
|
|
static void OnHealthChanged(void* arg, grpc_error* error) {
|
|
|
auto* self = static_cast<ConnectedSubchannelStateWatcher*>(arg);
|
|
|
- if (self->health_state_ == GRPC_CHANNEL_SHUTDOWN) {
|
|
|
- self->Unref();
|
|
|
- return;
|
|
|
- }
|
|
|
- grpc_subchannel* c = self->subchannel_;
|
|
|
- MutexLock lock(&c->mu);
|
|
|
- if (self->last_connectivity_state_ == GRPC_CHANNEL_READY) {
|
|
|
- grpc_connectivity_state_set(&c->state_and_health_tracker,
|
|
|
- self->health_state_, GRPC_ERROR_REF(error),
|
|
|
- "health_changed");
|
|
|
+ Subchannel* c = self->subchannel_;
|
|
|
+ {
|
|
|
+ MutexLock lock(&c->mu_);
|
|
|
+ if (self->health_state_ != GRPC_CHANNEL_SHUTDOWN) {
|
|
|
+ if (self->last_connectivity_state_ == GRPC_CHANNEL_READY) {
|
|
|
+ grpc_connectivity_state_set(&c->state_and_health_tracker_,
|
|
|
+ self->health_state_,
|
|
|
+ GRPC_ERROR_REF(error), "health_changed");
|
|
|
+ }
|
|
|
+ self->health_check_client_->NotifyOnHealthChange(
|
|
|
+ &self->health_state_, &self->on_health_changed_);
|
|
|
+ self = nullptr; // So we don't unref below.
|
|
|
+ }
|
|
|
}
|
|
|
- self->health_check_client_->NotifyOnHealthChange(&self->health_state_,
|
|
|
- &self->on_health_changed_);
|
|
|
+ // Don't unref until we've released the lock, because this might
|
|
|
+ // cause the subchannel (which contains the lock) to be destroyed.
|
|
|
+ if (self != nullptr) self->Unref();
|
|
|
}
|
|
|
|
|
|
- grpc_subchannel* subchannel_;
|
|
|
+ Subchannel* subchannel_;
|
|
|
grpc_closure on_connectivity_changed_;
|
|
|
grpc_connectivity_state pending_connectivity_state_ = GRPC_CHANNEL_READY;
|
|
|
grpc_connectivity_state last_connectivity_state_ = GRPC_CHANNEL_READY;
|
|
|
- grpc_core::OrphanablePtr<grpc_core::HealthCheckClient> health_check_client_;
|
|
|
+ OrphanablePtr<HealthCheckClient> health_check_client_;
|
|
|
grpc_closure on_health_changed_;
|
|
|
grpc_connectivity_state health_state_ = GRPC_CHANNEL_CONNECTING;
|
|
|
};
|
|
|
|
|
|
-} // namespace grpc_core
|
|
|
-
|
|
|
-#define SUBCHANNEL_CALL_TO_CALL_STACK(call) \
|
|
|
- (grpc_call_stack*)((char*)(call) + GPR_ROUND_UP_TO_ALIGNMENT_SIZE( \
|
|
|
- sizeof(grpc_subchannel_call)))
|
|
|
-#define CALLSTACK_TO_SUBCHANNEL_CALL(callstack) \
|
|
|
- (grpc_subchannel_call*)(((char*)(call_stack)) - \
|
|
|
- GPR_ROUND_UP_TO_ALIGNMENT_SIZE( \
|
|
|
- sizeof(grpc_subchannel_call)))
|
|
|
-
|
|
|
-static void on_subchannel_connected(void* subchannel, grpc_error* error);
|
|
|
-
|
|
|
-#ifndef NDEBUG
|
|
|
-#define REF_REASON reason
|
|
|
-#define REF_MUTATE_EXTRA_ARGS \
|
|
|
- GRPC_SUBCHANNEL_REF_EXTRA_ARGS, const char* purpose
|
|
|
-#define REF_MUTATE_PURPOSE(x) , file, line, reason, x
|
|
|
-#else
|
|
|
-#define REF_REASON ""
|
|
|
-#define REF_MUTATE_EXTRA_ARGS
|
|
|
-#define REF_MUTATE_PURPOSE(x)
|
|
|
-#endif
|
|
|
-
|
|
|
-/*
|
|
|
- * connection implementation
|
|
|
- */
|
|
|
-
|
|
|
-static void connection_destroy(void* arg, grpc_error* error) {
|
|
|
- grpc_channel_stack* stk = static_cast<grpc_channel_stack*>(arg);
|
|
|
- grpc_channel_stack_destroy(stk);
|
|
|
- gpr_free(stk);
|
|
|
-}
|
|
|
-
|
|
|
-/*
|
|
|
- * grpc_subchannel implementation
|
|
|
- */
|
|
|
-
|
|
|
-static void subchannel_destroy(void* arg, grpc_error* error) {
|
|
|
- grpc_subchannel* c = static_cast<grpc_subchannel*>(arg);
|
|
|
- if (c->channelz_subchannel != nullptr) {
|
|
|
- c->channelz_subchannel->AddTraceEvent(
|
|
|
- grpc_core::channelz::ChannelTrace::Severity::Info,
|
|
|
- grpc_slice_from_static_string("Subchannel destroyed"));
|
|
|
- c->channelz_subchannel->MarkSubchannelDestroyed();
|
|
|
- c->channelz_subchannel.reset();
|
|
|
- }
|
|
|
- c->health_check_service_name.reset();
|
|
|
- grpc_channel_args_destroy(c->args);
|
|
|
- grpc_connectivity_state_destroy(&c->state_tracker);
|
|
|
- grpc_connectivity_state_destroy(&c->state_and_health_tracker);
|
|
|
- grpc_connector_unref(c->connector);
|
|
|
- grpc_pollset_set_destroy(c->pollset_set);
|
|
|
- grpc_core::Delete(c->key);
|
|
|
- gpr_mu_destroy(&c->mu);
|
|
|
- gpr_free(c);
|
|
|
-}
|
|
|
+//
|
|
|
+// Subchannel::ExternalStateWatcher
|
|
|
+//
|
|
|
|
|
|
-static gpr_atm ref_mutate(grpc_subchannel* c, gpr_atm delta,
|
|
|
- int barrier REF_MUTATE_EXTRA_ARGS) {
|
|
|
- gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, delta)
|
|
|
- : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta);
|
|
|
-#ifndef NDEBUG
|
|
|
- if (grpc_trace_stream_refcount.enabled()) {
|
|
|
- gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
|
|
|
- "SUBCHANNEL: %p %12s 0x%" PRIxPTR " -> 0x%" PRIxPTR " [%s]", c,
|
|
|
- purpose, old_val, old_val + delta, reason);
|
|
|
+struct Subchannel::ExternalStateWatcher {
|
|
|
+ ExternalStateWatcher(Subchannel* subchannel, grpc_pollset_set* pollset_set,
|
|
|
+ grpc_closure* notify)
|
|
|
+ : subchannel(subchannel), pollset_set(pollset_set), notify(notify) {
|
|
|
+ GRPC_SUBCHANNEL_WEAK_REF(subchannel, "external_state_watcher+init");
|
|
|
+ GRPC_CLOSURE_INIT(&on_state_changed, OnStateChanged, this,
|
|
|
+ grpc_schedule_on_exec_ctx);
|
|
|
}
|
|
|
-#endif
|
|
|
- return old_val;
|
|
|
-}
|
|
|
-
|
|
|
-grpc_subchannel* grpc_subchannel_ref(
|
|
|
- grpc_subchannel* c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
- gpr_atm old_refs;
|
|
|
- old_refs = ref_mutate(c, (1 << INTERNAL_REF_BITS),
|
|
|
- 0 REF_MUTATE_PURPOSE("STRONG_REF"));
|
|
|
- GPR_ASSERT((old_refs & STRONG_REF_MASK) != 0);
|
|
|
- return c;
|
|
|
-}
|
|
|
|
|
|
-grpc_subchannel* grpc_subchannel_weak_ref(
|
|
|
- grpc_subchannel* c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
- gpr_atm old_refs;
|
|
|
- old_refs = ref_mutate(c, 1, 0 REF_MUTATE_PURPOSE("WEAK_REF"));
|
|
|
- GPR_ASSERT(old_refs != 0);
|
|
|
- return c;
|
|
|
-}
|
|
|
-
|
|
|
-grpc_subchannel* grpc_subchannel_ref_from_weak_ref(
|
|
|
- grpc_subchannel* c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
- if (!c) return nullptr;
|
|
|
- for (;;) {
|
|
|
- gpr_atm old_refs = gpr_atm_acq_load(&c->ref_pair);
|
|
|
- if (old_refs >= (1 << INTERNAL_REF_BITS)) {
|
|
|
- gpr_atm new_refs = old_refs + (1 << INTERNAL_REF_BITS);
|
|
|
- if (gpr_atm_rel_cas(&c->ref_pair, old_refs, new_refs)) {
|
|
|
- return c;
|
|
|
- }
|
|
|
- } else {
|
|
|
- return nullptr;
|
|
|
+ static void OnStateChanged(void* arg, grpc_error* error) {
|
|
|
+ ExternalStateWatcher* w = static_cast<ExternalStateWatcher*>(arg);
|
|
|
+ grpc_closure* follow_up = w->notify;
|
|
|
+ if (w->pollset_set != nullptr) {
|
|
|
+ grpc_pollset_set_del_pollset_set(w->subchannel->pollset_set_,
|
|
|
+ w->pollset_set);
|
|
|
}
|
|
|
+ gpr_mu_lock(&w->subchannel->mu_);
|
|
|
+ if (w->subchannel->external_state_watcher_list_ == w) {
|
|
|
+ w->subchannel->external_state_watcher_list_ = w->next;
|
|
|
+ }
|
|
|
+ if (w->next != nullptr) w->next->prev = w->prev;
|
|
|
+ if (w->prev != nullptr) w->prev->next = w->next;
|
|
|
+ gpr_mu_unlock(&w->subchannel->mu_);
|
|
|
+ GRPC_SUBCHANNEL_WEAK_UNREF(w->subchannel, "external_state_watcher+done");
|
|
|
+ Delete(w);
|
|
|
+ GRPC_CLOSURE_SCHED(follow_up, GRPC_ERROR_REF(error));
|
|
|
}
|
|
|
-}
|
|
|
|
|
|
-static void disconnect(grpc_subchannel* c) {
|
|
|
- // The subchannel_pool is only used once here in this subchannel, so the
|
|
|
- // access can be outside of the lock.
|
|
|
- if (c->subchannel_pool != nullptr) {
|
|
|
- c->subchannel_pool->UnregisterSubchannel(c->key);
|
|
|
- c->subchannel_pool.reset();
|
|
|
- }
|
|
|
- gpr_mu_lock(&c->mu);
|
|
|
- GPR_ASSERT(!c->disconnected);
|
|
|
- c->disconnected = true;
|
|
|
- grpc_connector_shutdown(c->connector, GRPC_ERROR_CREATE_FROM_STATIC_STRING(
|
|
|
- "Subchannel disconnected"));
|
|
|
- c->connected_subchannel.reset();
|
|
|
- c->connected_subchannel_watcher.reset();
|
|
|
- gpr_mu_unlock(&c->mu);
|
|
|
-}
|
|
|
+ Subchannel* subchannel;
|
|
|
+ grpc_pollset_set* pollset_set;
|
|
|
+ grpc_closure* notify;
|
|
|
+ grpc_closure on_state_changed;
|
|
|
+ ExternalStateWatcher* next = nullptr;
|
|
|
+ ExternalStateWatcher* prev = nullptr;
|
|
|
+};
|
|
|
|
|
|
-void grpc_subchannel_unref(grpc_subchannel* c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
- gpr_atm old_refs;
|
|
|
- // add a weak ref and subtract a strong ref (atomically)
|
|
|
- old_refs = ref_mutate(
|
|
|
- c, static_cast<gpr_atm>(1) - static_cast<gpr_atm>(1 << INTERNAL_REF_BITS),
|
|
|
- 1 REF_MUTATE_PURPOSE("STRONG_UNREF"));
|
|
|
- if ((old_refs & STRONG_REF_MASK) == (1 << INTERNAL_REF_BITS)) {
|
|
|
- disconnect(c);
|
|
|
- }
|
|
|
- GRPC_SUBCHANNEL_WEAK_UNREF(c, "strong-unref");
|
|
|
-}
|
|
|
+//
|
|
|
+// Subchannel
|
|
|
+//
|
|
|
|
|
|
-void grpc_subchannel_weak_unref(
|
|
|
- grpc_subchannel* c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
- gpr_atm old_refs;
|
|
|
- old_refs = ref_mutate(c, -static_cast<gpr_atm>(1),
|
|
|
- 1 REF_MUTATE_PURPOSE("WEAK_UNREF"));
|
|
|
- if (old_refs == 1) {
|
|
|
- GRPC_CLOSURE_SCHED(
|
|
|
- GRPC_CLOSURE_CREATE(subchannel_destroy, c, grpc_schedule_on_exec_ctx),
|
|
|
- GRPC_ERROR_NONE);
|
|
|
- }
|
|
|
-}
|
|
|
+namespace {
|
|
|
|
|
|
-static void parse_args_for_backoff_values(
|
|
|
- const grpc_channel_args* args, grpc_core::BackOff::Options* backoff_options,
|
|
|
- grpc_millis* min_connect_timeout_ms) {
|
|
|
+BackOff::Options ParseArgsForBackoffValues(
|
|
|
+ const grpc_channel_args* args, grpc_millis* min_connect_timeout_ms) {
|
|
|
grpc_millis initial_backoff_ms =
|
|
|
GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS * 1000;
|
|
|
*min_connect_timeout_ms =
|
|
|
@@ -510,7 +516,8 @@ static void parse_args_for_backoff_values(
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
- backoff_options->set_initial_backoff(initial_backoff_ms)
|
|
|
+ return BackOff::Options()
|
|
|
+ .set_initial_backoff(initial_backoff_ms)
|
|
|
.set_multiplier(fixed_reconnect_backoff
|
|
|
? 1.0
|
|
|
: GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER)
|
|
|
@@ -519,9 +526,6 @@ static void parse_args_for_backoff_values(
|
|
|
.set_max_backoff(max_backoff_ms);
|
|
|
}
|
|
|
|
|
|
-namespace grpc_core {
|
|
|
-namespace {
|
|
|
-
|
|
|
struct HealthCheckParams {
|
|
|
UniquePtr<char> service_name;
|
|
|
|
|
|
@@ -542,31 +546,19 @@ struct HealthCheckParams {
|
|
|
};
|
|
|
|
|
|
} // namespace
|
|
|
-} // namespace grpc_core
|
|
|
|
|
|
-grpc_subchannel* grpc_subchannel_create(grpc_connector* connector,
|
|
|
- const grpc_channel_args* args) {
|
|
|
- grpc_core::SubchannelKey* key =
|
|
|
- grpc_core::New<grpc_core::SubchannelKey>(args);
|
|
|
- grpc_core::SubchannelPoolInterface* subchannel_pool =
|
|
|
- grpc_core::SubchannelPoolInterface::GetSubchannelPoolFromChannelArgs(
|
|
|
- args);
|
|
|
- GPR_ASSERT(subchannel_pool != nullptr);
|
|
|
- grpc_subchannel* c = subchannel_pool->FindSubchannel(key);
|
|
|
- if (c != nullptr) {
|
|
|
- grpc_core::Delete(key);
|
|
|
- return c;
|
|
|
- }
|
|
|
+Subchannel::Subchannel(SubchannelKey* key, grpc_connector* connector,
|
|
|
+ const grpc_channel_args* args)
|
|
|
+ : key_(key),
|
|
|
+ connector_(connector),
|
|
|
+ backoff_(ParseArgsForBackoffValues(args, &min_connect_timeout_ms_)) {
|
|
|
GRPC_STATS_INC_CLIENT_SUBCHANNELS_CREATED();
|
|
|
- c = static_cast<grpc_subchannel*>(gpr_zalloc(sizeof(*c)));
|
|
|
- c->key = key;
|
|
|
- gpr_atm_no_barrier_store(&c->ref_pair, 1 << INTERNAL_REF_BITS);
|
|
|
- c->connector = connector;
|
|
|
- grpc_connector_ref(c->connector);
|
|
|
- c->pollset_set = grpc_pollset_set_create();
|
|
|
+ gpr_atm_no_barrier_store(&ref_pair_, 1 << INTERNAL_REF_BITS);
|
|
|
+ grpc_connector_ref(connector_);
|
|
|
+ pollset_set_ = grpc_pollset_set_create();
|
|
|
grpc_resolved_address* addr =
|
|
|
static_cast<grpc_resolved_address*>(gpr_malloc(sizeof(*addr)));
|
|
|
- grpc_get_subchannel_address_arg(args, addr);
|
|
|
+ GetAddressFromSubchannelAddressArg(args, addr);
|
|
|
grpc_resolved_address* new_address = nullptr;
|
|
|
grpc_channel_args* new_args = nullptr;
|
|
|
if (grpc_proxy_mappers_map_address(addr, args, &new_address, &new_args)) {
|
|
|
@@ -575,568 +567,492 @@ grpc_subchannel* grpc_subchannel_create(grpc_connector* connector,
|
|
|
addr = new_address;
|
|
|
}
|
|
|
static const char* keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS};
|
|
|
- grpc_arg new_arg = grpc_create_subchannel_address_arg(addr);
|
|
|
+ grpc_arg new_arg = CreateSubchannelAddressArg(addr);
|
|
|
gpr_free(addr);
|
|
|
- c->args = grpc_channel_args_copy_and_add_and_remove(
|
|
|
+ args_ = grpc_channel_args_copy_and_add_and_remove(
|
|
|
new_args != nullptr ? new_args : args, keys_to_remove,
|
|
|
GPR_ARRAY_SIZE(keys_to_remove), &new_arg, 1);
|
|
|
gpr_free(new_arg.value.string);
|
|
|
if (new_args != nullptr) grpc_channel_args_destroy(new_args);
|
|
|
- c->root_external_state_watcher.next = c->root_external_state_watcher.prev =
|
|
|
- &c->root_external_state_watcher;
|
|
|
- GRPC_CLOSURE_INIT(&c->on_connected, on_subchannel_connected, c,
|
|
|
+ GRPC_CLOSURE_INIT(&on_connecting_finished_, OnConnectingFinished, this,
|
|
|
grpc_schedule_on_exec_ctx);
|
|
|
- grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE,
|
|
|
+ grpc_connectivity_state_init(&state_tracker_, GRPC_CHANNEL_IDLE,
|
|
|
"subchannel");
|
|
|
- grpc_connectivity_state_init(&c->state_and_health_tracker, GRPC_CHANNEL_IDLE,
|
|
|
+ grpc_connectivity_state_init(&state_and_health_tracker_, GRPC_CHANNEL_IDLE,
|
|
|
"subchannel");
|
|
|
- grpc_core::BackOff::Options backoff_options;
|
|
|
- parse_args_for_backoff_values(args, &backoff_options,
|
|
|
- &c->min_connect_timeout_ms);
|
|
|
- c->backoff.Init(backoff_options);
|
|
|
- gpr_mu_init(&c->mu);
|
|
|
-
|
|
|
+ gpr_mu_init(&mu_);
|
|
|
// Check whether we should enable health checking.
|
|
|
const char* service_config_json = grpc_channel_arg_get_string(
|
|
|
- grpc_channel_args_find(c->args, GRPC_ARG_SERVICE_CONFIG));
|
|
|
+ grpc_channel_args_find(args_, GRPC_ARG_SERVICE_CONFIG));
|
|
|
if (service_config_json != nullptr) {
|
|
|
- grpc_core::UniquePtr<grpc_core::ServiceConfig> service_config =
|
|
|
- grpc_core::ServiceConfig::Create(service_config_json);
|
|
|
+ UniquePtr<ServiceConfig> service_config =
|
|
|
+ ServiceConfig::Create(service_config_json);
|
|
|
if (service_config != nullptr) {
|
|
|
- grpc_core::HealthCheckParams params;
|
|
|
- service_config->ParseGlobalParams(grpc_core::HealthCheckParams::Parse,
|
|
|
- ¶ms);
|
|
|
- c->health_check_service_name = std::move(params.service_name);
|
|
|
+ HealthCheckParams params;
|
|
|
+ service_config->ParseGlobalParams(HealthCheckParams::Parse, ¶ms);
|
|
|
+ health_check_service_name_ = std::move(params.service_name);
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
- const grpc_arg* arg =
|
|
|
- grpc_channel_args_find(c->args, GRPC_ARG_ENABLE_CHANNELZ);
|
|
|
- bool channelz_enabled =
|
|
|
+ const grpc_arg* arg = grpc_channel_args_find(args_, GRPC_ARG_ENABLE_CHANNELZ);
|
|
|
+ const bool channelz_enabled =
|
|
|
grpc_channel_arg_get_bool(arg, GRPC_ENABLE_CHANNELZ_DEFAULT);
|
|
|
arg = grpc_channel_args_find(
|
|
|
- c->args, GRPC_ARG_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE);
|
|
|
+ args_, GRPC_ARG_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE);
|
|
|
const grpc_integer_options options = {
|
|
|
GRPC_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE_DEFAULT, 0, INT_MAX};
|
|
|
size_t channel_tracer_max_memory =
|
|
|
(size_t)grpc_channel_arg_get_integer(arg, options);
|
|
|
if (channelz_enabled) {
|
|
|
- c->channelz_subchannel =
|
|
|
- grpc_core::MakeRefCounted<grpc_core::channelz::SubchannelNode>(
|
|
|
- c, channel_tracer_max_memory);
|
|
|
- c->channelz_subchannel->AddTraceEvent(
|
|
|
- grpc_core::channelz::ChannelTrace::Severity::Info,
|
|
|
- grpc_slice_from_static_string("Subchannel created"));
|
|
|
+ channelz_node_ = MakeRefCounted<channelz::SubchannelNode>(
|
|
|
+ this, channel_tracer_max_memory);
|
|
|
+ channelz_node_->AddTraceEvent(
|
|
|
+ channelz::ChannelTrace::Severity::Info,
|
|
|
+ grpc_slice_from_static_string("subchannel created"));
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+Subchannel::~Subchannel() {
|
|
|
+ if (channelz_node_ != nullptr) {
|
|
|
+ channelz_node_->AddTraceEvent(
|
|
|
+ channelz::ChannelTrace::Severity::Info,
|
|
|
+ grpc_slice_from_static_string("Subchannel destroyed"));
|
|
|
+ channelz_node_->MarkSubchannelDestroyed();
|
|
|
+ }
|
|
|
+ grpc_channel_args_destroy(args_);
|
|
|
+ grpc_connectivity_state_destroy(&state_tracker_);
|
|
|
+ grpc_connectivity_state_destroy(&state_and_health_tracker_);
|
|
|
+ grpc_connector_unref(connector_);
|
|
|
+ grpc_pollset_set_destroy(pollset_set_);
|
|
|
+ Delete(key_);
|
|
|
+ gpr_mu_destroy(&mu_);
|
|
|
+}
|
|
|
+
|
|
|
+Subchannel* Subchannel::Create(grpc_connector* connector,
|
|
|
+ const grpc_channel_args* args) {
|
|
|
+ SubchannelKey* key = New<SubchannelKey>(args);
|
|
|
+ SubchannelPoolInterface* subchannel_pool =
|
|
|
+ SubchannelPoolInterface::GetSubchannelPoolFromChannelArgs(args);
|
|
|
+ GPR_ASSERT(subchannel_pool != nullptr);
|
|
|
+ Subchannel* c = subchannel_pool->FindSubchannel(key);
|
|
|
+ if (c != nullptr) {
|
|
|
+ Delete(key);
|
|
|
+ return c;
|
|
|
}
|
|
|
+ c = New<Subchannel>(key, connector, args);
|
|
|
// Try to register the subchannel before setting the subchannel pool.
|
|
|
// Otherwise, in case of a registration race, unreffing c in
|
|
|
- // RegisterSubchannel() will cause c to be tried to be unregistered, while its
|
|
|
- // key maps to a different subchannel.
|
|
|
- grpc_subchannel* registered = subchannel_pool->RegisterSubchannel(key, c);
|
|
|
- if (registered == c) c->subchannel_pool = subchannel_pool->Ref();
|
|
|
+ // RegisterSubchannel() will cause c to be tried to be unregistered, while
|
|
|
+ // its key maps to a different subchannel.
|
|
|
+ Subchannel* registered = subchannel_pool->RegisterSubchannel(key, c);
|
|
|
+ if (registered == c) c->subchannel_pool_ = subchannel_pool->Ref();
|
|
|
return registered;
|
|
|
}
|
|
|
|
|
|
-grpc_core::channelz::SubchannelNode* grpc_subchannel_get_channelz_node(
|
|
|
- grpc_subchannel* subchannel) {
|
|
|
- return subchannel->channelz_subchannel.get();
|
|
|
+Subchannel* Subchannel::Ref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
+ gpr_atm old_refs;
|
|
|
+ old_refs = RefMutate((1 << INTERNAL_REF_BITS),
|
|
|
+ 0 GRPC_SUBCHANNEL_REF_MUTATE_PURPOSE("STRONG_REF"));
|
|
|
+ GPR_ASSERT((old_refs & STRONG_REF_MASK) != 0);
|
|
|
+ return this;
|
|
|
}
|
|
|
|
|
|
-intptr_t grpc_subchannel_get_child_socket_uuid(grpc_subchannel* subchannel) {
|
|
|
- if (subchannel->connected_subchannel != nullptr) {
|
|
|
- return subchannel->connected_subchannel->socket_uuid();
|
|
|
- } else {
|
|
|
- return 0;
|
|
|
+void Subchannel::Unref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
+ gpr_atm old_refs;
|
|
|
+ // add a weak ref and subtract a strong ref (atomically)
|
|
|
+ old_refs = RefMutate(
|
|
|
+ static_cast<gpr_atm>(1) - static_cast<gpr_atm>(1 << INTERNAL_REF_BITS),
|
|
|
+ 1 GRPC_SUBCHANNEL_REF_MUTATE_PURPOSE("STRONG_UNREF"));
|
|
|
+ if ((old_refs & STRONG_REF_MASK) == (1 << INTERNAL_REF_BITS)) {
|
|
|
+ Disconnect();
|
|
|
}
|
|
|
+ GRPC_SUBCHANNEL_WEAK_UNREF(this, "strong-unref");
|
|
|
}
|
|
|
|
|
|
-static void continue_connect_locked(grpc_subchannel* c) {
|
|
|
- grpc_connect_in_args args;
|
|
|
- args.interested_parties = c->pollset_set;
|
|
|
- const grpc_millis min_deadline =
|
|
|
- c->min_connect_timeout_ms + grpc_core::ExecCtx::Get()->Now();
|
|
|
- c->next_attempt_deadline = c->backoff->NextAttemptTime();
|
|
|
- args.deadline = std::max(c->next_attempt_deadline, min_deadline);
|
|
|
- args.channel_args = c->args;
|
|
|
- set_subchannel_connectivity_state_locked(c, GRPC_CHANNEL_CONNECTING,
|
|
|
- GRPC_ERROR_NONE, "connecting");
|
|
|
- grpc_connectivity_state_set(&c->state_and_health_tracker,
|
|
|
- GRPC_CHANNEL_CONNECTING, GRPC_ERROR_NONE,
|
|
|
- "connecting");
|
|
|
- grpc_connector_connect(c->connector, &args, &c->connecting_result,
|
|
|
- &c->on_connected);
|
|
|
+Subchannel* Subchannel::WeakRef(GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
+ gpr_atm old_refs;
|
|
|
+ old_refs = RefMutate(1, 0 GRPC_SUBCHANNEL_REF_MUTATE_PURPOSE("WEAK_REF"));
|
|
|
+ GPR_ASSERT(old_refs != 0);
|
|
|
+ return this;
|
|
|
}
|
|
|
|
|
|
-grpc_connectivity_state grpc_subchannel_check_connectivity(
|
|
|
- grpc_subchannel* c, grpc_error** error, bool inhibit_health_checks) {
|
|
|
- gpr_mu_lock(&c->mu);
|
|
|
- grpc_connectivity_state_tracker* tracker =
|
|
|
- inhibit_health_checks ? &c->state_tracker : &c->state_and_health_tracker;
|
|
|
- grpc_connectivity_state state = grpc_connectivity_state_get(tracker, error);
|
|
|
- gpr_mu_unlock(&c->mu);
|
|
|
- return state;
|
|
|
+namespace {
|
|
|
+
|
|
|
+void subchannel_destroy(void* arg, grpc_error* error) {
|
|
|
+ Subchannel* self = static_cast<Subchannel*>(arg);
|
|
|
+ Delete(self);
|
|
|
}
|
|
|
|
|
|
-static void on_external_state_watcher_done(void* arg, grpc_error* error) {
|
|
|
- external_state_watcher* w = static_cast<external_state_watcher*>(arg);
|
|
|
- grpc_closure* follow_up = w->notify;
|
|
|
- if (w->pollset_set != nullptr) {
|
|
|
- grpc_pollset_set_del_pollset_set(w->subchannel->pollset_set,
|
|
|
- w->pollset_set);
|
|
|
+} // namespace
|
|
|
+
|
|
|
+void Subchannel::WeakUnref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
+ gpr_atm old_refs;
|
|
|
+ old_refs = RefMutate(-static_cast<gpr_atm>(1),
|
|
|
+ 1 GRPC_SUBCHANNEL_REF_MUTATE_PURPOSE("WEAK_UNREF"));
|
|
|
+ if (old_refs == 1) {
|
|
|
+ GRPC_CLOSURE_SCHED(GRPC_CLOSURE_CREATE(subchannel_destroy, this,
|
|
|
+ grpc_schedule_on_exec_ctx),
|
|
|
+ GRPC_ERROR_NONE);
|
|
|
}
|
|
|
- gpr_mu_lock(&w->subchannel->mu);
|
|
|
- w->next->prev = w->prev;
|
|
|
- w->prev->next = w->next;
|
|
|
- gpr_mu_unlock(&w->subchannel->mu);
|
|
|
- GRPC_SUBCHANNEL_WEAK_UNREF(w->subchannel, "external_state_watcher");
|
|
|
- gpr_free(w);
|
|
|
- GRPC_CLOSURE_SCHED(follow_up, GRPC_ERROR_REF(error));
|
|
|
}
|
|
|
|
|
|
-static void on_alarm(void* arg, grpc_error* error) {
|
|
|
- grpc_subchannel* c = static_cast<grpc_subchannel*>(arg);
|
|
|
- gpr_mu_lock(&c->mu);
|
|
|
- c->have_alarm = false;
|
|
|
- if (c->disconnected) {
|
|
|
- error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING("Disconnected",
|
|
|
- &error, 1);
|
|
|
- } else if (c->retry_immediately) {
|
|
|
- c->retry_immediately = false;
|
|
|
- error = GRPC_ERROR_NONE;
|
|
|
- } else {
|
|
|
- GRPC_ERROR_REF(error);
|
|
|
- }
|
|
|
- if (error == GRPC_ERROR_NONE) {
|
|
|
- gpr_log(GPR_INFO, "Failed to connect to channel, retrying");
|
|
|
- continue_connect_locked(c);
|
|
|
- gpr_mu_unlock(&c->mu);
|
|
|
- } else {
|
|
|
- gpr_mu_unlock(&c->mu);
|
|
|
- GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting");
|
|
|
+Subchannel* Subchannel::RefFromWeakRef(GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
+ for (;;) {
|
|
|
+ gpr_atm old_refs = gpr_atm_acq_load(&ref_pair_);
|
|
|
+ if (old_refs >= (1 << INTERNAL_REF_BITS)) {
|
|
|
+ gpr_atm new_refs = old_refs + (1 << INTERNAL_REF_BITS);
|
|
|
+ if (gpr_atm_rel_cas(&ref_pair_, old_refs, new_refs)) {
|
|
|
+ return this;
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ return nullptr;
|
|
|
+ }
|
|
|
}
|
|
|
- GRPC_ERROR_UNREF(error);
|
|
|
}
|
|
|
|
|
|
-static void maybe_start_connecting_locked(grpc_subchannel* c) {
|
|
|
- if (c->disconnected) {
|
|
|
- /* Don't try to connect if we're already disconnected */
|
|
|
- return;
|
|
|
- }
|
|
|
- if (c->connecting) {
|
|
|
- /* Already connecting: don't restart */
|
|
|
- return;
|
|
|
- }
|
|
|
- if (c->connected_subchannel != nullptr) {
|
|
|
- /* Already connected: don't restart */
|
|
|
- return;
|
|
|
- }
|
|
|
- if (!grpc_connectivity_state_has_watchers(&c->state_tracker) &&
|
|
|
- !grpc_connectivity_state_has_watchers(&c->state_and_health_tracker)) {
|
|
|
- /* Nobody is interested in connecting: so don't just yet */
|
|
|
- return;
|
|
|
- }
|
|
|
- c->connecting = true;
|
|
|
- GRPC_SUBCHANNEL_WEAK_REF(c, "connecting");
|
|
|
- if (!c->backoff_begun) {
|
|
|
- c->backoff_begun = true;
|
|
|
- continue_connect_locked(c);
|
|
|
+intptr_t Subchannel::GetChildSocketUuid() {
|
|
|
+ if (connected_subchannel_ != nullptr) {
|
|
|
+ return connected_subchannel_->socket_uuid();
|
|
|
} else {
|
|
|
- GPR_ASSERT(!c->have_alarm);
|
|
|
- c->have_alarm = true;
|
|
|
- const grpc_millis time_til_next =
|
|
|
- c->next_attempt_deadline - grpc_core::ExecCtx::Get()->Now();
|
|
|
- if (time_til_next <= 0) {
|
|
|
- gpr_log(GPR_INFO, "Subchannel %p: Retry immediately", c);
|
|
|
- } else {
|
|
|
- gpr_log(GPR_INFO, "Subchannel %p: Retry in %" PRId64 " milliseconds", c,
|
|
|
- time_til_next);
|
|
|
- }
|
|
|
- GRPC_CLOSURE_INIT(&c->on_alarm, on_alarm, c, grpc_schedule_on_exec_ctx);
|
|
|
- grpc_timer_init(&c->alarm, c->next_attempt_deadline, &c->on_alarm);
|
|
|
+ return 0;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-void grpc_subchannel_notify_on_state_change(
|
|
|
- grpc_subchannel* c, grpc_pollset_set* interested_parties,
|
|
|
- grpc_connectivity_state* state, grpc_closure* notify,
|
|
|
- bool inhibit_health_checks) {
|
|
|
+const char* Subchannel::GetTargetAddress() {
|
|
|
+ const grpc_arg* addr_arg =
|
|
|
+ grpc_channel_args_find(args_, GRPC_ARG_SUBCHANNEL_ADDRESS);
|
|
|
+ const char* addr_str = grpc_channel_arg_get_string(addr_arg);
|
|
|
+ GPR_ASSERT(addr_str != nullptr); // Should have been set by LB policy.
|
|
|
+ return addr_str;
|
|
|
+}
|
|
|
+
|
|
|
+RefCountedPtr<ConnectedSubchannel> Subchannel::connected_subchannel() {
|
|
|
+ MutexLock lock(&mu_);
|
|
|
+ return connected_subchannel_;
|
|
|
+}
|
|
|
+
|
|
|
+channelz::SubchannelNode* Subchannel::channelz_node() {
|
|
|
+ return channelz_node_.get();
|
|
|
+}
|
|
|
+
|
|
|
+grpc_connectivity_state Subchannel::CheckConnectivity(
|
|
|
+ grpc_error** error, bool inhibit_health_checking) {
|
|
|
+ MutexLock lock(&mu_);
|
|
|
+ grpc_connectivity_state_tracker* tracker =
|
|
|
+ inhibit_health_checking ? &state_tracker_ : &state_and_health_tracker_;
|
|
|
+ grpc_connectivity_state state = grpc_connectivity_state_get(tracker, error);
|
|
|
+ return state;
|
|
|
+}
|
|
|
+
|
|
|
+void Subchannel::NotifyOnStateChange(grpc_pollset_set* interested_parties,
|
|
|
+ grpc_connectivity_state* state,
|
|
|
+ grpc_closure* notify,
|
|
|
+ bool inhibit_health_checking) {
|
|
|
grpc_connectivity_state_tracker* tracker =
|
|
|
- inhibit_health_checks ? &c->state_tracker : &c->state_and_health_tracker;
|
|
|
- external_state_watcher* w;
|
|
|
+ inhibit_health_checking ? &state_tracker_ : &state_and_health_tracker_;
|
|
|
+ ExternalStateWatcher* w;
|
|
|
if (state == nullptr) {
|
|
|
- gpr_mu_lock(&c->mu);
|
|
|
- for (w = c->root_external_state_watcher.next;
|
|
|
- w != &c->root_external_state_watcher; w = w->next) {
|
|
|
+ MutexLock lock(&mu_);
|
|
|
+ for (w = external_state_watcher_list_; w != nullptr; w = w->next) {
|
|
|
if (w->notify == notify) {
|
|
|
grpc_connectivity_state_notify_on_state_change(tracker, nullptr,
|
|
|
- &w->closure);
|
|
|
+ &w->on_state_changed);
|
|
|
}
|
|
|
}
|
|
|
- gpr_mu_unlock(&c->mu);
|
|
|
} else {
|
|
|
- w = static_cast<external_state_watcher*>(gpr_malloc(sizeof(*w)));
|
|
|
- w->subchannel = c;
|
|
|
- w->pollset_set = interested_parties;
|
|
|
- w->notify = notify;
|
|
|
- GRPC_CLOSURE_INIT(&w->closure, on_external_state_watcher_done, w,
|
|
|
- grpc_schedule_on_exec_ctx);
|
|
|
+ w = New<ExternalStateWatcher>(this, interested_parties, notify);
|
|
|
if (interested_parties != nullptr) {
|
|
|
- grpc_pollset_set_add_pollset_set(c->pollset_set, interested_parties);
|
|
|
+ grpc_pollset_set_add_pollset_set(pollset_set_, interested_parties);
|
|
|
+ }
|
|
|
+ MutexLock lock(&mu_);
|
|
|
+ if (external_state_watcher_list_ != nullptr) {
|
|
|
+ w->next = external_state_watcher_list_;
|
|
|
+ w->next->prev = w;
|
|
|
}
|
|
|
- GRPC_SUBCHANNEL_WEAK_REF(c, "external_state_watcher");
|
|
|
- gpr_mu_lock(&c->mu);
|
|
|
- w->next = &c->root_external_state_watcher;
|
|
|
- w->prev = w->next->prev;
|
|
|
- w->next->prev = w->prev->next = w;
|
|
|
- grpc_connectivity_state_notify_on_state_change(tracker, state, &w->closure);
|
|
|
- maybe_start_connecting_locked(c);
|
|
|
- gpr_mu_unlock(&c->mu);
|
|
|
+ external_state_watcher_list_ = w;
|
|
|
+ grpc_connectivity_state_notify_on_state_change(tracker, state,
|
|
|
+ &w->on_state_changed);
|
|
|
+ MaybeStartConnectingLocked();
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-static bool publish_transport_locked(grpc_subchannel* c) {
|
|
|
- /* construct channel stack */
|
|
|
- grpc_channel_stack_builder* builder = grpc_channel_stack_builder_create();
|
|
|
- grpc_channel_stack_builder_set_channel_arguments(
|
|
|
- builder, c->connecting_result.channel_args);
|
|
|
- grpc_channel_stack_builder_set_transport(builder,
|
|
|
- c->connecting_result.transport);
|
|
|
-
|
|
|
- if (!grpc_channel_init_create_stack(builder, GRPC_CLIENT_SUBCHANNEL)) {
|
|
|
- grpc_channel_stack_builder_destroy(builder);
|
|
|
- return false;
|
|
|
- }
|
|
|
- grpc_channel_stack* stk;
|
|
|
- grpc_error* error = grpc_channel_stack_builder_finish(
|
|
|
- builder, 0, 1, connection_destroy, nullptr,
|
|
|
- reinterpret_cast<void**>(&stk));
|
|
|
- if (error != GRPC_ERROR_NONE) {
|
|
|
- grpc_transport_destroy(c->connecting_result.transport);
|
|
|
- gpr_log(GPR_ERROR, "error initializing subchannel stack: %s",
|
|
|
- grpc_error_string(error));
|
|
|
- GRPC_ERROR_UNREF(error);
|
|
|
- return false;
|
|
|
- }
|
|
|
- intptr_t socket_uuid = c->connecting_result.socket_uuid;
|
|
|
- memset(&c->connecting_result, 0, sizeof(c->connecting_result));
|
|
|
-
|
|
|
- if (c->disconnected) {
|
|
|
- grpc_channel_stack_destroy(stk);
|
|
|
- gpr_free(stk);
|
|
|
- return false;
|
|
|
+void Subchannel::ResetBackoff() {
|
|
|
+ MutexLock lock(&mu_);
|
|
|
+ backoff_.Reset();
|
|
|
+ if (have_retry_alarm_) {
|
|
|
+ retry_immediately_ = true;
|
|
|
+ grpc_timer_cancel(&retry_alarm_);
|
|
|
+ } else {
|
|
|
+ backoff_begun_ = false;
|
|
|
+ MaybeStartConnectingLocked();
|
|
|
}
|
|
|
-
|
|
|
- /* publish */
|
|
|
- c->connected_subchannel.reset(grpc_core::New<grpc_core::ConnectedSubchannel>(
|
|
|
- stk, c->args, c->channelz_subchannel, socket_uuid));
|
|
|
- gpr_log(GPR_INFO, "New connected subchannel at %p for subchannel %p",
|
|
|
- c->connected_subchannel.get(), c);
|
|
|
-
|
|
|
- // Instantiate state watcher. Will clean itself up.
|
|
|
- c->connected_subchannel_watcher =
|
|
|
- grpc_core::MakeOrphanable<grpc_core::ConnectedSubchannelStateWatcher>(c);
|
|
|
-
|
|
|
- return true;
|
|
|
}
|
|
|
|
|
|
-static void on_subchannel_connected(void* arg, grpc_error* error) {
|
|
|
- grpc_subchannel* c = static_cast<grpc_subchannel*>(arg);
|
|
|
- grpc_channel_args* delete_channel_args = c->connecting_result.channel_args;
|
|
|
-
|
|
|
- GRPC_SUBCHANNEL_WEAK_REF(c, "on_subchannel_connected");
|
|
|
- gpr_mu_lock(&c->mu);
|
|
|
- c->connecting = false;
|
|
|
- if (c->connecting_result.transport != nullptr &&
|
|
|
- publish_transport_locked(c)) {
|
|
|
- /* do nothing, transport was published */
|
|
|
- } else if (c->disconnected) {
|
|
|
- GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting");
|
|
|
- } else {
|
|
|
- set_subchannel_connectivity_state_locked(
|
|
|
- c, GRPC_CHANNEL_TRANSIENT_FAILURE,
|
|
|
- grpc_error_set_int(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
|
|
|
- "Connect Failed", &error, 1),
|
|
|
- GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE),
|
|
|
- "connect_failed");
|
|
|
- grpc_connectivity_state_set(
|
|
|
- &c->state_and_health_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE,
|
|
|
- grpc_error_set_int(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
|
|
|
- "Connect Failed", &error, 1),
|
|
|
- GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE),
|
|
|
- "connect_failed");
|
|
|
-
|
|
|
- const char* errmsg = grpc_error_string(error);
|
|
|
- gpr_log(GPR_INFO, "Connect failed: %s", errmsg);
|
|
|
-
|
|
|
- maybe_start_connecting_locked(c);
|
|
|
- GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting");
|
|
|
- }
|
|
|
- gpr_mu_unlock(&c->mu);
|
|
|
- GRPC_SUBCHANNEL_WEAK_UNREF(c, "connected");
|
|
|
- grpc_channel_args_destroy(delete_channel_args);
|
|
|
+grpc_arg Subchannel::CreateSubchannelAddressArg(
|
|
|
+ const grpc_resolved_address* addr) {
|
|
|
+ return grpc_channel_arg_string_create(
|
|
|
+ (char*)GRPC_ARG_SUBCHANNEL_ADDRESS,
|
|
|
+ addr->len > 0 ? grpc_sockaddr_to_uri(addr) : gpr_strdup(""));
|
|
|
}
|
|
|
|
|
|
-void grpc_subchannel_reset_backoff(grpc_subchannel* subchannel) {
|
|
|
- gpr_mu_lock(&subchannel->mu);
|
|
|
- subchannel->backoff->Reset();
|
|
|
- if (subchannel->have_alarm) {
|
|
|
- subchannel->retry_immediately = true;
|
|
|
- grpc_timer_cancel(&subchannel->alarm);
|
|
|
- } else {
|
|
|
- subchannel->backoff_begun = false;
|
|
|
- maybe_start_connecting_locked(subchannel);
|
|
|
- }
|
|
|
- gpr_mu_unlock(&subchannel->mu);
|
|
|
+const char* Subchannel::GetUriFromSubchannelAddressArg(
|
|
|
+ const grpc_channel_args* args) {
|
|
|
+ const grpc_arg* addr_arg =
|
|
|
+ grpc_channel_args_find(args, GRPC_ARG_SUBCHANNEL_ADDRESS);
|
|
|
+ const char* addr_str = grpc_channel_arg_get_string(addr_arg);
|
|
|
+ GPR_ASSERT(addr_str != nullptr); // Should have been set by LB policy.
|
|
|
+ return addr_str;
|
|
|
}
|
|
|
|
|
|
-/*
|
|
|
- * grpc_subchannel_call implementation
|
|
|
- */
|
|
|
+namespace {
|
|
|
|
|
|
-static void subchannel_call_destroy(void* call, grpc_error* error) {
|
|
|
- GPR_TIMER_SCOPE("grpc_subchannel_call_unref.destroy", 0);
|
|
|
- grpc_subchannel_call* c = static_cast<grpc_subchannel_call*>(call);
|
|
|
- grpc_core::ConnectedSubchannel* connection = c->connection;
|
|
|
- grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(c), nullptr,
|
|
|
- c->schedule_closure_after_destroy);
|
|
|
- connection->Unref(DEBUG_LOCATION, "subchannel_call");
|
|
|
- c->~grpc_subchannel_call();
|
|
|
+void UriToSockaddr(const char* uri_str, grpc_resolved_address* addr) {
|
|
|
+ grpc_uri* uri = grpc_uri_parse(uri_str, 0 /* suppress_errors */);
|
|
|
+ GPR_ASSERT(uri != nullptr);
|
|
|
+ if (!grpc_parse_uri(uri, addr)) memset(addr, 0, sizeof(*addr));
|
|
|
+ grpc_uri_destroy(uri);
|
|
|
}
|
|
|
|
|
|
-void grpc_subchannel_call_set_cleanup_closure(grpc_subchannel_call* call,
|
|
|
- grpc_closure* closure) {
|
|
|
- GPR_ASSERT(call->schedule_closure_after_destroy == nullptr);
|
|
|
- GPR_ASSERT(closure != nullptr);
|
|
|
- call->schedule_closure_after_destroy = closure;
|
|
|
-}
|
|
|
+} // namespace
|
|
|
|
|
|
-grpc_subchannel_call* grpc_subchannel_call_ref(
|
|
|
- grpc_subchannel_call* c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
- GRPC_CALL_STACK_REF(SUBCHANNEL_CALL_TO_CALL_STACK(c), REF_REASON);
|
|
|
- return c;
|
|
|
+void Subchannel::GetAddressFromSubchannelAddressArg(
|
|
|
+ const grpc_channel_args* args, grpc_resolved_address* addr) {
|
|
|
+ const char* addr_uri_str = GetUriFromSubchannelAddressArg(args);
|
|
|
+ memset(addr, 0, sizeof(*addr));
|
|
|
+ if (*addr_uri_str != '\0') {
|
|
|
+ UriToSockaddr(addr_uri_str, addr);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
-void grpc_subchannel_call_unref(
|
|
|
- grpc_subchannel_call* c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
|
|
|
- GRPC_CALL_STACK_UNREF(SUBCHANNEL_CALL_TO_CALL_STACK(c), REF_REASON);
|
|
|
-}
|
|
|
+namespace {
|
|
|
|
|
|
-// Sets *status based on md_batch and error.
|
|
|
-static void get_call_status(grpc_subchannel_call* call,
|
|
|
- grpc_metadata_batch* md_batch, grpc_error* error,
|
|
|
- grpc_status_code* status) {
|
|
|
- if (error != GRPC_ERROR_NONE) {
|
|
|
- grpc_error_get_status(error, call->deadline, status, nullptr, nullptr,
|
|
|
- nullptr);
|
|
|
- } else {
|
|
|
- if (md_batch->idx.named.grpc_status != nullptr) {
|
|
|
- *status = grpc_get_status_code_from_metadata(
|
|
|
- md_batch->idx.named.grpc_status->md);
|
|
|
- } else {
|
|
|
- *status = GRPC_STATUS_UNKNOWN;
|
|
|
- }
|
|
|
+// Returns a string indicating the subchannel's connectivity state change to
|
|
|
+// \a state.
|
|
|
+const char* SubchannelConnectivityStateChangeString(
|
|
|
+ grpc_connectivity_state state) {
|
|
|
+ switch (state) {
|
|
|
+ case GRPC_CHANNEL_IDLE:
|
|
|
+ return "Subchannel state change to IDLE";
|
|
|
+ case GRPC_CHANNEL_CONNECTING:
|
|
|
+ return "Subchannel state change to CONNECTING";
|
|
|
+ case GRPC_CHANNEL_READY:
|
|
|
+ return "Subchannel state change to READY";
|
|
|
+ case GRPC_CHANNEL_TRANSIENT_FAILURE:
|
|
|
+ return "Subchannel state change to TRANSIENT_FAILURE";
|
|
|
+ case GRPC_CHANNEL_SHUTDOWN:
|
|
|
+ return "Subchannel state change to SHUTDOWN";
|
|
|
}
|
|
|
- GRPC_ERROR_UNREF(error);
|
|
|
+ GPR_UNREACHABLE_CODE(return "UNKNOWN");
|
|
|
}
|
|
|
|
|
|
-static void recv_trailing_metadata_ready(void* arg, grpc_error* error) {
|
|
|
- grpc_subchannel_call* call = static_cast<grpc_subchannel_call*>(arg);
|
|
|
- GPR_ASSERT(call->recv_trailing_metadata != nullptr);
|
|
|
- grpc_status_code status = GRPC_STATUS_OK;
|
|
|
- grpc_metadata_batch* md_batch = call->recv_trailing_metadata;
|
|
|
- get_call_status(call, md_batch, GRPC_ERROR_REF(error), &status);
|
|
|
- grpc_core::channelz::SubchannelNode* channelz_subchannel =
|
|
|
- call->connection->channelz_subchannel();
|
|
|
- GPR_ASSERT(channelz_subchannel != nullptr);
|
|
|
- if (status == GRPC_STATUS_OK) {
|
|
|
- channelz_subchannel->RecordCallSucceeded();
|
|
|
- } else {
|
|
|
- channelz_subchannel->RecordCallFailed();
|
|
|
+} // namespace
|
|
|
+
|
|
|
+void Subchannel::SetConnectivityStateLocked(grpc_connectivity_state state,
|
|
|
+ grpc_error* error,
|
|
|
+ const char* reason) {
|
|
|
+ if (channelz_node_ != nullptr) {
|
|
|
+ channelz_node_->AddTraceEvent(
|
|
|
+ channelz::ChannelTrace::Severity::Info,
|
|
|
+ grpc_slice_from_static_string(
|
|
|
+ SubchannelConnectivityStateChangeString(state)));
|
|
|
}
|
|
|
- GRPC_CLOSURE_RUN(call->original_recv_trailing_metadata,
|
|
|
- GRPC_ERROR_REF(error));
|
|
|
+ grpc_connectivity_state_set(&state_tracker_, state, error, reason);
|
|
|
}
|
|
|
|
|
|
-// If channelz is enabled, intercept recv_trailing so that we may check the
|
|
|
-// status and associate it to a subchannel.
|
|
|
-static void maybe_intercept_recv_trailing_metadata(
|
|
|
- grpc_subchannel_call* call, grpc_transport_stream_op_batch* batch) {
|
|
|
- // only intercept payloads with recv trailing.
|
|
|
- if (!batch->recv_trailing_metadata) {
|
|
|
+void Subchannel::MaybeStartConnectingLocked() {
|
|
|
+ if (disconnected_) {
|
|
|
+ // Don't try to connect if we're already disconnected.
|
|
|
return;
|
|
|
}
|
|
|
- // only add interceptor is channelz is enabled.
|
|
|
- if (call->connection->channelz_subchannel() == nullptr) {
|
|
|
+ if (connecting_) {
|
|
|
+ // Already connecting: don't restart.
|
|
|
return;
|
|
|
}
|
|
|
- GRPC_CLOSURE_INIT(&call->recv_trailing_metadata_ready,
|
|
|
- recv_trailing_metadata_ready, call,
|
|
|
- grpc_schedule_on_exec_ctx);
|
|
|
- // save some state needed for the interception callback.
|
|
|
- GPR_ASSERT(call->recv_trailing_metadata == nullptr);
|
|
|
- call->recv_trailing_metadata =
|
|
|
- batch->payload->recv_trailing_metadata.recv_trailing_metadata;
|
|
|
- call->original_recv_trailing_metadata =
|
|
|
- batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
|
|
|
- batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
|
|
|
- &call->recv_trailing_metadata_ready;
|
|
|
-}
|
|
|
-
|
|
|
-void grpc_subchannel_call_process_op(grpc_subchannel_call* call,
|
|
|
- grpc_transport_stream_op_batch* batch) {
|
|
|
- GPR_TIMER_SCOPE("grpc_subchannel_call_process_op", 0);
|
|
|
- maybe_intercept_recv_trailing_metadata(call, batch);
|
|
|
- grpc_call_stack* call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call);
|
|
|
- grpc_call_element* top_elem = grpc_call_stack_element(call_stack, 0);
|
|
|
- GRPC_CALL_LOG_OP(GPR_INFO, top_elem, batch);
|
|
|
- top_elem->filter->start_transport_stream_op_batch(top_elem, batch);
|
|
|
+ if (connected_subchannel_ != nullptr) {
|
|
|
+ // Already connected: don't restart.
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (!grpc_connectivity_state_has_watchers(&state_tracker_) &&
|
|
|
+ !grpc_connectivity_state_has_watchers(&state_and_health_tracker_)) {
|
|
|
+ // Nobody is interested in connecting: so don't just yet.
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ connecting_ = true;
|
|
|
+ GRPC_SUBCHANNEL_WEAK_REF(this, "connecting");
|
|
|
+ if (!backoff_begun_) {
|
|
|
+ backoff_begun_ = true;
|
|
|
+ ContinueConnectingLocked();
|
|
|
+ } else {
|
|
|
+ GPR_ASSERT(!have_retry_alarm_);
|
|
|
+ have_retry_alarm_ = true;
|
|
|
+ const grpc_millis time_til_next =
|
|
|
+ next_attempt_deadline_ - ExecCtx::Get()->Now();
|
|
|
+ if (time_til_next <= 0) {
|
|
|
+ gpr_log(GPR_INFO, "Subchannel %p: Retry immediately", this);
|
|
|
+ } else {
|
|
|
+ gpr_log(GPR_INFO, "Subchannel %p: Retry in %" PRId64 " milliseconds",
|
|
|
+ this, time_til_next);
|
|
|
+ }
|
|
|
+ GRPC_CLOSURE_INIT(&on_retry_alarm_, OnRetryAlarm, this,
|
|
|
+ grpc_schedule_on_exec_ctx);
|
|
|
+ grpc_timer_init(&retry_alarm_, next_attempt_deadline_, &on_retry_alarm_);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
-grpc_core::RefCountedPtr<grpc_core::ConnectedSubchannel>
|
|
|
-grpc_subchannel_get_connected_subchannel(grpc_subchannel* c) {
|
|
|
- gpr_mu_lock(&c->mu);
|
|
|
- auto copy = c->connected_subchannel;
|
|
|
- gpr_mu_unlock(&c->mu);
|
|
|
- return copy;
|
|
|
+void Subchannel::OnRetryAlarm(void* arg, grpc_error* error) {
|
|
|
+ Subchannel* c = static_cast<Subchannel*>(arg);
|
|
|
+ gpr_mu_lock(&c->mu_);
|
|
|
+ c->have_retry_alarm_ = false;
|
|
|
+ if (c->disconnected_) {
|
|
|
+ error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING("Disconnected",
|
|
|
+ &error, 1);
|
|
|
+ } else if (c->retry_immediately_) {
|
|
|
+ c->retry_immediately_ = false;
|
|
|
+ error = GRPC_ERROR_NONE;
|
|
|
+ } else {
|
|
|
+ GRPC_ERROR_REF(error);
|
|
|
+ }
|
|
|
+ if (error == GRPC_ERROR_NONE) {
|
|
|
+ gpr_log(GPR_INFO, "Failed to connect to channel, retrying");
|
|
|
+ c->ContinueConnectingLocked();
|
|
|
+ gpr_mu_unlock(&c->mu_);
|
|
|
+ } else {
|
|
|
+ gpr_mu_unlock(&c->mu_);
|
|
|
+ GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting");
|
|
|
+ }
|
|
|
+ GRPC_ERROR_UNREF(error);
|
|
|
}
|
|
|
|
|
|
-void* grpc_connected_subchannel_call_get_parent_data(
|
|
|
- grpc_subchannel_call* subchannel_call) {
|
|
|
- grpc_channel_stack* chanstk = subchannel_call->connection->channel_stack();
|
|
|
- return (char*)subchannel_call + sizeof(grpc_subchannel_call) +
|
|
|
- chanstk->call_stack_size;
|
|
|
+void Subchannel::ContinueConnectingLocked() {
|
|
|
+ grpc_connect_in_args args;
|
|
|
+ args.interested_parties = pollset_set_;
|
|
|
+ const grpc_millis min_deadline =
|
|
|
+ min_connect_timeout_ms_ + ExecCtx::Get()->Now();
|
|
|
+ next_attempt_deadline_ = backoff_.NextAttemptTime();
|
|
|
+ args.deadline = std::max(next_attempt_deadline_, min_deadline);
|
|
|
+ args.channel_args = args_;
|
|
|
+ SetConnectivityStateLocked(GRPC_CHANNEL_CONNECTING, GRPC_ERROR_NONE,
|
|
|
+ "connecting");
|
|
|
+ grpc_connectivity_state_set(&state_and_health_tracker_,
|
|
|
+ GRPC_CHANNEL_CONNECTING, GRPC_ERROR_NONE,
|
|
|
+ "connecting");
|
|
|
+ grpc_connector_connect(connector_, &args, &connecting_result_,
|
|
|
+ &on_connecting_finished_);
|
|
|
}
|
|
|
|
|
|
-grpc_call_stack* grpc_subchannel_call_get_call_stack(
|
|
|
- grpc_subchannel_call* subchannel_call) {
|
|
|
- return SUBCHANNEL_CALL_TO_CALL_STACK(subchannel_call);
|
|
|
-}
|
|
|
+void Subchannel::OnConnectingFinished(void* arg, grpc_error* error) {
|
|
|
+ auto* c = static_cast<Subchannel*>(arg);
|
|
|
+ grpc_channel_args* delete_channel_args = c->connecting_result_.channel_args;
|
|
|
+ GRPC_SUBCHANNEL_WEAK_REF(c, "on_connecting_finished");
|
|
|
+ gpr_mu_lock(&c->mu_);
|
|
|
+ c->connecting_ = false;
|
|
|
+ if (c->connecting_result_.transport != nullptr &&
|
|
|
+ c->PublishTransportLocked()) {
|
|
|
+ // Do nothing, transport was published.
|
|
|
+ } else if (c->disconnected_) {
|
|
|
+ GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting");
|
|
|
+ } else {
|
|
|
+ c->SetConnectivityStateLocked(
|
|
|
+ GRPC_CHANNEL_TRANSIENT_FAILURE,
|
|
|
+ grpc_error_set_int(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
|
|
|
+ "Connect Failed", &error, 1),
|
|
|
+ GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE),
|
|
|
+ "connect_failed");
|
|
|
+ grpc_connectivity_state_set(
|
|
|
+ &c->state_and_health_tracker_, GRPC_CHANNEL_TRANSIENT_FAILURE,
|
|
|
+ grpc_error_set_int(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
|
|
|
+ "Connect Failed", &error, 1),
|
|
|
+ GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE),
|
|
|
+ "connect_failed");
|
|
|
|
|
|
-static void grpc_uri_to_sockaddr(const char* uri_str,
|
|
|
- grpc_resolved_address* addr) {
|
|
|
- grpc_uri* uri = grpc_uri_parse(uri_str, 0 /* suppress_errors */);
|
|
|
- GPR_ASSERT(uri != nullptr);
|
|
|
- if (!grpc_parse_uri(uri, addr)) memset(addr, 0, sizeof(*addr));
|
|
|
- grpc_uri_destroy(uri);
|
|
|
-}
|
|
|
+ const char* errmsg = grpc_error_string(error);
|
|
|
+ gpr_log(GPR_INFO, "Connect failed: %s", errmsg);
|
|
|
|
|
|
-void grpc_get_subchannel_address_arg(const grpc_channel_args* args,
|
|
|
- grpc_resolved_address* addr) {
|
|
|
- const char* addr_uri_str = grpc_get_subchannel_address_uri_arg(args);
|
|
|
- memset(addr, 0, sizeof(*addr));
|
|
|
- if (*addr_uri_str != '\0') {
|
|
|
- grpc_uri_to_sockaddr(addr_uri_str, addr);
|
|
|
+ c->MaybeStartConnectingLocked();
|
|
|
+ GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting");
|
|
|
}
|
|
|
+ gpr_mu_unlock(&c->mu_);
|
|
|
+ GRPC_SUBCHANNEL_WEAK_UNREF(c, "on_connecting_finished");
|
|
|
+ grpc_channel_args_destroy(delete_channel_args);
|
|
|
}
|
|
|
|
|
|
-const char* grpc_subchannel_get_target(grpc_subchannel* subchannel) {
|
|
|
- const grpc_arg* addr_arg =
|
|
|
- grpc_channel_args_find(subchannel->args, GRPC_ARG_SUBCHANNEL_ADDRESS);
|
|
|
- const char* addr_str = grpc_channel_arg_get_string(addr_arg);
|
|
|
- GPR_ASSERT(addr_str != nullptr); // Should have been set by LB policy.
|
|
|
- return addr_str;
|
|
|
-}
|
|
|
-
|
|
|
-const char* grpc_get_subchannel_address_uri_arg(const grpc_channel_args* args) {
|
|
|
- const grpc_arg* addr_arg =
|
|
|
- grpc_channel_args_find(args, GRPC_ARG_SUBCHANNEL_ADDRESS);
|
|
|
- const char* addr_str = grpc_channel_arg_get_string(addr_arg);
|
|
|
- GPR_ASSERT(addr_str != nullptr); // Should have been set by LB policy.
|
|
|
- return addr_str;
|
|
|
-}
|
|
|
-
|
|
|
-grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address* addr) {
|
|
|
- return grpc_channel_arg_string_create(
|
|
|
- (char*)GRPC_ARG_SUBCHANNEL_ADDRESS,
|
|
|
- addr->len > 0 ? grpc_sockaddr_to_uri(addr) : gpr_strdup(""));
|
|
|
-}
|
|
|
-
|
|
|
-namespace grpc_core {
|
|
|
-
|
|
|
-ConnectedSubchannel::ConnectedSubchannel(
|
|
|
- grpc_channel_stack* channel_stack, const grpc_channel_args* args,
|
|
|
- grpc_core::RefCountedPtr<grpc_core::channelz::SubchannelNode>
|
|
|
- channelz_subchannel,
|
|
|
- intptr_t socket_uuid)
|
|
|
- : RefCounted<ConnectedSubchannel>(&grpc_trace_stream_refcount),
|
|
|
- channel_stack_(channel_stack),
|
|
|
- args_(grpc_channel_args_copy(args)),
|
|
|
- channelz_subchannel_(std::move(channelz_subchannel)),
|
|
|
- socket_uuid_(socket_uuid) {}
|
|
|
+namespace {
|
|
|
|
|
|
-ConnectedSubchannel::~ConnectedSubchannel() {
|
|
|
- grpc_channel_args_destroy(args_);
|
|
|
- GRPC_CHANNEL_STACK_UNREF(channel_stack_, "connected_subchannel_dtor");
|
|
|
+void ConnectionDestroy(void* arg, grpc_error* error) {
|
|
|
+ grpc_channel_stack* stk = static_cast<grpc_channel_stack*>(arg);
|
|
|
+ grpc_channel_stack_destroy(stk);
|
|
|
+ gpr_free(stk);
|
|
|
}
|
|
|
|
|
|
-void ConnectedSubchannel::NotifyOnStateChange(
|
|
|
- grpc_pollset_set* interested_parties, grpc_connectivity_state* state,
|
|
|
- grpc_closure* closure) {
|
|
|
- grpc_transport_op* op = grpc_make_transport_op(nullptr);
|
|
|
- grpc_channel_element* elem;
|
|
|
- op->connectivity_state = state;
|
|
|
- op->on_connectivity_state_change = closure;
|
|
|
- op->bind_pollset_set = interested_parties;
|
|
|
- elem = grpc_channel_stack_element(channel_stack_, 0);
|
|
|
- elem->filter->start_transport_op(elem, op);
|
|
|
-}
|
|
|
+} // namespace
|
|
|
|
|
|
-void ConnectedSubchannel::Ping(grpc_closure* on_initiate,
|
|
|
- grpc_closure* on_ack) {
|
|
|
- grpc_transport_op* op = grpc_make_transport_op(nullptr);
|
|
|
- grpc_channel_element* elem;
|
|
|
- op->send_ping.on_initiate = on_initiate;
|
|
|
- op->send_ping.on_ack = on_ack;
|
|
|
- elem = grpc_channel_stack_element(channel_stack_, 0);
|
|
|
- elem->filter->start_transport_op(elem, op);
|
|
|
+bool Subchannel::PublishTransportLocked() {
|
|
|
+ // Construct channel stack.
|
|
|
+ grpc_channel_stack_builder* builder = grpc_channel_stack_builder_create();
|
|
|
+ grpc_channel_stack_builder_set_channel_arguments(
|
|
|
+ builder, connecting_result_.channel_args);
|
|
|
+ grpc_channel_stack_builder_set_transport(builder,
|
|
|
+ connecting_result_.transport);
|
|
|
+ if (!grpc_channel_init_create_stack(builder, GRPC_CLIENT_SUBCHANNEL)) {
|
|
|
+ grpc_channel_stack_builder_destroy(builder);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ grpc_channel_stack* stk;
|
|
|
+ grpc_error* error = grpc_channel_stack_builder_finish(
|
|
|
+ builder, 0, 1, ConnectionDestroy, nullptr,
|
|
|
+ reinterpret_cast<void**>(&stk));
|
|
|
+ if (error != GRPC_ERROR_NONE) {
|
|
|
+ grpc_transport_destroy(connecting_result_.transport);
|
|
|
+ gpr_log(GPR_ERROR, "error initializing subchannel stack: %s",
|
|
|
+ grpc_error_string(error));
|
|
|
+ GRPC_ERROR_UNREF(error);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ intptr_t socket_uuid = connecting_result_.socket_uuid;
|
|
|
+ memset(&connecting_result_, 0, sizeof(connecting_result_));
|
|
|
+ if (disconnected_) {
|
|
|
+ grpc_channel_stack_destroy(stk);
|
|
|
+ gpr_free(stk);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ // Publish.
|
|
|
+ connected_subchannel_.reset(
|
|
|
+ New<ConnectedSubchannel>(stk, args_, channelz_node_, socket_uuid));
|
|
|
+ gpr_log(GPR_INFO, "New connected subchannel at %p for subchannel %p",
|
|
|
+ connected_subchannel_.get(), this);
|
|
|
+ // Instantiate state watcher. Will clean itself up.
|
|
|
+ connected_subchannel_watcher_ =
|
|
|
+ MakeOrphanable<ConnectedSubchannelStateWatcher>(this);
|
|
|
+ return true;
|
|
|
}
|
|
|
|
|
|
-grpc_error* ConnectedSubchannel::CreateCall(const CallArgs& args,
|
|
|
- grpc_subchannel_call** call) {
|
|
|
- const size_t allocation_size =
|
|
|
- GetInitialCallSizeEstimate(args.parent_data_size);
|
|
|
- *call = new (gpr_arena_alloc(args.arena, allocation_size))
|
|
|
- grpc_subchannel_call(this, args);
|
|
|
- grpc_call_stack* callstk = SUBCHANNEL_CALL_TO_CALL_STACK(*call);
|
|
|
- RefCountedPtr<ConnectedSubchannel> connection =
|
|
|
- Ref(DEBUG_LOCATION, "subchannel_call");
|
|
|
- connection.release(); // Ref is passed to the grpc_subchannel_call object.
|
|
|
- const grpc_call_element_args call_args = {
|
|
|
- callstk, /* call_stack */
|
|
|
- nullptr, /* server_transport_data */
|
|
|
- args.context, /* context */
|
|
|
- args.path, /* path */
|
|
|
- args.start_time, /* start_time */
|
|
|
- args.deadline, /* deadline */
|
|
|
- args.arena, /* arena */
|
|
|
- args.call_combiner /* call_combiner */
|
|
|
- };
|
|
|
- grpc_error* error = grpc_call_stack_init(
|
|
|
- channel_stack_, 1, subchannel_call_destroy, *call, &call_args);
|
|
|
- if (GPR_UNLIKELY(error != GRPC_ERROR_NONE)) {
|
|
|
- const char* error_string = grpc_error_string(error);
|
|
|
- gpr_log(GPR_ERROR, "error: %s", error_string);
|
|
|
- return error;
|
|
|
- }
|
|
|
- grpc_call_stack_set_pollset_or_pollset_set(callstk, args.pollent);
|
|
|
- if (channelz_subchannel_ != nullptr) {
|
|
|
- channelz_subchannel_->RecordCallStarted();
|
|
|
+void Subchannel::Disconnect() {
|
|
|
+ // The subchannel_pool is only used once here in this subchannel, so the
|
|
|
+ // access can be outside of the lock.
|
|
|
+ if (subchannel_pool_ != nullptr) {
|
|
|
+ subchannel_pool_->UnregisterSubchannel(key_);
|
|
|
+ subchannel_pool_.reset();
|
|
|
}
|
|
|
- return GRPC_ERROR_NONE;
|
|
|
+ MutexLock lock(&mu_);
|
|
|
+ GPR_ASSERT(!disconnected_);
|
|
|
+ disconnected_ = true;
|
|
|
+ grpc_connector_shutdown(connector_, GRPC_ERROR_CREATE_FROM_STATIC_STRING(
|
|
|
+ "Subchannel disconnected"));
|
|
|
+ connected_subchannel_.reset();
|
|
|
+ connected_subchannel_watcher_.reset();
|
|
|
}
|
|
|
|
|
|
-size_t ConnectedSubchannel::GetInitialCallSizeEstimate(
|
|
|
- size_t parent_data_size) const {
|
|
|
- size_t allocation_size =
|
|
|
- GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_subchannel_call));
|
|
|
- if (parent_data_size > 0) {
|
|
|
- allocation_size +=
|
|
|
- GPR_ROUND_UP_TO_ALIGNMENT_SIZE(channel_stack_->call_stack_size) +
|
|
|
- parent_data_size;
|
|
|
- } else {
|
|
|
- allocation_size += channel_stack_->call_stack_size;
|
|
|
+gpr_atm Subchannel::RefMutate(
|
|
|
+ gpr_atm delta, int barrier GRPC_SUBCHANNEL_REF_MUTATE_EXTRA_ARGS) {
|
|
|
+ gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&ref_pair_, delta)
|
|
|
+ : gpr_atm_no_barrier_fetch_add(&ref_pair_, delta);
|
|
|
+#ifndef NDEBUG
|
|
|
+ if (grpc_trace_stream_refcount.enabled()) {
|
|
|
+ gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
|
|
|
+ "SUBCHANNEL: %p %12s 0x%" PRIxPTR " -> 0x%" PRIxPTR " [%s]", this,
|
|
|
+ purpose, old_val, old_val + delta, reason);
|
|
|
}
|
|
|
- return allocation_size;
|
|
|
+#endif
|
|
|
+ return old_val;
|
|
|
}
|
|
|
|
|
|
} // namespace grpc_core
|
Jan Tattermusch