Remove CAS loops in global subchannel pool and simplify subchannel refcounting (#25485)

reviewable/pr25609/r2^2
apolcyn 4 years ago committed by GitHub
parent e6e82104e2
commit 0901c9914b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 26
      src/core/ext/filters/client_channel/client_channel.cc
  2. 3
      src/core/ext/filters/client_channel/client_channel_factory.h
  3. 166
      src/core/ext/filters/client_channel/global_subchannel_pool.cc
  4. 25
      src/core/ext/filters/client_channel/global_subchannel_pool.h
  5. 94
      src/core/ext/filters/client_channel/local_subchannel_pool.cc
  6. 19
      src/core/ext/filters/client_channel/local_subchannel_pool.h
  7. 209
      src/core/ext/filters/client_channel/subchannel.cc
  8. 65
      src/core/ext/filters/client_channel/subchannel.h
  9. 15
      src/core/ext/filters/client_channel/subchannel_pool_interface.cc
  10. 18
      src/core/ext/filters/client_channel/subchannel_pool_interface.h
  11. 5
      src/core/ext/transport/chttp2/client/insecure/channel_create.cc
  12. 5
      src/core/ext/transport/chttp2/client/secure/secure_channel_create.cc
  13. 2
      src/core/lib/gprpp/dual_ref_counted.h
  14. 2
      test/cpp/microbenchmarks/bm_call_create.cc

@ -1258,27 +1258,28 @@ grpc_error* DynamicTerminationFilterChannelData::Init(
// control plane work_serializer.
class ChannelData::SubchannelWrapper : public SubchannelInterface {
public:
SubchannelWrapper(ChannelData* chand, Subchannel* subchannel,
SubchannelWrapper(ChannelData* chand, RefCountedPtr<Subchannel> subchannel,
absl::optional<std::string> health_check_service_name)
: SubchannelInterface(
GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)
? "SubchannelWrapper"
: nullptr),
chand_(chand),
subchannel_(subchannel),
subchannel_(std::move(subchannel)),
health_check_service_name_(std::move(health_check_service_name)) {
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
gpr_log(GPR_INFO,
"chand=%p: creating subchannel wrapper %p for subchannel %p",
chand, this, subchannel_);
chand, this, subchannel_.get());
}
GRPC_CHANNEL_STACK_REF(chand_->owning_stack_, "SubchannelWrapper");
auto* subchannel_node = subchannel_->channelz_node();
if (subchannel_node != nullptr) {
auto it = chand_->subchannel_refcount_map_.find(subchannel_);
auto it = chand_->subchannel_refcount_map_.find(subchannel_.get());
if (it == chand_->subchannel_refcount_map_.end()) {
chand_->channelz_node_->AddChildSubchannel(subchannel_node->uuid());
it = chand_->subchannel_refcount_map_.emplace(subchannel_, 0).first;
it = chand_->subchannel_refcount_map_.emplace(subchannel_.get(), 0)
.first;
}
++it->second;
}
@ -1289,12 +1290,12 @@ class ChannelData::SubchannelWrapper : public SubchannelInterface {
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
gpr_log(GPR_INFO,
"chand=%p: destroying subchannel wrapper %p for subchannel %p",
chand_, this, subchannel_);
chand_, this, subchannel_.get());
}
chand_->subchannel_wrappers_.erase(this);
auto* subchannel_node = subchannel_->channelz_node();
if (subchannel_node != nullptr) {
auto it = chand_->subchannel_refcount_map_.find(subchannel_);
auto it = chand_->subchannel_refcount_map_.find(subchannel_.get());
GPR_ASSERT(it != chand_->subchannel_refcount_map_.end());
--it->second;
if (it->second == 0) {
@ -1302,7 +1303,6 @@ class ChannelData::SubchannelWrapper : public SubchannelInterface {
chand_->subchannel_refcount_map_.erase(it);
}
}
GRPC_SUBCHANNEL_UNREF(subchannel_, "unref from LB");
GRPC_CHANNEL_STACK_UNREF(chand_->owning_stack_, "SubchannelWrapper");
}
@ -1436,7 +1436,7 @@ class ChannelData::SubchannelWrapper : public SubchannelInterface {
gpr_log(GPR_INFO,
"chand=%p: connectivity change for subchannel wrapper %p "
"subchannel %p; hopping into work_serializer",
parent_->chand_, parent_.get(), parent_->subchannel_);
parent_->chand_, parent_.get(), parent_->subchannel_.get());
}
Ref().release(); // ref owned by lambda
parent_->chand_->work_serializer_->Run(
@ -1470,7 +1470,7 @@ class ChannelData::SubchannelWrapper : public SubchannelInterface {
"chand=%p: processing connectivity change in work serializer "
"for subchannel wrapper %p subchannel %p "
"watcher=%p",
parent_->chand_, parent_.get(), parent_->subchannel_,
parent_->chand_, parent_.get(), parent_->subchannel_.get(),
watcher_.get());
}
ConnectivityStateChange state_change = PopConnectivityStateChange();
@ -1539,7 +1539,7 @@ class ChannelData::SubchannelWrapper : public SubchannelInterface {
}
ChannelData* chand_;
Subchannel* subchannel_;
RefCountedPtr<Subchannel> subchannel_;
absl::optional<std::string> health_check_service_name_;
// Maps from the address of the watcher passed to us by the LB policy
// to the address of the WrapperWatcher that we passed to the underlying
@ -1758,7 +1758,7 @@ class ChannelData::ClientChannelControlHelper
args_to_add.data(), args_to_add.size());
gpr_free(args_to_add[0].value.string);
// Create subchannel.
Subchannel* subchannel =
RefCountedPtr<Subchannel> subchannel =
chand_->client_channel_factory_->CreateSubchannel(new_args);
grpc_channel_args_destroy(new_args);
if (subchannel == nullptr) return nullptr;
@ -1766,7 +1766,7 @@ class ChannelData::ClientChannelControlHelper
subchannel->ThrottleKeepaliveTime(chand_->keepalive_time_);
// Create and return wrapper for the subchannel.
return MakeRefCounted<SubchannelWrapper>(
chand_, subchannel, std::move(health_check_service_name));
chand_, std::move(subchannel), std::move(health_check_service_name));
}
void UpdateState(

@ -32,7 +32,8 @@ class ClientChannelFactory {
virtual ~ClientChannelFactory() = default;
// Creates a subchannel with the specified args.
virtual Subchannel* CreateSubchannel(const grpc_channel_args* args) = 0;
virtual RefCountedPtr<Subchannel> CreateSubchannel(
const grpc_channel_args* args) = 0;
// Returns a channel arg containing the specified factory.
static grpc_arg CreateChannelArg(ClientChannelFactory* factory);

@ -27,16 +27,6 @@ namespace grpc_core {
#define GRPC_REGISTER_SUBCHANNEL_CALM_DOWN_AFTER_ATTEMPTS 100
#define GRPC_REGISTER_SUBCHANNEL_CALM_DOWN_MICROS 10
GlobalSubchannelPool::GlobalSubchannelPool() {
subchannel_map_ = grpc_avl_create(&subchannel_avl_vtable_);
gpr_mu_init(&mu_);
}
GlobalSubchannelPool::~GlobalSubchannelPool() {
gpr_mu_destroy(&mu_);
grpc_avl_unref(subchannel_map_, nullptr);
}
void GlobalSubchannelPool::Init() {
instance_ = new RefCountedPtr<GlobalSubchannelPool>(
MakeRefCounted<GlobalSubchannelPool>());
@ -57,145 +47,37 @@ RefCountedPtr<GlobalSubchannelPool> GlobalSubchannelPool::instance() {
return *instance_;
}
Subchannel* GlobalSubchannelPool::RegisterSubchannel(SubchannelKey* key,
Subchannel* constructed) {
Subchannel* c = nullptr;
// Compare and swap (CAS) loop:
for (int attempt_count = 0; c == nullptr; attempt_count++) {
// Ref the shared map to have a local copy.
gpr_mu_lock(&mu_);
grpc_avl old_map = grpc_avl_ref(subchannel_map_, nullptr);
gpr_mu_unlock(&mu_);
// Check to see if a subchannel already exists.
c = static_cast<Subchannel*>(grpc_avl_get(old_map, key, nullptr));
if (c != nullptr) {
// The subchannel already exists. Try to reuse it.
c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(c, "subchannel_register+reuse");
if (c != nullptr) {
GRPC_SUBCHANNEL_UNREF(constructed,
"subchannel_register+found_existing");
// Exit the CAS loop without modifying the shared map.
} else {
// Reuse of the subchannel failed, so retry CAS loop
if (attempt_count >=
GRPC_REGISTER_SUBCHANNEL_CALM_DOWN_AFTER_ATTEMPTS) {
// GRPC_SUBCHANNEL_REF_FROM_WEAK_REF returning nullptr means that the
// subchannel we got is no longer valid and it's going to be removed
// from the AVL tree soon. Spinning here excesively here can actually
// prevent another thread from removing the subchannel, basically
// resulting in a live lock. See b/157516542 for more details.
// TODO(jtattermusch): the entire ref-counting mechanism for
// subchannels should be overhaulded, but the current workaround
// is fine for short-term.
// TODO(jtattermusch): gpr does not support thread yield operation,
// so a very short wait is the best we can do.
gpr_sleep_until(gpr_time_add(
gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_micros(GRPC_REGISTER_SUBCHANNEL_CALM_DOWN_MICROS,
GPR_TIMESPAN)));
}
}
} else {
// There hasn't been such subchannel. Add one.
// Note that we should ref the old map first because grpc_avl_add() will
// unref it while we still need to access it later.
grpc_avl new_map = grpc_avl_add(
grpc_avl_ref(old_map, nullptr), new SubchannelKey(*key),
GRPC_SUBCHANNEL_WEAK_REF(constructed, "subchannel_register+new"),
nullptr);
// Try to publish the change to the shared map. It may happen (but
// unlikely) that some other thread has changed the shared map, so compare
// to make sure it's unchanged before swapping. Retry if it's changed.
gpr_mu_lock(&mu_);
if (old_map.root == subchannel_map_.root) {
GPR_SWAP(grpc_avl, new_map, subchannel_map_);
c = constructed;
}
gpr_mu_unlock(&mu_);
grpc_avl_unref(new_map, nullptr);
}
grpc_avl_unref(old_map, nullptr);
}
return c;
}
void GlobalSubchannelPool::UnregisterSubchannel(SubchannelKey* key) {
bool done = false;
// Compare and swap (CAS) loop:
while (!done) {
// Ref the shared map to have a local copy.
gpr_mu_lock(&mu_);
grpc_avl old_map = grpc_avl_ref(subchannel_map_, nullptr);
gpr_mu_unlock(&mu_);
// Remove the subchannel.
// Note that we should ref the old map first because grpc_avl_remove() will
// unref it while we still need to access it later.
grpc_avl new_map =
grpc_avl_remove(grpc_avl_ref(old_map, nullptr), key, nullptr);
// Try to publish the change to the shared map. It may happen (but
// unlikely) that some other thread has changed the shared map, so compare
// to make sure it's unchanged before swapping. Retry if it's changed.
gpr_mu_lock(&mu_);
if (old_map.root == subchannel_map_.root) {
GPR_SWAP(grpc_avl, new_map, subchannel_map_);
done = true;
}
gpr_mu_unlock(&mu_);
grpc_avl_unref(new_map, nullptr);
grpc_avl_unref(old_map, nullptr);
RefCountedPtr<Subchannel> GlobalSubchannelPool::RegisterSubchannel(
const SubchannelKey& key, RefCountedPtr<Subchannel> constructed) {
MutexLock lock(&mu_);
auto it = subchannel_map_.find(key);
if (it != subchannel_map_.end()) {
RefCountedPtr<Subchannel> existing = it->second->RefIfNonZero();
if (existing != nullptr) return existing;
}
}
Subchannel* GlobalSubchannelPool::FindSubchannel(SubchannelKey* key) {
// Lock, and take a reference to the subchannel map.
// We don't need to do the search under a lock as AVL's are immutable.
gpr_mu_lock(&mu_);
grpc_avl index = grpc_avl_ref(subchannel_map_, nullptr);
gpr_mu_unlock(&mu_);
Subchannel* c = static_cast<Subchannel*>(grpc_avl_get(index, key, nullptr));
if (c != nullptr) c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(c, "found_from_pool");
grpc_avl_unref(index, nullptr);
return c;
subchannel_map_[key] = constructed.get();
return constructed;
}
RefCountedPtr<GlobalSubchannelPool>* GlobalSubchannelPool::instance_ = nullptr;
namespace {
void sck_avl_destroy(void* p, void* /*user_data*/) {
SubchannelKey* key = static_cast<SubchannelKey*>(p);
delete key;
}
void* sck_avl_copy(void* p, void* /*unused*/) {
const SubchannelKey* key = static_cast<const SubchannelKey*>(p);
auto* new_key = new SubchannelKey(*key);
return static_cast<void*>(new_key);
}
long sck_avl_compare(void* a, void* b, void* /*unused*/) {
const SubchannelKey* key_a = static_cast<const SubchannelKey*>(a);
const SubchannelKey* key_b = static_cast<const SubchannelKey*>(b);
return key_a->Cmp(*key_b);
}
void scv_avl_destroy(void* p, void* /*user_data*/) {
GRPC_SUBCHANNEL_WEAK_UNREF((Subchannel*)p, "global_subchannel_pool");
void GlobalSubchannelPool::UnregisterSubchannel(const SubchannelKey& key,
Subchannel* subchannel) {
MutexLock lock(&mu_);
auto it = subchannel_map_.find(key);
// delete only if key hasn't been re-registered to a different subchannel
// between strong-unreffing and unregistration of subchannel.
if (it != subchannel_map_.end() && it->second == subchannel) {
subchannel_map_.erase(it);
}
}
void* scv_avl_copy(void* p, void* /*unused*/) {
GRPC_SUBCHANNEL_WEAK_REF((Subchannel*)p, "global_subchannel_pool");
return p;
RefCountedPtr<Subchannel> GlobalSubchannelPool::FindSubchannel(
const SubchannelKey& key) {
MutexLock lock(&mu_);
auto it = subchannel_map_.find(key);
if (it == subchannel_map_.end()) return nullptr;
return it->second->RefIfNonZero();
}
} // namespace
const grpc_avl_vtable GlobalSubchannelPool::subchannel_avl_vtable_ = {
sck_avl_destroy, // destroy_key
sck_avl_copy, // copy_key
sck_avl_compare, // compare_keys
scv_avl_destroy, // destroy_value
scv_avl_copy // copy_value
};
} // namespace grpc_core

@ -21,7 +21,10 @@
#include <grpc/support/port_platform.h>
#include <map>
#include "src/core/ext/filters/client_channel/subchannel_pool_interface.h"
#include "src/core/lib/gprpp/sync.h"
namespace grpc_core {
@ -33,8 +36,8 @@ namespace grpc_core {
class GlobalSubchannelPool final : public SubchannelPoolInterface {
public:
// The ctor and dtor are not intended to use directly.
GlobalSubchannelPool();
~GlobalSubchannelPool() override;
GlobalSubchannelPool() {}
~GlobalSubchannelPool() override {}
// Should be called exactly once at filter initialization time.
static void Init();
@ -45,22 +48,24 @@ class GlobalSubchannelPool final : public SubchannelPoolInterface {
static RefCountedPtr<GlobalSubchannelPool> instance();
// Implements interface methods.
Subchannel* RegisterSubchannel(SubchannelKey* key,
Subchannel* constructed) override;
void UnregisterSubchannel(SubchannelKey* key) override;
Subchannel* FindSubchannel(SubchannelKey* key) override;
RefCountedPtr<Subchannel> RegisterSubchannel(
const SubchannelKey& key, RefCountedPtr<Subchannel> constructed) override
ABSL_LOCKS_EXCLUDED(mu_);
void UnregisterSubchannel(const SubchannelKey& key,
Subchannel* subchannel) override
ABSL_LOCKS_EXCLUDED(mu_);
RefCountedPtr<Subchannel> FindSubchannel(const SubchannelKey& key) override
ABSL_LOCKS_EXCLUDED(mu_);
private:
// The singleton instance. (It's a pointer to RefCountedPtr so that this
// non-local static object can be trivially destructible.)
static RefCountedPtr<GlobalSubchannelPool>* instance_;
// The vtable for subchannel operations in an AVL tree.
static const grpc_avl_vtable subchannel_avl_vtable_;
// A map from subchannel key to subchannel.
grpc_avl subchannel_map_;
std::map<SubchannelKey, Subchannel*> subchannel_map_ ABSL_GUARDED_BY(mu_);
// To protect subchannel_map_.
gpr_mu mu_;
Mutex mu_;
};
} // namespace grpc_core

@ -24,73 +24,33 @@
namespace grpc_core {
LocalSubchannelPool::LocalSubchannelPool() {
subchannel_map_ = grpc_avl_create(&subchannel_avl_vtable_);
RefCountedPtr<Subchannel> LocalSubchannelPool::RegisterSubchannel(
const SubchannelKey& key, RefCountedPtr<Subchannel> constructed) {
auto it = subchannel_map_.find(key);
// Because this pool is only accessed under the client channel's work
// serializer, and because FindSubchannel is checked before invoking
// RegisterSubchannel, no such subchannel should exist in the map.
GPR_ASSERT(it == subchannel_map_.end());
subchannel_map_[key] = constructed.get();
return constructed;
}
void LocalSubchannelPool::UnregisterSubchannel(const SubchannelKey& key,
Subchannel* subchannel) {
auto it = subchannel_map_.find(key);
// Because this subchannel pool is accessed only under the client
// channel's work serializer, any subchannel created by RegisterSubchannel
// will be deleted from the map in UnregisterSubchannel.
GPR_ASSERT(it != subchannel_map_.end());
GPR_ASSERT(it->second == subchannel);
subchannel_map_.erase(it);
}
RefCountedPtr<Subchannel> LocalSubchannelPool::FindSubchannel(
const SubchannelKey& key) {
auto it = subchannel_map_.find(key);
if (it == subchannel_map_.end()) return nullptr;
return it->second->Ref();
}
LocalSubchannelPool::~LocalSubchannelPool() {
grpc_avl_unref(subchannel_map_, nullptr);
}
Subchannel* LocalSubchannelPool::RegisterSubchannel(SubchannelKey* key,
Subchannel* constructed) {
// Check to see if a subchannel already exists.
Subchannel* c =
static_cast<Subchannel*>(grpc_avl_get(subchannel_map_, key, nullptr));
if (c != nullptr) {
// The subchannel already exists. Reuse it.
c = GRPC_SUBCHANNEL_REF(c, "subchannel_register+reuse");
GRPC_SUBCHANNEL_UNREF(constructed, "subchannel_register+found_existing");
} else {
// There hasn't been such subchannel. Add one.
subchannel_map_ = grpc_avl_add(subchannel_map_, new SubchannelKey(*key),
constructed, nullptr);
c = constructed;
}
return c;
}
void LocalSubchannelPool::UnregisterSubchannel(SubchannelKey* key) {
subchannel_map_ = grpc_avl_remove(subchannel_map_, key, nullptr);
}
Subchannel* LocalSubchannelPool::FindSubchannel(SubchannelKey* key) {
Subchannel* c =
static_cast<Subchannel*>(grpc_avl_get(subchannel_map_, key, nullptr));
return c == nullptr ? c : GRPC_SUBCHANNEL_REF(c, "found_from_pool");
}
namespace {
void sck_avl_destroy(void* p, void* /*user_data*/) {
SubchannelKey* key = static_cast<SubchannelKey*>(p);
delete key;
}
void* sck_avl_copy(void* p, void* /*unused*/) {
const SubchannelKey* key = static_cast<const SubchannelKey*>(p);
auto new_key = new SubchannelKey(*key);
return static_cast<void*>(new_key);
}
long sck_avl_compare(void* a, void* b, void* /*unused*/) {
const SubchannelKey* key_a = static_cast<const SubchannelKey*>(a);
const SubchannelKey* key_b = static_cast<const SubchannelKey*>(b);
return key_a->Cmp(*key_b);
}
void scv_avl_destroy(void* /*p*/, void* /*user_data*/) {}
void* scv_avl_copy(void* p, void* /*unused*/) { return p; }
} // namespace
const grpc_avl_vtable LocalSubchannelPool::subchannel_avl_vtable_ = {
sck_avl_destroy, // destroy_key
sck_avl_copy, // copy_key
sck_avl_compare, // compare_keys
scv_avl_destroy, // destroy_value
scv_avl_copy // copy_value
};
} // namespace grpc_core

@ -21,6 +21,8 @@
#include <grpc/support/port_platform.h>
#include <map>
#include "src/core/ext/filters/client_channel/subchannel_pool_interface.h"
namespace grpc_core {
@ -34,22 +36,21 @@ namespace grpc_core {
// Thread-unsafe.
class LocalSubchannelPool final : public SubchannelPoolInterface {
public:
LocalSubchannelPool();
~LocalSubchannelPool() override;
LocalSubchannelPool() {}
~LocalSubchannelPool() override {}
// Implements interface methods.
// Thread-unsafe. Intended to be invoked within the client_channel work
// serializer.
Subchannel* RegisterSubchannel(SubchannelKey* key,
Subchannel* constructed) override;
void UnregisterSubchannel(SubchannelKey* key) override;
Subchannel* FindSubchannel(SubchannelKey* key) override;
RefCountedPtr<Subchannel> RegisterSubchannel(
const SubchannelKey& key, RefCountedPtr<Subchannel> constructed) override;
void UnregisterSubchannel(const SubchannelKey& key,
Subchannel* subchannel) override;
RefCountedPtr<Subchannel> FindSubchannel(const SubchannelKey& key) override;
private:
// The vtable for subchannel operations in an AVL tree.
static const grpc_avl_vtable subchannel_avl_vtable_;
// A map from subchannel key to subchannel.
grpc_avl subchannel_map_;
std::map<SubchannelKey, Subchannel*> subchannel_map_;
};
} // namespace grpc_core

@ -303,20 +303,17 @@ class Subchannel::ConnectedSubchannelStateWatcher
: public AsyncConnectivityStateWatcherInterface {
public:
// Must be instantiated while holding c->mu.
explicit ConnectedSubchannelStateWatcher(Subchannel* c) : subchannel_(c) {
// Steal subchannel ref for connecting.
GRPC_SUBCHANNEL_WEAK_REF(subchannel_, "state_watcher");
GRPC_SUBCHANNEL_WEAK_UNREF(subchannel_, "connecting");
}
explicit ConnectedSubchannelStateWatcher(WeakRefCountedPtr<Subchannel> c)
: subchannel_(std::move(c)) {}
~ConnectedSubchannelStateWatcher() override {
GRPC_SUBCHANNEL_WEAK_UNREF(subchannel_, "state_watcher");
subchannel_.reset(DEBUG_LOCATION, "state_watcher");
}
private:
void OnConnectivityStateChange(grpc_connectivity_state new_state,
const absl::Status& status) override {
Subchannel* c = subchannel_;
Subchannel* c = subchannel_.get();
MutexLock lock(&c->mu_);
switch (new_state) {
case GRPC_CHANNEL_TRANSIENT_FAILURE:
@ -357,7 +354,7 @@ class Subchannel::ConnectedSubchannelStateWatcher
}
}
Subchannel* subchannel_;
WeakRefCountedPtr<Subchannel> subchannel_;
};
// Asynchronously notifies the \a watcher of a change in the connectvity state
@ -424,19 +421,19 @@ void Subchannel::ConnectivityStateWatcherList::NotifyLocked(
class Subchannel::HealthWatcherMap::HealthWatcher
: public AsyncConnectivityStateWatcherInterface {
public:
HealthWatcher(Subchannel* c, std::string health_check_service_name,
grpc_connectivity_state subchannel_state)
: subchannel_(c),
HealthWatcher(WeakRefCountedPtr<Subchannel> c,
std::string health_check_service_name)
: subchannel_(std::move(c)),
health_check_service_name_(std::move(health_check_service_name)),
state_(subchannel_state == GRPC_CHANNEL_READY ? GRPC_CHANNEL_CONNECTING
: subchannel_state) {
GRPC_SUBCHANNEL_WEAK_REF(subchannel_, "health_watcher");
state_(subchannel_->state_ == GRPC_CHANNEL_READY
? GRPC_CHANNEL_CONNECTING
: subchannel_->state_) {
// If the subchannel is already connected, start health checking.
if (subchannel_state == GRPC_CHANNEL_READY) StartHealthCheckingLocked();
if (subchannel_->state_ == GRPC_CHANNEL_READY) StartHealthCheckingLocked();
}
~HealthWatcher() override {
GRPC_SUBCHANNEL_WEAK_UNREF(subchannel_, "health_watcher");
subchannel_.reset(DEBUG_LOCATION, "health_watcher");
}
const std::string& health_check_service_name() const {
@ -449,7 +446,8 @@ class Subchannel::HealthWatcherMap::HealthWatcher
grpc_connectivity_state initial_state,
RefCountedPtr<Subchannel::ConnectivityStateWatcherInterface> watcher) {
if (state_ != initial_state) {
new AsyncWatcherNotifierLocked(watcher, subchannel_, state_, status_);
new AsyncWatcherNotifierLocked(watcher, subchannel_.get(), state_,
status_);
}
watcher_list_.AddWatcherLocked(std::move(watcher));
}
@ -470,14 +468,14 @@ class Subchannel::HealthWatcherMap::HealthWatcher
if (state_ != GRPC_CHANNEL_CONNECTING) {
state_ = GRPC_CHANNEL_CONNECTING;
status_ = status;
watcher_list_.NotifyLocked(subchannel_, state_, status);
watcher_list_.NotifyLocked(subchannel_.get(), state_, status);
}
// If we've become connected, start health checking.
StartHealthCheckingLocked();
} else {
state_ = state;
status_ = status;
watcher_list_.NotifyLocked(subchannel_, state_, status);
watcher_list_.NotifyLocked(subchannel_.get(), state_, status);
// We're not connected, so stop health checking.
health_check_client_.reset();
}
@ -496,7 +494,7 @@ class Subchannel::HealthWatcherMap::HealthWatcher
if (new_state != GRPC_CHANNEL_SHUTDOWN && health_check_client_ != nullptr) {
state_ = new_state;
status_ = status;
watcher_list_.NotifyLocked(subchannel_, new_state, status);
watcher_list_.NotifyLocked(subchannel_.get(), new_state, status);
}
}
@ -507,7 +505,7 @@ class Subchannel::HealthWatcherMap::HealthWatcher
subchannel_->pollset_set_, subchannel_->channelz_node_, Ref());
}
Subchannel* subchannel_;
WeakRefCountedPtr<Subchannel> subchannel_;
std::string health_check_service_name_;
OrphanablePtr<HealthCheckClient> health_check_client_;
grpc_connectivity_state state_;
@ -520,7 +518,8 @@ class Subchannel::HealthWatcherMap::HealthWatcher
//
void Subchannel::HealthWatcherMap::AddWatcherLocked(
Subchannel* subchannel, grpc_connectivity_state initial_state,
WeakRefCountedPtr<Subchannel> subchannel,
grpc_connectivity_state initial_state,
const std::string& health_check_service_name,
RefCountedPtr<ConnectivityStateWatcherInterface> watcher) {
// If the health check service name is not already present in the map,
@ -528,8 +527,8 @@ void Subchannel::HealthWatcherMap::AddWatcherLocked(
auto it = map_.find(health_check_service_name);
HealthWatcher* health_watcher;
if (it == map_.end()) {
auto w = MakeOrphanable<HealthWatcher>(
subchannel, health_check_service_name, subchannel->state_);
auto w = MakeOrphanable<HealthWatcher>(std::move(subchannel),
health_check_service_name);
health_watcher = w.get();
map_.emplace(health_check_service_name, std::move(w));
} else {
@ -647,14 +646,16 @@ Subchannel::ConnectivityStateWatcherInterface::PopConnectivityStateChange() {
return state_change;
}
Subchannel::Subchannel(SubchannelKey* key,
Subchannel::Subchannel(SubchannelKey key,
OrphanablePtr<SubchannelConnector> connector,
const grpc_channel_args* args)
: key_(key),
: DualRefCounted<Subchannel>(
GRPC_TRACE_FLAG_ENABLED(grpc_trace_subchannel_refcount) ? "Subchannel"
: nullptr),
key_(std::move(key)),
connector_(std::move(connector)),
backoff_(ParseArgsForBackoffValues(args, &min_connect_timeout_ms_)) {
GRPC_STATS_INC_CLIENT_SUBCHANNELS_CREATED();
gpr_atm_no_barrier_store(&ref_pair_, 1 << INTERNAL_REF_BITS);
pollset_set_ = grpc_pollset_set_create();
grpc_resolved_address* addr =
static_cast<grpc_resolved_address*>(gpr_malloc(sizeof(*addr)));
@ -704,26 +705,26 @@ Subchannel::~Subchannel() {
grpc_channel_args_destroy(args_);
connector_.reset();
grpc_pollset_set_destroy(pollset_set_);
delete key_;
}
Subchannel* Subchannel::Create(OrphanablePtr<SubchannelConnector> connector,
const grpc_channel_args* args) {
SubchannelKey* key = new SubchannelKey(args);
RefCountedPtr<Subchannel> Subchannel::Create(
OrphanablePtr<SubchannelConnector> connector,
const grpc_channel_args* args) {
SubchannelKey key(args);
SubchannelPoolInterface* subchannel_pool =
SubchannelPoolInterface::GetSubchannelPoolFromChannelArgs(args);
GPR_ASSERT(subchannel_pool != nullptr);
Subchannel* c = subchannel_pool->FindSubchannel(key);
RefCountedPtr<Subchannel> c = subchannel_pool->FindSubchannel(key);
if (c != nullptr) {
delete key;
return c;
}
c = new Subchannel(key, std::move(connector), args);
c = MakeRefCounted<Subchannel>(std::move(key), std::move(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.
Subchannel* registered = subchannel_pool->RegisterSubchannel(key, c);
RefCountedPtr<Subchannel> registered =
subchannel_pool->RegisterSubchannel(c->key_, c);
if (registered == c) c->subchannel_pool_ = subchannel_pool->Ref();
return registered;
}
@ -747,68 +748,6 @@ void Subchannel::ThrottleKeepaliveTime(int new_keepalive_time) {
}
}
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;
}
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");
}
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;
}
namespace {
void subchannel_destroy(void* arg, grpc_error* /*error*/) {
Subchannel* self = static_cast<Subchannel*>(arg);
delete self;
}
} // 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) {
ExecCtx::Run(DEBUG_LOCATION,
GRPC_CLOSURE_CREATE(subchannel_destroy, this,
grpc_schedule_on_exec_ctx),
GRPC_ERROR_NONE);
}
}
Subchannel* Subchannel::RefFromWeakRef() {
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;
}
}
}
const char* Subchannel::GetTargetAddress() {
const grpc_arg* addr_arg =
grpc_channel_args_find(args_, GRPC_ARG_SUBCHANNEL_ADDRESS);
@ -854,7 +793,8 @@ void Subchannel::WatchConnectivityState(
watcher_list_.AddWatcherLocked(std::move(watcher));
} else {
health_watcher_map_.AddWatcherLocked(
this, initial_state, *health_check_service_name, std::move(watcher));
WeakRef(DEBUG_LOCATION, "health_watcher"), initial_state,
*health_check_service_name, std::move(watcher));
}
}
@ -891,6 +831,21 @@ void Subchannel::ResetBackoff() {
}
}
void Subchannel::Orphan() {
// 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_, this);
subchannel_pool_.reset();
}
MutexLock lock(&mu_);
GPR_ASSERT(!disconnected_);
disconnected_ = true;
connector_.reset();
connected_subchannel_.reset();
health_watcher_map_.ShutdownLocked();
}
grpc_arg Subchannel::CreateSubchannelAddressArg(
const grpc_resolved_address* addr) {
return grpc_channel_arg_string_create(
@ -984,7 +939,8 @@ void Subchannel::MaybeStartConnectingLocked() {
return;
}
connecting_ = true;
GRPC_SUBCHANNEL_WEAK_REF(this, "connecting");
WeakRef(DEBUG_LOCATION, "connecting")
.release(); // ref held by pending connect
if (!backoff_begun_) {
backoff_begun_ = true;
ContinueConnectingLocked();
@ -1006,10 +962,8 @@ void Subchannel::MaybeStartConnectingLocked() {
}
void Subchannel::OnRetryAlarm(void* arg, grpc_error* error) {
Subchannel* c = static_cast<Subchannel*>(arg);
// TODO(soheilhy): Once subchannel refcounting is simplified, we can get use
// MutexLock instead of ReleasableMutexLock, here.
ReleasableMutexLock lock(&c->mu_);
WeakRefCountedPtr<Subchannel> c(static_cast<Subchannel*>(arg));
MutexLock lock(&c->mu_);
c->have_retry_alarm_ = false;
if (c->disconnected_) {
error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING("Disconnected",
@ -1023,10 +977,9 @@ void Subchannel::OnRetryAlarm(void* arg, grpc_error* error) {
if (error == GRPC_ERROR_NONE) {
gpr_log(GPR_INFO, "Failed to connect to channel, retrying");
c->ContinueConnectingLocked();
lock.Release();
} else {
lock.Release();
GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting");
// Still connecting, keep ref around. Note that this stolen ref won't
// be dropped without first acquiring c->mu_.
c.release();
}
GRPC_ERROR_UNREF(error);
}
@ -1044,27 +997,23 @@ void Subchannel::ContinueConnectingLocked() {
}
void Subchannel::OnConnectingFinished(void* arg, grpc_error* error) {
auto* c = static_cast<Subchannel*>(arg);
WeakRefCountedPtr<Subchannel> c(static_cast<Subchannel*>(arg));
const grpc_channel_args* delete_channel_args =
c->connecting_result_.channel_args;
GRPC_SUBCHANNEL_WEAK_REF(c, "on_connecting_finished");
{
MutexLock 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 {
} else if (!c->disconnected_) {
gpr_log(GPR_INFO, "Connect failed: %s", grpc_error_string(error));
c->SetConnectivityStateLocked(GRPC_CHANNEL_TRANSIENT_FAILURE,
grpc_error_to_absl_status(error));
GRPC_SUBCHANNEL_WEAK_UNREF(c, "connecting");
}
}
GRPC_SUBCHANNEL_WEAK_UNREF(c, "on_connecting_finished");
grpc_channel_args_destroy(delete_channel_args);
c.reset(DEBUG_LOCATION, "connecting");
}
namespace {
@ -1117,39 +1066,11 @@ bool Subchannel::PublishTransportLocked() {
}
// Start watching connected subchannel.
connected_subchannel_->StartWatch(
pollset_set_, MakeOrphanable<ConnectedSubchannelStateWatcher>(this));
pollset_set_, MakeOrphanable<ConnectedSubchannelStateWatcher>(
WeakRef(DEBUG_LOCATION, "state_watcher")));
// Report initial state.
SetConnectivityStateLocked(GRPC_CHANNEL_READY, absl::Status());
return true;
}
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();
}
MutexLock lock(&mu_);
GPR_ASSERT(!disconnected_);
disconnected_ = true;
connector_.reset();
connected_subchannel_.reset();
health_watcher_map_.ShutdownLocked();
}
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_subchannel_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);
}
#endif
return old_val;
}
} // namespace grpc_core

@ -30,6 +30,7 @@
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/gpr/time_precise.h"
#include "src/core/lib/gprpp/arena.h"
#include "src/core/lib/gprpp/dual_ref_counted.h"
#include "src/core/lib/gprpp/ref_counted.h"
#include "src/core/lib/gprpp/ref_counted_ptr.h"
#include "src/core/lib/gprpp/sync.h"
@ -41,31 +42,6 @@
// Channel arg containing a URI indicating the address to connect to.
#define GRPC_ARG_SUBCHANNEL_ADDRESS "grpc.subchannel_address"
// For debugging refcounting.
#ifndef NDEBUG
#define GRPC_SUBCHANNEL_REF(p, r) (p)->Ref(__FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(p, r) (p)->RefFromWeakRef()
#define GRPC_SUBCHANNEL_UNREF(p, r) (p)->Unref(__FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_WEAK_REF(p, r) (p)->WeakRef(__FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_WEAK_UNREF(p, r) (p)->WeakUnref(__FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_REF_EXTRA_ARGS \
const char *file, int line, const char *reason
#define GRPC_SUBCHANNEL_REF_REASON reason
#define GRPC_SUBCHANNEL_REF_MUTATE_EXTRA_ARGS \
, GRPC_SUBCHANNEL_REF_EXTRA_ARGS, const char* purpose
#define GRPC_SUBCHANNEL_REF_MUTATE_PURPOSE(x) , file, line, reason, x
#else
#define GRPC_SUBCHANNEL_REF(p, r) (p)->Ref()
#define GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(p, r) (p)->RefFromWeakRef()
#define GRPC_SUBCHANNEL_UNREF(p, r) (p)->Unref()
#define GRPC_SUBCHANNEL_WEAK_REF(p, r) (p)->WeakRef()
#define GRPC_SUBCHANNEL_WEAK_UNREF(p, r) (p)->WeakUnref()
#define GRPC_SUBCHANNEL_REF_EXTRA_ARGS
#define GRPC_SUBCHANNEL_REF_REASON ""
#define GRPC_SUBCHANNEL_REF_MUTATE_EXTRA_ARGS
#define GRPC_SUBCHANNEL_REF_MUTATE_PURPOSE(x)
#endif
namespace grpc_core {
class SubchannelCall;
@ -168,7 +144,7 @@ class SubchannelCall {
// different from the SubchannelInterface that is exposed to LB policy
// implementations. The client channel provides an adaptor class
// (SubchannelWrapper) that "converts" between the two.
class Subchannel {
class Subchannel : public DualRefCounted<Subchannel> {
public:
class ConnectivityStateWatcherInterface
: public RefCounted<ConnectivityStateWatcherInterface> {
@ -213,29 +189,20 @@ class Subchannel {
};
// The ctor and dtor are not intended to use directly.
Subchannel(SubchannelKey* key, OrphanablePtr<SubchannelConnector> connector,
Subchannel(SubchannelKey key, OrphanablePtr<SubchannelConnector> connector,
const grpc_channel_args* args);
~Subchannel();
~Subchannel() override;
// Creates a subchannel given \a connector and \a args.
static Subchannel* Create(OrphanablePtr<SubchannelConnector> connector,
const grpc_channel_args* args);
static RefCountedPtr<Subchannel> Create(
OrphanablePtr<SubchannelConnector> connector,
const grpc_channel_args* args);
// Throttles keepalive time to \a new_keepalive_time iff \a new_keepalive_time
// is larger than the subchannel's current keepalive time. The updated value
// will have an affect when the subchannel creates a new ConnectedSubchannel.
void ThrottleKeepaliveTime(int new_keepalive_time);
// Strong and weak refcounting.
Subchannel* Ref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
void Unref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
Subchannel* WeakRef(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
void WeakUnref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
// Attempts to return a strong ref when only the weak refcount is guaranteed
// non-zero. If the strong refcount is zero, does not alter the refcount and
// returns null.
Subchannel* RefFromWeakRef();
// Gets the string representing the subchannel address.
// Caller doesn't take ownership.
const char* GetTargetAddress();
@ -282,6 +249,9 @@ class Subchannel {
// go away.
void ResetBackoff();
// Tears down any existing connection, and arranges for destruction
void Orphan() override;
// Returns a new channel arg encoding the subchannel address as a URI
// string. Caller is responsible for freeing the string.
static grpc_arg CreateSubchannelAddressArg(const grpc_resolved_address* addr);
@ -333,7 +303,8 @@ class Subchannel {
class HealthWatcherMap {
public:
void AddWatcherLocked(
Subchannel* subchannel, grpc_connectivity_state initial_state,
WeakRefCountedPtr<Subchannel> subchannel,
grpc_connectivity_state initial_state,
const std::string& health_check_service_name,
RefCountedPtr<ConnectivityStateWatcherInterface> watcher);
void RemoveWatcherLocked(const std::string& health_check_service_name,
@ -368,28 +339,18 @@ class Subchannel {
void ContinueConnectingLocked();
static void OnConnectingFinished(void* arg, grpc_error* error);
bool PublishTransportLocked();
void Disconnect();
gpr_atm RefMutate(gpr_atm delta,
int barrier GRPC_SUBCHANNEL_REF_MUTATE_EXTRA_ARGS);
// The subchannel pool this subchannel is in.
RefCountedPtr<SubchannelPoolInterface> subchannel_pool_;
// TODO(juanlishen): Consider using args_ as key_ directly.
// Subchannel key that identifies this subchannel in the subchannel pool.
SubchannelKey* key_;
const SubchannelKey key_;
// Channel args.
grpc_channel_args* args_;
// pollset_set tracking who's interested in a connection being setup.
grpc_pollset_set* pollset_set_;
// Protects the other members.
Mutex mu_;
// 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_;
// Connection states.
OrphanablePtr<SubchannelConnector> connector_;

@ -49,8 +49,19 @@ SubchannelKey& SubchannelKey::operator=(const SubchannelKey& other) {
return *this;
}
int SubchannelKey::Cmp(const SubchannelKey& other) const {
return grpc_channel_args_compare(args_, other.args_);
SubchannelKey::SubchannelKey(SubchannelKey&& other) noexcept {
args_ = other.args_;
other.args_ = nullptr;
}
SubchannelKey& SubchannelKey::operator=(SubchannelKey&& other) noexcept {
args_ = other.args_;
other.args_ = nullptr;
return *this;
}
bool SubchannelKey::operator<(const SubchannelKey& other) const {
return grpc_channel_args_compare(args_, other.args_) < 0;
}
void SubchannelKey::Init(

@ -41,11 +41,11 @@ class SubchannelKey {
// Copyable.
SubchannelKey(const SubchannelKey& other);
SubchannelKey& operator=(const SubchannelKey& other);
// Not movable.
SubchannelKey(SubchannelKey&&) = delete;
SubchannelKey& operator=(SubchannelKey&&) = delete;
// Movable
SubchannelKey(SubchannelKey&&) noexcept;
SubchannelKey& operator=(SubchannelKey&&) noexcept;
int Cmp(const SubchannelKey& other) const;
bool operator<(const SubchannelKey& other) const;
private:
// Initializes the subchannel key with the given \a args and the function to
@ -72,15 +72,17 @@ class SubchannelPoolInterface : public RefCounted<SubchannelPoolInterface> {
// Registers a subchannel against a key. Returns the subchannel registered
// with \a key, which may be different from \a constructed because we reuse
// (instead of update) any existing subchannel already registered with \a key.
virtual Subchannel* RegisterSubchannel(SubchannelKey* key,
Subchannel* constructed) = 0;
virtual RefCountedPtr<Subchannel> RegisterSubchannel(
const SubchannelKey& key, RefCountedPtr<Subchannel> constructed) = 0;
// Removes the registered subchannel found by \a key.
virtual void UnregisterSubchannel(SubchannelKey* key) = 0;
virtual void UnregisterSubchannel(const SubchannelKey& key,
Subchannel* subchannel) = 0;
// Finds the subchannel registered for the given subchannel key. Returns NULL
// if no such channel exists. Thread-safe.
virtual Subchannel* FindSubchannel(SubchannelKey* key) = 0;
virtual RefCountedPtr<Subchannel> FindSubchannel(
const SubchannelKey& key) = 0;
// Creates a channel arg from \a subchannel pool.
static grpc_arg CreateChannelArg(SubchannelPoolInterface* subchannel_pool);

@ -37,10 +37,11 @@ namespace grpc_core {
class Chttp2InsecureClientChannelFactory : public ClientChannelFactory {
public:
Subchannel* CreateSubchannel(const grpc_channel_args* args) override {
RefCountedPtr<Subchannel> CreateSubchannel(
const grpc_channel_args* args) override {
grpc_channel_args* new_args =
grpc_default_authority_add_if_not_present(args);
Subchannel* s =
RefCountedPtr<Subchannel> s =
Subchannel::Create(MakeOrphanable<Chttp2Connector>(), new_args);
grpc_channel_args_destroy(new_args);
return s;

@ -43,14 +43,15 @@ namespace grpc_core {
class Chttp2SecureClientChannelFactory : public ClientChannelFactory {
public:
Subchannel* CreateSubchannel(const grpc_channel_args* args) override {
RefCountedPtr<Subchannel> CreateSubchannel(
const grpc_channel_args* args) override {
grpc_channel_args* new_args = GetSecureNamingChannelArgs(args);
if (new_args == nullptr) {
gpr_log(GPR_ERROR,
"Failed to create channel args during subchannel creation.");
return nullptr;
}
Subchannel* s =
RefCountedPtr<Subchannel> s =
Subchannel::Create(MakeOrphanable<Chttp2Connector>(), new_args);
grpc_channel_args_destroy(new_args);
return s;

@ -195,7 +195,7 @@ class DualRefCounted : public Orphanable {
#ifndef NDEBUG
const uint32_t weak_refs = GetWeakRefs(prev_ref_pair);
const uint32_t strong_refs = GetStrongRefs(prev_ref_pair);
if (trace_ != nullptr) {
if (trace != nullptr) {
gpr_log(GPR_INFO, "%s:%p %s:%d weak_unref %d -> %d (refs=%d) %s", trace,
this, location.file(), location.line(), weak_refs, weak_refs - 1,
strong_refs, reason);

@ -318,7 +318,7 @@ static void DoNothing(void* /*arg*/, grpc_error* /*error*/) {}
class FakeClientChannelFactory : public grpc_core::ClientChannelFactory {
public:
grpc_core::Subchannel* CreateSubchannel(
grpc_core::RefCountedPtr<grpc_core::Subchannel> CreateSubchannel(
const grpc_channel_args* /*args*/) override {
return nullptr;
}

Loading…
Cancel
Save