|
|
|
|
@ -106,15 +106,183 @@ class UnimplementedAsyncRequestContext { |
|
|
|
|
|
|
|
|
|
} // namespace
|
|
|
|
|
|
|
|
|
|
ServerInterface::BaseAsyncRequest::BaseAsyncRequest( |
|
|
|
|
ServerInterface* server, ServerContext* context, |
|
|
|
|
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, |
|
|
|
|
ServerCompletionQueue* notification_cq, void* tag, bool delete_on_finalize) |
|
|
|
|
: server_(server), |
|
|
|
|
context_(context), |
|
|
|
|
stream_(stream), |
|
|
|
|
call_cq_(call_cq), |
|
|
|
|
notification_cq_(notification_cq), |
|
|
|
|
tag_(tag), |
|
|
|
|
delete_on_finalize_(delete_on_finalize), |
|
|
|
|
call_(nullptr), |
|
|
|
|
done_intercepting_(false) { |
|
|
|
|
/* Set up interception state partially for the receive ops. call_wrapper_ is
|
|
|
|
|
* not filled at this point, but it will be filled before the interceptors are |
|
|
|
|
* run. */ |
|
|
|
|
interceptor_methods_.SetCall(&call_wrapper_); |
|
|
|
|
interceptor_methods_.SetReverse(); |
|
|
|
|
call_cq_->RegisterAvalanching(); // This op will trigger more ops
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ServerInterface::BaseAsyncRequest::~BaseAsyncRequest() { |
|
|
|
|
call_cq_->CompleteAvalanching(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag, |
|
|
|
|
bool* status) { |
|
|
|
|
if (done_intercepting_) { |
|
|
|
|
*tag = tag_; |
|
|
|
|
if (delete_on_finalize_) { |
|
|
|
|
delete this; |
|
|
|
|
} |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
context_->set_call(call_); |
|
|
|
|
context_->cq_ = call_cq_; |
|
|
|
|
if (call_wrapper_.call() == nullptr) { |
|
|
|
|
// Fill it since it is empty.
|
|
|
|
|
call_wrapper_ = internal::Call( |
|
|
|
|
call_, server_, call_cq_, server_->max_receive_message_size(), nullptr); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// just the pointers inside call are copied here
|
|
|
|
|
stream_->BindCall(&call_wrapper_); |
|
|
|
|
|
|
|
|
|
if (*status && call_ && call_wrapper_.server_rpc_info()) { |
|
|
|
|
done_intercepting_ = true; |
|
|
|
|
// Set interception point for RECV INITIAL METADATA
|
|
|
|
|
interceptor_methods_.AddInterceptionHookPoint( |
|
|
|
|
experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA); |
|
|
|
|
interceptor_methods_.SetRecvInitialMetadata(&context_->client_metadata_); |
|
|
|
|
if (interceptor_methods_.RunInterceptors( |
|
|
|
|
[this]() { ContinueFinalizeResultAfterInterception(); })) { |
|
|
|
|
// There are no interceptors to run. Continue
|
|
|
|
|
} else { |
|
|
|
|
// There were interceptors to be run, so
|
|
|
|
|
// ContinueFinalizeResultAfterInterception will be run when interceptors
|
|
|
|
|
// are done.
|
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
if (*status && call_) { |
|
|
|
|
context_->BeginCompletionOp(&call_wrapper_, nullptr, nullptr); |
|
|
|
|
} |
|
|
|
|
*tag = tag_; |
|
|
|
|
if (delete_on_finalize_) { |
|
|
|
|
delete this; |
|
|
|
|
} |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void ServerInterface::BaseAsyncRequest:: |
|
|
|
|
ContinueFinalizeResultAfterInterception() { |
|
|
|
|
context_->BeginCompletionOp(&call_wrapper_, nullptr, nullptr); |
|
|
|
|
// Queue a tag which will be returned immediately
|
|
|
|
|
grpc_core::ExecCtx exec_ctx; |
|
|
|
|
grpc_cq_begin_op(notification_cq_->cq(), this); |
|
|
|
|
grpc_cq_end_op( |
|
|
|
|
notification_cq_->cq(), this, GRPC_ERROR_NONE, |
|
|
|
|
[](void* arg, grpc_cq_completion* completion) { delete completion; }, |
|
|
|
|
nullptr, new grpc_cq_completion()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ServerInterface::RegisteredAsyncRequest::RegisteredAsyncRequest( |
|
|
|
|
ServerInterface* server, ServerContext* context, |
|
|
|
|
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, |
|
|
|
|
ServerCompletionQueue* notification_cq, void* tag, const char* name, |
|
|
|
|
internal::RpcMethod::RpcType type) |
|
|
|
|
: BaseAsyncRequest(server, context, stream, call_cq, notification_cq, tag, |
|
|
|
|
true), |
|
|
|
|
name_(name), |
|
|
|
|
type_(type) {} |
|
|
|
|
|
|
|
|
|
void ServerInterface::RegisteredAsyncRequest::IssueRequest( |
|
|
|
|
void* registered_method, grpc_byte_buffer** payload, |
|
|
|
|
ServerCompletionQueue* notification_cq) { |
|
|
|
|
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_registered_call( |
|
|
|
|
server_->server(), registered_method, &call_, |
|
|
|
|
&context_->deadline_, |
|
|
|
|
context_->client_metadata_.arr(), payload, |
|
|
|
|
call_cq_->cq(), notification_cq->cq(), this)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ServerInterface::GenericAsyncRequest::GenericAsyncRequest( |
|
|
|
|
ServerInterface* server, GenericServerContext* context, |
|
|
|
|
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, |
|
|
|
|
ServerCompletionQueue* notification_cq, void* tag, bool delete_on_finalize) |
|
|
|
|
: BaseAsyncRequest(server, context, stream, call_cq, notification_cq, tag, |
|
|
|
|
delete_on_finalize) { |
|
|
|
|
grpc_call_details_init(&call_details_); |
|
|
|
|
GPR_ASSERT(notification_cq); |
|
|
|
|
GPR_ASSERT(call_cq); |
|
|
|
|
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( |
|
|
|
|
server->server(), &call_, &call_details_, |
|
|
|
|
context->client_metadata_.arr(), call_cq->cq(), |
|
|
|
|
notification_cq->cq(), this)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
bool ServerInterface::GenericAsyncRequest::FinalizeResult(void** tag, |
|
|
|
|
bool* status) { |
|
|
|
|
// If we are done intercepting, there is nothing more for us to do
|
|
|
|
|
if (done_intercepting_) { |
|
|
|
|
return BaseAsyncRequest::FinalizeResult(tag, status); |
|
|
|
|
} |
|
|
|
|
// TODO(yangg) remove the copy here.
|
|
|
|
|
if (*status) { |
|
|
|
|
static_cast<GenericServerContext*>(context_)->method_ = |
|
|
|
|
StringFromCopiedSlice(call_details_.method); |
|
|
|
|
static_cast<GenericServerContext*>(context_)->host_ = |
|
|
|
|
StringFromCopiedSlice(call_details_.host); |
|
|
|
|
context_->deadline_ = call_details_.deadline; |
|
|
|
|
} |
|
|
|
|
grpc_slice_unref(call_details_.method); |
|
|
|
|
grpc_slice_unref(call_details_.host); |
|
|
|
|
call_wrapper_ = internal::Call( |
|
|
|
|
call_, server_, call_cq_, server_->max_receive_message_size(), |
|
|
|
|
context_->set_server_rpc_info( |
|
|
|
|
static_cast<GenericServerContext*>(context_)->method_.c_str(), |
|
|
|
|
internal::RpcMethod::BIDI_STREAMING, |
|
|
|
|
*server_->interceptor_creators())); |
|
|
|
|
return BaseAsyncRequest::FinalizeResult(tag, status); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
namespace { |
|
|
|
|
class ShutdownCallback : public grpc_experimental_completion_queue_functor { |
|
|
|
|
public: |
|
|
|
|
ShutdownCallback() { functor_run = &ShutdownCallback::Run; } |
|
|
|
|
// TakeCQ takes ownership of the cq into the shutdown callback
|
|
|
|
|
// so that the shutdown callback will be responsible for destroying it
|
|
|
|
|
void TakeCQ(CompletionQueue* cq) { cq_ = cq; } |
|
|
|
|
|
|
|
|
|
// The Run function will get invoked by the completion queue library
|
|
|
|
|
// when the shutdown is actually complete
|
|
|
|
|
static void Run(grpc_experimental_completion_queue_functor* cb, int) { |
|
|
|
|
auto* callback = static_cast<ShutdownCallback*>(cb); |
|
|
|
|
delete callback->cq_; |
|
|
|
|
delete callback; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
CompletionQueue* cq_ = nullptr; |
|
|
|
|
}; |
|
|
|
|
} // namespace
|
|
|
|
|
|
|
|
|
|
} // namespace grpc
|
|
|
|
|
|
|
|
|
|
namespace grpc_impl { |
|
|
|
|
|
|
|
|
|
/// Use private inheritance rather than composition only to establish order
|
|
|
|
|
/// of construction, since the public base class should be constructed after the
|
|
|
|
|
/// elements belonging to the private base class are constructed. This is not
|
|
|
|
|
/// possible using true composition.
|
|
|
|
|
class Server::UnimplementedAsyncRequest final |
|
|
|
|
: private UnimplementedAsyncRequestContext, |
|
|
|
|
: private grpc::UnimplementedAsyncRequestContext, |
|
|
|
|
public GenericAsyncRequest { |
|
|
|
|
public: |
|
|
|
|
UnimplementedAsyncRequest(Server* server, ServerCompletionQueue* cq) |
|
|
|
|
UnimplementedAsyncRequest(Server* server, grpc::ServerCompletionQueue* cq) |
|
|
|
|
: GenericAsyncRequest(server, &server_context_, &generic_stream_, cq, cq, |
|
|
|
|
nullptr, false), |
|
|
|
|
server_(server), |
|
|
|
|
@ -122,27 +290,29 @@ class Server::UnimplementedAsyncRequest final |
|
|
|
|
|
|
|
|
|
bool FinalizeResult(void** tag, bool* status) override; |
|
|
|
|
|
|
|
|
|
ServerContext* context() { return &server_context_; } |
|
|
|
|
GenericServerAsyncReaderWriter* stream() { return &generic_stream_; } |
|
|
|
|
grpc::ServerContext* context() { return &server_context_; } |
|
|
|
|
grpc::GenericServerAsyncReaderWriter* stream() { return &generic_stream_; } |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
Server* const server_; |
|
|
|
|
ServerCompletionQueue* const cq_; |
|
|
|
|
grpc::ServerCompletionQueue* const cq_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
/// UnimplementedAsyncResponse should not post user-visible completions to the
|
|
|
|
|
/// C++ completion queue, but is generated as a CQ event by the core
|
|
|
|
|
class Server::UnimplementedAsyncResponse final |
|
|
|
|
: public internal::CallOpSet<internal::CallOpSendInitialMetadata, |
|
|
|
|
internal::CallOpServerSendStatus> { |
|
|
|
|
: public grpc::internal::CallOpSet< |
|
|
|
|
grpc::internal::CallOpSendInitialMetadata, |
|
|
|
|
grpc::internal::CallOpServerSendStatus> { |
|
|
|
|
public: |
|
|
|
|
UnimplementedAsyncResponse(UnimplementedAsyncRequest* request); |
|
|
|
|
~UnimplementedAsyncResponse() { delete request_; } |
|
|
|
|
|
|
|
|
|
bool FinalizeResult(void** tag, bool* status) override { |
|
|
|
|
if (internal::CallOpSet< |
|
|
|
|
internal::CallOpSendInitialMetadata, |
|
|
|
|
internal::CallOpServerSendStatus>::FinalizeResult(tag, status)) { |
|
|
|
|
if (grpc::internal::CallOpSet< |
|
|
|
|
grpc::internal::CallOpSendInitialMetadata, |
|
|
|
|
grpc::internal::CallOpServerSendStatus>::FinalizeResult(tag, |
|
|
|
|
status)) { |
|
|
|
|
delete this; |
|
|
|
|
} else { |
|
|
|
|
// The tag was swallowed due to interception. We will see it again.
|
|
|
|
|
@ -154,15 +324,16 @@ class Server::UnimplementedAsyncResponse final |
|
|
|
|
UnimplementedAsyncRequest* const request_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
class Server::SyncRequest final : public internal::CompletionQueueTag { |
|
|
|
|
class Server::SyncRequest final : public grpc::internal::CompletionQueueTag { |
|
|
|
|
public: |
|
|
|
|
SyncRequest(internal::RpcServiceMethod* method, void* method_tag) |
|
|
|
|
SyncRequest(grpc::internal::RpcServiceMethod* method, void* method_tag) |
|
|
|
|
: method_(method), |
|
|
|
|
method_tag_(method_tag), |
|
|
|
|
in_flight_(false), |
|
|
|
|
has_request_payload_( |
|
|
|
|
method->method_type() == internal::RpcMethod::NORMAL_RPC || |
|
|
|
|
method->method_type() == internal::RpcMethod::SERVER_STREAMING), |
|
|
|
|
has_request_payload_(method->method_type() == |
|
|
|
|
grpc::internal::RpcMethod::NORMAL_RPC || |
|
|
|
|
method->method_type() == |
|
|
|
|
grpc::internal::RpcMethod::SERVER_STREAMING), |
|
|
|
|
call_details_(nullptr), |
|
|
|
|
cq_(nullptr) { |
|
|
|
|
grpc_metadata_array_init(&request_metadata_); |
|
|
|
|
@ -273,7 +444,8 @@ class Server::SyncRequest final : public internal::CompletionQueueTag { |
|
|
|
|
interceptor_methods_.SetReverse(); |
|
|
|
|
// Set interception point for RECV INITIAL METADATA
|
|
|
|
|
interceptor_methods_.AddInterceptionHookPoint( |
|
|
|
|
experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA); |
|
|
|
|
grpc::experimental::InterceptionHookPoints:: |
|
|
|
|
POST_RECV_INITIAL_METADATA); |
|
|
|
|
interceptor_methods_.SetRecvInitialMetadata(&ctx_.client_metadata_); |
|
|
|
|
|
|
|
|
|
if (has_request_payload_) { |
|
|
|
|
@ -285,7 +457,7 @@ class Server::SyncRequest final : public internal::CompletionQueueTag { |
|
|
|
|
|
|
|
|
|
request_payload_ = nullptr; |
|
|
|
|
interceptor_methods_.AddInterceptionHookPoint( |
|
|
|
|
experimental::InterceptionHookPoints::POST_RECV_MESSAGE); |
|
|
|
|
grpc::experimental::InterceptionHookPoints::POST_RECV_MESSAGE); |
|
|
|
|
interceptor_methods_.SetRecvMessage(request_, nullptr); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -304,40 +476,40 @@ class Server::SyncRequest final : public internal::CompletionQueueTag { |
|
|
|
|
global_callbacks_->PreSynchronousRequest(&ctx_); |
|
|
|
|
auto* handler = resources_ ? method_->handler() |
|
|
|
|
: server_->resource_exhausted_handler_.get(); |
|
|
|
|
handler->RunHandler(internal::MethodHandler::HandlerParameter( |
|
|
|
|
handler->RunHandler(grpc::internal::MethodHandler::HandlerParameter( |
|
|
|
|
&call_, &ctx_, request_, request_status_, nullptr, nullptr)); |
|
|
|
|
request_ = nullptr; |
|
|
|
|
global_callbacks_->PostSynchronousRequest(&ctx_); |
|
|
|
|
|
|
|
|
|
cq_.Shutdown(); |
|
|
|
|
|
|
|
|
|
internal::CompletionQueueTag* op_tag = ctx_.GetCompletionOpTag(); |
|
|
|
|
grpc::internal::CompletionQueueTag* op_tag = ctx_.GetCompletionOpTag(); |
|
|
|
|
cq_.TryPluck(op_tag, gpr_inf_future(GPR_CLOCK_REALTIME)); |
|
|
|
|
|
|
|
|
|
/* Ensure the cq_ is shutdown */ |
|
|
|
|
DummyTag ignored_tag; |
|
|
|
|
grpc::DummyTag ignored_tag; |
|
|
|
|
GPR_ASSERT(cq_.Pluck(&ignored_tag) == false); |
|
|
|
|
} |
|
|
|
|
delete this; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
CompletionQueue cq_; |
|
|
|
|
ServerContext ctx_; |
|
|
|
|
grpc::CompletionQueue cq_; |
|
|
|
|
grpc::ServerContext ctx_; |
|
|
|
|
const bool has_request_payload_; |
|
|
|
|
grpc_byte_buffer* request_payload_; |
|
|
|
|
void* request_; |
|
|
|
|
Status request_status_; |
|
|
|
|
internal::RpcServiceMethod* const method_; |
|
|
|
|
internal::Call call_; |
|
|
|
|
grpc::Status request_status_; |
|
|
|
|
grpc::internal::RpcServiceMethod* const method_; |
|
|
|
|
grpc::internal::Call call_; |
|
|
|
|
Server* server_; |
|
|
|
|
std::shared_ptr<GlobalCallbacks> global_callbacks_; |
|
|
|
|
bool resources_; |
|
|
|
|
internal::InterceptorBatchMethodsImpl interceptor_methods_; |
|
|
|
|
grpc::internal::InterceptorBatchMethodsImpl interceptor_methods_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
internal::RpcServiceMethod* const method_; |
|
|
|
|
grpc::internal::RpcServiceMethod* const method_; |
|
|
|
|
void* const method_tag_; |
|
|
|
|
bool in_flight_; |
|
|
|
|
const bool has_request_payload_; |
|
|
|
|
@ -349,7 +521,7 @@ class Server::SyncRequest final : public internal::CompletionQueueTag { |
|
|
|
|
grpc_completion_queue* cq_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
class Server::CallbackRequestBase : public internal::CompletionQueueTag { |
|
|
|
|
class Server::CallbackRequestBase : public grpc::internal::CompletionQueueTag { |
|
|
|
|
public: |
|
|
|
|
virtual ~CallbackRequestBase() {} |
|
|
|
|
virtual bool Request() = 0; |
|
|
|
|
@ -358,7 +530,7 @@ class Server::CallbackRequestBase : public internal::CompletionQueueTag { |
|
|
|
|
template <class ServerContextType> |
|
|
|
|
class Server::CallbackRequest final : public Server::CallbackRequestBase { |
|
|
|
|
public: |
|
|
|
|
static_assert(std::is_base_of<ServerContext, ServerContextType>::value, |
|
|
|
|
static_assert(std::is_base_of<grpc::ServerContext, ServerContextType>::value, |
|
|
|
|
"ServerContextType must be derived from ServerContext"); |
|
|
|
|
|
|
|
|
|
// The constructor needs to know the server for this callback request and its
|
|
|
|
|
@ -368,15 +540,16 @@ class Server::CallbackRequest final : public Server::CallbackRequestBase { |
|
|
|
|
// requested. For generic services, method and method_tag are nullptr since
|
|
|
|
|
// these services don't have pre-defined methods or method registration tags.
|
|
|
|
|
CallbackRequest(Server* server, size_t method_idx, |
|
|
|
|
internal::RpcServiceMethod* method, void* method_tag) |
|
|
|
|
grpc::internal::RpcServiceMethod* method, void* method_tag) |
|
|
|
|
: server_(server), |
|
|
|
|
method_index_(method_idx), |
|
|
|
|
method_(method), |
|
|
|
|
method_tag_(method_tag), |
|
|
|
|
has_request_payload_( |
|
|
|
|
method_ != nullptr && |
|
|
|
|
(method->method_type() == internal::RpcMethod::NORMAL_RPC || |
|
|
|
|
method->method_type() == internal::RpcMethod::SERVER_STREAMING)), |
|
|
|
|
(method->method_type() == grpc::internal::RpcMethod::NORMAL_RPC || |
|
|
|
|
method->method_type() == |
|
|
|
|
grpc::internal::RpcMethod::SERVER_STREAMING)), |
|
|
|
|
cq_(server->CallbackCQ()), |
|
|
|
|
tag_(this) { |
|
|
|
|
server_->callback_reqs_outstanding_++; |
|
|
|
|
@ -440,7 +613,7 @@ class Server::CallbackRequest final : public Server::CallbackRequestBase { |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
Server::CallbackRequest<ServerContextType>* req_; |
|
|
|
|
internal::Call* call_; |
|
|
|
|
grpc::internal::Call* call_; |
|
|
|
|
|
|
|
|
|
static void StaticRun(grpc_experimental_completion_queue_functor* cb, |
|
|
|
|
int ok) { |
|
|
|
|
@ -491,21 +664,24 @@ class Server::CallbackRequest final : public Server::CallbackRequestBase { |
|
|
|
|
req_->request_metadata_.count = 0; |
|
|
|
|
|
|
|
|
|
// Create a C++ Call to control the underlying core call
|
|
|
|
|
call_ = new (grpc_call_arena_alloc(req_->call_, sizeof(internal::Call))) |
|
|
|
|
internal::Call(req_->call_, req_->server_, req_->cq_, |
|
|
|
|
req_->server_->max_receive_message_size(), |
|
|
|
|
req_->ctx_.set_server_rpc_info( |
|
|
|
|
req_->method_name(), |
|
|
|
|
(req_->method_ != nullptr) |
|
|
|
|
? req_->method_->method_type() |
|
|
|
|
: internal::RpcMethod::BIDI_STREAMING, |
|
|
|
|
req_->server_->interceptor_creators_)); |
|
|
|
|
call_ = |
|
|
|
|
new (grpc_call_arena_alloc(req_->call_, sizeof(grpc::internal::Call))) |
|
|
|
|
grpc::internal::Call( |
|
|
|
|
req_->call_, req_->server_, req_->cq_, |
|
|
|
|
req_->server_->max_receive_message_size(), |
|
|
|
|
req_->ctx_.set_server_rpc_info( |
|
|
|
|
req_->method_name(), |
|
|
|
|
(req_->method_ != nullptr) |
|
|
|
|
? req_->method_->method_type() |
|
|
|
|
: grpc::internal::RpcMethod::BIDI_STREAMING, |
|
|
|
|
req_->server_->interceptor_creators_)); |
|
|
|
|
|
|
|
|
|
req_->interceptor_methods_.SetCall(call_); |
|
|
|
|
req_->interceptor_methods_.SetReverse(); |
|
|
|
|
// Set interception point for RECV INITIAL METADATA
|
|
|
|
|
req_->interceptor_methods_.AddInterceptionHookPoint( |
|
|
|
|
experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA); |
|
|
|
|
grpc::experimental::InterceptionHookPoints:: |
|
|
|
|
POST_RECV_INITIAL_METADATA); |
|
|
|
|
req_->interceptor_methods_.SetRecvInitialMetadata( |
|
|
|
|
&req_->ctx_.client_metadata_); |
|
|
|
|
|
|
|
|
|
@ -516,7 +692,7 @@ class Server::CallbackRequest final : public Server::CallbackRequestBase { |
|
|
|
|
&req_->handler_data_); |
|
|
|
|
req_->request_payload_ = nullptr; |
|
|
|
|
req_->interceptor_methods_.AddInterceptionHookPoint( |
|
|
|
|
experimental::InterceptionHookPoints::POST_RECV_MESSAGE); |
|
|
|
|
grpc::experimental::InterceptionHookPoints::POST_RECV_MESSAGE); |
|
|
|
|
req_->interceptor_methods_.SetRecvMessage(req_->request_, nullptr); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -532,7 +708,7 @@ class Server::CallbackRequest final : public Server::CallbackRequestBase { |
|
|
|
|
auto* handler = (req_->method_ != nullptr) |
|
|
|
|
? req_->method_->handler() |
|
|
|
|
: req_->server_->generic_handler_.get(); |
|
|
|
|
handler->RunHandler(internal::MethodHandler::HandlerParameter( |
|
|
|
|
handler->RunHandler(grpc::internal::MethodHandler::HandlerParameter( |
|
|
|
|
call_, &req_->ctx_, req_->request_, req_->request_status_, |
|
|
|
|
req_->handler_data_, [this] { |
|
|
|
|
// Recycle this request if there aren't too many outstanding.
|
|
|
|
|
@ -580,41 +756,41 @@ class Server::CallbackRequest final : public Server::CallbackRequestBase { |
|
|
|
|
request_payload_ = nullptr; |
|
|
|
|
request_ = nullptr; |
|
|
|
|
handler_data_ = nullptr; |
|
|
|
|
request_status_ = Status(); |
|
|
|
|
request_status_ = grpc::Status(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
Server* const server_; |
|
|
|
|
const size_t method_index_; |
|
|
|
|
internal::RpcServiceMethod* const method_; |
|
|
|
|
grpc::internal::RpcServiceMethod* const method_; |
|
|
|
|
void* const method_tag_; |
|
|
|
|
const bool has_request_payload_; |
|
|
|
|
grpc_byte_buffer* request_payload_; |
|
|
|
|
void* request_; |
|
|
|
|
void* handler_data_; |
|
|
|
|
Status request_status_; |
|
|
|
|
grpc::Status request_status_; |
|
|
|
|
grpc_call_details* call_details_ = nullptr; |
|
|
|
|
grpc_call* call_; |
|
|
|
|
gpr_timespec deadline_; |
|
|
|
|
grpc_metadata_array request_metadata_; |
|
|
|
|
CompletionQueue* cq_; |
|
|
|
|
grpc::CompletionQueue* cq_; |
|
|
|
|
CallbackCallTag tag_; |
|
|
|
|
ServerContextType ctx_; |
|
|
|
|
internal::InterceptorBatchMethodsImpl interceptor_methods_; |
|
|
|
|
grpc::internal::InterceptorBatchMethodsImpl interceptor_methods_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
template <> |
|
|
|
|
bool Server::CallbackRequest<ServerContext>::FinalizeResult(void** tag, |
|
|
|
|
bool* status) { |
|
|
|
|
bool Server::CallbackRequest<grpc::ServerContext>::FinalizeResult( |
|
|
|
|
void** tag, bool* status) { |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
template <> |
|
|
|
|
bool Server::CallbackRequest<GenericServerContext>::FinalizeResult( |
|
|
|
|
bool Server::CallbackRequest<grpc::GenericServerContext>::FinalizeResult( |
|
|
|
|
void** tag, bool* status) { |
|
|
|
|
if (*status) { |
|
|
|
|
// TODO(yangg) remove the copy here
|
|
|
|
|
ctx_.method_ = StringFromCopiedSlice(call_details_->method); |
|
|
|
|
ctx_.host_ = StringFromCopiedSlice(call_details_->host); |
|
|
|
|
ctx_.method_ = grpc::StringFromCopiedSlice(call_details_->method); |
|
|
|
|
ctx_.host_ = grpc::StringFromCopiedSlice(call_details_->host); |
|
|
|
|
} |
|
|
|
|
grpc_slice_unref(call_details_->method); |
|
|
|
|
grpc_slice_unref(call_details_->host); |
|
|
|
|
@ -622,21 +798,22 @@ bool Server::CallbackRequest<GenericServerContext>::FinalizeResult( |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
template <> |
|
|
|
|
const char* Server::CallbackRequest<ServerContext>::method_name() const { |
|
|
|
|
const char* Server::CallbackRequest<grpc::ServerContext>::method_name() const { |
|
|
|
|
return method_->name(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
template <> |
|
|
|
|
const char* Server::CallbackRequest<GenericServerContext>::method_name() const { |
|
|
|
|
const char* Server::CallbackRequest<grpc::GenericServerContext>::method_name() |
|
|
|
|
const { |
|
|
|
|
return ctx_.method().c_str(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Implementation of ThreadManager. Each instance of SyncRequestThreadManager
|
|
|
|
|
// manages a pool of threads that poll for incoming Sync RPCs and call the
|
|
|
|
|
// appropriate RPC handlers
|
|
|
|
|
class Server::SyncRequestThreadManager : public ThreadManager { |
|
|
|
|
class Server::SyncRequestThreadManager : public grpc::ThreadManager { |
|
|
|
|
public: |
|
|
|
|
SyncRequestThreadManager(Server* server, CompletionQueue* server_cq, |
|
|
|
|
SyncRequestThreadManager(Server* server, grpc::CompletionQueue* server_cq, |
|
|
|
|
std::shared_ptr<GlobalCallbacks> global_callbacks, |
|
|
|
|
grpc_resource_quota* rq, int min_pollers, |
|
|
|
|
int max_pollers, int cq_timeout_msec) |
|
|
|
|
@ -655,11 +832,11 @@ class Server::SyncRequestThreadManager : public ThreadManager { |
|
|
|
|
gpr_time_from_millis(cq_timeout_msec_, GPR_TIMESPAN)); |
|
|
|
|
|
|
|
|
|
switch (server_cq_->AsyncNext(tag, ok, deadline)) { |
|
|
|
|
case CompletionQueue::TIMEOUT: |
|
|
|
|
case grpc::CompletionQueue::TIMEOUT: |
|
|
|
|
return TIMEOUT; |
|
|
|
|
case CompletionQueue::SHUTDOWN: |
|
|
|
|
case grpc::CompletionQueue::SHUTDOWN: |
|
|
|
|
return SHUTDOWN; |
|
|
|
|
case CompletionQueue::GOT_EVENT: |
|
|
|
|
case grpc::CompletionQueue::GOT_EVENT: |
|
|
|
|
return WORK_FOUND; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -694,15 +871,15 @@ class Server::SyncRequestThreadManager : public ThreadManager { |
|
|
|
|
// object
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void AddSyncMethod(internal::RpcServiceMethod* method, void* tag) { |
|
|
|
|
void AddSyncMethod(grpc::internal::RpcServiceMethod* method, void* tag) { |
|
|
|
|
sync_requests_.emplace_back(new SyncRequest(method, tag)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void AddUnknownSyncMethod() { |
|
|
|
|
if (!sync_requests_.empty()) { |
|
|
|
|
unknown_method_.reset(new internal::RpcServiceMethod( |
|
|
|
|
"unknown", internal::RpcMethod::BIDI_STREAMING, |
|
|
|
|
new internal::UnknownMethodHandler)); |
|
|
|
|
unknown_method_.reset(new grpc::internal::RpcServiceMethod( |
|
|
|
|
"unknown", grpc::internal::RpcMethod::BIDI_STREAMING, |
|
|
|
|
new grpc::internal::UnknownMethodHandler)); |
|
|
|
|
sync_requests_.emplace_back( |
|
|
|
|
new SyncRequest(unknown_method_.get(), nullptr)); |
|
|
|
|
} |
|
|
|
|
@ -746,22 +923,22 @@ class Server::SyncRequestThreadManager : public ThreadManager { |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
Server* server_; |
|
|
|
|
CompletionQueue* server_cq_; |
|
|
|
|
grpc::CompletionQueue* server_cq_; |
|
|
|
|
int cq_timeout_msec_; |
|
|
|
|
std::vector<std::unique_ptr<SyncRequest>> sync_requests_; |
|
|
|
|
std::unique_ptr<internal::RpcServiceMethod> unknown_method_; |
|
|
|
|
std::unique_ptr<grpc::internal::RpcServiceMethod> unknown_method_; |
|
|
|
|
std::shared_ptr<Server::GlobalCallbacks> global_callbacks_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
static internal::GrpcLibraryInitializer g_gli_initializer; |
|
|
|
|
static grpc::internal::GrpcLibraryInitializer g_gli_initializer; |
|
|
|
|
Server::Server( |
|
|
|
|
int max_receive_message_size, ChannelArguments* args, |
|
|
|
|
std::shared_ptr<std::vector<std::unique_ptr<ServerCompletionQueue>>> |
|
|
|
|
int max_receive_message_size, grpc::ChannelArguments* args, |
|
|
|
|
std::shared_ptr<std::vector<std::unique_ptr<grpc::ServerCompletionQueue>>> |
|
|
|
|
sync_server_cqs, |
|
|
|
|
int min_pollers, int max_pollers, int sync_cq_timeout_msec, |
|
|
|
|
grpc_resource_quota* server_rq, |
|
|
|
|
std::vector< |
|
|
|
|
std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> |
|
|
|
|
std::unique_ptr<grpc::experimental::ServerInterceptorFactoryInterface>> |
|
|
|
|
interceptor_creators) |
|
|
|
|
: interceptor_creators_(std::move(interceptor_creators)), |
|
|
|
|
max_receive_message_size_(max_receive_message_size), |
|
|
|
|
@ -773,8 +950,8 @@ Server::Server( |
|
|
|
|
server_initializer_(new grpc_impl::ServerInitializer(this)), |
|
|
|
|
health_check_service_disabled_(false) { |
|
|
|
|
g_gli_initializer.summon(); |
|
|
|
|
gpr_once_init(&g_once_init_callbacks, InitGlobalCallbacks); |
|
|
|
|
global_callbacks_ = g_callbacks; |
|
|
|
|
gpr_once_init(&grpc::g_once_init_callbacks, grpc::InitGlobalCallbacks); |
|
|
|
|
global_callbacks_ = grpc::g_callbacks; |
|
|
|
|
global_callbacks_->UpdateArguments(args); |
|
|
|
|
|
|
|
|
|
if (sync_server_cqs_ != nullptr) { |
|
|
|
|
@ -801,13 +978,14 @@ Server::Server( |
|
|
|
|
args->SetChannelArgs(&channel_args); |
|
|
|
|
|
|
|
|
|
for (size_t i = 0; i < channel_args.num_args; i++) { |
|
|
|
|
if (0 == |
|
|
|
|
strcmp(channel_args.args[i].key, kHealthCheckServiceInterfaceArg)) { |
|
|
|
|
if (0 == strcmp(channel_args.args[i].key, |
|
|
|
|
grpc::kHealthCheckServiceInterfaceArg)) { |
|
|
|
|
if (channel_args.args[i].value.pointer.p == nullptr) { |
|
|
|
|
health_check_service_disabled_ = true; |
|
|
|
|
} else { |
|
|
|
|
health_check_service_.reset(static_cast<HealthCheckServiceInterface*>( |
|
|
|
|
channel_args.args[i].value.pointer.p)); |
|
|
|
|
health_check_service_.reset( |
|
|
|
|
static_cast<grpc::HealthCheckServiceInterface*>( |
|
|
|
|
channel_args.args[i].value.pointer.p)); |
|
|
|
|
} |
|
|
|
|
break; |
|
|
|
|
} |
|
|
|
|
@ -844,49 +1022,49 @@ Server::~Server() { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Server::SetGlobalCallbacks(GlobalCallbacks* callbacks) { |
|
|
|
|
GPR_ASSERT(!g_callbacks); |
|
|
|
|
GPR_ASSERT(!grpc::g_callbacks); |
|
|
|
|
GPR_ASSERT(callbacks); |
|
|
|
|
g_callbacks.reset(callbacks); |
|
|
|
|
grpc::g_callbacks.reset(callbacks); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_server* Server::c_server() { return server_; } |
|
|
|
|
|
|
|
|
|
std::shared_ptr<Channel> Server::InProcessChannel( |
|
|
|
|
const ChannelArguments& args) { |
|
|
|
|
std::shared_ptr<grpc::Channel> Server::InProcessChannel( |
|
|
|
|
const grpc::ChannelArguments& args) { |
|
|
|
|
grpc_channel_args channel_args = args.c_channel_args(); |
|
|
|
|
return ::grpc::CreateChannelInternal( |
|
|
|
|
return grpc::CreateChannelInternal( |
|
|
|
|
"inproc", grpc_inproc_channel_create(server_, &channel_args, nullptr), |
|
|
|
|
std::vector< |
|
|
|
|
std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>()); |
|
|
|
|
std::vector<std::unique_ptr< |
|
|
|
|
grpc::experimental::ClientInterceptorFactoryInterface>>()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
std::shared_ptr<Channel> |
|
|
|
|
std::shared_ptr<grpc::Channel> |
|
|
|
|
Server::experimental_type::InProcessChannelWithInterceptors( |
|
|
|
|
const ChannelArguments& args, |
|
|
|
|
const grpc::ChannelArguments& args, |
|
|
|
|
std::vector< |
|
|
|
|
std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> |
|
|
|
|
std::unique_ptr<grpc::experimental::ClientInterceptorFactoryInterface>> |
|
|
|
|
interceptor_creators) { |
|
|
|
|
grpc_channel_args channel_args = args.c_channel_args(); |
|
|
|
|
return ::grpc::CreateChannelInternal( |
|
|
|
|
return grpc::CreateChannelInternal( |
|
|
|
|
"inproc", |
|
|
|
|
grpc_inproc_channel_create(server_->server_, &channel_args, nullptr), |
|
|
|
|
std::move(interceptor_creators)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static grpc_server_register_method_payload_handling PayloadHandlingForMethod( |
|
|
|
|
internal::RpcServiceMethod* method) { |
|
|
|
|
grpc::internal::RpcServiceMethod* method) { |
|
|
|
|
switch (method->method_type()) { |
|
|
|
|
case internal::RpcMethod::NORMAL_RPC: |
|
|
|
|
case internal::RpcMethod::SERVER_STREAMING: |
|
|
|
|
case grpc::internal::RpcMethod::NORMAL_RPC: |
|
|
|
|
case grpc::internal::RpcMethod::SERVER_STREAMING: |
|
|
|
|
return GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER; |
|
|
|
|
case internal::RpcMethod::CLIENT_STREAMING: |
|
|
|
|
case internal::RpcMethod::BIDI_STREAMING: |
|
|
|
|
case grpc::internal::RpcMethod::CLIENT_STREAMING: |
|
|
|
|
case grpc::internal::RpcMethod::BIDI_STREAMING: |
|
|
|
|
return GRPC_SRM_PAYLOAD_NONE; |
|
|
|
|
} |
|
|
|
|
GPR_UNREACHABLE_CODE(return GRPC_SRM_PAYLOAD_NONE;); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
bool Server::RegisterService(const grpc::string* host, Service* service) { |
|
|
|
|
bool Server::RegisterService(const grpc::string* host, grpc::Service* service) { |
|
|
|
|
bool has_async_methods = service->has_async_methods(); |
|
|
|
|
if (has_async_methods) { |
|
|
|
|
GPR_ASSERT(service->server_ == nullptr && |
|
|
|
|
@ -902,7 +1080,7 @@ bool Server::RegisterService(const grpc::string* host, Service* service) { |
|
|
|
|
continue; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
internal::RpcServiceMethod* method = it->get(); |
|
|
|
|
grpc::internal::RpcServiceMethod* method = it->get(); |
|
|
|
|
void* method_registration_tag = grpc_server_register_method( |
|
|
|
|
server_, method->name(), host ? host->c_str() : nullptr, |
|
|
|
|
PayloadHandlingForMethod(method), 0); |
|
|
|
|
@ -915,7 +1093,7 @@ bool Server::RegisterService(const grpc::string* host, Service* service) { |
|
|
|
|
if (method->handler() == nullptr) { // Async method without handler
|
|
|
|
|
method->set_server_tag(method_registration_tag); |
|
|
|
|
} else if (method->api_type() == |
|
|
|
|
internal::RpcServiceMethod::ApiType::SYNC) { |
|
|
|
|
grpc::internal::RpcServiceMethod::ApiType::SYNC) { |
|
|
|
|
for (auto it = sync_req_mgrs_.begin(); it != sync_req_mgrs_.end(); it++) { |
|
|
|
|
(*it)->AddSyncMethod(method, method_registration_tag); |
|
|
|
|
} |
|
|
|
|
@ -925,8 +1103,9 @@ bool Server::RegisterService(const grpc::string* host, Service* service) { |
|
|
|
|
auto method_index = callback_unmatched_reqs_count_.size() - 1; |
|
|
|
|
// TODO(vjpai): Register these dynamically based on need
|
|
|
|
|
for (int i = 0; i < DEFAULT_CALLBACK_REQS_PER_METHOD; i++) { |
|
|
|
|
callback_reqs_to_start_.push_back(new CallbackRequest<ServerContext>( |
|
|
|
|
this, method_index, method, method_registration_tag)); |
|
|
|
|
callback_reqs_to_start_.push_back( |
|
|
|
|
new CallbackRequest<grpc::ServerContext>(this, method_index, method, |
|
|
|
|
method_registration_tag)); |
|
|
|
|
} |
|
|
|
|
// Enqueue it so that it will be Request'ed later after all request
|
|
|
|
|
// matchers are created at core server startup
|
|
|
|
|
@ -947,7 +1126,7 @@ bool Server::RegisterService(const grpc::string* host, Service* service) { |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Server::RegisterAsyncGenericService(AsyncGenericService* service) { |
|
|
|
|
void Server::RegisterAsyncGenericService(grpc::AsyncGenericService* service) { |
|
|
|
|
GPR_ASSERT(service->server_ == nullptr && |
|
|
|
|
"Can only register an async generic service against one server."); |
|
|
|
|
service->server_ = this; |
|
|
|
|
@ -955,7 +1134,7 @@ void Server::RegisterAsyncGenericService(AsyncGenericService* service) { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Server::RegisterCallbackGenericService( |
|
|
|
|
experimental::CallbackGenericService* service) { |
|
|
|
|
grpc::experimental::CallbackGenericService* service) { |
|
|
|
|
GPR_ASSERT( |
|
|
|
|
service->server_ == nullptr && |
|
|
|
|
"Can only register a callback generic service against one server."); |
|
|
|
|
@ -967,44 +1146,45 @@ void Server::RegisterCallbackGenericService( |
|
|
|
|
auto method_index = callback_unmatched_reqs_count_.size() - 1; |
|
|
|
|
// TODO(vjpai): Register these dynamically based on need
|
|
|
|
|
for (int i = 0; i < DEFAULT_CALLBACK_REQS_PER_METHOD; i++) { |
|
|
|
|
callback_reqs_to_start_.push_back(new CallbackRequest<GenericServerContext>( |
|
|
|
|
this, method_index, nullptr, nullptr)); |
|
|
|
|
callback_reqs_to_start_.push_back( |
|
|
|
|
new CallbackRequest<grpc::GenericServerContext>(this, method_index, |
|
|
|
|
nullptr, nullptr)); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
int Server::AddListeningPort(const grpc::string& addr, |
|
|
|
|
ServerCredentials* creds) { |
|
|
|
|
grpc::ServerCredentials* creds) { |
|
|
|
|
GPR_ASSERT(!started_); |
|
|
|
|
int port = creds->AddPortToServer(addr, server_); |
|
|
|
|
global_callbacks_->AddPort(this, addr, creds, port); |
|
|
|
|
return port; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Server::Start(ServerCompletionQueue** cqs, size_t num_cqs) { |
|
|
|
|
void Server::Start(grpc::ServerCompletionQueue** cqs, size_t num_cqs) { |
|
|
|
|
GPR_ASSERT(!started_); |
|
|
|
|
global_callbacks_->PreServerStart(this); |
|
|
|
|
started_ = true; |
|
|
|
|
|
|
|
|
|
// Only create default health check service when user did not provide an
|
|
|
|
|
// explicit one.
|
|
|
|
|
ServerCompletionQueue* health_check_cq = nullptr; |
|
|
|
|
DefaultHealthCheckService::HealthCheckServiceImpl* |
|
|
|
|
grpc::ServerCompletionQueue* health_check_cq = nullptr; |
|
|
|
|
grpc::DefaultHealthCheckService::HealthCheckServiceImpl* |
|
|
|
|
default_health_check_service_impl = nullptr; |
|
|
|
|
if (health_check_service_ == nullptr && !health_check_service_disabled_ && |
|
|
|
|
DefaultHealthCheckServiceEnabled()) { |
|
|
|
|
auto* default_hc_service = new DefaultHealthCheckService; |
|
|
|
|
grpc::DefaultHealthCheckServiceEnabled()) { |
|
|
|
|
auto* default_hc_service = new grpc::DefaultHealthCheckService; |
|
|
|
|
health_check_service_.reset(default_hc_service); |
|
|
|
|
// We create a non-polling CQ to avoid impacting application
|
|
|
|
|
// performance. This ensures that we don't introduce thread hops
|
|
|
|
|
// for application requests that wind up on this CQ, which is polled
|
|
|
|
|
// in its own thread.
|
|
|
|
|
health_check_cq = |
|
|
|
|
new ServerCompletionQueue(GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING, nullptr); |
|
|
|
|
health_check_cq = new grpc::ServerCompletionQueue( |
|
|
|
|
GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING, nullptr); |
|
|
|
|
grpc_server_register_completion_queue(server_, health_check_cq->cq(), |
|
|
|
|
nullptr); |
|
|
|
|
default_health_check_service_impl = |
|
|
|
|
default_hc_service->GetHealthCheckService( |
|
|
|
|
std::unique_ptr<ServerCompletionQueue>(health_check_cq)); |
|
|
|
|
std::unique_ptr<grpc::ServerCompletionQueue>(health_check_cq)); |
|
|
|
|
RegisterService(nullptr, default_health_check_service_impl); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -1012,7 +1192,8 @@ void Server::Start(ServerCompletionQueue** cqs, size_t num_cqs) { |
|
|
|
|
// service to handle any unimplemented methods using the default reactor
|
|
|
|
|
// creator
|
|
|
|
|
if (!callback_reqs_to_start_.empty() && !has_callback_generic_service_) { |
|
|
|
|
unimplemented_service_.reset(new experimental::CallbackGenericService); |
|
|
|
|
unimplemented_service_.reset( |
|
|
|
|
new grpc::experimental::CallbackGenericService); |
|
|
|
|
RegisterCallbackGenericService(unimplemented_service_.get()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -1037,7 +1218,8 @@ void Server::Start(ServerCompletionQueue** cqs, size_t num_cqs) { |
|
|
|
|
// server CQs), make sure that we have a ResourceExhausted handler
|
|
|
|
|
// to deal with the case of thread exhaustion
|
|
|
|
|
if (sync_server_cqs_ != nullptr && !sync_server_cqs_->empty()) { |
|
|
|
|
resource_exhausted_handler_.reset(new internal::ResourceExhaustedHandler); |
|
|
|
|
resource_exhausted_handler_.reset( |
|
|
|
|
new grpc::internal::ResourceExhaustedHandler); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
for (auto it = sync_req_mgrs_.begin(); it != sync_req_mgrs_.end(); it++) { |
|
|
|
|
@ -1063,20 +1245,20 @@ void Server::ShutdownInternal(gpr_timespec deadline) { |
|
|
|
|
shutdown_ = true; |
|
|
|
|
|
|
|
|
|
/// The completion queue to use for server shutdown completion notification
|
|
|
|
|
CompletionQueue shutdown_cq; |
|
|
|
|
ShutdownTag shutdown_tag; // Dummy shutdown tag
|
|
|
|
|
grpc::CompletionQueue shutdown_cq; |
|
|
|
|
grpc::ShutdownTag shutdown_tag; // Dummy shutdown tag
|
|
|
|
|
grpc_server_shutdown_and_notify(server_, shutdown_cq.cq(), &shutdown_tag); |
|
|
|
|
|
|
|
|
|
shutdown_cq.Shutdown(); |
|
|
|
|
|
|
|
|
|
void* tag; |
|
|
|
|
bool ok; |
|
|
|
|
CompletionQueue::NextStatus status = |
|
|
|
|
grpc::CompletionQueue::NextStatus status = |
|
|
|
|
shutdown_cq.AsyncNext(&tag, &ok, deadline); |
|
|
|
|
|
|
|
|
|
// If this timed out, it means we are done with the grace period for a clean
|
|
|
|
|
// shutdown. We should force a shutdown now by cancelling all inflight calls
|
|
|
|
|
if (status == CompletionQueue::NextStatus::TIMEOUT) { |
|
|
|
|
if (status == grpc::CompletionQueue::NextStatus::TIMEOUT) { |
|
|
|
|
grpc_server_cancel_all_calls(server_); |
|
|
|
|
} |
|
|
|
|
// Else in case of SHUTDOWN or GOT_EVENT, it means that the server has
|
|
|
|
|
@ -1128,154 +1310,11 @@ void Server::Wait() { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Server::PerformOpsOnCall(internal::CallOpSetInterface* ops, |
|
|
|
|
internal::Call* call) { |
|
|
|
|
void Server::PerformOpsOnCall(grpc::internal::CallOpSetInterface* ops, |
|
|
|
|
grpc::internal::Call* call) { |
|
|
|
|
ops->FillOps(call); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ServerInterface::BaseAsyncRequest::BaseAsyncRequest( |
|
|
|
|
ServerInterface* server, ServerContext* context, |
|
|
|
|
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, |
|
|
|
|
ServerCompletionQueue* notification_cq, void* tag, bool delete_on_finalize) |
|
|
|
|
: server_(server), |
|
|
|
|
context_(context), |
|
|
|
|
stream_(stream), |
|
|
|
|
call_cq_(call_cq), |
|
|
|
|
notification_cq_(notification_cq), |
|
|
|
|
tag_(tag), |
|
|
|
|
delete_on_finalize_(delete_on_finalize), |
|
|
|
|
call_(nullptr), |
|
|
|
|
done_intercepting_(false) { |
|
|
|
|
/* Set up interception state partially for the receive ops. call_wrapper_ is
|
|
|
|
|
* not filled at this point, but it will be filled before the interceptors are |
|
|
|
|
* run. */ |
|
|
|
|
interceptor_methods_.SetCall(&call_wrapper_); |
|
|
|
|
interceptor_methods_.SetReverse(); |
|
|
|
|
call_cq_->RegisterAvalanching(); // This op will trigger more ops
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ServerInterface::BaseAsyncRequest::~BaseAsyncRequest() { |
|
|
|
|
call_cq_->CompleteAvalanching(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag, |
|
|
|
|
bool* status) { |
|
|
|
|
if (done_intercepting_) { |
|
|
|
|
*tag = tag_; |
|
|
|
|
if (delete_on_finalize_) { |
|
|
|
|
delete this; |
|
|
|
|
} |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
context_->set_call(call_); |
|
|
|
|
context_->cq_ = call_cq_; |
|
|
|
|
if (call_wrapper_.call() == nullptr) { |
|
|
|
|
// Fill it since it is empty.
|
|
|
|
|
call_wrapper_ = internal::Call( |
|
|
|
|
call_, server_, call_cq_, server_->max_receive_message_size(), nullptr); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// just the pointers inside call are copied here
|
|
|
|
|
stream_->BindCall(&call_wrapper_); |
|
|
|
|
|
|
|
|
|
if (*status && call_ && call_wrapper_.server_rpc_info()) { |
|
|
|
|
done_intercepting_ = true; |
|
|
|
|
// Set interception point for RECV INITIAL METADATA
|
|
|
|
|
interceptor_methods_.AddInterceptionHookPoint( |
|
|
|
|
experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA); |
|
|
|
|
interceptor_methods_.SetRecvInitialMetadata(&context_->client_metadata_); |
|
|
|
|
if (interceptor_methods_.RunInterceptors( |
|
|
|
|
[this]() { ContinueFinalizeResultAfterInterception(); })) { |
|
|
|
|
// There are no interceptors to run. Continue
|
|
|
|
|
} else { |
|
|
|
|
// There were interceptors to be run, so
|
|
|
|
|
// ContinueFinalizeResultAfterInterception will be run when interceptors
|
|
|
|
|
// are done.
|
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
if (*status && call_) { |
|
|
|
|
context_->BeginCompletionOp(&call_wrapper_, nullptr, nullptr); |
|
|
|
|
} |
|
|
|
|
*tag = tag_; |
|
|
|
|
if (delete_on_finalize_) { |
|
|
|
|
delete this; |
|
|
|
|
} |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void ServerInterface::BaseAsyncRequest:: |
|
|
|
|
ContinueFinalizeResultAfterInterception() { |
|
|
|
|
context_->BeginCompletionOp(&call_wrapper_, nullptr, nullptr); |
|
|
|
|
// Queue a tag which will be returned immediately
|
|
|
|
|
grpc_core::ExecCtx exec_ctx; |
|
|
|
|
grpc_cq_begin_op(notification_cq_->cq(), this); |
|
|
|
|
grpc_cq_end_op( |
|
|
|
|
notification_cq_->cq(), this, GRPC_ERROR_NONE, |
|
|
|
|
[](void* arg, grpc_cq_completion* completion) { delete completion; }, |
|
|
|
|
nullptr, new grpc_cq_completion()); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ServerInterface::RegisteredAsyncRequest::RegisteredAsyncRequest( |
|
|
|
|
ServerInterface* server, ServerContext* context, |
|
|
|
|
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, |
|
|
|
|
ServerCompletionQueue* notification_cq, void* tag, const char* name, |
|
|
|
|
internal::RpcMethod::RpcType type) |
|
|
|
|
: BaseAsyncRequest(server, context, stream, call_cq, notification_cq, tag, |
|
|
|
|
true), |
|
|
|
|
name_(name), |
|
|
|
|
type_(type) {} |
|
|
|
|
|
|
|
|
|
void ServerInterface::RegisteredAsyncRequest::IssueRequest( |
|
|
|
|
void* registered_method, grpc_byte_buffer** payload, |
|
|
|
|
ServerCompletionQueue* notification_cq) { |
|
|
|
|
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_registered_call( |
|
|
|
|
server_->server(), registered_method, &call_, |
|
|
|
|
&context_->deadline_, |
|
|
|
|
context_->client_metadata_.arr(), payload, |
|
|
|
|
call_cq_->cq(), notification_cq->cq(), this)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ServerInterface::GenericAsyncRequest::GenericAsyncRequest( |
|
|
|
|
ServerInterface* server, GenericServerContext* context, |
|
|
|
|
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, |
|
|
|
|
ServerCompletionQueue* notification_cq, void* tag, bool delete_on_finalize) |
|
|
|
|
: BaseAsyncRequest(server, context, stream, call_cq, notification_cq, tag, |
|
|
|
|
delete_on_finalize) { |
|
|
|
|
grpc_call_details_init(&call_details_); |
|
|
|
|
GPR_ASSERT(notification_cq); |
|
|
|
|
GPR_ASSERT(call_cq); |
|
|
|
|
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call( |
|
|
|
|
server->server(), &call_, &call_details_, |
|
|
|
|
context->client_metadata_.arr(), call_cq->cq(), |
|
|
|
|
notification_cq->cq(), this)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
bool ServerInterface::GenericAsyncRequest::FinalizeResult(void** tag, |
|
|
|
|
bool* status) { |
|
|
|
|
// If we are done intercepting, there is nothing more for us to do
|
|
|
|
|
if (done_intercepting_) { |
|
|
|
|
return BaseAsyncRequest::FinalizeResult(tag, status); |
|
|
|
|
} |
|
|
|
|
// TODO(yangg) remove the copy here.
|
|
|
|
|
if (*status) { |
|
|
|
|
static_cast<GenericServerContext*>(context_)->method_ = |
|
|
|
|
StringFromCopiedSlice(call_details_.method); |
|
|
|
|
static_cast<GenericServerContext*>(context_)->host_ = |
|
|
|
|
StringFromCopiedSlice(call_details_.host); |
|
|
|
|
context_->deadline_ = call_details_.deadline; |
|
|
|
|
} |
|
|
|
|
grpc_slice_unref(call_details_.method); |
|
|
|
|
grpc_slice_unref(call_details_.host); |
|
|
|
|
call_wrapper_ = internal::Call( |
|
|
|
|
call_, server_, call_cq_, server_->max_receive_message_size(), |
|
|
|
|
context_->set_server_rpc_info( |
|
|
|
|
static_cast<GenericServerContext*>(context_)->method_.c_str(), |
|
|
|
|
internal::RpcMethod::BIDI_STREAMING, |
|
|
|
|
*server_->interceptor_creators())); |
|
|
|
|
return BaseAsyncRequest::FinalizeResult(tag, status); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
bool Server::UnimplementedAsyncRequest::FinalizeResult(void** tag, |
|
|
|
|
bool* status) { |
|
|
|
|
if (GenericAsyncRequest::FinalizeResult(tag, status)) { |
|
|
|
|
@ -1295,41 +1334,22 @@ bool Server::UnimplementedAsyncRequest::FinalizeResult(void** tag, |
|
|
|
|
Server::UnimplementedAsyncResponse::UnimplementedAsyncResponse( |
|
|
|
|
UnimplementedAsyncRequest* request) |
|
|
|
|
: request_(request) { |
|
|
|
|
Status status(StatusCode::UNIMPLEMENTED, ""); |
|
|
|
|
internal::UnknownMethodHandler::FillOps(request_->context(), this); |
|
|
|
|
grpc::Status status(grpc::StatusCode::UNIMPLEMENTED, ""); |
|
|
|
|
grpc::internal::UnknownMethodHandler::FillOps(request_->context(), this); |
|
|
|
|
request_->stream()->call_.PerformOps(this); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ServerInitializer* Server::initializer() { return server_initializer_.get(); } |
|
|
|
|
|
|
|
|
|
namespace { |
|
|
|
|
class ShutdownCallback : public grpc_experimental_completion_queue_functor { |
|
|
|
|
public: |
|
|
|
|
ShutdownCallback() { functor_run = &ShutdownCallback::Run; } |
|
|
|
|
// TakeCQ takes ownership of the cq into the shutdown callback
|
|
|
|
|
// so that the shutdown callback will be responsible for destroying it
|
|
|
|
|
void TakeCQ(CompletionQueue* cq) { cq_ = cq; } |
|
|
|
|
|
|
|
|
|
// The Run function will get invoked by the completion queue library
|
|
|
|
|
// when the shutdown is actually complete
|
|
|
|
|
static void Run(grpc_experimental_completion_queue_functor* cb, int) { |
|
|
|
|
auto* callback = static_cast<ShutdownCallback*>(cb); |
|
|
|
|
delete callback->cq_; |
|
|
|
|
delete callback; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
CompletionQueue* cq_ = nullptr; |
|
|
|
|
}; |
|
|
|
|
} // namespace
|
|
|
|
|
grpc::ServerInitializer* Server::initializer() { |
|
|
|
|
return server_initializer_.get(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
CompletionQueue* Server::CallbackCQ() { |
|
|
|
|
grpc::CompletionQueue* Server::CallbackCQ() { |
|
|
|
|
// TODO(vjpai): Consider using a single global CQ for the default CQ
|
|
|
|
|
// if there is no explicit per-server CQ registered
|
|
|
|
|
grpc::internal::MutexLock l(&mu_); |
|
|
|
|
if (callback_cq_ == nullptr) { |
|
|
|
|
auto* shutdown_callback = new ShutdownCallback; |
|
|
|
|
callback_cq_ = new CompletionQueue(grpc_completion_queue_attributes{ |
|
|
|
|
auto* shutdown_callback = new grpc::ShutdownCallback; |
|
|
|
|
callback_cq_ = new grpc::CompletionQueue(grpc_completion_queue_attributes{ |
|
|
|
|
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_CALLBACK, GRPC_CQ_DEFAULT_POLLING, |
|
|
|
|
shutdown_callback}); |
|
|
|
|
|
|
|
|
|
@ -1339,4 +1359,4 @@ CompletionQueue* Server::CallbackCQ() { |
|
|
|
|
return callback_cq_; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} // namespace grpc
|
|
|
|
|
} // namespace grpc_impl
|
|
|
|
|
|
Karthik Ravi Shankar
6 years ago