2 // Copyright 2015-2016 gRPC authors.
4 // Licensed under the Apache License, Version 2.0 (the "License");
5 // you may not use this file except in compliance with the License.
6 // You may obtain a copy of the License at
8 // http://www.apache.org/licenses/LICENSE-2.0
10 // Unless required by applicable law or agreed to in writing, software
11 // distributed under the License is distributed on an "AS IS" BASIS,
12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 // See the License for the specific language governing permissions and
14 // limitations under the License.
17 #include <grpc/support/port_platform.h>
19 #include "src/core/lib/surface/server.h"
33 #include "absl/memory/memory.h"
34 #include "absl/types/optional.h"
36 #include <grpc/support/alloc.h>
37 #include <grpc/support/log.h>
38 #include <grpc/support/string_util.h>
40 #include "src/core/lib/channel/channel_args.h"
41 #include "src/core/lib/channel/channelz.h"
42 #include "src/core/lib/channel/connected_channel.h"
43 #include "src/core/lib/debug/stats.h"
44 #include "src/core/lib/gpr/spinlock.h"
45 #include "src/core/lib/gpr/string.h"
46 #include "src/core/lib/gprpp/mpscq.h"
47 #include "src/core/lib/iomgr/executor.h"
48 #include "src/core/lib/iomgr/iomgr.h"
49 #include "src/core/lib/slice/slice_internal.h"
50 #include "src/core/lib/surface/api_trace.h"
51 #include "src/core/lib/surface/call.h"
52 #include "src/core/lib/surface/channel.h"
53 #include "src/core/lib/surface/completion_queue.h"
54 #include "src/core/lib/surface/init.h"
55 #include "src/core/lib/transport/metadata.h"
56 #include "src/core/lib/transport/static_metadata.h"
60 TraceFlag grpc_server_channel_trace(false, "server_channel");
63 // Server::RequestedCall
66 struct Server::RequestedCall {
67 enum class Type { BATCH_CALL, REGISTERED_CALL };
69 RequestedCall(void* tag_arg, grpc_completion_queue* call_cq,
70 grpc_call** call_arg, grpc_metadata_array* initial_md,
71 grpc_call_details* details)
72 : type(Type::BATCH_CALL),
74 cq_bound_to_call(call_cq),
76 initial_metadata(initial_md) {
77 details->reserved = nullptr;
78 data.batch.details = details;
81 RequestedCall(void* tag_arg, grpc_completion_queue* call_cq,
82 grpc_call** call_arg, grpc_metadata_array* initial_md,
83 RegisteredMethod* rm, gpr_timespec* deadline,
84 grpc_byte_buffer** optional_payload)
85 : type(Type::REGISTERED_CALL),
87 cq_bound_to_call(call_cq),
89 initial_metadata(initial_md) {
90 data.registered.method = rm;
91 data.registered.deadline = deadline;
92 data.registered.optional_payload = optional_payload;
95 MultiProducerSingleConsumerQueue::Node mpscq_node;
98 grpc_completion_queue* const cq_bound_to_call;
99 grpc_call** const call;
100 grpc_cq_completion completion;
101 grpc_metadata_array* const initial_metadata;
104 grpc_call_details* details;
107 RegisteredMethod* method;
108 gpr_timespec* deadline;
109 grpc_byte_buffer** optional_payload;
115 // Server::RegisteredMethod
118 struct Server::RegisteredMethod {
120 const char* method_arg, const char* host_arg,
121 grpc_server_register_method_payload_handling payload_handling_arg,
123 : method(method_arg == nullptr ? "" : method_arg),
124 host(host_arg == nullptr ? "" : host_arg),
125 payload_handling(payload_handling_arg),
128 ~RegisteredMethod() = default;
130 const std::string method;
131 const std::string host;
132 const grpc_server_register_method_payload_handling payload_handling;
133 const uint32_t flags;
134 // One request matcher per method.
135 std::unique_ptr<RequestMatcherInterface> matcher;
139 // Server::RequestMatcherInterface
142 // RPCs that come in from the transport must be matched against RPC requests
143 // from the application. An incoming request from the application can be matched
144 // to an RPC that has already arrived or can be queued up for later use.
145 // Likewise, an RPC coming in from the transport can either be matched to a
146 // request that already arrived from the application or can be queued up for
147 // later use (marked pending). If there is a match, the request's tag is posted
148 // on the request's notification CQ.
150 // RequestMatcherInterface is the base class to provide this functionality.
151 class Server::RequestMatcherInterface {
153 virtual ~RequestMatcherInterface() {}
155 // Unref the calls associated with any incoming RPCs in the pending queue (not
156 // yet matched to an application-requested RPC).
157 virtual void ZombifyPending() = 0;
159 // Mark all application-requested RPCs failed if they have not been matched to
160 // an incoming RPC. The error parameter indicates why the RPCs are being
161 // failed (always server shutdown in all current implementations).
162 virtual void KillRequests(grpc_error* error) = 0;
164 // How many request queues are supported by this matcher. This is an abstract
165 // concept that essentially maps to gRPC completion queues.
166 virtual size_t request_queue_count() const = 0;
168 // This function is invoked when the application requests a new RPC whose
169 // information is in the call parameter. The request_queue_index marks the
170 // queue onto which to place this RPC, and is typically associated with a gRPC
171 // CQ. If there are pending RPCs waiting to be matched, publish one (match it
172 // and notify the CQ).
173 virtual void RequestCallWithPossiblePublish(size_t request_queue_index,
174 RequestedCall* call) = 0;
176 // This function is invoked on an incoming RPC, represented by the calld
177 // object. The RequestMatcher will try to match it against an
178 // application-requested RPC if possible or will place it in the pending queue
179 // otherwise. To enable some measure of fairness between server CQs, the match
180 // is done starting at the start_request_queue_index parameter in a cyclic
181 // order rather than always starting at 0.
182 virtual void MatchOrQueue(size_t start_request_queue_index,
183 CallData* calld) = 0;
185 // Returns the server associated with this request matcher
186 virtual Server* server() const = 0;
189 // The RealRequestMatcher is an implementation of RequestMatcherInterface that
190 // actually uses all the features of RequestMatcherInterface: expecting the
191 // application to explicitly request RPCs and then matching those to incoming
192 // RPCs, along with a slow path by which incoming RPCs are put on a locked
193 // pending list if they aren't able to be matched to an application request.
194 class Server::RealRequestMatcher : public RequestMatcherInterface {
196 explicit RealRequestMatcher(Server* server)
197 : server_(server), requests_per_cq_(server->cqs_.size()) {}
199 ~RealRequestMatcher() override {
200 for (LockedMultiProducerSingleConsumerQueue& queue : requests_per_cq_) {
201 GPR_ASSERT(queue.Pop() == nullptr);
205 void ZombifyPending() override {
206 while (!pending_.empty()) {
207 CallData* calld = pending_.front();
208 calld->SetState(CallData::CallState::ZOMBIED);
214 void KillRequests(grpc_error* error) override {
215 for (size_t i = 0; i < requests_per_cq_.size(); i++) {
217 while ((rc = reinterpret_cast<RequestedCall*>(
218 requests_per_cq_[i].Pop())) != nullptr) {
219 server_->FailCall(i, rc, GRPC_ERROR_REF(error));
222 GRPC_ERROR_UNREF(error);
225 size_t request_queue_count() const override {
226 return requests_per_cq_.size();
229 void RequestCallWithPossiblePublish(size_t request_queue_index,
230 RequestedCall* call) override {
231 if (requests_per_cq_[request_queue_index].Push(&call->mpscq_node)) {
232 /* this was the first queued request: we need to lock and start
235 RequestedCall* rc = nullptr;
238 auto pop_next_pending = [this, request_queue_index] {
239 PendingCall pending_call;
241 MutexLock lock(&server_->mu_call_);
242 if (!pending_.empty()) {
243 pending_call.rc = reinterpret_cast<RequestedCall*>(
244 requests_per_cq_[request_queue_index].Pop());
245 if (pending_call.rc != nullptr) {
246 pending_call.calld = pending_.front();
254 PendingCall next_pending = pop_next_pending();
255 if (next_pending.rc == nullptr) break;
256 if (!next_pending.calld->MaybeActivate()) {
258 next_pending.calld->KillZombie();
260 next_pending.calld->Publish(request_queue_index, next_pending.rc);
266 void MatchOrQueue(size_t start_request_queue_index,
267 CallData* calld) override {
268 for (size_t i = 0; i < requests_per_cq_.size(); i++) {
269 size_t cq_idx = (start_request_queue_index + i) % requests_per_cq_.size();
271 reinterpret_cast<RequestedCall*>(requests_per_cq_[cq_idx].TryPop());
273 GRPC_STATS_INC_SERVER_CQS_CHECKED(i);
274 calld->SetState(CallData::CallState::ACTIVATED);
275 calld->Publish(cq_idx, rc);
279 // No cq to take the request found; queue it on the slow list.
280 GRPC_STATS_INC_SERVER_SLOWPATH_REQUESTS_QUEUED();
281 // We need to ensure that all the queues are empty. We do this under
282 // the server mu_call_ lock to ensure that if something is added to
283 // an empty request queue, it will block until the call is actually
284 // added to the pending list.
285 RequestedCall* rc = nullptr;
289 MutexLock lock(&server_->mu_call_);
290 for (loop_count = 0; loop_count < requests_per_cq_.size(); loop_count++) {
292 (start_request_queue_index + loop_count) % requests_per_cq_.size();
293 rc = reinterpret_cast<RequestedCall*>(requests_per_cq_[cq_idx].Pop());
299 calld->SetState(CallData::CallState::PENDING);
300 pending_.push(calld);
304 GRPC_STATS_INC_SERVER_CQS_CHECKED(loop_count + requests_per_cq_.size());
305 calld->SetState(CallData::CallState::ACTIVATED);
306 calld->Publish(cq_idx, rc);
309 Server* server() const override { return server_; }
312 Server* const server_;
313 std::queue<CallData*> pending_;
314 std::vector<LockedMultiProducerSingleConsumerQueue> requests_per_cq_;
317 // AllocatingRequestMatchers don't allow the application to request an RPC in
318 // advance or queue up any incoming RPC for later match. Instead, MatchOrQueue
319 // will call out to an allocation function passed in at the construction of the
320 // object. These request matchers are designed for the C++ callback API, so they
321 // only support 1 completion queue (passed in at the constructor).
322 class Server::AllocatingRequestMatcherBase : public RequestMatcherInterface {
324 AllocatingRequestMatcherBase(Server* server, grpc_completion_queue* cq)
325 : server_(server), cq_(cq) {
327 for (idx = 0; idx < server->cqs_.size(); idx++) {
328 if (server->cqs_[idx] == cq) {
332 GPR_ASSERT(idx < server->cqs_.size());
336 void ZombifyPending() override {}
338 void KillRequests(grpc_error* error) override { GRPC_ERROR_UNREF(error); }
340 size_t request_queue_count() const override { return 0; }
342 void RequestCallWithPossiblePublish(size_t /*request_queue_index*/,
343 RequestedCall* /*call*/) final {
347 Server* server() const override { return server_; }
349 // Supply the completion queue related to this request matcher
350 grpc_completion_queue* cq() const { return cq_; }
352 // Supply the completion queue's index relative to the server.
353 size_t cq_idx() const { return cq_idx_; }
356 Server* const server_;
357 grpc_completion_queue* const cq_;
361 // An allocating request matcher for non-registered methods (used for generic
362 // API and unimplemented RPCs).
363 class Server::AllocatingRequestMatcherBatch
364 : public AllocatingRequestMatcherBase {
366 AllocatingRequestMatcherBatch(Server* server, grpc_completion_queue* cq,
367 std::function<BatchCallAllocation()> allocator)
368 : AllocatingRequestMatcherBase(server, cq),
369 allocator_(std::move(allocator)) {}
371 void MatchOrQueue(size_t /*start_request_queue_index*/,
372 CallData* calld) override {
373 BatchCallAllocation call_info = allocator_();
374 GPR_ASSERT(server()->ValidateServerRequest(
375 cq(), static_cast<void*>(call_info.tag), nullptr, nullptr) ==
377 RequestedCall* rc = new RequestedCall(
378 static_cast<void*>(call_info.tag), cq(), call_info.call,
379 call_info.initial_metadata, call_info.details);
380 calld->SetState(CallData::CallState::ACTIVATED);
381 calld->Publish(cq_idx(), rc);
385 std::function<BatchCallAllocation()> allocator_;
388 // An allocating request matcher for registered methods.
389 class Server::AllocatingRequestMatcherRegistered
390 : public AllocatingRequestMatcherBase {
392 AllocatingRequestMatcherRegistered(
393 Server* server, grpc_completion_queue* cq, RegisteredMethod* rm,
394 std::function<RegisteredCallAllocation()> allocator)
395 : AllocatingRequestMatcherBase(server, cq),
396 registered_method_(rm),
397 allocator_(std::move(allocator)) {}
399 void MatchOrQueue(size_t /*start_request_queue_index*/,
400 CallData* calld) override {
401 RegisteredCallAllocation call_info = allocator_();
403 server()->ValidateServerRequest(cq(), static_cast<void*>(call_info.tag),
404 call_info.optional_payload,
405 registered_method_) == GRPC_CALL_OK);
406 RequestedCall* rc = new RequestedCall(
407 static_cast<void*>(call_info.tag), cq(), call_info.call,
408 call_info.initial_metadata, registered_method_, call_info.deadline,
409 call_info.optional_payload);
410 calld->SetState(CallData::CallState::ACTIVATED);
411 calld->Publish(cq_idx(), rc);
415 RegisteredMethod* const registered_method_;
416 std::function<RegisteredCallAllocation()> allocator_;
420 // ChannelBroadcaster
425 class ChannelBroadcaster {
427 // This can have an empty constructor and destructor since we want to control
428 // when the actual setup and shutdown broadcast take place.
430 // Copies over the channels from the locked server.
431 void FillChannelsLocked(std::vector<grpc_channel*> channels) {
432 GPR_DEBUG_ASSERT(channels_.empty());
433 channels_ = std::move(channels);
436 // Broadcasts a shutdown on each channel.
437 void BroadcastShutdown(bool send_goaway, grpc_error* force_disconnect) {
438 for (grpc_channel* channel : channels_) {
439 SendShutdown(channel, send_goaway, GRPC_ERROR_REF(force_disconnect));
440 GRPC_CHANNEL_INTERNAL_UNREF(channel, "broadcast");
442 channels_.clear(); // just for safety against double broadcast
443 GRPC_ERROR_UNREF(force_disconnect);
447 struct ShutdownCleanupArgs {
448 grpc_closure closure;
452 static void ShutdownCleanup(void* arg, grpc_error* /*error*/) {
453 ShutdownCleanupArgs* a = static_cast<ShutdownCleanupArgs*>(arg);
454 grpc_slice_unref_internal(a->slice);
458 static void SendShutdown(grpc_channel* channel, bool send_goaway,
459 grpc_error* send_disconnect) {
460 ShutdownCleanupArgs* sc = new ShutdownCleanupArgs;
461 GRPC_CLOSURE_INIT(&sc->closure, ShutdownCleanup, sc,
462 grpc_schedule_on_exec_ctx);
463 grpc_transport_op* op = grpc_make_transport_op(&sc->closure);
464 grpc_channel_element* elem;
467 ? grpc_error_set_int(
468 GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server shutdown"),
469 GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_OK)
471 op->set_accept_stream = true;
472 sc->slice = grpc_slice_from_copied_string("Server shutdown");
473 op->disconnect_with_error = send_disconnect;
475 grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
476 elem->filter->start_transport_op(elem, op);
479 std::vector<grpc_channel*> channels_;
488 const grpc_channel_filter Server::kServerTopFilter = {
489 Server::CallData::StartTransportStreamOpBatch,
490 grpc_channel_next_op,
491 sizeof(Server::CallData),
492 Server::CallData::InitCallElement,
493 grpc_call_stack_ignore_set_pollset_or_pollset_set,
494 Server::CallData::DestroyCallElement,
495 sizeof(Server::ChannelData),
496 Server::ChannelData::InitChannelElement,
497 Server::ChannelData::DestroyChannelElement,
498 grpc_channel_next_get_info,
504 grpc_resource_user* CreateDefaultResourceUser(const grpc_channel_args* args) {
505 if (args != nullptr) {
506 grpc_resource_quota* resource_quota =
507 grpc_resource_quota_from_channel_args(args, false /* create */);
508 if (resource_quota != nullptr) {
509 return grpc_resource_user_create(resource_quota, "default");
515 RefCountedPtr<channelz::ServerNode> CreateChannelzNode(
516 const grpc_channel_args* args) {
517 RefCountedPtr<channelz::ServerNode> channelz_node;
518 if (grpc_channel_args_find_bool(args, GRPC_ARG_ENABLE_CHANNELZ,
519 GRPC_ENABLE_CHANNELZ_DEFAULT)) {
520 size_t channel_tracer_max_memory = grpc_channel_args_find_integer(
521 args, GRPC_ARG_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE,
522 {GRPC_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE_DEFAULT, 0, INT_MAX});
524 MakeRefCounted<channelz::ServerNode>(channel_tracer_max_memory);
525 channelz_node->AddTraceEvent(
526 channelz::ChannelTrace::Severity::Info,
527 grpc_slice_from_static_string("Server created"));
529 return channelz_node;
534 Server::Server(const grpc_channel_args* args)
535 : channel_args_(grpc_channel_args_copy(args)),
536 default_resource_user_(CreateDefaultResourceUser(args)),
537 channelz_node_(CreateChannelzNode(args)) {}
540 grpc_channel_args_destroy(channel_args_);
541 // Remove the cq pollsets from the config_fetcher.
542 if (started_ && config_fetcher_ != nullptr &&
543 config_fetcher_->interested_parties() != nullptr) {
544 for (grpc_pollset* pollset : pollsets_) {
545 grpc_pollset_set_del_pollset(config_fetcher_->interested_parties(),
549 for (size_t i = 0; i < cqs_.size(); i++) {
550 GRPC_CQ_INTERNAL_UNREF(cqs_[i], "server");
554 void Server::AddListener(OrphanablePtr<ListenerInterface> listener) {
555 channelz::ListenSocketNode* listen_socket_node =
556 listener->channelz_listen_socket_node();
557 if (listen_socket_node != nullptr && channelz_node_ != nullptr) {
558 channelz_node_->AddChildListenSocket(listen_socket_node->Ref());
560 listeners_.emplace_back(std::move(listener));
563 void Server::Start() {
565 for (grpc_completion_queue* cq : cqs_) {
566 if (grpc_cq_can_listen(cq)) {
567 pollsets_.push_back(grpc_cq_pollset(cq));
570 if (unregistered_request_matcher_ == nullptr) {
571 unregistered_request_matcher_ = absl::make_unique<RealRequestMatcher>(this);
573 for (std::unique_ptr<RegisteredMethod>& rm : registered_methods_) {
574 if (rm->matcher == nullptr) {
575 rm->matcher = absl::make_unique<RealRequestMatcher>(this);
579 MutexLock lock(&mu_global_);
582 // Register the interested parties from the config fetcher to the cq pollsets
583 // before starting listeners so that config fetcher is being polled when the
584 // listeners start watch the fetcher.
585 if (config_fetcher_ != nullptr &&
586 config_fetcher_->interested_parties() != nullptr) {
587 for (grpc_pollset* pollset : pollsets_) {
588 grpc_pollset_set_add_pollset(config_fetcher_->interested_parties(),
592 for (auto& listener : listeners_) {
593 listener.listener->Start(this, &pollsets_);
595 MutexLock lock(&mu_global_);
597 starting_cv_.Signal();
600 grpc_error* Server::SetupTransport(
601 grpc_transport* transport, grpc_pollset* accepting_pollset,
602 const grpc_channel_args* args,
603 const RefCountedPtr<grpc_core::channelz::SocketNode>& socket_node,
604 grpc_resource_user* resource_user) {
606 grpc_error* error = GRPC_ERROR_NONE;
607 grpc_channel* channel = grpc_channel_create(
608 nullptr, args, GRPC_SERVER_CHANNEL, transport, resource_user, &error);
609 if (channel == nullptr) {
612 ChannelData* chand = static_cast<ChannelData*>(
613 grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0)
617 for (cq_idx = 0; cq_idx < cqs_.size(); cq_idx++) {
618 if (grpc_cq_pollset(cqs_[cq_idx]) == accepting_pollset) break;
620 if (cq_idx == cqs_.size()) {
621 // Completion queue not found. Pick a random one to publish new calls to.
622 cq_idx = static_cast<size_t>(rand()) % cqs_.size();
624 // Set up channelz node.
625 intptr_t channelz_socket_uuid = 0;
626 if (socket_node != nullptr) {
627 channelz_socket_uuid = socket_node->uuid();
628 channelz_node_->AddChildSocket(socket_node);
631 chand->InitTransport(Ref(), channel, cq_idx, transport, channelz_socket_uuid);
632 return GRPC_ERROR_NONE;
635 bool Server::HasOpenConnections() {
636 MutexLock lock(&mu_global_);
637 return !channels_.empty();
640 void Server::SetRegisteredMethodAllocator(
641 grpc_completion_queue* cq, void* method_tag,
642 std::function<RegisteredCallAllocation()> allocator) {
643 RegisteredMethod* rm = static_cast<RegisteredMethod*>(method_tag);
644 rm->matcher = absl::make_unique<AllocatingRequestMatcherRegistered>(
645 this, cq, rm, std::move(allocator));
648 void Server::SetBatchMethodAllocator(
649 grpc_completion_queue* cq, std::function<BatchCallAllocation()> allocator) {
650 GPR_DEBUG_ASSERT(unregistered_request_matcher_ == nullptr);
651 unregistered_request_matcher_ =
652 absl::make_unique<AllocatingRequestMatcherBatch>(this, cq,
653 std::move(allocator));
656 void Server::RegisterCompletionQueue(grpc_completion_queue* cq) {
657 for (grpc_completion_queue* queue : cqs_) {
658 if (queue == cq) return;
660 GRPC_CQ_INTERNAL_REF(cq, "server");
666 bool streq(const std::string& a, const char* b) {
667 return (a.empty() && b == nullptr) ||
668 ((b != nullptr) && !strcmp(a.c_str(), b));
673 Server::RegisteredMethod* Server::RegisterMethod(
674 const char* method, const char* host,
675 grpc_server_register_method_payload_handling payload_handling,
679 "grpc_server_register_method method string cannot be NULL");
682 for (std::unique_ptr<RegisteredMethod>& m : registered_methods_) {
683 if (streq(m->method, method) && streq(m->host, host)) {
684 gpr_log(GPR_ERROR, "duplicate registration for %s@%s", method,
689 if ((flags & ~GRPC_INITIAL_METADATA_USED_MASK) != 0) {
690 gpr_log(GPR_ERROR, "grpc_server_register_method invalid flags 0x%08x",
694 registered_methods_.emplace_back(absl::make_unique<RegisteredMethod>(
695 method, host, payload_handling, flags));
696 return registered_methods_.back().get();
699 void Server::DoneRequestEvent(void* req, grpc_cq_completion* /*c*/) {
700 delete static_cast<RequestedCall*>(req);
703 void Server::FailCall(size_t cq_idx, RequestedCall* rc, grpc_error* error) {
705 rc->initial_metadata->count = 0;
706 GPR_ASSERT(error != GRPC_ERROR_NONE);
707 grpc_cq_end_op(cqs_[cq_idx], rc->tag, error, DoneRequestEvent, rc,
711 // Before calling MaybeFinishShutdown(), we must hold mu_global_ and not
713 void Server::MaybeFinishShutdown() {
714 if (!shutdown_flag_.load(std::memory_order_acquire) || shutdown_published_) {
718 MutexLock lock(&mu_call_);
719 KillPendingWorkLocked(
720 GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
722 if (!channels_.empty() || listeners_destroyed_ < listeners_.size()) {
723 if (gpr_time_cmp(gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME),
724 last_shutdown_message_time_),
725 gpr_time_from_seconds(1, GPR_TIMESPAN)) >= 0) {
726 last_shutdown_message_time_ = gpr_now(GPR_CLOCK_REALTIME);
728 "Waiting for %" PRIuPTR " channels and %" PRIuPTR "/%" PRIuPTR
729 " listeners to be destroyed before shutting down server",
730 channels_.size(), listeners_.size() - listeners_destroyed_,
735 shutdown_published_ = true;
736 for (auto& shutdown_tag : shutdown_tags_) {
738 grpc_cq_end_op(shutdown_tag.cq, shutdown_tag.tag, GRPC_ERROR_NONE,
739 DoneShutdownEvent, this, &shutdown_tag.completion);
743 void Server::KillPendingWorkLocked(grpc_error* error) {
745 unregistered_request_matcher_->KillRequests(GRPC_ERROR_REF(error));
746 unregistered_request_matcher_->ZombifyPending();
747 for (std::unique_ptr<RegisteredMethod>& rm : registered_methods_) {
748 rm->matcher->KillRequests(GRPC_ERROR_REF(error));
749 rm->matcher->ZombifyPending();
752 GRPC_ERROR_UNREF(error);
755 std::vector<grpc_channel*> Server::GetChannelsLocked() const {
756 std::vector<grpc_channel*> channels;
757 channels.reserve(channels_.size());
758 for (const ChannelData* chand : channels_) {
759 channels.push_back(chand->channel());
760 GRPC_CHANNEL_INTERNAL_REF(chand->channel(), "broadcast");
765 void Server::ListenerDestroyDone(void* arg, grpc_error* /*error*/) {
766 Server* server = static_cast<Server*>(arg);
767 MutexLock lock(&server->mu_global_);
768 server->listeners_destroyed_++;
769 server->MaybeFinishShutdown();
774 void DonePublishedShutdown(void* /*done_arg*/, grpc_cq_completion* storage) {
780 // - Kills all pending requests-for-incoming-RPC-calls (i.e., the requests made
781 // via grpc_server_request_call() and grpc_server_request_registered_call()
782 // will now be cancelled). See KillPendingWorkLocked().
784 // - Shuts down the listeners (i.e., the server will no longer listen on the
785 // port for new incoming channels).
787 // - Iterates through all channels on the server and sends shutdown msg (see
788 // ChannelBroadcaster::BroadcastShutdown() for details) to the clients via
789 // the transport layer. The transport layer then guarantees the following:
790 // -- Sends shutdown to the client (e.g., HTTP2 transport sends GOAWAY).
791 // -- If the server has outstanding calls that are in the process, the
792 // connection is NOT closed until the server is done with all those calls.
793 // -- Once there are no more calls in progress, the channel is closed.
794 void Server::ShutdownAndNotify(grpc_completion_queue* cq, void* tag) {
795 ChannelBroadcaster broadcaster;
797 // Wait for startup to be finished. Locks mu_global.
798 MutexLock lock(&mu_global_);
799 WaitUntil(&starting_cv_, &mu_global_, [this] { return !starting_; });
800 // Stay locked, and gather up some stuff to do.
801 GPR_ASSERT(grpc_cq_begin_op(cq, tag));
802 if (shutdown_published_) {
803 grpc_cq_end_op(cq, tag, GRPC_ERROR_NONE, DonePublishedShutdown, nullptr,
804 new grpc_cq_completion);
807 shutdown_tags_.emplace_back(tag, cq);
808 if (shutdown_flag_.load(std::memory_order_acquire)) {
811 last_shutdown_message_time_ = gpr_now(GPR_CLOCK_REALTIME);
812 broadcaster.FillChannelsLocked(GetChannelsLocked());
813 shutdown_flag_.store(true, std::memory_order_release);
814 // Collect all unregistered then registered calls.
816 MutexLock lock(&mu_call_);
817 KillPendingWorkLocked(
818 GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
820 MaybeFinishShutdown();
822 // Shutdown listeners.
823 for (auto& listener : listeners_) {
824 channelz::ListenSocketNode* channelz_listen_socket_node =
825 listener.listener->channelz_listen_socket_node();
826 if (channelz_node_ != nullptr && channelz_listen_socket_node != nullptr) {
827 channelz_node_->RemoveChildListenSocket(
828 channelz_listen_socket_node->uuid());
830 GRPC_CLOSURE_INIT(&listener.destroy_done, ListenerDestroyDone, this,
831 grpc_schedule_on_exec_ctx);
832 listener.listener->SetOnDestroyDone(&listener.destroy_done);
833 listener.listener.reset();
835 broadcaster.BroadcastShutdown(/*send_goaway=*/true, GRPC_ERROR_NONE);
838 void Server::CancelAllCalls() {
839 ChannelBroadcaster broadcaster;
841 MutexLock lock(&mu_global_);
842 broadcaster.FillChannelsLocked(GetChannelsLocked());
844 broadcaster.BroadcastShutdown(
845 /*send_goaway=*/false,
846 GRPC_ERROR_CREATE_FROM_STATIC_STRING("Cancelling all calls"));
849 void Server::Orphan() {
851 MutexLock lock(&mu_global_);
852 GPR_ASSERT(shutdown_flag_.load(std::memory_order_acquire) ||
854 GPR_ASSERT(listeners_destroyed_ == listeners_.size());
856 if (default_resource_user_ != nullptr) {
857 grpc_resource_quota_unref(grpc_resource_user_quota(default_resource_user_));
858 grpc_resource_user_shutdown(default_resource_user_);
859 grpc_resource_user_unref(default_resource_user_);
864 grpc_call_error Server::ValidateServerRequest(
865 grpc_completion_queue* cq_for_notification, void* tag,
866 grpc_byte_buffer** optional_payload, RegisteredMethod* rm) {
867 if ((rm == nullptr && optional_payload != nullptr) ||
868 ((rm != nullptr) && ((optional_payload == nullptr) !=
869 (rm->payload_handling == GRPC_SRM_PAYLOAD_NONE)))) {
870 return GRPC_CALL_ERROR_PAYLOAD_TYPE_MISMATCH;
872 if (grpc_cq_begin_op(cq_for_notification, tag) == false) {
873 return GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN;
878 grpc_call_error Server::ValidateServerRequestAndCq(
879 size_t* cq_idx, grpc_completion_queue* cq_for_notification, void* tag,
880 grpc_byte_buffer** optional_payload, RegisteredMethod* rm) {
882 for (idx = 0; idx < cqs_.size(); idx++) {
883 if (cqs_[idx] == cq_for_notification) {
887 if (idx == cqs_.size()) {
888 return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
890 grpc_call_error error =
891 ValidateServerRequest(cq_for_notification, tag, optional_payload, rm);
892 if (error != GRPC_CALL_OK) {
899 grpc_call_error Server::QueueRequestedCall(size_t cq_idx, RequestedCall* rc) {
900 if (shutdown_flag_.load(std::memory_order_acquire)) {
902 GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
905 RequestMatcherInterface* rm;
907 case RequestedCall::Type::BATCH_CALL:
908 rm = unregistered_request_matcher_.get();
910 case RequestedCall::Type::REGISTERED_CALL:
911 rm = rc->data.registered.method->matcher.get();
914 rm->RequestCallWithPossiblePublish(cq_idx, rc);
918 grpc_call_error Server::RequestCall(grpc_call** call,
919 grpc_call_details* details,
920 grpc_metadata_array* request_metadata,
921 grpc_completion_queue* cq_bound_to_call,
922 grpc_completion_queue* cq_for_notification,
925 grpc_call_error error = ValidateServerRequestAndCq(
926 &cq_idx, cq_for_notification, tag, nullptr, nullptr);
927 if (error != GRPC_CALL_OK) {
931 new RequestedCall(tag, cq_bound_to_call, call, request_metadata, details);
932 return QueueRequestedCall(cq_idx, rc);
935 grpc_call_error Server::RequestRegisteredCall(
936 RegisteredMethod* rm, grpc_call** call, gpr_timespec* deadline,
937 grpc_metadata_array* request_metadata, grpc_byte_buffer** optional_payload,
938 grpc_completion_queue* cq_bound_to_call,
939 grpc_completion_queue* cq_for_notification, void* tag_new) {
941 grpc_call_error error = ValidateServerRequestAndCq(
942 &cq_idx, cq_for_notification, tag_new, optional_payload, rm);
943 if (error != GRPC_CALL_OK) {
947 new RequestedCall(tag_new, cq_bound_to_call, call, request_metadata, rm,
948 deadline, optional_payload);
949 return QueueRequestedCall(cq_idx, rc);
953 // Server::ChannelData::ConnectivityWatcher
956 class Server::ChannelData::ConnectivityWatcher
957 : public AsyncConnectivityStateWatcherInterface {
959 explicit ConnectivityWatcher(ChannelData* chand) : chand_(chand) {
960 GRPC_CHANNEL_INTERNAL_REF(chand_->channel_, "connectivity");
963 ~ConnectivityWatcher() override {
964 GRPC_CHANNEL_INTERNAL_UNREF(chand_->channel_, "connectivity");
968 void OnConnectivityStateChange(grpc_connectivity_state new_state,
969 const absl::Status& /*status*/) override {
970 // Don't do anything until we are being shut down.
971 if (new_state != GRPC_CHANNEL_SHUTDOWN) return;
972 // Shut down channel.
973 MutexLock lock(&chand_->server_->mu_global_);
981 // Server::ChannelData
984 Server::ChannelData::~ChannelData() {
985 if (registered_methods_ != nullptr) {
986 for (const ChannelRegisteredMethod& crm : *registered_methods_) {
987 grpc_slice_unref_internal(crm.method);
988 GPR_DEBUG_ASSERT(crm.method.refcount == &kNoopRefcount ||
989 crm.method.refcount == nullptr);
991 grpc_slice_unref_internal(crm.host);
992 GPR_DEBUG_ASSERT(crm.host.refcount == &kNoopRefcount ||
993 crm.host.refcount == nullptr);
996 registered_methods_.reset();
998 if (server_ != nullptr) {
999 if (server_->channelz_node_ != nullptr && channelz_socket_uuid_ != 0) {
1000 server_->channelz_node_->RemoveChildSocket(channelz_socket_uuid_);
1003 MutexLock lock(&server_->mu_global_);
1004 if (list_position_.has_value()) {
1005 server_->channels_.erase(*list_position_);
1006 list_position_.reset();
1008 server_->MaybeFinishShutdown();
1013 void Server::ChannelData::InitTransport(RefCountedPtr<Server> server,
1014 grpc_channel* channel, size_t cq_idx,
1015 grpc_transport* transport,
1016 intptr_t channelz_socket_uuid) {
1017 server_ = std::move(server);
1020 channelz_socket_uuid_ = channelz_socket_uuid;
1021 // Build a lookup table phrased in terms of mdstr's in this channels context
1022 // to quickly find registered methods.
1023 size_t num_registered_methods = server_->registered_methods_.size();
1024 if (num_registered_methods > 0) {
1025 uint32_t max_probes = 0;
1026 size_t slots = 2 * num_registered_methods;
1027 registered_methods_ =
1028 absl::make_unique<std::vector<ChannelRegisteredMethod>>(slots);
1029 for (std::unique_ptr<RegisteredMethod>& rm : server_->registered_methods_) {
1030 ExternallyManagedSlice host;
1031 ExternallyManagedSlice method(rm->method.c_str());
1032 const bool has_host = !rm->host.empty();
1034 host = ExternallyManagedSlice(rm->host.c_str());
1037 GRPC_MDSTR_KV_HASH(has_host ? host.Hash() : 0, method.Hash());
1038 uint32_t probes = 0;
1039 for (probes = 0; (*registered_methods_)[(hash + probes) % slots]
1040 .server_registered_method != nullptr;
1043 if (probes > max_probes) max_probes = probes;
1044 ChannelRegisteredMethod* crm =
1045 &(*registered_methods_)[(hash + probes) % slots];
1046 crm->server_registered_method = rm.get();
1047 crm->flags = rm->flags;
1048 crm->has_host = has_host;
1052 crm->method = method;
1054 GPR_ASSERT(slots <= UINT32_MAX);
1055 registered_method_max_probes_ = max_probes;
1059 MutexLock lock(&server_->mu_global_);
1060 server_->channels_.push_front(this);
1061 list_position_ = server_->channels_.begin();
1063 // Start accept_stream transport op.
1064 grpc_transport_op* op = grpc_make_transport_op(nullptr);
1065 op->set_accept_stream = true;
1066 op->set_accept_stream_fn = AcceptStream;
1067 op->set_accept_stream_user_data = this;
1068 op->start_connectivity_watch = MakeOrphanable<ConnectivityWatcher>(this);
1069 if (server_->shutdown_flag_.load(std::memory_order_acquire)) {
1070 op->disconnect_with_error =
1071 GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server shutdown");
1073 grpc_transport_perform_op(transport, op);
1076 Server::ChannelRegisteredMethod* Server::ChannelData::GetRegisteredMethod(
1077 const grpc_slice& host, const grpc_slice& path, bool is_idempotent) {
1078 if (registered_methods_ == nullptr) return nullptr;
1079 /* TODO(ctiller): unify these two searches */
1080 /* check for an exact match with host */
1081 uint32_t hash = GRPC_MDSTR_KV_HASH(grpc_slice_hash_internal(host),
1082 grpc_slice_hash_internal(path));
1083 for (size_t i = 0; i <= registered_method_max_probes_; i++) {
1084 ChannelRegisteredMethod* rm =
1085 &(*registered_methods_)[(hash + i) % registered_methods_->size()];
1086 if (rm->server_registered_method == nullptr) break;
1087 if (!rm->has_host) continue;
1088 if (rm->host != host) continue;
1089 if (rm->method != path) continue;
1090 if ((rm->flags & GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) &&
1096 /* check for a wildcard method definition (no host set) */
1097 hash = GRPC_MDSTR_KV_HASH(0, grpc_slice_hash_internal(path));
1098 for (size_t i = 0; i <= registered_method_max_probes_; i++) {
1099 ChannelRegisteredMethod* rm =
1100 &(*registered_methods_)[(hash + i) % registered_methods_->size()];
1101 if (rm->server_registered_method == nullptr) break;
1102 if (rm->has_host) continue;
1103 if (rm->method != path) continue;
1104 if ((rm->flags & GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) &&
1113 void Server::ChannelData::AcceptStream(void* arg, grpc_transport* /*transport*/,
1114 const void* transport_server_data) {
1115 auto* chand = static_cast<Server::ChannelData*>(arg);
1117 grpc_call_create_args args;
1118 args.channel = chand->channel_;
1119 args.server = chand->server_.get();
1120 args.parent = nullptr;
1121 args.propagation_mask = 0;
1123 args.pollset_set_alternative = nullptr;
1124 args.server_transport_data = transport_server_data;
1125 args.add_initial_metadata = nullptr;
1126 args.add_initial_metadata_count = 0;
1127 args.send_deadline = GRPC_MILLIS_INF_FUTURE;
1129 grpc_error* error = grpc_call_create(&args, &call);
1130 grpc_call_element* elem =
1131 grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
1132 auto* calld = static_cast<Server::CallData*>(elem->call_data);
1133 if (error != GRPC_ERROR_NONE) {
1134 GRPC_ERROR_UNREF(error);
1135 calld->FailCallCreation();
1141 void Server::ChannelData::FinishDestroy(void* arg, grpc_error* /*error*/) {
1142 auto* chand = static_cast<Server::ChannelData*>(arg);
1143 Server* server = chand->server_.get();
1144 GRPC_CHANNEL_INTERNAL_UNREF(chand->channel_, "server");
1148 void Server::ChannelData::Destroy() {
1149 if (!list_position_.has_value()) return;
1150 GPR_ASSERT(server_ != nullptr);
1151 server_->channels_.erase(*list_position_);
1152 list_position_.reset();
1153 server_->Ref().release();
1154 server_->MaybeFinishShutdown();
1155 GRPC_CLOSURE_INIT(&finish_destroy_channel_closure_, FinishDestroy, this,
1156 grpc_schedule_on_exec_ctx);
1157 if (GRPC_TRACE_FLAG_ENABLED(grpc_server_channel_trace)) {
1158 gpr_log(GPR_INFO, "Disconnected client");
1160 grpc_transport_op* op =
1161 grpc_make_transport_op(&finish_destroy_channel_closure_);
1162 op->set_accept_stream = true;
1163 grpc_channel_next_op(
1164 grpc_channel_stack_element(grpc_channel_get_channel_stack(channel_), 0),
1168 grpc_error* Server::ChannelData::InitChannelElement(
1169 grpc_channel_element* elem, grpc_channel_element_args* args) {
1170 GPR_ASSERT(args->is_first);
1171 GPR_ASSERT(!args->is_last);
1172 new (elem->channel_data) ChannelData();
1173 return GRPC_ERROR_NONE;
1176 void Server::ChannelData::DestroyChannelElement(grpc_channel_element* elem) {
1177 auto* chand = static_cast<ChannelData*>(elem->channel_data);
1178 chand->~ChannelData();
1185 Server::CallData::CallData(grpc_call_element* elem,
1186 const grpc_call_element_args& args,
1187 RefCountedPtr<Server> server)
1188 : server_(std::move(server)),
1189 call_(grpc_call_from_top_element(elem)),
1190 call_combiner_(args.call_combiner) {
1191 GRPC_CLOSURE_INIT(&recv_initial_metadata_ready_, RecvInitialMetadataReady,
1192 elem, grpc_schedule_on_exec_ctx);
1193 GRPC_CLOSURE_INIT(&recv_trailing_metadata_ready_, RecvTrailingMetadataReady,
1194 elem, grpc_schedule_on_exec_ctx);
1197 Server::CallData::~CallData() {
1198 GPR_ASSERT(state_.Load(MemoryOrder::RELAXED) != CallState::PENDING);
1199 GRPC_ERROR_UNREF(recv_initial_metadata_error_);
1200 if (host_.has_value()) {
1201 grpc_slice_unref_internal(*host_);
1203 if (path_.has_value()) {
1204 grpc_slice_unref_internal(*path_);
1206 grpc_metadata_array_destroy(&initial_metadata_);
1207 grpc_byte_buffer_destroy(payload_);
1210 void Server::CallData::SetState(CallState state) {
1211 state_.Store(state, MemoryOrder::RELAXED);
1214 bool Server::CallData::MaybeActivate() {
1215 CallState expected = CallState::PENDING;
1216 return state_.CompareExchangeStrong(&expected, CallState::ACTIVATED,
1217 MemoryOrder::ACQ_REL,
1218 MemoryOrder::RELAXED);
1221 void Server::CallData::FailCallCreation() {
1222 CallState expected_not_started = CallState::NOT_STARTED;
1223 CallState expected_pending = CallState::PENDING;
1224 if (state_.CompareExchangeStrong(&expected_not_started, CallState::ZOMBIED,
1225 MemoryOrder::ACQ_REL,
1226 MemoryOrder::ACQUIRE)) {
1228 } else if (state_.CompareExchangeStrong(&expected_pending, CallState::ZOMBIED,
1229 MemoryOrder::ACQ_REL,
1230 MemoryOrder::RELAXED)) {
1231 // Zombied call will be destroyed when it's removed from the pending
1236 void Server::CallData::Start(grpc_call_element* elem) {
1238 op.op = GRPC_OP_RECV_INITIAL_METADATA;
1240 op.reserved = nullptr;
1241 op.data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_;
1242 GRPC_CLOSURE_INIT(&recv_initial_metadata_batch_complete_,
1243 RecvInitialMetadataBatchComplete, elem,
1244 grpc_schedule_on_exec_ctx);
1245 grpc_call_start_batch_and_execute(call_, &op, 1,
1246 &recv_initial_metadata_batch_complete_);
1249 void Server::CallData::Publish(size_t cq_idx, RequestedCall* rc) {
1250 grpc_call_set_completion_queue(call_, rc->cq_bound_to_call);
1252 cq_new_ = server_->cqs_[cq_idx];
1253 GPR_SWAP(grpc_metadata_array, *rc->initial_metadata, initial_metadata_);
1255 case RequestedCall::Type::BATCH_CALL:
1256 GPR_ASSERT(host_.has_value());
1257 GPR_ASSERT(path_.has_value());
1258 rc->data.batch.details->host = grpc_slice_ref_internal(*host_);
1259 rc->data.batch.details->method = grpc_slice_ref_internal(*path_);
1260 rc->data.batch.details->deadline =
1261 grpc_millis_to_timespec(deadline_, GPR_CLOCK_MONOTONIC);
1262 rc->data.batch.details->flags = recv_initial_metadata_flags_;
1264 case RequestedCall::Type::REGISTERED_CALL:
1265 *rc->data.registered.deadline =
1266 grpc_millis_to_timespec(deadline_, GPR_CLOCK_MONOTONIC);
1267 if (rc->data.registered.optional_payload != nullptr) {
1268 *rc->data.registered.optional_payload = payload_;
1273 GPR_UNREACHABLE_CODE(return );
1275 grpc_cq_end_op(cq_new_, rc->tag, GRPC_ERROR_NONE, Server::DoneRequestEvent,
1276 rc, &rc->completion, true);
1279 void Server::CallData::PublishNewRpc(void* arg, grpc_error* error) {
1280 grpc_call_element* call_elem = static_cast<grpc_call_element*>(arg);
1281 auto* calld = static_cast<Server::CallData*>(call_elem->call_data);
1282 auto* chand = static_cast<Server::ChannelData*>(call_elem->channel_data);
1283 RequestMatcherInterface* rm = calld->matcher_;
1284 Server* server = rm->server();
1285 if (error != GRPC_ERROR_NONE ||
1286 server->shutdown_flag_.load(std::memory_order_acquire)) {
1287 calld->state_.Store(CallState::ZOMBIED, MemoryOrder::RELAXED);
1288 calld->KillZombie();
1291 rm->MatchOrQueue(chand->cq_idx(), calld);
1296 void KillZombieClosure(void* call, grpc_error* /*error*/) {
1297 grpc_call_unref(static_cast<grpc_call*>(call));
1302 void Server::CallData::KillZombie() {
1303 GRPC_CLOSURE_INIT(&kill_zombie_closure_, KillZombieClosure, call_,
1304 grpc_schedule_on_exec_ctx);
1305 ExecCtx::Run(DEBUG_LOCATION, &kill_zombie_closure_, GRPC_ERROR_NONE);
1308 void Server::CallData::StartNewRpc(grpc_call_element* elem) {
1309 auto* chand = static_cast<ChannelData*>(elem->channel_data);
1310 if (server_->shutdown_flag_.load(std::memory_order_acquire)) {
1311 state_.Store(CallState::ZOMBIED, MemoryOrder::RELAXED);
1315 // Find request matcher.
1316 matcher_ = server_->unregistered_request_matcher_.get();
1317 grpc_server_register_method_payload_handling payload_handling =
1318 GRPC_SRM_PAYLOAD_NONE;
1319 if (path_.has_value() && host_.has_value()) {
1320 ChannelRegisteredMethod* rm =
1321 chand->GetRegisteredMethod(*host_, *path_,
1322 (recv_initial_metadata_flags_ &
1323 GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST));
1324 if (rm != nullptr) {
1325 matcher_ = rm->server_registered_method->matcher.get();
1326 payload_handling = rm->server_registered_method->payload_handling;
1329 // Start recv_message op if needed.
1330 switch (payload_handling) {
1331 case GRPC_SRM_PAYLOAD_NONE:
1332 PublishNewRpc(elem, GRPC_ERROR_NONE);
1334 case GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER: {
1336 op.op = GRPC_OP_RECV_MESSAGE;
1338 op.reserved = nullptr;
1339 op.data.recv_message.recv_message = &payload_;
1340 GRPC_CLOSURE_INIT(&publish_, PublishNewRpc, elem,
1341 grpc_schedule_on_exec_ctx);
1342 grpc_call_start_batch_and_execute(call_, &op, 1, &publish_);
1348 void Server::CallData::RecvInitialMetadataBatchComplete(void* arg,
1349 grpc_error* error) {
1350 grpc_call_element* elem = static_cast<grpc_call_element*>(arg);
1351 auto* calld = static_cast<Server::CallData*>(elem->call_data);
1352 if (error != GRPC_ERROR_NONE) {
1353 calld->FailCallCreation();
1356 calld->StartNewRpc(elem);
1359 void Server::CallData::StartTransportStreamOpBatchImpl(
1360 grpc_call_element* elem, grpc_transport_stream_op_batch* batch) {
1361 if (batch->recv_initial_metadata) {
1362 GPR_ASSERT(batch->payload->recv_initial_metadata.recv_flags == nullptr);
1363 recv_initial_metadata_ =
1364 batch->payload->recv_initial_metadata.recv_initial_metadata;
1365 original_recv_initial_metadata_ready_ =
1366 batch->payload->recv_initial_metadata.recv_initial_metadata_ready;
1367 batch->payload->recv_initial_metadata.recv_initial_metadata_ready =
1368 &recv_initial_metadata_ready_;
1369 batch->payload->recv_initial_metadata.recv_flags =
1370 &recv_initial_metadata_flags_;
1372 if (batch->recv_trailing_metadata) {
1373 original_recv_trailing_metadata_ready_ =
1374 batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
1375 batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
1376 &recv_trailing_metadata_ready_;
1378 grpc_call_next_op(elem, batch);
1381 void Server::CallData::RecvInitialMetadataReady(void* arg, grpc_error* error) {
1382 grpc_call_element* elem = static_cast<grpc_call_element*>(arg);
1383 CallData* calld = static_cast<CallData*>(elem->call_data);
1384 grpc_millis op_deadline;
1385 if (error == GRPC_ERROR_NONE) {
1386 GPR_DEBUG_ASSERT(calld->recv_initial_metadata_->idx.named.path != nullptr);
1387 GPR_DEBUG_ASSERT(calld->recv_initial_metadata_->idx.named.authority !=
1389 calld->path_.emplace(grpc_slice_ref_internal(
1390 GRPC_MDVALUE(calld->recv_initial_metadata_->idx.named.path->md)));
1391 calld->host_.emplace(grpc_slice_ref_internal(
1392 GRPC_MDVALUE(calld->recv_initial_metadata_->idx.named.authority->md)));
1393 grpc_metadata_batch_remove(calld->recv_initial_metadata_, GRPC_BATCH_PATH);
1394 grpc_metadata_batch_remove(calld->recv_initial_metadata_,
1395 GRPC_BATCH_AUTHORITY);
1397 GRPC_ERROR_REF(error);
1399 op_deadline = calld->recv_initial_metadata_->deadline;
1400 if (op_deadline != GRPC_MILLIS_INF_FUTURE) {
1401 calld->deadline_ = op_deadline;
1403 if (calld->host_.has_value() && calld->path_.has_value()) {
1406 /* Pass the error reference to calld->recv_initial_metadata_error */
1407 grpc_error* src_error = error;
1408 error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
1409 "Missing :authority or :path", &src_error, 1);
1410 GRPC_ERROR_UNREF(src_error);
1411 calld->recv_initial_metadata_error_ = GRPC_ERROR_REF(error);
1413 grpc_closure* closure = calld->original_recv_initial_metadata_ready_;
1414 calld->original_recv_initial_metadata_ready_ = nullptr;
1415 if (calld->seen_recv_trailing_metadata_ready_) {
1416 GRPC_CALL_COMBINER_START(calld->call_combiner_,
1417 &calld->recv_trailing_metadata_ready_,
1418 calld->recv_trailing_metadata_error_,
1419 "continue server recv_trailing_metadata_ready");
1421 Closure::Run(DEBUG_LOCATION, closure, error);
1424 void Server::CallData::RecvTrailingMetadataReady(void* arg, grpc_error* error) {
1425 grpc_call_element* elem = static_cast<grpc_call_element*>(arg);
1426 CallData* calld = static_cast<CallData*>(elem->call_data);
1427 if (calld->original_recv_initial_metadata_ready_ != nullptr) {
1428 calld->recv_trailing_metadata_error_ = GRPC_ERROR_REF(error);
1429 calld->seen_recv_trailing_metadata_ready_ = true;
1430 GRPC_CLOSURE_INIT(&calld->recv_trailing_metadata_ready_,
1431 RecvTrailingMetadataReady, elem,
1432 grpc_schedule_on_exec_ctx);
1433 GRPC_CALL_COMBINER_STOP(calld->call_combiner_,
1434 "deferring server recv_trailing_metadata_ready "
1435 "until after recv_initial_metadata_ready");
1439 grpc_error_add_child(GRPC_ERROR_REF(error),
1440 GRPC_ERROR_REF(calld->recv_initial_metadata_error_));
1441 Closure::Run(DEBUG_LOCATION, calld->original_recv_trailing_metadata_ready_,
1445 grpc_error* Server::CallData::InitCallElement(
1446 grpc_call_element* elem, const grpc_call_element_args* args) {
1447 auto* chand = static_cast<ChannelData*>(elem->channel_data);
1448 new (elem->call_data) Server::CallData(elem, *args, chand->server());
1449 return GRPC_ERROR_NONE;
1452 void Server::CallData::DestroyCallElement(
1453 grpc_call_element* elem, const grpc_call_final_info* /*final_info*/,
1454 grpc_closure* /*ignored*/) {
1455 auto* calld = static_cast<CallData*>(elem->call_data);
1459 void Server::CallData::StartTransportStreamOpBatch(
1460 grpc_call_element* elem, grpc_transport_stream_op_batch* batch) {
1461 auto* calld = static_cast<CallData*>(elem->call_data);
1462 calld->StartTransportStreamOpBatchImpl(elem, batch);
1465 } // namespace grpc_core
1471 grpc_server* grpc_server_create(const grpc_channel_args* args, void* reserved) {
1472 grpc_core::ExecCtx exec_ctx;
1473 GRPC_API_TRACE("grpc_server_create(%p, %p)", 2, (args, reserved));
1474 grpc_server* c_server = new grpc_server;
1475 c_server->core_server = grpc_core::MakeOrphanable<grpc_core::Server>(args);
1479 void grpc_server_register_completion_queue(grpc_server* server,
1480 grpc_completion_queue* cq,
1483 "grpc_server_register_completion_queue(server=%p, cq=%p, reserved=%p)", 3,
1484 (server, cq, reserved));
1485 GPR_ASSERT(!reserved);
1486 auto cq_type = grpc_get_cq_completion_type(cq);
1487 if (cq_type != GRPC_CQ_NEXT && cq_type != GRPC_CQ_CALLBACK) {
1489 "Completion queue of type %d is being registered as a "
1490 "server-completion-queue",
1491 static_cast<int>(cq_type));
1492 /* Ideally we should log an error and abort but ruby-wrapped-language API
1493 calls grpc_completion_queue_pluck() on server completion queues */
1495 server->core_server->RegisterCompletionQueue(cq);
1498 void* grpc_server_register_method(
1499 grpc_server* server, const char* method, const char* host,
1500 grpc_server_register_method_payload_handling payload_handling,
1503 "grpc_server_register_method(server=%p, method=%s, host=%s, "
1505 4, (server, method, host, flags));
1506 return server->core_server->RegisterMethod(method, host, payload_handling,
1510 void grpc_server_start(grpc_server* server) {
1511 grpc_core::ExecCtx exec_ctx;
1512 GRPC_API_TRACE("grpc_server_start(server=%p)", 1, (server));
1513 server->core_server->Start();
1516 void grpc_server_shutdown_and_notify(grpc_server* server,
1517 grpc_completion_queue* cq, void* tag) {
1518 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1519 grpc_core::ExecCtx exec_ctx;
1520 GRPC_API_TRACE("grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", 3,
1522 server->core_server->ShutdownAndNotify(cq, tag);
1525 void grpc_server_cancel_all_calls(grpc_server* server) {
1526 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1527 grpc_core::ExecCtx exec_ctx;
1528 GRPC_API_TRACE("grpc_server_cancel_all_calls(server=%p)", 1, (server));
1529 server->core_server->CancelAllCalls();
1532 void grpc_server_destroy(grpc_server* server) {
1533 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1534 grpc_core::ExecCtx exec_ctx;
1535 GRPC_API_TRACE("grpc_server_destroy(server=%p)", 1, (server));
1539 grpc_call_error grpc_server_request_call(
1540 grpc_server* server, grpc_call** call, grpc_call_details* details,
1541 grpc_metadata_array* request_metadata,
1542 grpc_completion_queue* cq_bound_to_call,
1543 grpc_completion_queue* cq_for_notification, void* tag) {
1544 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1545 grpc_core::ExecCtx exec_ctx;
1546 GRPC_STATS_INC_SERVER_REQUESTED_CALLS();
1548 "grpc_server_request_call("
1549 "server=%p, call=%p, details=%p, initial_metadata=%p, "
1550 "cq_bound_to_call=%p, cq_for_notification=%p, tag=%p)",
1552 (server, call, details, request_metadata, cq_bound_to_call,
1553 cq_for_notification, tag));
1554 return server->core_server->RequestCall(call, details, request_metadata,
1555 cq_bound_to_call, cq_for_notification,
1559 grpc_call_error grpc_server_request_registered_call(
1560 grpc_server* server, void* registered_method, grpc_call** call,
1561 gpr_timespec* deadline, grpc_metadata_array* request_metadata,
1562 grpc_byte_buffer** optional_payload,
1563 grpc_completion_queue* cq_bound_to_call,
1564 grpc_completion_queue* cq_for_notification, void* tag_new) {
1565 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1566 grpc_core::ExecCtx exec_ctx;
1567 GRPC_STATS_INC_SERVER_REQUESTED_CALLS();
1569 static_cast<grpc_core::Server::RegisteredMethod*>(registered_method);
1571 "grpc_server_request_registered_call("
1572 "server=%p, registered_method=%p, call=%p, deadline=%p, "
1573 "request_metadata=%p, "
1574 "optional_payload=%p, cq_bound_to_call=%p, cq_for_notification=%p, "
1577 (server, registered_method, call, deadline, request_metadata,
1578 optional_payload, cq_bound_to_call, cq_for_notification, tag_new));
1579 return server->core_server->RequestRegisteredCall(
1580 rm, call, deadline, request_metadata, optional_payload, cq_bound_to_call,
1581 cq_for_notification, tag_new);
1584 void grpc_server_set_config_fetcher(
1585 grpc_server* server, grpc_server_config_fetcher* server_config_fetcher) {
1586 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1587 grpc_core::ExecCtx exec_ctx;
1588 GRPC_API_TRACE("grpc_server_set_config_fetcher(server=%p, config_fetcher=%p)",
1589 2, (server, server_config_fetcher));
1590 server->core_server->set_config_fetcher(
1591 std::unique_ptr<grpc_server_config_fetcher>(server_config_fetcher));
1594 void grpc_server_config_fetcher_destroy(
1595 grpc_server_config_fetcher* server_config_fetcher) {
1596 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1597 grpc_core::ExecCtx exec_ctx;
1598 GRPC_API_TRACE("grpc_server_config_fetcher_destroy(config_fetcher=%p)", 1,
1599 (server_config_fetcher));
1600 delete server_config_fetcher;