|
|
|
|
@ -26,6 +26,8 @@ |
|
|
|
|
#include <algorithm> |
|
|
|
|
#include <cstring> |
|
|
|
|
|
|
|
|
|
#include "absl/strings/str_format.h" |
|
|
|
|
|
|
|
|
|
#include <grpc/support/alloc.h> |
|
|
|
|
#include <grpc/support/string_util.h> |
|
|
|
|
|
|
|
|
|
@ -325,7 +327,8 @@ class Subchannel::ConnectedSubchannelStateWatcher |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
void OnConnectivityStateChange(grpc_connectivity_state new_state) override { |
|
|
|
|
void OnConnectivityStateChange(grpc_connectivity_state new_state, |
|
|
|
|
const absl::Status& status) override { |
|
|
|
|
Subchannel* c = subchannel_; |
|
|
|
|
MutexLock lock(&c->mu_); |
|
|
|
|
switch (new_state) { |
|
|
|
|
@ -343,7 +346,15 @@ class Subchannel::ConnectedSubchannelStateWatcher |
|
|
|
|
if (c->channelz_node() != nullptr) { |
|
|
|
|
c->channelz_node()->SetChildSocket(nullptr); |
|
|
|
|
} |
|
|
|
|
c->SetConnectivityStateLocked(GRPC_CHANNEL_TRANSIENT_FAILURE); |
|
|
|
|
// We need to construct our own status if the underlying state was
|
|
|
|
|
// shutdown since the accompanying status will be StatusCode::OK
|
|
|
|
|
// otherwise.
|
|
|
|
|
c->SetConnectivityStateLocked( |
|
|
|
|
GRPC_CHANNEL_TRANSIENT_FAILURE, |
|
|
|
|
new_state == GRPC_CHANNEL_SHUTDOWN |
|
|
|
|
? absl::Status(absl::StatusCode::kUnavailable, |
|
|
|
|
"Subchannel has disconnected.") |
|
|
|
|
: status); |
|
|
|
|
c->backoff_begun_ = false; |
|
|
|
|
c->backoff_.Reset(); |
|
|
|
|
} |
|
|
|
|
@ -354,7 +365,7 @@ class Subchannel::ConnectedSubchannelStateWatcher |
|
|
|
|
// a callback for READY, because that was the state we started
|
|
|
|
|
// this watch from. And a connected subchannel should never go
|
|
|
|
|
// from READY to CONNECTING or IDLE.
|
|
|
|
|
c->SetConnectivityStateLocked(new_state); |
|
|
|
|
c->SetConnectivityStateLocked(new_state, status); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
@ -368,14 +379,15 @@ class Subchannel::AsyncWatcherNotifierLocked { |
|
|
|
|
public: |
|
|
|
|
AsyncWatcherNotifierLocked( |
|
|
|
|
RefCountedPtr<Subchannel::ConnectivityStateWatcherInterface> watcher, |
|
|
|
|
Subchannel* subchannel, grpc_connectivity_state state) |
|
|
|
|
Subchannel* subchannel, grpc_connectivity_state state, |
|
|
|
|
const absl::Status& status) |
|
|
|
|
: watcher_(std::move(watcher)) { |
|
|
|
|
RefCountedPtr<ConnectedSubchannel> connected_subchannel; |
|
|
|
|
if (state == GRPC_CHANNEL_READY) { |
|
|
|
|
connected_subchannel = subchannel->connected_subchannel_; |
|
|
|
|
} |
|
|
|
|
watcher_->PushConnectivityStateChange( |
|
|
|
|
{state, std::move(connected_subchannel)}); |
|
|
|
|
{state, status, std::move(connected_subchannel)}); |
|
|
|
|
ExecCtx::Run( |
|
|
|
|
DEBUG_LOCATION, |
|
|
|
|
GRPC_CLOSURE_INIT(&closure_, |
|
|
|
|
@ -409,9 +421,10 @@ void Subchannel::ConnectivityStateWatcherList::RemoveWatcherLocked( |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Subchannel::ConnectivityStateWatcherList::NotifyLocked( |
|
|
|
|
Subchannel* subchannel, grpc_connectivity_state state) { |
|
|
|
|
Subchannel* subchannel, grpc_connectivity_state state, |
|
|
|
|
const absl::Status& status) { |
|
|
|
|
for (const auto& p : watchers_) { |
|
|
|
|
new AsyncWatcherNotifierLocked(p.second, subchannel, state); |
|
|
|
|
new AsyncWatcherNotifierLocked(p.second, subchannel, state, status); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -450,7 +463,7 @@ class Subchannel::HealthWatcherMap::HealthWatcher |
|
|
|
|
grpc_connectivity_state initial_state, |
|
|
|
|
RefCountedPtr<Subchannel::ConnectivityStateWatcherInterface> watcher) { |
|
|
|
|
if (state_ != initial_state) { |
|
|
|
|
new AsyncWatcherNotifierLocked(watcher, subchannel_, state_); |
|
|
|
|
new AsyncWatcherNotifierLocked(watcher, subchannel_, state_, status_); |
|
|
|
|
} |
|
|
|
|
watcher_list_.AddWatcherLocked(std::move(watcher)); |
|
|
|
|
} |
|
|
|
|
@ -462,7 +475,7 @@ class Subchannel::HealthWatcherMap::HealthWatcher |
|
|
|
|
|
|
|
|
|
bool HasWatchers() const { return !watcher_list_.empty(); } |
|
|
|
|
|
|
|
|
|
void NotifyLocked(grpc_connectivity_state state) { |
|
|
|
|
void NotifyLocked(grpc_connectivity_state state, const absl::Status& status) { |
|
|
|
|
if (state == GRPC_CHANNEL_READY) { |
|
|
|
|
// If we had not already notified for CONNECTING state, do so now.
|
|
|
|
|
// (We may have missed this earlier, because if the transition
|
|
|
|
|
@ -470,13 +483,15 @@ class Subchannel::HealthWatcherMap::HealthWatcher |
|
|
|
|
// subchannel may not have sent us a notification for CONNECTING.)
|
|
|
|
|
if (state_ != GRPC_CHANNEL_CONNECTING) { |
|
|
|
|
state_ = GRPC_CHANNEL_CONNECTING; |
|
|
|
|
watcher_list_.NotifyLocked(subchannel_, state_); |
|
|
|
|
status_ = status; |
|
|
|
|
watcher_list_.NotifyLocked(subchannel_, state_, status); |
|
|
|
|
} |
|
|
|
|
// If we've become connected, start health checking.
|
|
|
|
|
StartHealthCheckingLocked(); |
|
|
|
|
} else { |
|
|
|
|
state_ = state; |
|
|
|
|
watcher_list_.NotifyLocked(subchannel_, state_); |
|
|
|
|
status_ = status; |
|
|
|
|
watcher_list_.NotifyLocked(subchannel_, state_, status); |
|
|
|
|
// We're not connected, so stop health checking.
|
|
|
|
|
health_check_client_.reset(); |
|
|
|
|
} |
|
|
|
|
@ -489,11 +504,13 @@ class Subchannel::HealthWatcherMap::HealthWatcher |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
void OnConnectivityStateChange(grpc_connectivity_state new_state) override { |
|
|
|
|
void OnConnectivityStateChange(grpc_connectivity_state new_state, |
|
|
|
|
const absl::Status& status) override { |
|
|
|
|
MutexLock lock(&subchannel_->mu_); |
|
|
|
|
if (new_state != GRPC_CHANNEL_SHUTDOWN && health_check_client_ != nullptr) { |
|
|
|
|
state_ = new_state; |
|
|
|
|
watcher_list_.NotifyLocked(subchannel_, new_state); |
|
|
|
|
status_ = status; |
|
|
|
|
watcher_list_.NotifyLocked(subchannel_, new_state, status); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -508,6 +525,7 @@ class Subchannel::HealthWatcherMap::HealthWatcher |
|
|
|
|
grpc_core::UniquePtr<char> health_check_service_name_; |
|
|
|
|
OrphanablePtr<HealthCheckClient> health_check_client_; |
|
|
|
|
grpc_connectivity_state state_; |
|
|
|
|
absl::Status status_; |
|
|
|
|
ConnectivityStateWatcherList watcher_list_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
@ -547,9 +565,10 @@ void Subchannel::HealthWatcherMap::RemoveWatcherLocked( |
|
|
|
|
if (!it->second->HasWatchers()) map_.erase(it); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Subchannel::HealthWatcherMap::NotifyLocked(grpc_connectivity_state state) { |
|
|
|
|
void Subchannel::HealthWatcherMap::NotifyLocked(grpc_connectivity_state state, |
|
|
|
|
const absl::Status& status) { |
|
|
|
|
for (const auto& p : map_) { |
|
|
|
|
p.second->NotifyLocked(state); |
|
|
|
|
p.second->NotifyLocked(state, status); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -826,7 +845,7 @@ void Subchannel::WatchConnectivityState( |
|
|
|
|
} |
|
|
|
|
if (health_check_service_name == nullptr) { |
|
|
|
|
if (state_ != initial_state) { |
|
|
|
|
new AsyncWatcherNotifierLocked(watcher, this, state_); |
|
|
|
|
new AsyncWatcherNotifierLocked(watcher, this, state_, status_); |
|
|
|
|
} |
|
|
|
|
watcher_list_.AddWatcherLocked(std::move(watcher)); |
|
|
|
|
} else { |
|
|
|
|
@ -928,8 +947,10 @@ const char* SubchannelConnectivityStateChangeString( |
|
|
|
|
} // namespace
|
|
|
|
|
|
|
|
|
|
// Note: Must be called with a state that is different from the current state.
|
|
|
|
|
void Subchannel::SetConnectivityStateLocked(grpc_connectivity_state state) { |
|
|
|
|
void Subchannel::SetConnectivityStateLocked(grpc_connectivity_state state, |
|
|
|
|
const absl::Status& status) { |
|
|
|
|
state_ = state; |
|
|
|
|
status_ = status; |
|
|
|
|
if (channelz_node_ != nullptr) { |
|
|
|
|
channelz_node_->UpdateConnectivityState(state); |
|
|
|
|
channelz_node_->AddTraceEvent( |
|
|
|
|
@ -938,9 +959,9 @@ void Subchannel::SetConnectivityStateLocked(grpc_connectivity_state state) { |
|
|
|
|
SubchannelConnectivityStateChangeString(state))); |
|
|
|
|
} |
|
|
|
|
// Notify non-health watchers.
|
|
|
|
|
watcher_list_.NotifyLocked(this, state); |
|
|
|
|
watcher_list_.NotifyLocked(this, state, status); |
|
|
|
|
// Notify health watchers.
|
|
|
|
|
health_watcher_map_.NotifyLocked(state); |
|
|
|
|
health_watcher_map_.NotifyLocked(state, status); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Subchannel::MaybeStartConnectingLocked() { |
|
|
|
|
@ -1012,7 +1033,7 @@ void Subchannel::ContinueConnectingLocked() { |
|
|
|
|
next_attempt_deadline_ = backoff_.NextAttemptTime(); |
|
|
|
|
args.deadline = std::max(next_attempt_deadline_, min_deadline); |
|
|
|
|
args.channel_args = args_; |
|
|
|
|
SetConnectivityStateLocked(GRPC_CHANNEL_CONNECTING); |
|
|
|
|
SetConnectivityStateLocked(GRPC_CHANNEL_CONNECTING, absl::Status()); |
|
|
|
|
connector_->Connect(args, &connecting_result_, &on_connecting_finished_); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -1031,7 +1052,8 @@ void Subchannel::OnConnectingFinished(void* arg, grpc_error* error) { |
|
|
|
|
GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting"); |
|
|
|
|
} else { |
|
|
|
|
gpr_log(GPR_INFO, "Connect failed: %s", grpc_error_string(error)); |
|
|
|
|
c->SetConnectivityStateLocked(GRPC_CHANNEL_TRANSIENT_FAILURE); |
|
|
|
|
c->SetConnectivityStateLocked(GRPC_CHANNEL_TRANSIENT_FAILURE, |
|
|
|
|
grpc_error_to_absl_status(error)); |
|
|
|
|
GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting"); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
@ -1091,7 +1113,7 @@ bool Subchannel::PublishTransportLocked() { |
|
|
|
|
connected_subchannel_->StartWatch( |
|
|
|
|
pollset_set_, MakeOrphanable<ConnectedSubchannelStateWatcher>(this)); |
|
|
|
|
// Report initial state.
|
|
|
|
|
SetConnectivityStateLocked(GRPC_CHANNEL_READY); |
|
|
|
|
SetConnectivityStateLocked(GRPC_CHANNEL_READY, absl::Status()); |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
Yash Tibrewal
5 years ago