|
|
|
|
@ -38,202 +38,162 @@ |
|
|
|
|
#include "src/core/lib/iomgr/tcp_client.h" |
|
|
|
|
#include "src/core/lib/slice/slice_internal.h" |
|
|
|
|
|
|
|
|
|
typedef struct { |
|
|
|
|
grpc_connector base; |
|
|
|
|
namespace grpc_core { |
|
|
|
|
|
|
|
|
|
gpr_mu mu; |
|
|
|
|
gpr_refcount refs; |
|
|
|
|
|
|
|
|
|
bool shutdown; |
|
|
|
|
bool connecting; |
|
|
|
|
|
|
|
|
|
grpc_closure* notify; |
|
|
|
|
grpc_connect_in_args args; |
|
|
|
|
grpc_connect_out_args* result; |
|
|
|
|
|
|
|
|
|
grpc_endpoint* endpoint; // Non-NULL until handshaking starts.
|
|
|
|
|
|
|
|
|
|
grpc_closure connected; |
|
|
|
|
|
|
|
|
|
grpc_core::RefCountedPtr<grpc_core::HandshakeManager> handshake_mgr; |
|
|
|
|
} chttp2_connector; |
|
|
|
|
Chttp2Connector::Chttp2Connector() { |
|
|
|
|
GRPC_CLOSURE_INIT(&connected_, Connected, this, grpc_schedule_on_exec_ctx); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void chttp2_connector_ref(grpc_connector* con) { |
|
|
|
|
chttp2_connector* c = reinterpret_cast<chttp2_connector*>(con); |
|
|
|
|
gpr_ref(&c->refs); |
|
|
|
|
Chttp2Connector::~Chttp2Connector() { |
|
|
|
|
if (endpoint_ != nullptr) grpc_endpoint_destroy(endpoint_); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void chttp2_connector_unref(grpc_connector* con) { |
|
|
|
|
chttp2_connector* c = reinterpret_cast<chttp2_connector*>(con); |
|
|
|
|
if (gpr_unref(&c->refs)) { |
|
|
|
|
gpr_mu_destroy(&c->mu); |
|
|
|
|
// If handshaking is not yet in progress, destroy the endpoint.
|
|
|
|
|
// Otherwise, the handshaker will do this for us.
|
|
|
|
|
if (c->endpoint != nullptr) grpc_endpoint_destroy(c->endpoint); |
|
|
|
|
gpr_free(c); |
|
|
|
|
void Chttp2Connector::Connect(const Args& args, Result* result, |
|
|
|
|
grpc_closure* notify) { |
|
|
|
|
grpc_resolved_address addr; |
|
|
|
|
Subchannel::GetAddressFromSubchannelAddressArg(args.channel_args, &addr); |
|
|
|
|
grpc_endpoint** ep; |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&mu_); |
|
|
|
|
GPR_ASSERT(notify_ == nullptr); |
|
|
|
|
args_ = args; |
|
|
|
|
result_ = result; |
|
|
|
|
notify_ = notify; |
|
|
|
|
GPR_ASSERT(!connecting_); |
|
|
|
|
connecting_ = true; |
|
|
|
|
GPR_ASSERT(endpoint_ == nullptr); |
|
|
|
|
ep = &endpoint_; |
|
|
|
|
} |
|
|
|
|
// In some implementations, the closure can be flushed before
|
|
|
|
|
// grpc_tcp_client_connect() returns, and since the closure requires access
|
|
|
|
|
// to mu_, this can result in a deadlock (see
|
|
|
|
|
// https://github.com/grpc/grpc/issues/16427 for details).
|
|
|
|
|
// grpc_tcp_client_connect() will fill endpoint_ with proper contents, and we
|
|
|
|
|
// make sure that we still exist at that point by taking a ref.
|
|
|
|
|
Ref().release(); // Ref held by callback.
|
|
|
|
|
grpc_tcp_client_connect(&connected_, ep, args.interested_parties, |
|
|
|
|
args.channel_args, &addr, args.deadline); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void chttp2_connector_shutdown(grpc_connector* con, grpc_error* why) { |
|
|
|
|
chttp2_connector* c = reinterpret_cast<chttp2_connector*>(con); |
|
|
|
|
gpr_mu_lock(&c->mu); |
|
|
|
|
c->shutdown = true; |
|
|
|
|
if (c->handshake_mgr != nullptr) { |
|
|
|
|
c->handshake_mgr->Shutdown(GRPC_ERROR_REF(why)); |
|
|
|
|
void Chttp2Connector::Shutdown(grpc_error* error) { |
|
|
|
|
MutexLock lock(&mu_); |
|
|
|
|
shutdown_ = true; |
|
|
|
|
if (handshake_mgr_ != nullptr) { |
|
|
|
|
handshake_mgr_->Shutdown(GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
// If handshaking is not yet in progress, shutdown the endpoint.
|
|
|
|
|
// Otherwise, the handshaker will do this for us.
|
|
|
|
|
if (!c->connecting && c->endpoint != nullptr) { |
|
|
|
|
grpc_endpoint_shutdown(c->endpoint, GRPC_ERROR_REF(why)); |
|
|
|
|
if (!connecting_ && endpoint_ != nullptr) { |
|
|
|
|
grpc_endpoint_shutdown(endpoint_, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
gpr_mu_unlock(&c->mu); |
|
|
|
|
GRPC_ERROR_UNREF(why); |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_handshake_done(void* arg, grpc_error* error) { |
|
|
|
|
auto* args = static_cast<grpc_core::HandshakerArgs*>(arg); |
|
|
|
|
chttp2_connector* c = static_cast<chttp2_connector*>(args->user_data); |
|
|
|
|
gpr_mu_lock(&c->mu); |
|
|
|
|
if (error != GRPC_ERROR_NONE || c->shutdown) { |
|
|
|
|
if (error == GRPC_ERROR_NONE) { |
|
|
|
|
error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("connector shutdown"); |
|
|
|
|
// We were shut down after handshaking completed successfully, so
|
|
|
|
|
// destroy the endpoint here.
|
|
|
|
|
// TODO(ctiller): It is currently necessary to shutdown endpoints
|
|
|
|
|
// before destroying them, even if we know that there are no
|
|
|
|
|
// pending read/write callbacks. This should be fixed, at which
|
|
|
|
|
// point this can be removed.
|
|
|
|
|
grpc_endpoint_shutdown(args->endpoint, GRPC_ERROR_REF(error)); |
|
|
|
|
grpc_endpoint_destroy(args->endpoint); |
|
|
|
|
grpc_channel_args_destroy(args->args); |
|
|
|
|
grpc_slice_buffer_destroy_internal(args->read_buffer); |
|
|
|
|
gpr_free(args->read_buffer); |
|
|
|
|
void Chttp2Connector::Connected(void* arg, grpc_error* error) { |
|
|
|
|
Chttp2Connector* self = static_cast<Chttp2Connector*>(arg); |
|
|
|
|
bool unref = false; |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&self->mu_); |
|
|
|
|
GPR_ASSERT(self->connecting_); |
|
|
|
|
self->connecting_ = false; |
|
|
|
|
if (error != GRPC_ERROR_NONE || self->shutdown_) { |
|
|
|
|
if (error == GRPC_ERROR_NONE) { |
|
|
|
|
error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("connector shutdown"); |
|
|
|
|
} else { |
|
|
|
|
error = GRPC_ERROR_REF(error); |
|
|
|
|
} |
|
|
|
|
if (self->endpoint_ != nullptr) { |
|
|
|
|
grpc_endpoint_shutdown(self->endpoint_, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
self->result_->Reset(); |
|
|
|
|
grpc_closure* notify = self->notify_; |
|
|
|
|
self->notify_ = nullptr; |
|
|
|
|
ExecCtx::Run(DEBUG_LOCATION, notify, error); |
|
|
|
|
unref = true; |
|
|
|
|
} else { |
|
|
|
|
error = GRPC_ERROR_REF(error); |
|
|
|
|
GPR_ASSERT(self->endpoint_ != nullptr); |
|
|
|
|
self->StartHandshakeLocked(); |
|
|
|
|
} |
|
|
|
|
c->result->reset(); |
|
|
|
|
} else { |
|
|
|
|
grpc_endpoint_delete_from_pollset_set(args->endpoint, |
|
|
|
|
c->args.interested_parties); |
|
|
|
|
c->result->transport = |
|
|
|
|
grpc_create_chttp2_transport(args->args, args->endpoint, true); |
|
|
|
|
c->result->socket = |
|
|
|
|
grpc_chttp2_transport_get_socket_node(c->result->transport); |
|
|
|
|
GPR_ASSERT(c->result->transport); |
|
|
|
|
// TODO(roth): We ideally want to wait until we receive HTTP/2
|
|
|
|
|
// settings from the server before we consider the connection
|
|
|
|
|
// established. If that doesn't happen before the connection
|
|
|
|
|
// timeout expires, then we should consider the connection attempt a
|
|
|
|
|
// failure and feed that information back into the backoff code.
|
|
|
|
|
// We could pass a notify_on_receive_settings callback to
|
|
|
|
|
// grpc_chttp2_transport_start_reading() to let us know when
|
|
|
|
|
// settings are received, but we would need to figure out how to use
|
|
|
|
|
// that information here.
|
|
|
|
|
//
|
|
|
|
|
// Unfortunately, we don't currently have a way to split apart the two
|
|
|
|
|
// effects of scheduling c->notify: we start sending RPCs immediately
|
|
|
|
|
// (which we want to do) and we consider the connection attempt successful
|
|
|
|
|
// (which we don't want to do until we get the notify_on_receive_settings
|
|
|
|
|
// callback from the transport). If we could split those things
|
|
|
|
|
// apart, then we could start sending RPCs but then wait for our
|
|
|
|
|
// timeout before deciding if the connection attempt is successful.
|
|
|
|
|
// If the attempt is not successful, then we would tear down the
|
|
|
|
|
// transport and feed the failure back into the backoff code.
|
|
|
|
|
//
|
|
|
|
|
// In addition, even if we did that, we would probably not want to do
|
|
|
|
|
// so until after transparent retries is implemented. Otherwise, any
|
|
|
|
|
// RPC that we attempt to send on the connection before the timeout
|
|
|
|
|
// would fail instead of being retried on a subsequent attempt.
|
|
|
|
|
grpc_chttp2_transport_start_reading(c->result->transport, args->read_buffer, |
|
|
|
|
nullptr); |
|
|
|
|
c->result->channel_args = args->args; |
|
|
|
|
} |
|
|
|
|
grpc_closure* notify = c->notify; |
|
|
|
|
c->notify = nullptr; |
|
|
|
|
grpc_core::ExecCtx::Run(DEBUG_LOCATION, notify, error); |
|
|
|
|
c->handshake_mgr.reset(); |
|
|
|
|
gpr_mu_unlock(&c->mu); |
|
|
|
|
chttp2_connector_unref(reinterpret_cast<grpc_connector*>(c)); |
|
|
|
|
if (unref) self->Unref(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void start_handshake_locked(chttp2_connector* c) { |
|
|
|
|
c->handshake_mgr = grpc_core::MakeRefCounted<grpc_core::HandshakeManager>(); |
|
|
|
|
grpc_core::HandshakerRegistry::AddHandshakers( |
|
|
|
|
grpc_core::HANDSHAKER_CLIENT, c->args.channel_args, |
|
|
|
|
c->args.interested_parties, c->handshake_mgr.get()); |
|
|
|
|
grpc_endpoint_add_to_pollset_set(c->endpoint, c->args.interested_parties); |
|
|
|
|
c->handshake_mgr->DoHandshake(c->endpoint, c->args.channel_args, |
|
|
|
|
c->args.deadline, nullptr /* acceptor */, |
|
|
|
|
on_handshake_done, c); |
|
|
|
|
c->endpoint = nullptr; // Endpoint handed off to handshake manager.
|
|
|
|
|
void Chttp2Connector::StartHandshakeLocked() { |
|
|
|
|
handshake_mgr_ = MakeRefCounted<HandshakeManager>(); |
|
|
|
|
HandshakerRegistry::AddHandshakers(HANDSHAKER_CLIENT, args_.channel_args, |
|
|
|
|
args_.interested_parties, |
|
|
|
|
handshake_mgr_.get()); |
|
|
|
|
grpc_endpoint_add_to_pollset_set(endpoint_, args_.interested_parties); |
|
|
|
|
handshake_mgr_->DoHandshake(endpoint_, args_.channel_args, args_.deadline, |
|
|
|
|
nullptr /* acceptor */, OnHandshakeDone, this); |
|
|
|
|
endpoint_ = nullptr; // Endpoint handed off to handshake manager.
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void connected(void* arg, grpc_error* error) { |
|
|
|
|
chttp2_connector* c = static_cast<chttp2_connector*>(arg); |
|
|
|
|
gpr_mu_lock(&c->mu); |
|
|
|
|
GPR_ASSERT(c->connecting); |
|
|
|
|
c->connecting = false; |
|
|
|
|
if (error != GRPC_ERROR_NONE || c->shutdown) { |
|
|
|
|
if (error == GRPC_ERROR_NONE) { |
|
|
|
|
error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("connector shutdown"); |
|
|
|
|
void Chttp2Connector::OnHandshakeDone(void* arg, grpc_error* error) { |
|
|
|
|
auto* args = static_cast<HandshakerArgs*>(arg); |
|
|
|
|
Chttp2Connector* self = static_cast<Chttp2Connector*>(args->user_data); |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&self->mu_); |
|
|
|
|
if (error != GRPC_ERROR_NONE || self->shutdown_) { |
|
|
|
|
if (error == GRPC_ERROR_NONE) { |
|
|
|
|
error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("connector shutdown"); |
|
|
|
|
// We were shut down after handshaking completed successfully, so
|
|
|
|
|
// destroy the endpoint here.
|
|
|
|
|
// TODO(ctiller): It is currently necessary to shutdown endpoints
|
|
|
|
|
// before destroying them, even if we know that there are no
|
|
|
|
|
// pending read/write callbacks. This should be fixed, at which
|
|
|
|
|
// point this can be removed.
|
|
|
|
|
grpc_endpoint_shutdown(args->endpoint, GRPC_ERROR_REF(error)); |
|
|
|
|
grpc_endpoint_destroy(args->endpoint); |
|
|
|
|
grpc_channel_args_destroy(args->args); |
|
|
|
|
grpc_slice_buffer_destroy_internal(args->read_buffer); |
|
|
|
|
gpr_free(args->read_buffer); |
|
|
|
|
} else { |
|
|
|
|
error = GRPC_ERROR_REF(error); |
|
|
|
|
} |
|
|
|
|
self->result_->Reset(); |
|
|
|
|
} else { |
|
|
|
|
error = GRPC_ERROR_REF(error); |
|
|
|
|
grpc_endpoint_delete_from_pollset_set(args->endpoint, |
|
|
|
|
self->args_.interested_parties); |
|
|
|
|
self->result_->transport = |
|
|
|
|
grpc_create_chttp2_transport(args->args, args->endpoint, true); |
|
|
|
|
self->result_->socket_node = |
|
|
|
|
grpc_chttp2_transport_get_socket_node(self->result_->transport); |
|
|
|
|
GPR_ASSERT(self->result_->transport != nullptr); |
|
|
|
|
// TODO(roth): We ideally want to wait until we receive HTTP/2
|
|
|
|
|
// settings from the server before we consider the connection
|
|
|
|
|
// established. If that doesn't happen before the connection
|
|
|
|
|
// timeout expires, then we should consider the connection attempt a
|
|
|
|
|
// failure and feed that information back into the backoff code.
|
|
|
|
|
// We could pass a notify_on_receive_settings callback to
|
|
|
|
|
// grpc_chttp2_transport_start_reading() to let us know when
|
|
|
|
|
// settings are received, but we would need to figure out how to use
|
|
|
|
|
// that information here.
|
|
|
|
|
//
|
|
|
|
|
// Unfortunately, we don't currently have a way to split apart the two
|
|
|
|
|
// effects of scheduling c->notify: we start sending RPCs immediately
|
|
|
|
|
// (which we want to do) and we consider the connection attempt successful
|
|
|
|
|
// (which we don't want to do until we get the notify_on_receive_settings
|
|
|
|
|
// callback from the transport). If we could split those things
|
|
|
|
|
// apart, then we could start sending RPCs but then wait for our
|
|
|
|
|
// timeout before deciding if the connection attempt is successful.
|
|
|
|
|
// If the attempt is not successful, then we would tear down the
|
|
|
|
|
// transport and feed the failure back into the backoff code.
|
|
|
|
|
//
|
|
|
|
|
// In addition, even if we did that, we would probably not want to do
|
|
|
|
|
// so until after transparent retries is implemented. Otherwise, any
|
|
|
|
|
// RPC that we attempt to send on the connection before the timeout
|
|
|
|
|
// would fail instead of being retried on a subsequent attempt.
|
|
|
|
|
grpc_chttp2_transport_start_reading(self->result_->transport, |
|
|
|
|
args->read_buffer, nullptr); |
|
|
|
|
self->result_->channel_args = args->args; |
|
|
|
|
} |
|
|
|
|
c->result->reset(); |
|
|
|
|
grpc_closure* notify = c->notify; |
|
|
|
|
c->notify = nullptr; |
|
|
|
|
grpc_core::ExecCtx::Run(DEBUG_LOCATION, notify, error); |
|
|
|
|
if (c->endpoint != nullptr) { |
|
|
|
|
grpc_endpoint_shutdown(c->endpoint, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
gpr_mu_unlock(&c->mu); |
|
|
|
|
chttp2_connector_unref(static_cast<grpc_connector*>(arg)); |
|
|
|
|
} else { |
|
|
|
|
GPR_ASSERT(c->endpoint != nullptr); |
|
|
|
|
start_handshake_locked(c); |
|
|
|
|
gpr_mu_unlock(&c->mu); |
|
|
|
|
grpc_closure* notify = self->notify_; |
|
|
|
|
self->notify_ = nullptr; |
|
|
|
|
ExecCtx::Run(DEBUG_LOCATION, notify, error); |
|
|
|
|
self->handshake_mgr_.reset(); |
|
|
|
|
} |
|
|
|
|
self->Unref(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void chttp2_connector_connect(grpc_connector* con, |
|
|
|
|
const grpc_connect_in_args* args, |
|
|
|
|
grpc_connect_out_args* result, |
|
|
|
|
grpc_closure* notify) { |
|
|
|
|
chttp2_connector* c = reinterpret_cast<chttp2_connector*>(con); |
|
|
|
|
grpc_resolved_address addr; |
|
|
|
|
grpc_core::Subchannel::GetAddressFromSubchannelAddressArg(args->channel_args, |
|
|
|
|
&addr); |
|
|
|
|
gpr_mu_lock(&c->mu); |
|
|
|
|
GPR_ASSERT(c->notify == nullptr); |
|
|
|
|
c->notify = notify; |
|
|
|
|
c->args = *args; |
|
|
|
|
c->result = result; |
|
|
|
|
GPR_ASSERT(c->endpoint == nullptr); |
|
|
|
|
chttp2_connector_ref(con); // Ref taken for callback.
|
|
|
|
|
GRPC_CLOSURE_INIT(&c->connected, connected, c, grpc_schedule_on_exec_ctx); |
|
|
|
|
GPR_ASSERT(!c->connecting); |
|
|
|
|
c->connecting = true; |
|
|
|
|
grpc_closure* closure = &c->connected; |
|
|
|
|
grpc_endpoint** ep = &c->endpoint; |
|
|
|
|
gpr_mu_unlock(&c->mu); |
|
|
|
|
// In some implementations, the closure can be flushed before
|
|
|
|
|
// grpc_tcp_client_connect and since the closure requires access to c->mu,
|
|
|
|
|
// this can result in a deadlock. Refer
|
|
|
|
|
// https://github.com/grpc/grpc/issues/16427
|
|
|
|
|
// grpc_tcp_client_connect would fill c->endpoint with proper contents and we
|
|
|
|
|
// make sure that we would still exist at that point by taking a ref.
|
|
|
|
|
grpc_tcp_client_connect(closure, ep, args->interested_parties, |
|
|
|
|
args->channel_args, &addr, args->deadline); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static const grpc_connector_vtable chttp2_connector_vtable = { |
|
|
|
|
chttp2_connector_ref, chttp2_connector_unref, chttp2_connector_shutdown, |
|
|
|
|
chttp2_connector_connect}; |
|
|
|
|
|
|
|
|
|
grpc_connector* grpc_chttp2_connector_create() { |
|
|
|
|
chttp2_connector* c = static_cast<chttp2_connector*>(gpr_zalloc(sizeof(*c))); |
|
|
|
|
c->base.vtable = &chttp2_connector_vtable; |
|
|
|
|
gpr_mu_init(&c->mu); |
|
|
|
|
gpr_ref_init(&c->refs, 1); |
|
|
|
|
return &c->base; |
|
|
|
|
} |
|
|
|
|
} // namespace grpc_core
|
|
|
|
|
|
Mark D. Roth
5 years ago
committed by
GitHub