|
|
|
|
@ -39,6 +39,8 @@ |
|
|
|
|
#include "src/core/lib/channel/channel_args.h" |
|
|
|
|
#include "src/core/lib/channel/handshaker.h" |
|
|
|
|
#include "src/core/lib/channel/handshaker_registry.h" |
|
|
|
|
#include "src/core/lib/gprpp/ref_counted.h" |
|
|
|
|
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
|
|
|
|
#include "src/core/lib/iomgr/endpoint.h" |
|
|
|
|
#include "src/core/lib/iomgr/resolve_address.h" |
|
|
|
|
#include "src/core/lib/iomgr/resource_quota.h" |
|
|
|
|
@ -47,85 +49,159 @@ |
|
|
|
|
#include "src/core/lib/surface/api_trace.h" |
|
|
|
|
#include "src/core/lib/surface/server.h" |
|
|
|
|
|
|
|
|
|
struct server_state { |
|
|
|
|
grpc_server* server; |
|
|
|
|
grpc_tcp_server* tcp_server; |
|
|
|
|
grpc_channel_args* args; |
|
|
|
|
gpr_mu mu; |
|
|
|
|
bool shutdown; |
|
|
|
|
grpc_closure tcp_server_shutdown_complete; |
|
|
|
|
grpc_closure* server_destroy_listener_done; |
|
|
|
|
grpc_core::HandshakeManager* pending_handshake_mgrs; |
|
|
|
|
grpc_core::RefCountedPtr<grpc_core::channelz::ListenSocketNode> |
|
|
|
|
channelz_listen_socket; |
|
|
|
|
}; |
|
|
|
|
namespace grpc_core { |
|
|
|
|
namespace { |
|
|
|
|
|
|
|
|
|
class Chttp2ServerListener : public ServerListenerInterface { |
|
|
|
|
public: |
|
|
|
|
static grpc_error* Create(grpc_server* server, const char* addr, |
|
|
|
|
grpc_channel_args* args, int* port_num); |
|
|
|
|
|
|
|
|
|
static grpc_error* CreateWithAcceptor(grpc_server* server, const char* name, |
|
|
|
|
grpc_channel_args* args); |
|
|
|
|
|
|
|
|
|
// Do not instantiate directly. Use one of the factory methods above.
|
|
|
|
|
Chttp2ServerListener(grpc_server* server, grpc_channel_args* args); |
|
|
|
|
~Chttp2ServerListener(); |
|
|
|
|
|
|
|
|
|
void Start(grpc_server* server, grpc_pollset** pollsets, |
|
|
|
|
size_t npollsets) override; |
|
|
|
|
|
|
|
|
|
channelz::ListenSocketNode* channelz_listen_socket_node() const override { |
|
|
|
|
return channelz_listen_socket_.get(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void SetOnDestroyDone(grpc_closure* on_destroy_done) override; |
|
|
|
|
|
|
|
|
|
void Orphan() override; |
|
|
|
|
|
|
|
|
|
struct server_connection_state { |
|
|
|
|
gpr_refcount refs; |
|
|
|
|
server_state* svr_state; |
|
|
|
|
grpc_pollset* accepting_pollset; |
|
|
|
|
grpc_tcp_server_acceptor* acceptor; |
|
|
|
|
grpc_core::RefCountedPtr<grpc_core::HandshakeManager> handshake_mgr; |
|
|
|
|
private: |
|
|
|
|
class ConnectionState : public RefCounted<ConnectionState> { |
|
|
|
|
public: |
|
|
|
|
ConnectionState(Chttp2ServerListener* listener, |
|
|
|
|
grpc_pollset* accepting_pollset, |
|
|
|
|
grpc_tcp_server_acceptor* acceptor, |
|
|
|
|
RefCountedPtr<HandshakeManager> handshake_mgr, |
|
|
|
|
grpc_channel_args* args, grpc_endpoint* endpoint); |
|
|
|
|
|
|
|
|
|
~ConnectionState(); |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
static void OnTimeout(void* arg, grpc_error* error); |
|
|
|
|
static void OnReceiveSettings(void* arg, grpc_error* error); |
|
|
|
|
static void OnHandshakeDone(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
Chttp2ServerListener* const listener_; |
|
|
|
|
grpc_pollset* const accepting_pollset_; |
|
|
|
|
grpc_tcp_server_acceptor* const acceptor_; |
|
|
|
|
RefCountedPtr<HandshakeManager> handshake_mgr_; |
|
|
|
|
// State for enforcing handshake timeout on receiving HTTP/2 settings.
|
|
|
|
|
grpc_chttp2_transport* transport; |
|
|
|
|
grpc_millis deadline; |
|
|
|
|
grpc_timer timer; |
|
|
|
|
grpc_closure on_timeout; |
|
|
|
|
grpc_closure on_receive_settings; |
|
|
|
|
grpc_pollset_set* interested_parties; |
|
|
|
|
grpc_chttp2_transport* transport_ = nullptr; |
|
|
|
|
grpc_millis deadline_; |
|
|
|
|
grpc_timer timer_; |
|
|
|
|
grpc_closure on_timeout_; |
|
|
|
|
grpc_closure on_receive_settings_; |
|
|
|
|
grpc_pollset_set* const interested_parties_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
static void server_connection_state_unref( |
|
|
|
|
server_connection_state* connection_state) { |
|
|
|
|
if (gpr_unref(&connection_state->refs)) { |
|
|
|
|
if (connection_state->transport != nullptr) { |
|
|
|
|
GRPC_CHTTP2_UNREF_TRANSPORT(connection_state->transport, |
|
|
|
|
"receive settings timeout"); |
|
|
|
|
static void OnAccept(void* arg, grpc_endpoint* tcp, |
|
|
|
|
grpc_pollset* accepting_pollset, |
|
|
|
|
grpc_tcp_server_acceptor* acceptor); |
|
|
|
|
|
|
|
|
|
RefCountedPtr<HandshakeManager> CreateHandshakeManager(); |
|
|
|
|
|
|
|
|
|
static void TcpServerShutdownComplete(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
static void DestroyListener(grpc_server* /*server*/, void* arg, |
|
|
|
|
grpc_closure* destroy_done); |
|
|
|
|
|
|
|
|
|
grpc_server* const server_; |
|
|
|
|
grpc_channel_args* const args_; |
|
|
|
|
grpc_tcp_server* tcp_server_; |
|
|
|
|
Mutex mu_; |
|
|
|
|
bool shutdown_ = true; |
|
|
|
|
grpc_closure tcp_server_shutdown_complete_; |
|
|
|
|
grpc_closure* on_destroy_done_ = nullptr; |
|
|
|
|
HandshakeManager* pending_handshake_mgrs_ = nullptr; |
|
|
|
|
RefCountedPtr<channelz::ListenSocketNode> channelz_listen_socket_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
// Chttp2ServerListener::ConnectionState
|
|
|
|
|
//
|
|
|
|
|
|
|
|
|
|
grpc_millis GetConnectionDeadline(const grpc_channel_args* args) { |
|
|
|
|
int timeout_ms = |
|
|
|
|
grpc_channel_args_find_integer(args, GRPC_ARG_SERVER_HANDSHAKE_TIMEOUT_MS, |
|
|
|
|
{120 * GPR_MS_PER_SEC, 1, INT_MAX}); |
|
|
|
|
return ExecCtx::Get()->Now() + timeout_ms; |
|
|
|
|
} |
|
|
|
|
grpc_pollset_set_del_pollset(connection_state->interested_parties, |
|
|
|
|
connection_state->accepting_pollset); |
|
|
|
|
grpc_pollset_set_destroy(connection_state->interested_parties); |
|
|
|
|
gpr_free(connection_state); |
|
|
|
|
|
|
|
|
|
Chttp2ServerListener::ConnectionState::ConnectionState( |
|
|
|
|
Chttp2ServerListener* listener, grpc_pollset* accepting_pollset, |
|
|
|
|
grpc_tcp_server_acceptor* acceptor, |
|
|
|
|
RefCountedPtr<HandshakeManager> handshake_mgr, grpc_channel_args* args, |
|
|
|
|
grpc_endpoint* endpoint) |
|
|
|
|
: listener_(listener), |
|
|
|
|
accepting_pollset_(accepting_pollset), |
|
|
|
|
acceptor_(acceptor), |
|
|
|
|
handshake_mgr_(std::move(handshake_mgr)), |
|
|
|
|
deadline_(GetConnectionDeadline(args)), |
|
|
|
|
interested_parties_(grpc_pollset_set_create()) { |
|
|
|
|
grpc_pollset_set_add_pollset(interested_parties_, accepting_pollset_); |
|
|
|
|
HandshakerRegistry::AddHandshakers(HANDSHAKER_SERVER, args, |
|
|
|
|
interested_parties_, handshake_mgr_.get()); |
|
|
|
|
handshake_mgr_->DoHandshake(endpoint, args, deadline_, acceptor_, |
|
|
|
|
OnHandshakeDone, this); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
Chttp2ServerListener::ConnectionState::~ConnectionState() { |
|
|
|
|
if (transport_ != nullptr) { |
|
|
|
|
GRPC_CHTTP2_UNREF_TRANSPORT(transport_, "receive settings timeout"); |
|
|
|
|
} |
|
|
|
|
grpc_pollset_set_del_pollset(interested_parties_, accepting_pollset_); |
|
|
|
|
grpc_pollset_set_destroy(interested_parties_); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_timeout(void* arg, grpc_error* error) { |
|
|
|
|
server_connection_state* connection_state = |
|
|
|
|
static_cast<server_connection_state*>(arg); |
|
|
|
|
void Chttp2ServerListener::ConnectionState::OnTimeout(void* arg, |
|
|
|
|
grpc_error* error) { |
|
|
|
|
ConnectionState* self = static_cast<ConnectionState*>(arg); |
|
|
|
|
// Note that we may be called with GRPC_ERROR_NONE when the timer fires
|
|
|
|
|
// or with an error indicating that the timer system is being shut down.
|
|
|
|
|
if (error != GRPC_ERROR_CANCELLED) { |
|
|
|
|
grpc_transport_op* op = grpc_make_transport_op(nullptr); |
|
|
|
|
op->disconnect_with_error = GRPC_ERROR_CREATE_FROM_STATIC_STRING( |
|
|
|
|
"Did not receive HTTP/2 settings before handshake timeout"); |
|
|
|
|
grpc_transport_perform_op(&connection_state->transport->base, op); |
|
|
|
|
grpc_transport_perform_op(&self->transport_->base, op); |
|
|
|
|
} |
|
|
|
|
server_connection_state_unref(connection_state); |
|
|
|
|
self->Unref(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_receive_settings(void* arg, grpc_error* error) { |
|
|
|
|
server_connection_state* connection_state = |
|
|
|
|
static_cast<server_connection_state*>(arg); |
|
|
|
|
void Chttp2ServerListener::ConnectionState::OnReceiveSettings( |
|
|
|
|
void* arg, grpc_error* error) { |
|
|
|
|
ConnectionState* self = static_cast<ConnectionState*>(arg); |
|
|
|
|
if (error == GRPC_ERROR_NONE) { |
|
|
|
|
grpc_timer_cancel(&connection_state->timer); |
|
|
|
|
grpc_timer_cancel(&self->timer_); |
|
|
|
|
} |
|
|
|
|
server_connection_state_unref(connection_state); |
|
|
|
|
self->Unref(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_handshake_done(void* arg, grpc_error* error) { |
|
|
|
|
auto* args = static_cast<grpc_core::HandshakerArgs*>(arg); |
|
|
|
|
server_connection_state* connection_state = |
|
|
|
|
static_cast<server_connection_state*>(args->user_data); |
|
|
|
|
gpr_mu_lock(&connection_state->svr_state->mu); |
|
|
|
|
grpc_resource_user* resource_user = grpc_server_get_default_resource_user( |
|
|
|
|
connection_state->svr_state->server); |
|
|
|
|
if (error != GRPC_ERROR_NONE || connection_state->svr_state->shutdown) { |
|
|
|
|
void Chttp2ServerListener::ConnectionState::OnHandshakeDone(void* arg, |
|
|
|
|
grpc_error* error) { |
|
|
|
|
auto* args = static_cast<HandshakerArgs*>(arg); |
|
|
|
|
ConnectionState* self = static_cast<ConnectionState*>(args->user_data); |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&self->listener_->mu_); |
|
|
|
|
grpc_resource_user* resource_user = |
|
|
|
|
grpc_server_get_default_resource_user(self->listener_->server_); |
|
|
|
|
if (error != GRPC_ERROR_NONE || self->listener_->shutdown_) { |
|
|
|
|
const char* error_str = grpc_error_string(error); |
|
|
|
|
gpr_log(GPR_DEBUG, "Handshaking failed: %s", error_str); |
|
|
|
|
grpc_resource_user* resource_user = grpc_server_get_default_resource_user( |
|
|
|
|
connection_state->svr_state->server); |
|
|
|
|
grpc_resource_user* resource_user = |
|
|
|
|
grpc_server_get_default_resource_user(self->listener_->server_); |
|
|
|
|
if (resource_user != nullptr) { |
|
|
|
|
grpc_resource_user_free(resource_user, GRPC_RESOURCE_QUOTA_CHANNEL_SIZE); |
|
|
|
|
grpc_resource_user_free(resource_user, |
|
|
|
|
GRPC_RESOURCE_QUOTA_CHANNEL_SIZE); |
|
|
|
|
} |
|
|
|
|
if (error == GRPC_ERROR_NONE && args->endpoint != nullptr) { |
|
|
|
|
// We were shut down after handshaking completed successfully, so
|
|
|
|
|
@ -148,27 +224,31 @@ static void on_handshake_done(void* arg, grpc_error* error) { |
|
|
|
|
grpc_transport* transport = grpc_create_chttp2_transport( |
|
|
|
|
args->args, args->endpoint, false, resource_user); |
|
|
|
|
grpc_server_setup_transport( |
|
|
|
|
connection_state->svr_state->server, transport, |
|
|
|
|
connection_state->accepting_pollset, args->args, |
|
|
|
|
grpc_chttp2_transport_get_socket_node(transport), resource_user); |
|
|
|
|
self->listener_->server_, transport, self->accepting_pollset_, |
|
|
|
|
args->args, grpc_chttp2_transport_get_socket_node(transport), |
|
|
|
|
resource_user); |
|
|
|
|
// Use notify_on_receive_settings callback to enforce the
|
|
|
|
|
// handshake deadline.
|
|
|
|
|
connection_state->transport = |
|
|
|
|
reinterpret_cast<grpc_chttp2_transport*>(transport); |
|
|
|
|
gpr_ref(&connection_state->refs); |
|
|
|
|
GRPC_CLOSURE_INIT(&connection_state->on_receive_settings, |
|
|
|
|
on_receive_settings, connection_state, |
|
|
|
|
// Note: The reinterpret_cast<>s here are safe, because
|
|
|
|
|
// grpc_chttp2_transport is a C-style extension of
|
|
|
|
|
// grpc_transport, so this is morally equivalent of a
|
|
|
|
|
// static_cast<> to a derived class.
|
|
|
|
|
// TODO(roth): Change to static_cast<> when we C++-ify the
|
|
|
|
|
// transport API.
|
|
|
|
|
self->transport_ = reinterpret_cast<grpc_chttp2_transport*>(transport); |
|
|
|
|
self->Ref().release(); // Held by OnReceiveSettings().
|
|
|
|
|
GRPC_CLOSURE_INIT(&self->on_receive_settings_, OnReceiveSettings, self, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
grpc_chttp2_transport_start_reading( |
|
|
|
|
transport, args->read_buffer, &connection_state->on_receive_settings); |
|
|
|
|
grpc_chttp2_transport_start_reading(transport, args->read_buffer, |
|
|
|
|
&self->on_receive_settings_); |
|
|
|
|
grpc_channel_args_destroy(args->args); |
|
|
|
|
gpr_ref(&connection_state->refs); |
|
|
|
|
GRPC_CHTTP2_REF_TRANSPORT((grpc_chttp2_transport*)transport, |
|
|
|
|
self->Ref().release(); // Held by OnTimeout().
|
|
|
|
|
GRPC_CHTTP2_REF_TRANSPORT( |
|
|
|
|
reinterpret_cast<grpc_chttp2_transport*>(transport), |
|
|
|
|
"receive settings timeout"); |
|
|
|
|
GRPC_CLOSURE_INIT(&connection_state->on_timeout, on_timeout, |
|
|
|
|
connection_state, grpc_schedule_on_exec_ctx); |
|
|
|
|
grpc_timer_init(&connection_state->timer, connection_state->deadline, |
|
|
|
|
&connection_state->on_timeout); |
|
|
|
|
GRPC_CLOSURE_INIT(&self->on_timeout_, OnTimeout, self, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
grpc_timer_init(&self->timer_, self->deadline_, &self->on_timeout_); |
|
|
|
|
} else { |
|
|
|
|
if (resource_user != nullptr) { |
|
|
|
|
grpc_resource_user_free(resource_user, |
|
|
|
|
@ -176,276 +256,232 @@ static void on_handshake_done(void* arg, grpc_error* error) { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
connection_state->handshake_mgr->RemoveFromPendingMgrList( |
|
|
|
|
&connection_state->svr_state->pending_handshake_mgrs); |
|
|
|
|
gpr_mu_unlock(&connection_state->svr_state->mu); |
|
|
|
|
connection_state->handshake_mgr.reset(); |
|
|
|
|
gpr_free(connection_state->acceptor); |
|
|
|
|
grpc_tcp_server_unref(connection_state->svr_state->tcp_server); |
|
|
|
|
server_connection_state_unref(connection_state); |
|
|
|
|
self->handshake_mgr_->RemoveFromPendingMgrList( |
|
|
|
|
&self->listener_->pending_handshake_mgrs_); |
|
|
|
|
} |
|
|
|
|
self->handshake_mgr_.reset(); |
|
|
|
|
gpr_free(self->acceptor_); |
|
|
|
|
grpc_tcp_server_unref(self->listener_->tcp_server_); |
|
|
|
|
self->Unref(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_accept(void* arg, grpc_endpoint* tcp, |
|
|
|
|
grpc_pollset* accepting_pollset, |
|
|
|
|
grpc_tcp_server_acceptor* acceptor) { |
|
|
|
|
server_state* state = static_cast<server_state*>(arg); |
|
|
|
|
gpr_mu_lock(&state->mu); |
|
|
|
|
if (state->shutdown) { |
|
|
|
|
gpr_mu_unlock(&state->mu); |
|
|
|
|
grpc_endpoint_shutdown(tcp, GRPC_ERROR_NONE); |
|
|
|
|
grpc_endpoint_destroy(tcp); |
|
|
|
|
gpr_free(acceptor); |
|
|
|
|
return; |
|
|
|
|
//
|
|
|
|
|
// Chttp2ServerListener
|
|
|
|
|
//
|
|
|
|
|
|
|
|
|
|
grpc_error* Chttp2ServerListener::Create(grpc_server* server, const char* addr, |
|
|
|
|
grpc_channel_args* args, |
|
|
|
|
int* port_num) { |
|
|
|
|
std::vector<grpc_error*> error_list; |
|
|
|
|
grpc_resolved_addresses* resolved = nullptr; |
|
|
|
|
Chttp2ServerListener* listener = nullptr; |
|
|
|
|
// The bulk of this method is inside of a lambda to make cleanup
|
|
|
|
|
// easier without using goto.
|
|
|
|
|
grpc_error* error = [&]() { |
|
|
|
|
*port_num = -1; |
|
|
|
|
/* resolve address */ |
|
|
|
|
grpc_error* error = grpc_blocking_resolve_address(addr, "https", &resolved); |
|
|
|
|
if (error != GRPC_ERROR_NONE) return error; |
|
|
|
|
// Create Chttp2ServerListener.
|
|
|
|
|
listener = new Chttp2ServerListener(server, args); |
|
|
|
|
error = grpc_tcp_server_create(&listener->tcp_server_shutdown_complete_, |
|
|
|
|
args, &listener->tcp_server_); |
|
|
|
|
if (error != GRPC_ERROR_NONE) return error; |
|
|
|
|
for (size_t i = 0; i < resolved->naddrs; i++) { |
|
|
|
|
int port_temp; |
|
|
|
|
error = grpc_tcp_server_add_port(listener->tcp_server_, |
|
|
|
|
&resolved->addrs[i], &port_temp); |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
error_list.push_back(error); |
|
|
|
|
} else { |
|
|
|
|
if (*port_num == -1) { |
|
|
|
|
*port_num = port_temp; |
|
|
|
|
} else { |
|
|
|
|
GPR_ASSERT(*port_num == port_temp); |
|
|
|
|
} |
|
|
|
|
grpc_resource_user* resource_user = |
|
|
|
|
grpc_server_get_default_resource_user(state->server); |
|
|
|
|
if (resource_user != nullptr && |
|
|
|
|
!grpc_resource_user_safe_alloc(resource_user, |
|
|
|
|
GRPC_RESOURCE_QUOTA_CHANNEL_SIZE)) { |
|
|
|
|
gpr_log( |
|
|
|
|
GPR_ERROR, |
|
|
|
|
"Memory quota exhausted, rejecting the connection, no handshaking."); |
|
|
|
|
gpr_mu_unlock(&state->mu); |
|
|
|
|
grpc_endpoint_shutdown(tcp, GRPC_ERROR_NONE); |
|
|
|
|
grpc_endpoint_destroy(tcp); |
|
|
|
|
gpr_free(acceptor); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
auto handshake_mgr = grpc_core::MakeRefCounted<grpc_core::HandshakeManager>(); |
|
|
|
|
handshake_mgr->AddToPendingMgrList(&state->pending_handshake_mgrs); |
|
|
|
|
grpc_tcp_server_ref(state->tcp_server); |
|
|
|
|
gpr_mu_unlock(&state->mu); |
|
|
|
|
server_connection_state* connection_state = |
|
|
|
|
static_cast<server_connection_state*>( |
|
|
|
|
gpr_zalloc(sizeof(*connection_state))); |
|
|
|
|
gpr_ref_init(&connection_state->refs, 1); |
|
|
|
|
connection_state->svr_state = state; |
|
|
|
|
connection_state->accepting_pollset = accepting_pollset; |
|
|
|
|
connection_state->acceptor = acceptor; |
|
|
|
|
connection_state->handshake_mgr = handshake_mgr; |
|
|
|
|
connection_state->interested_parties = grpc_pollset_set_create(); |
|
|
|
|
grpc_pollset_set_add_pollset(connection_state->interested_parties, |
|
|
|
|
connection_state->accepting_pollset); |
|
|
|
|
grpc_core::HandshakerRegistry::AddHandshakers( |
|
|
|
|
grpc_core::HANDSHAKER_SERVER, state->args, |
|
|
|
|
connection_state->interested_parties, |
|
|
|
|
connection_state->handshake_mgr.get()); |
|
|
|
|
const grpc_arg* timeout_arg = |
|
|
|
|
grpc_channel_args_find(state->args, GRPC_ARG_SERVER_HANDSHAKE_TIMEOUT_MS); |
|
|
|
|
connection_state->deadline = |
|
|
|
|
grpc_core::ExecCtx::Get()->Now() + |
|
|
|
|
grpc_channel_arg_get_integer(timeout_arg, |
|
|
|
|
{120 * GPR_MS_PER_SEC, 1, INT_MAX}); |
|
|
|
|
connection_state->handshake_mgr->DoHandshake( |
|
|
|
|
tcp, state->args, connection_state->deadline, acceptor, on_handshake_done, |
|
|
|
|
connection_state); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/* Server callback: start listening on our ports */ |
|
|
|
|
static void server_start_listener(grpc_server* /*server*/, void* arg, |
|
|
|
|
grpc_pollset** pollsets, |
|
|
|
|
size_t pollset_count) { |
|
|
|
|
server_state* state = static_cast<server_state*>(arg); |
|
|
|
|
gpr_mu_lock(&state->mu); |
|
|
|
|
state->shutdown = false; |
|
|
|
|
gpr_mu_unlock(&state->mu); |
|
|
|
|
grpc_tcp_server_start(state->tcp_server, pollsets, pollset_count, on_accept, |
|
|
|
|
state); |
|
|
|
|
if (error_list.size() == resolved->naddrs) { |
|
|
|
|
std::string msg = |
|
|
|
|
absl::StrFormat("No address added out of total %" PRIuPTR " resolved", |
|
|
|
|
resolved->naddrs); |
|
|
|
|
return GRPC_ERROR_CREATE_REFERENCING_FROM_COPIED_STRING( |
|
|
|
|
msg.c_str(), error_list.data(), error_list.size()); |
|
|
|
|
} else if (!error_list.empty()) { |
|
|
|
|
std::string msg = absl::StrFormat( |
|
|
|
|
"Only %" PRIuPTR " addresses added out of total %" PRIuPTR |
|
|
|
|
" resolved", |
|
|
|
|
resolved->naddrs - error_list.size(), resolved->naddrs); |
|
|
|
|
error = GRPC_ERROR_CREATE_REFERENCING_FROM_COPIED_STRING( |
|
|
|
|
msg.c_str(), error_list.data(), error_list.size()); |
|
|
|
|
gpr_log(GPR_INFO, "WARNING: %s", grpc_error_string(error)); |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
/* we managed to bind some addresses: continue */ |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void tcp_server_shutdown_complete(void* arg, grpc_error* error) { |
|
|
|
|
server_state* state = static_cast<server_state*>(arg); |
|
|
|
|
/* ensure all threads have unlocked */ |
|
|
|
|
gpr_mu_lock(&state->mu); |
|
|
|
|
grpc_closure* destroy_done = state->server_destroy_listener_done; |
|
|
|
|
GPR_ASSERT(state->shutdown); |
|
|
|
|
if (state->pending_handshake_mgrs != nullptr) { |
|
|
|
|
state->pending_handshake_mgrs->ShutdownAllPending(GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
state->channelz_listen_socket.reset(); |
|
|
|
|
gpr_mu_unlock(&state->mu); |
|
|
|
|
// Flush queued work before destroying handshaker factory, since that
|
|
|
|
|
// may do a synchronous unref.
|
|
|
|
|
grpc_core::ExecCtx::Get()->Flush(); |
|
|
|
|
if (destroy_done != nullptr) { |
|
|
|
|
grpc_core::ExecCtx::Run(DEBUG_LOCATION, destroy_done, |
|
|
|
|
GRPC_ERROR_REF(error)); |
|
|
|
|
grpc_core::ExecCtx::Get()->Flush(); |
|
|
|
|
// Create channelz node.
|
|
|
|
|
if (grpc_channel_args_find_bool(args, GRPC_ARG_ENABLE_CHANNELZ, |
|
|
|
|
GRPC_ENABLE_CHANNELZ_DEFAULT)) { |
|
|
|
|
listener->channelz_listen_socket_ = |
|
|
|
|
MakeRefCounted<channelz::ListenSocketNode>( |
|
|
|
|
addr, absl::StrFormat("chttp2 listener %s", addr)); |
|
|
|
|
} |
|
|
|
|
/* Register with the server only upon success */ |
|
|
|
|
grpc_server_add_listener(server, |
|
|
|
|
OrphanablePtr<ServerListenerInterface>(listener)); |
|
|
|
|
return GRPC_ERROR_NONE; |
|
|
|
|
}(); |
|
|
|
|
if (resolved != nullptr) { |
|
|
|
|
grpc_resolved_addresses_destroy(resolved); |
|
|
|
|
} |
|
|
|
|
grpc_channel_args_destroy(state->args); |
|
|
|
|
gpr_mu_destroy(&state->mu); |
|
|
|
|
gpr_free(state); |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
if (listener != nullptr) { |
|
|
|
|
if (listener->tcp_server_ != nullptr) { |
|
|
|
|
grpc_tcp_server_unref(listener->tcp_server_); |
|
|
|
|
} else { |
|
|
|
|
delete listener; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/* Server callback: destroy the tcp listener (so we don't generate further
|
|
|
|
|
callbacks) */ |
|
|
|
|
static void server_destroy_listener(grpc_server* /*server*/, void* arg, |
|
|
|
|
grpc_closure* destroy_done) { |
|
|
|
|
server_state* state = static_cast<server_state*>(arg); |
|
|
|
|
gpr_mu_lock(&state->mu); |
|
|
|
|
state->shutdown = true; |
|
|
|
|
state->server_destroy_listener_done = destroy_done; |
|
|
|
|
grpc_tcp_server* tcp_server = state->tcp_server; |
|
|
|
|
gpr_mu_unlock(&state->mu); |
|
|
|
|
grpc_tcp_server_shutdown_listeners(tcp_server); |
|
|
|
|
grpc_tcp_server_unref(tcp_server); |
|
|
|
|
} else { |
|
|
|
|
grpc_channel_args_destroy(args); |
|
|
|
|
} |
|
|
|
|
*port_num = 0; |
|
|
|
|
} |
|
|
|
|
for (grpc_error* error : error_list) { |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
} |
|
|
|
|
return error; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static grpc_error* chttp2_server_add_acceptor(grpc_server* server, |
|
|
|
|
grpc_error* Chttp2ServerListener::CreateWithAcceptor(grpc_server* server, |
|
|
|
|
const char* name, |
|
|
|
|
grpc_channel_args* args) { |
|
|
|
|
grpc_tcp_server* tcp_server = nullptr; |
|
|
|
|
grpc_error* err = GRPC_ERROR_NONE; |
|
|
|
|
server_state* state = nullptr; |
|
|
|
|
const grpc_arg* arg = nullptr; |
|
|
|
|
grpc_core::TcpServerFdHandler** arg_val = nullptr; |
|
|
|
|
state = static_cast<server_state*>(gpr_zalloc(sizeof(*state))); |
|
|
|
|
GRPC_CLOSURE_INIT(&state->tcp_server_shutdown_complete, |
|
|
|
|
tcp_server_shutdown_complete, state, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
err = grpc_tcp_server_create(&state->tcp_server_shutdown_complete, args, |
|
|
|
|
&tcp_server); |
|
|
|
|
if (err != GRPC_ERROR_NONE) { |
|
|
|
|
goto error; |
|
|
|
|
} |
|
|
|
|
state->server = server; |
|
|
|
|
state->tcp_server = tcp_server; |
|
|
|
|
state->args = args; |
|
|
|
|
state->shutdown = true; |
|
|
|
|
gpr_mu_init(&state->mu); |
|
|
|
|
Chttp2ServerListener* listener = new Chttp2ServerListener(server, args); |
|
|
|
|
grpc_error* error = grpc_tcp_server_create( |
|
|
|
|
&listener->tcp_server_shutdown_complete_, args, &listener->tcp_server_); |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
delete listener; |
|
|
|
|
return error; |
|
|
|
|
} |
|
|
|
|
// TODO(yangg) channelz
|
|
|
|
|
arg = grpc_channel_args_find(args, name); |
|
|
|
|
GPR_ASSERT(arg->type == GRPC_ARG_POINTER); |
|
|
|
|
arg_val = static_cast<grpc_core::TcpServerFdHandler**>(arg->value.pointer.p); |
|
|
|
|
*arg_val = grpc_tcp_server_create_fd_handler(tcp_server); |
|
|
|
|
|
|
|
|
|
grpc_server_add_listener(server, state, server_start_listener, |
|
|
|
|
server_destroy_listener, /* node */ nullptr); |
|
|
|
|
return err; |
|
|
|
|
|
|
|
|
|
/* Error path: cleanup and return */ |
|
|
|
|
error: |
|
|
|
|
GPR_ASSERT(err != GRPC_ERROR_NONE); |
|
|
|
|
if (tcp_server) { |
|
|
|
|
grpc_tcp_server_unref(tcp_server); |
|
|
|
|
} else { |
|
|
|
|
grpc_channel_args_destroy(args); |
|
|
|
|
gpr_free(state); |
|
|
|
|
TcpServerFdHandler** arg_val = |
|
|
|
|
grpc_channel_args_find_pointer<TcpServerFdHandler*>(args, name); |
|
|
|
|
*arg_val = grpc_tcp_server_create_fd_handler(listener->tcp_server_); |
|
|
|
|
grpc_server_add_listener(server, |
|
|
|
|
OrphanablePtr<ServerListenerInterface>(listener)); |
|
|
|
|
return GRPC_ERROR_NONE; |
|
|
|
|
} |
|
|
|
|
return err; |
|
|
|
|
|
|
|
|
|
Chttp2ServerListener::Chttp2ServerListener(grpc_server* server, |
|
|
|
|
grpc_channel_args* args) |
|
|
|
|
: server_(server), args_(args) { |
|
|
|
|
GRPC_CLOSURE_INIT(&tcp_server_shutdown_complete_, TcpServerShutdownComplete, |
|
|
|
|
this, grpc_schedule_on_exec_ctx); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_error* grpc_chttp2_server_add_port(grpc_server* server, const char* addr, |
|
|
|
|
grpc_channel_args* args, |
|
|
|
|
int* port_num) { |
|
|
|
|
grpc_resolved_addresses* resolved = nullptr; |
|
|
|
|
grpc_tcp_server* tcp_server = nullptr; |
|
|
|
|
size_t i; |
|
|
|
|
size_t count = 0; |
|
|
|
|
int port_temp; |
|
|
|
|
grpc_error* err = GRPC_ERROR_NONE; |
|
|
|
|
server_state* state = nullptr; |
|
|
|
|
grpc_error** errors = nullptr; |
|
|
|
|
size_t naddrs = 0; |
|
|
|
|
const grpc_arg* arg = nullptr; |
|
|
|
|
Chttp2ServerListener::~Chttp2ServerListener() { |
|
|
|
|
grpc_channel_args_destroy(args_); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
*port_num = -1; |
|
|
|
|
/* Server callback: start listening on our ports */ |
|
|
|
|
void Chttp2ServerListener::Start(grpc_server* /*server*/, |
|
|
|
|
grpc_pollset** pollsets, |
|
|
|
|
size_t pollset_count) { |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&mu_); |
|
|
|
|
shutdown_ = false; |
|
|
|
|
} |
|
|
|
|
grpc_tcp_server_start(tcp_server_, pollsets, pollset_count, OnAccept, this); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if (strncmp(addr, "external:", 9) == 0) { |
|
|
|
|
return chttp2_server_add_acceptor(server, addr, args); |
|
|
|
|
void Chttp2ServerListener::SetOnDestroyDone(grpc_closure* on_destroy_done) { |
|
|
|
|
MutexLock lock(&mu_); |
|
|
|
|
on_destroy_done_ = on_destroy_done; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/* resolve address */ |
|
|
|
|
err = grpc_blocking_resolve_address(addr, "https", &resolved); |
|
|
|
|
if (err != GRPC_ERROR_NONE) { |
|
|
|
|
goto error; |
|
|
|
|
RefCountedPtr<HandshakeManager> Chttp2ServerListener::CreateHandshakeManager() { |
|
|
|
|
MutexLock lock(&mu_); |
|
|
|
|
if (shutdown_) return nullptr; |
|
|
|
|
grpc_resource_user* resource_user = |
|
|
|
|
grpc_server_get_default_resource_user(server_); |
|
|
|
|
if (resource_user != nullptr && |
|
|
|
|
!grpc_resource_user_safe_alloc(resource_user, |
|
|
|
|
GRPC_RESOURCE_QUOTA_CHANNEL_SIZE)) { |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"Memory quota exhausted, rejecting connection, no handshaking."); |
|
|
|
|
return nullptr; |
|
|
|
|
} |
|
|
|
|
state = static_cast<server_state*>(gpr_zalloc(sizeof(*state))); |
|
|
|
|
GRPC_CLOSURE_INIT(&state->tcp_server_shutdown_complete, |
|
|
|
|
tcp_server_shutdown_complete, state, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
err = grpc_tcp_server_create(&state->tcp_server_shutdown_complete, args, |
|
|
|
|
&tcp_server); |
|
|
|
|
if (err != GRPC_ERROR_NONE) { |
|
|
|
|
goto error; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
state->server = server; |
|
|
|
|
state->tcp_server = tcp_server; |
|
|
|
|
state->args = args; |
|
|
|
|
state->shutdown = true; |
|
|
|
|
gpr_mu_init(&state->mu); |
|
|
|
|
|
|
|
|
|
naddrs = resolved->naddrs; |
|
|
|
|
errors = static_cast<grpc_error**>(gpr_malloc(sizeof(*errors) * naddrs)); |
|
|
|
|
for (i = 0; i < naddrs; i++) { |
|
|
|
|
errors[i] = |
|
|
|
|
grpc_tcp_server_add_port(tcp_server, &resolved->addrs[i], &port_temp); |
|
|
|
|
if (errors[i] == GRPC_ERROR_NONE) { |
|
|
|
|
if (*port_num == -1) { |
|
|
|
|
*port_num = port_temp; |
|
|
|
|
} else { |
|
|
|
|
GPR_ASSERT(*port_num == port_temp); |
|
|
|
|
auto handshake_mgr = MakeRefCounted<HandshakeManager>(); |
|
|
|
|
handshake_mgr->AddToPendingMgrList(&pending_handshake_mgrs_); |
|
|
|
|
grpc_tcp_server_ref(tcp_server_); // Ref held by ConnectionState.
|
|
|
|
|
return handshake_mgr; |
|
|
|
|
} |
|
|
|
|
count++; |
|
|
|
|
|
|
|
|
|
void Chttp2ServerListener::OnAccept(void* arg, grpc_endpoint* tcp, |
|
|
|
|
grpc_pollset* accepting_pollset, |
|
|
|
|
grpc_tcp_server_acceptor* acceptor) { |
|
|
|
|
Chttp2ServerListener* self = static_cast<Chttp2ServerListener*>(arg); |
|
|
|
|
RefCountedPtr<HandshakeManager> handshake_mgr = |
|
|
|
|
self->CreateHandshakeManager(); |
|
|
|
|
if (handshake_mgr == nullptr) { |
|
|
|
|
grpc_endpoint_shutdown(tcp, GRPC_ERROR_NONE); |
|
|
|
|
grpc_endpoint_destroy(tcp); |
|
|
|
|
gpr_free(acceptor); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
// Deletes itself when done.
|
|
|
|
|
new ConnectionState(self, accepting_pollset, acceptor, |
|
|
|
|
std::move(handshake_mgr), self->args_, tcp); |
|
|
|
|
} |
|
|
|
|
if (count == 0) { |
|
|
|
|
char* msg; |
|
|
|
|
gpr_asprintf(&msg, "No address added out of total %" PRIuPTR " resolved", |
|
|
|
|
naddrs); |
|
|
|
|
err = GRPC_ERROR_CREATE_REFERENCING_FROM_COPIED_STRING(msg, errors, naddrs); |
|
|
|
|
gpr_free(msg); |
|
|
|
|
goto error; |
|
|
|
|
} else if (count != naddrs) { |
|
|
|
|
char* msg; |
|
|
|
|
gpr_asprintf(&msg, |
|
|
|
|
"Only %" PRIuPTR " addresses added out of total %" PRIuPTR |
|
|
|
|
" resolved", |
|
|
|
|
count, naddrs); |
|
|
|
|
err = GRPC_ERROR_CREATE_REFERENCING_FROM_COPIED_STRING(msg, errors, naddrs); |
|
|
|
|
gpr_free(msg); |
|
|
|
|
|
|
|
|
|
const char* warning_message = grpc_error_string(err); |
|
|
|
|
gpr_log(GPR_INFO, "WARNING: %s", warning_message); |
|
|
|
|
|
|
|
|
|
/* we managed to bind some addresses: continue */ |
|
|
|
|
void Chttp2ServerListener::TcpServerShutdownComplete(void* arg, |
|
|
|
|
grpc_error* error) { |
|
|
|
|
Chttp2ServerListener* self = static_cast<Chttp2ServerListener*>(arg); |
|
|
|
|
/* ensure all threads have unlocked */ |
|
|
|
|
grpc_closure* destroy_done = nullptr; |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&self->mu_); |
|
|
|
|
destroy_done = self->on_destroy_done_; |
|
|
|
|
GPR_ASSERT(self->shutdown_); |
|
|
|
|
if (self->pending_handshake_mgrs_ != nullptr) { |
|
|
|
|
self->pending_handshake_mgrs_->ShutdownAllPending(GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
grpc_resolved_addresses_destroy(resolved); |
|
|
|
|
|
|
|
|
|
arg = grpc_channel_args_find(args, GRPC_ARG_ENABLE_CHANNELZ); |
|
|
|
|
if (grpc_channel_arg_get_bool(arg, GRPC_ENABLE_CHANNELZ_DEFAULT)) { |
|
|
|
|
state->channelz_listen_socket = |
|
|
|
|
grpc_core::MakeRefCounted<grpc_core::channelz::ListenSocketNode>( |
|
|
|
|
addr, absl::StrFormat("chttp2 listener %s", addr)); |
|
|
|
|
self->channelz_listen_socket_.reset(); |
|
|
|
|
} |
|
|
|
|
// Flush queued work before destroying handshaker factory, since that
|
|
|
|
|
// may do a synchronous unref.
|
|
|
|
|
ExecCtx::Get()->Flush(); |
|
|
|
|
if (destroy_done != nullptr) { |
|
|
|
|
ExecCtx::Run(DEBUG_LOCATION, destroy_done, GRPC_ERROR_REF(error)); |
|
|
|
|
ExecCtx::Get()->Flush(); |
|
|
|
|
} |
|
|
|
|
delete self; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/* Register with the server only upon success */ |
|
|
|
|
grpc_server_add_listener(server, state, server_start_listener, |
|
|
|
|
server_destroy_listener, |
|
|
|
|
state->channelz_listen_socket); |
|
|
|
|
goto done; |
|
|
|
|
|
|
|
|
|
/* Error path: cleanup and return */ |
|
|
|
|
error: |
|
|
|
|
GPR_ASSERT(err != GRPC_ERROR_NONE); |
|
|
|
|
if (resolved) { |
|
|
|
|
grpc_resolved_addresses_destroy(resolved); |
|
|
|
|
/* Server callback: destroy the tcp listener (so we don't generate further
|
|
|
|
|
callbacks) */ |
|
|
|
|
void Chttp2ServerListener::Orphan() { |
|
|
|
|
grpc_tcp_server* tcp_server; |
|
|
|
|
{ |
|
|
|
|
MutexLock lock(&mu_); |
|
|
|
|
shutdown_ = true; |
|
|
|
|
tcp_server = tcp_server_; |
|
|
|
|
} |
|
|
|
|
if (tcp_server) { |
|
|
|
|
grpc_tcp_server_shutdown_listeners(tcp_server); |
|
|
|
|
grpc_tcp_server_unref(tcp_server); |
|
|
|
|
} else { |
|
|
|
|
grpc_channel_args_destroy(args); |
|
|
|
|
gpr_free(state); |
|
|
|
|
} |
|
|
|
|
*port_num = 0; |
|
|
|
|
|
|
|
|
|
done: |
|
|
|
|
if (errors != nullptr) { |
|
|
|
|
for (i = 0; i < naddrs; i++) { |
|
|
|
|
GRPC_ERROR_UNREF(errors[i]); |
|
|
|
|
} |
|
|
|
|
gpr_free(errors); |
|
|
|
|
} // namespace
|
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
// Chttp2ServerAddPort()
|
|
|
|
|
//
|
|
|
|
|
|
|
|
|
|
grpc_error* Chttp2ServerAddPort(grpc_server* server, const char* addr, |
|
|
|
|
grpc_channel_args* args, int* port_num) { |
|
|
|
|
if (strncmp(addr, "external:", 9) == 0) { |
|
|
|
|
return grpc_core::Chttp2ServerListener::CreateWithAcceptor(server, addr, |
|
|
|
|
args); |
|
|
|
|
} |
|
|
|
|
return err; |
|
|
|
|
return grpc_core::Chttp2ServerListener::Create(server, addr, args, port_num); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} // namespace grpc_core
|
|
|
|
|
|
Mark D. Roth
5 years ago
committed by
GitHub