|
|
|
|
@ -35,7 +35,6 @@ |
|
|
|
|
#include <grpc/support/time.h> |
|
|
|
|
|
|
|
|
|
#include "src/core/lib/address_utils/sockaddr_utils.h" |
|
|
|
|
#include "src/core/lib/channel/channel_args.h" |
|
|
|
|
#include "src/core/lib/gpr/string.h" |
|
|
|
|
#include "src/core/lib/iomgr/ev_posix.h" |
|
|
|
|
#include "src/core/lib/iomgr/executor.h" |
|
|
|
|
@ -51,6 +50,8 @@ |
|
|
|
|
|
|
|
|
|
extern grpc_core::TraceFlag grpc_tcp_trace; |
|
|
|
|
|
|
|
|
|
using ::grpc_event_engine::experimental::EndpointConfig; |
|
|
|
|
|
|
|
|
|
struct async_connect { |
|
|
|
|
gpr_mu mu; |
|
|
|
|
grpc_fd* fd; |
|
|
|
|
@ -62,9 +63,9 @@ struct async_connect { |
|
|
|
|
std::string addr_str; |
|
|
|
|
grpc_endpoint** ep; |
|
|
|
|
grpc_closure* closure; |
|
|
|
|
grpc_channel_args* channel_args; |
|
|
|
|
int64_t connection_handle; |
|
|
|
|
bool connect_cancelled; |
|
|
|
|
grpc_core::PosixTcpOptions options; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
struct ConnectionShard { |
|
|
|
|
@ -90,9 +91,9 @@ void grpc_tcp_client_global_init() { |
|
|
|
|
gpr_once_init(&g_tcp_client_posix_init, do_tcp_client_global_init); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static grpc_error_handle prepare_socket(const grpc_resolved_address* addr, |
|
|
|
|
int fd, |
|
|
|
|
const grpc_channel_args* channel_args) { |
|
|
|
|
static grpc_error_handle prepare_socket( |
|
|
|
|
const grpc_resolved_address* addr, int fd, |
|
|
|
|
const grpc_core::PosixTcpOptions& options) { |
|
|
|
|
grpc_error_handle err = GRPC_ERROR_NONE; |
|
|
|
|
|
|
|
|
|
GPR_ASSERT(fd >= 0); |
|
|
|
|
@ -106,15 +107,14 @@ static grpc_error_handle prepare_socket(const grpc_resolved_address* addr, |
|
|
|
|
if (!GRPC_ERROR_IS_NONE(err)) goto error; |
|
|
|
|
err = grpc_set_socket_reuse_addr(fd, 1); |
|
|
|
|
if (!GRPC_ERROR_IS_NONE(err)) goto error; |
|
|
|
|
err = grpc_set_socket_tcp_user_timeout(fd, channel_args, |
|
|
|
|
true /* is_client */); |
|
|
|
|
err = grpc_set_socket_tcp_user_timeout(fd, options, true /* is_client */); |
|
|
|
|
if (!GRPC_ERROR_IS_NONE(err)) goto error; |
|
|
|
|
} |
|
|
|
|
err = grpc_set_socket_no_sigpipe_if_possible(fd); |
|
|
|
|
if (!GRPC_ERROR_IS_NONE(err)) goto error; |
|
|
|
|
|
|
|
|
|
err = grpc_apply_socket_mutator_in_args(fd, GRPC_FD_CLIENT_CONNECTION_USAGE, |
|
|
|
|
channel_args); |
|
|
|
|
options); |
|
|
|
|
if (!GRPC_ERROR_IS_NONE(err)) goto error; |
|
|
|
|
|
|
|
|
|
goto done; |
|
|
|
|
@ -143,15 +143,20 @@ static void tc_on_alarm(void* acp, grpc_error_handle error) { |
|
|
|
|
gpr_mu_unlock(&ac->mu); |
|
|
|
|
if (done) { |
|
|
|
|
gpr_mu_destroy(&ac->mu); |
|
|
|
|
grpc_channel_args_destroy(ac->channel_args); |
|
|
|
|
delete ac; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_endpoint* grpc_tcp_client_create_from_fd( |
|
|
|
|
grpc_fd* fd, const grpc_channel_args* channel_args, |
|
|
|
|
static grpc_endpoint* grpc_tcp_client_create_from_fd( |
|
|
|
|
grpc_fd* fd, const grpc_core::PosixTcpOptions& options, |
|
|
|
|
absl::string_view addr_str) { |
|
|
|
|
return grpc_tcp_create(fd, options, addr_str); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_endpoint* grpc_tcp_create_from_fd( |
|
|
|
|
grpc_fd* fd, const grpc_event_engine::experimental::EndpointConfig& config, |
|
|
|
|
absl::string_view addr_str) { |
|
|
|
|
return grpc_tcp_create(fd, channel_args, addr_str); |
|
|
|
|
return grpc_tcp_create(fd, TcpOptionsFromEndpointConfig(config), addr_str); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_writable(void* acp, grpc_error_handle error) { |
|
|
|
|
@ -207,7 +212,7 @@ static void on_writable(void* acp, grpc_error_handle error) { |
|
|
|
|
switch (so_error) { |
|
|
|
|
case 0: |
|
|
|
|
grpc_pollset_set_del_fd(ac->interested_parties, fd); |
|
|
|
|
*ep = grpc_tcp_client_create_from_fd(fd, ac->channel_args, ac->addr_str); |
|
|
|
|
*ep = grpc_tcp_client_create_from_fd(fd, ac->options, ac->addr_str); |
|
|
|
|
fd = nullptr; |
|
|
|
|
break; |
|
|
|
|
case ENOBUFS: |
|
|
|
|
@ -269,7 +274,6 @@ finish: |
|
|
|
|
// This is safe even outside the lock, because "done", the sentinel, is
|
|
|
|
|
// populated *inside* the lock.
|
|
|
|
|
gpr_mu_destroy(&ac->mu); |
|
|
|
|
grpc_channel_args_destroy(ac->channel_args); |
|
|
|
|
delete ac; |
|
|
|
|
} |
|
|
|
|
// Push async connect closure to the executor since this may actually be
|
|
|
|
|
@ -284,8 +288,9 @@ finish: |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_error_handle grpc_tcp_client_prepare_fd( |
|
|
|
|
const grpc_channel_args* channel_args, const grpc_resolved_address* addr, |
|
|
|
|
grpc_resolved_address* mapped_addr, int* fd) { |
|
|
|
|
const grpc_core::PosixTcpOptions& options, |
|
|
|
|
const grpc_resolved_address* addr, grpc_resolved_address* mapped_addr, |
|
|
|
|
int* fd) { |
|
|
|
|
grpc_dualstack_mode dsmode; |
|
|
|
|
grpc_error_handle error; |
|
|
|
|
*fd = -1; |
|
|
|
|
@ -306,8 +311,7 @@ grpc_error_handle grpc_tcp_client_prepare_fd( |
|
|
|
|
memcpy(mapped_addr, addr, sizeof(*mapped_addr)); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
if ((error = prepare_socket(mapped_addr, *fd, channel_args)) != |
|
|
|
|
GRPC_ERROR_NONE) { |
|
|
|
|
if ((error = prepare_socket(mapped_addr, *fd, options)) != GRPC_ERROR_NONE) { |
|
|
|
|
return error; |
|
|
|
|
} |
|
|
|
|
return GRPC_ERROR_NONE; |
|
|
|
|
@ -315,8 +319,9 @@ grpc_error_handle grpc_tcp_client_prepare_fd( |
|
|
|
|
|
|
|
|
|
int64_t grpc_tcp_client_create_from_prepared_fd( |
|
|
|
|
grpc_pollset_set* interested_parties, grpc_closure* closure, const int fd, |
|
|
|
|
const grpc_channel_args* channel_args, const grpc_resolved_address* addr, |
|
|
|
|
grpc_core::Timestamp deadline, grpc_endpoint** ep) { |
|
|
|
|
const grpc_core::PosixTcpOptions& options, |
|
|
|
|
const grpc_resolved_address* addr, grpc_core::Timestamp deadline, |
|
|
|
|
grpc_endpoint** ep) { |
|
|
|
|
int err; |
|
|
|
|
do { |
|
|
|
|
err = connect(fd, reinterpret_cast<const grpc_sockaddr*>(addr->addr), |
|
|
|
|
@ -342,7 +347,7 @@ int64_t grpc_tcp_client_create_from_prepared_fd( |
|
|
|
|
if (err >= 0) { |
|
|
|
|
// Connection already succeded. Return 0 to discourage any cancellation
|
|
|
|
|
// attempts.
|
|
|
|
|
*ep = grpc_tcp_client_create_from_fd(fdobj, channel_args, addr_uri.value()); |
|
|
|
|
*ep = grpc_tcp_client_create_from_fd(fdobj, options, addr_uri.value()); |
|
|
|
|
grpc_core::ExecCtx::Run(DEBUG_LOCATION, closure, GRPC_ERROR_NONE); |
|
|
|
|
return 0; |
|
|
|
|
} |
|
|
|
|
@ -371,7 +376,7 @@ int64_t grpc_tcp_client_create_from_prepared_fd( |
|
|
|
|
ac->refs = 2; |
|
|
|
|
GRPC_CLOSURE_INIT(&ac->write_closure, on_writable, ac, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
ac->channel_args = grpc_channel_args_copy(channel_args); |
|
|
|
|
ac->options = options; |
|
|
|
|
|
|
|
|
|
if (GRPC_TRACE_FLAG_ENABLED(grpc_tcp_trace)) { |
|
|
|
|
gpr_log(GPR_INFO, "CLIENT_CONNECT: %s: asynchronously connecting fd %p", |
|
|
|
|
@ -395,21 +400,21 @@ int64_t grpc_tcp_client_create_from_prepared_fd( |
|
|
|
|
|
|
|
|
|
static int64_t tcp_connect(grpc_closure* closure, grpc_endpoint** ep, |
|
|
|
|
grpc_pollset_set* interested_parties, |
|
|
|
|
const grpc_channel_args* channel_args, |
|
|
|
|
const EndpointConfig& config, |
|
|
|
|
const grpc_resolved_address* addr, |
|
|
|
|
grpc_core::Timestamp deadline) { |
|
|
|
|
grpc_resolved_address mapped_addr; |
|
|
|
|
grpc_core::PosixTcpOptions options(TcpOptionsFromEndpointConfig(config)); |
|
|
|
|
int fd = -1; |
|
|
|
|
grpc_error_handle error; |
|
|
|
|
*ep = nullptr; |
|
|
|
|
if ((error = grpc_tcp_client_prepare_fd(channel_args, addr, &mapped_addr, |
|
|
|
|
&fd)) != GRPC_ERROR_NONE) { |
|
|
|
|
if ((error = grpc_tcp_client_prepare_fd(options, addr, &mapped_addr, &fd)) != |
|
|
|
|
GRPC_ERROR_NONE) { |
|
|
|
|
grpc_core::ExecCtx::Run(DEBUG_LOCATION, closure, error); |
|
|
|
|
return 0; |
|
|
|
|
} |
|
|
|
|
return grpc_tcp_client_create_from_prepared_fd(interested_parties, closure, |
|
|
|
|
fd, channel_args, &mapped_addr, |
|
|
|
|
deadline, ep); |
|
|
|
|
return grpc_tcp_client_create_from_prepared_fd( |
|
|
|
|
interested_parties, closure, fd, options, &mapped_addr, deadline, ep); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static bool tcp_cancel_connect(int64_t connection_handle) { |
|
|
|
|
@ -458,7 +463,6 @@ static bool tcp_cancel_connect(int64_t connection_handle) { |
|
|
|
|
// This is safe even outside the lock, because "done", the sentinel, is
|
|
|
|
|
// populated *inside* the lock.
|
|
|
|
|
gpr_mu_destroy(&ac->mu); |
|
|
|
|
grpc_channel_args_destroy(ac->channel_args); |
|
|
|
|
delete ac; |
|
|
|
|
} |
|
|
|
|
return connection_cancel_success; |
|
|
|
|
|
Vignesh Babu
2 years ago
committed by
GitHub