1 //
2 // Copyright 2015-2016 gRPC authors.
3 //
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
7 //
8 //     http://www.apache.org/licenses/LICENSE-2.0
9 //
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.
15 //
16 
17 #include <grpc/support/port_platform.h>
18 
19 #include "src/core/lib/surface/server.h"
20 
21 #include <limits.h>
22 #include <stdlib.h>
23 #include <string.h>
24 
25 #include <algorithm>
26 #include <atomic>
27 #include <iterator>
28 #include <list>
29 #include <queue>
30 #include <utility>
31 #include <vector>
32 
33 #include "absl/memory/memory.h"
34 #include "absl/types/optional.h"
35 
36 #include <grpc/support/alloc.h>
37 #include <grpc/support/log.h>
38 #include <grpc/support/string_util.h>
39 
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"
57 
58 namespace grpc_core {
59 
60 TraceFlag grpc_server_channel_trace(false, "server_channel");
61 
62 //
63 // Server::RequestedCall
64 //
65 
66 struct Server::RequestedCall {
67   enum class Type { BATCH_CALL, REGISTERED_CALL };
68 
RequestedCallgrpc_core::Server::RequestedCall69   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),
73         tag(tag_arg),
74         cq_bound_to_call(call_cq),
75         call(call_arg),
76         initial_metadata(initial_md) {
77     details->reserved = nullptr;
78     data.batch.details = details;
79   }
80 
RequestedCallgrpc_core::Server::RequestedCall81   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),
86         tag(tag_arg),
87         cq_bound_to_call(call_cq),
88         call(call_arg),
89         initial_metadata(initial_md) {
90     data.registered.method = rm;
91     data.registered.deadline = deadline;
92     data.registered.optional_payload = optional_payload;
93   }
94 
95   MultiProducerSingleConsumerQueue::Node mpscq_node;
96   const Type type;
97   void* const tag;
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;
102   union {
103     struct {
104       grpc_call_details* details;
105     } batch;
106     struct {
107       RegisteredMethod* method;
108       gpr_timespec* deadline;
109       grpc_byte_buffer** optional_payload;
110     } registered;
111   } data;
112 };
113 
114 //
115 // Server::RegisteredMethod
116 //
117 
118 struct Server::RegisteredMethod {
RegisteredMethodgrpc_core::Server::RegisteredMethod119   RegisteredMethod(
120       const char* method_arg, const char* host_arg,
121       grpc_server_register_method_payload_handling payload_handling_arg,
122       uint32_t flags_arg)
123       : method(method_arg == nullptr ? "" : method_arg),
124         host(host_arg == nullptr ? "" : host_arg),
125         payload_handling(payload_handling_arg),
126         flags(flags_arg) {}
127 
128   ~RegisteredMethod() = default;
129 
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;
136 };
137 
138 //
139 // Server::RequestMatcherInterface
140 //
141 
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.
149 //
150 // RequestMatcherInterface is the base class to provide this functionality.
151 class Server::RequestMatcherInterface {
152  public:
~RequestMatcherInterface()153   virtual ~RequestMatcherInterface() {}
154 
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;
158 
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;
163 
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;
167 
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;
175 
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;
184 
185   // Returns the server associated with this request matcher
186   virtual Server* server() const = 0;
187 };
188 
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 {
195  public:
RealRequestMatcher(Server * server)196   explicit RealRequestMatcher(Server* server)
197       : server_(server), requests_per_cq_(server->cqs_.size()) {}
198 
~RealRequestMatcher()199   ~RealRequestMatcher() override {
200     for (LockedMultiProducerSingleConsumerQueue& queue : requests_per_cq_) {
201       GPR_ASSERT(queue.Pop() == nullptr);
202     }
203   }
204 
ZombifyPending()205   void ZombifyPending() override {
206     while (!pending_.empty()) {
207       CallData* calld = pending_.front();
208       calld->SetState(CallData::CallState::ZOMBIED);
209       calld->KillZombie();
210       pending_.pop();
211     }
212   }
213 
KillRequests(grpc_error * error)214   void KillRequests(grpc_error* error) override {
215     for (size_t i = 0; i < requests_per_cq_.size(); i++) {
216       RequestedCall* rc;
217       while ((rc = reinterpret_cast<RequestedCall*>(
218                   requests_per_cq_[i].Pop())) != nullptr) {
219         server_->FailCall(i, rc, GRPC_ERROR_REF(error));
220       }
221     }
222     GRPC_ERROR_UNREF(error);
223   }
224 
request_queue_count() const225   size_t request_queue_count() const override {
226     return requests_per_cq_.size();
227   }
228 
RequestCallWithPossiblePublish(size_t request_queue_index,RequestedCall * call)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
233          matching calls */
234       struct PendingCall {
235         RequestedCall* rc = nullptr;
236         CallData* calld;
237       };
238       auto pop_next_pending = [this, request_queue_index] {
239         PendingCall pending_call;
240         {
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();
247               pending_.pop();
248             }
249           }
250         }
251         return pending_call;
252       };
253       while (true) {
254         PendingCall next_pending = pop_next_pending();
255         if (next_pending.rc == nullptr) break;
256         if (!next_pending.calld->MaybeActivate()) {
257           // Zombied Call
258           next_pending.calld->KillZombie();
259         } else {
260           next_pending.calld->Publish(request_queue_index, next_pending.rc);
261         }
262       }
263     }
264   }
265 
MatchOrQueue(size_t start_request_queue_index,CallData * calld)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();
270       RequestedCall* rc =
271           reinterpret_cast<RequestedCall*>(requests_per_cq_[cq_idx].TryPop());
272       if (rc != nullptr) {
273         GRPC_STATS_INC_SERVER_CQS_CHECKED(i);
274         calld->SetState(CallData::CallState::ACTIVATED);
275         calld->Publish(cq_idx, rc);
276         return;
277       }
278     }
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;
286     size_t cq_idx = 0;
287     size_t loop_count;
288     {
289       MutexLock lock(&server_->mu_call_);
290       for (loop_count = 0; loop_count < requests_per_cq_.size(); loop_count++) {
291         cq_idx =
292             (start_request_queue_index + loop_count) % requests_per_cq_.size();
293         rc = reinterpret_cast<RequestedCall*>(requests_per_cq_[cq_idx].Pop());
294         if (rc != nullptr) {
295           break;
296         }
297       }
298       if (rc == nullptr) {
299         calld->SetState(CallData::CallState::PENDING);
300         pending_.push(calld);
301         return;
302       }
303     }
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);
307   }
308 
server() const309   Server* server() const override { return server_; }
310 
311  private:
312   Server* const server_;
313   std::queue<CallData*> pending_;
314   std::vector<LockedMultiProducerSingleConsumerQueue> requests_per_cq_;
315 };
316 
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 {
323  public:
AllocatingRequestMatcherBase(Server * server,grpc_completion_queue * cq)324   AllocatingRequestMatcherBase(Server* server, grpc_completion_queue* cq)
325       : server_(server), cq_(cq) {
326     size_t idx;
327     for (idx = 0; idx < server->cqs_.size(); idx++) {
328       if (server->cqs_[idx] == cq) {
329         break;
330       }
331     }
332     GPR_ASSERT(idx < server->cqs_.size());
333     cq_idx_ = idx;
334   }
335 
ZombifyPending()336   void ZombifyPending() override {}
337 
KillRequests(grpc_error * error)338   void KillRequests(grpc_error* error) override { GRPC_ERROR_UNREF(error); }
339 
request_queue_count() const340   size_t request_queue_count() const override { return 0; }
341 
RequestCallWithPossiblePublish(size_t,RequestedCall *)342   void RequestCallWithPossiblePublish(size_t /*request_queue_index*/,
343                                       RequestedCall* /*call*/) final {
344     GPR_ASSERT(false);
345   }
346 
server() const347   Server* server() const override { return server_; }
348 
349   // Supply the completion queue related to this request matcher
cq() const350   grpc_completion_queue* cq() const { return cq_; }
351 
352   // Supply the completion queue's index relative to the server.
cq_idx() const353   size_t cq_idx() const { return cq_idx_; }
354 
355  private:
356   Server* const server_;
357   grpc_completion_queue* const cq_;
358   size_t cq_idx_;
359 };
360 
361 // An allocating request matcher for non-registered methods (used for generic
362 // API and unimplemented RPCs).
363 class Server::AllocatingRequestMatcherBatch
364     : public AllocatingRequestMatcherBase {
365  public:
AllocatingRequestMatcherBatch(Server * server,grpc_completion_queue * cq,std::function<BatchCallAllocation ()> allocator)366   AllocatingRequestMatcherBatch(Server* server, grpc_completion_queue* cq,
367                                 std::function<BatchCallAllocation()> allocator)
368       : AllocatingRequestMatcherBase(server, cq),
369         allocator_(std::move(allocator)) {}
370 
MatchOrQueue(size_t,CallData * calld)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) ==
376                GRPC_CALL_OK);
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);
382   }
383 
384  private:
385   std::function<BatchCallAllocation()> allocator_;
386 };
387 
388 // An allocating request matcher for registered methods.
389 class Server::AllocatingRequestMatcherRegistered
390     : public AllocatingRequestMatcherBase {
391  public:
AllocatingRequestMatcherRegistered(Server * server,grpc_completion_queue * cq,RegisteredMethod * rm,std::function<RegisteredCallAllocation ()> allocator)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)) {}
398 
MatchOrQueue(size_t,CallData * calld)399   void MatchOrQueue(size_t /*start_request_queue_index*/,
400                     CallData* calld) override {
401     RegisteredCallAllocation call_info = allocator_();
402     GPR_ASSERT(
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);
412   }
413 
414  private:
415   RegisteredMethod* const registered_method_;
416   std::function<RegisteredCallAllocation()> allocator_;
417 };
418 
419 //
420 // ChannelBroadcaster
421 //
422 
423 namespace {
424 
425 class ChannelBroadcaster {
426  public:
427   // This can have an empty constructor and destructor since we want to control
428   // when the actual setup and shutdown broadcast take place.
429 
430   // Copies over the channels from the locked server.
FillChannelsLocked(std::vector<grpc_channel * > channels)431   void FillChannelsLocked(std::vector<grpc_channel*> channels) {
432     GPR_DEBUG_ASSERT(channels_.empty());
433     channels_ = std::move(channels);
434   }
435 
436   // Broadcasts a shutdown on each channel.
BroadcastShutdown(bool send_goaway,grpc_error * force_disconnect)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");
441     }
442     channels_.clear();  // just for safety against double broadcast
443     GRPC_ERROR_UNREF(force_disconnect);
444   }
445 
446  private:
447   struct ShutdownCleanupArgs {
448     grpc_closure closure;
449     grpc_slice slice;
450   };
451 
ShutdownCleanup(void * arg,grpc_error *)452   static void ShutdownCleanup(void* arg, grpc_error* /*error*/) {
453     ShutdownCleanupArgs* a = static_cast<ShutdownCleanupArgs*>(arg);
454     grpc_slice_unref_internal(a->slice);
455     delete a;
456   }
457 
SendShutdown(grpc_channel * channel,bool send_goaway,grpc_error * send_disconnect)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;
465     op->goaway_error =
466         send_goaway
467             ? grpc_error_set_int(
468                   GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server shutdown"),
469                   GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_OK)
470             : GRPC_ERROR_NONE;
471     op->set_accept_stream = true;
472     sc->slice = grpc_slice_from_copied_string("Server shutdown");
473     op->disconnect_with_error = send_disconnect;
474     elem =
475         grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
476     elem->filter->start_transport_op(elem, op);
477   }
478 
479   std::vector<grpc_channel*> channels_;
480 };
481 
482 }  // namespace
483 
484 //
485 // Server
486 //
487 
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,
499     "server",
500 };
501 
502 namespace {
503 
CreateDefaultResourceUser(const grpc_channel_args * args)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");
510     }
511   }
512   return nullptr;
513 }
514 
CreateChannelzNode(Server * server,const grpc_channel_args * args)515 RefCountedPtr<channelz::ServerNode> CreateChannelzNode(
516     Server* server, 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});
523     channelz_node =
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"));
528   }
529   return channelz_node;
530 }
531 
532 }  // namespace
533 
Server(const grpc_channel_args * args)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(this, args)) {}
538 
~Server()539 Server::~Server() {
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(),
546                                    pollset);
547     }
548   }
549   for (size_t i = 0; i < cqs_.size(); i++) {
550     GRPC_CQ_INTERNAL_UNREF(cqs_[i], "server");
551   }
552 }
553 
AddListener(OrphanablePtr<ListenerInterface> listener)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());
559   }
560   listeners_.emplace_back(std::move(listener));
561 }
562 
Start()563 void Server::Start() {
564   started_ = true;
565   for (grpc_completion_queue* cq : cqs_) {
566     if (grpc_cq_can_listen(cq)) {
567       pollsets_.push_back(grpc_cq_pollset(cq));
568     }
569   }
570   if (unregistered_request_matcher_ == nullptr) {
571     unregistered_request_matcher_ = absl::make_unique<RealRequestMatcher>(this);
572   }
573   for (std::unique_ptr<RegisteredMethod>& rm : registered_methods_) {
574     if (rm->matcher == nullptr) {
575       rm->matcher = absl::make_unique<RealRequestMatcher>(this);
576     }
577   }
578   {
579     MutexLock lock(&mu_global_);
580     starting_ = true;
581   }
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(),
589                                    pollset);
590     }
591   }
592   for (auto& listener : listeners_) {
593     listener.listener->Start(this, &pollsets_);
594   }
595   MutexLock lock(&mu_global_);
596   starting_ = false;
597   starting_cv_.Signal();
598 }
599 
SetupTransport(grpc_transport * transport,grpc_pollset * accepting_pollset,const grpc_channel_args * args,const RefCountedPtr<grpc_core::channelz::SocketNode> & socket_node,grpc_resource_user * resource_user)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) {
605   // Create channel.
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) {
610     return error;
611   }
612   ChannelData* chand = static_cast<ChannelData*>(
613       grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0)
614           ->channel_data);
615   // Set up CQs.
616   size_t cq_idx;
617   for (cq_idx = 0; cq_idx < cqs_.size(); cq_idx++) {
618     if (grpc_cq_pollset(cqs_[cq_idx]) == accepting_pollset) break;
619   }
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();
623   }
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);
629   }
630   // Initialize chand.
631   chand->InitTransport(Ref(), channel, cq_idx, transport, channelz_socket_uuid);
632   return GRPC_ERROR_NONE;
633 }
634 
HasOpenConnections()635 bool Server::HasOpenConnections() {
636   MutexLock lock(&mu_global_);
637   return !channels_.empty();
638 }
639 
SetRegisteredMethodAllocator(grpc_completion_queue * cq,void * method_tag,std::function<RegisteredCallAllocation ()> allocator)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));
646 }
647 
SetBatchMethodAllocator(grpc_completion_queue * cq,std::function<BatchCallAllocation ()> 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));
654 }
655 
RegisterCompletionQueue(grpc_completion_queue * cq)656 void Server::RegisterCompletionQueue(grpc_completion_queue* cq) {
657   for (grpc_completion_queue* queue : cqs_) {
658     if (queue == cq) return;
659   }
660   GRPC_CQ_INTERNAL_REF(cq, "server");
661   cqs_.push_back(cq);
662 }
663 
664 namespace {
665 
streq(const std::string & a,const char * b)666 bool streq(const std::string& a, const char* b) {
667   return (a.empty() && b == nullptr) ||
668          ((b != nullptr) && !strcmp(a.c_str(), b));
669 }
670 
671 }  // namespace
672 
RegisterMethod(const char * method,const char * host,grpc_server_register_method_payload_handling payload_handling,uint32_t flags)673 Server::RegisteredMethod* Server::RegisterMethod(
674     const char* method, const char* host,
675     grpc_server_register_method_payload_handling payload_handling,
676     uint32_t flags) {
677   if (!method) {
678     gpr_log(GPR_ERROR,
679             "grpc_server_register_method method string cannot be NULL");
680     return nullptr;
681   }
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,
685               host ? host : "*");
686       return nullptr;
687     }
688   }
689   if ((flags & ~GRPC_INITIAL_METADATA_USED_MASK) != 0) {
690     gpr_log(GPR_ERROR, "grpc_server_register_method invalid flags 0x%08x",
691             flags);
692     return nullptr;
693   }
694   registered_methods_.emplace_back(absl::make_unique<RegisteredMethod>(
695       method, host, payload_handling, flags));
696   return registered_methods_.back().get();
697 }
698 
DoneRequestEvent(void * req,grpc_cq_completion *)699 void Server::DoneRequestEvent(void* req, grpc_cq_completion* /*c*/) {
700   delete static_cast<RequestedCall*>(req);
701 }
702 
FailCall(size_t cq_idx,RequestedCall * rc,grpc_error * error)703 void Server::FailCall(size_t cq_idx, RequestedCall* rc, grpc_error* error) {
704   *rc->call = nullptr;
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,
708                  &rc->completion);
709 }
710 
711 // Before calling MaybeFinishShutdown(), we must hold mu_global_ and not
712 // hold mu_call_.
MaybeFinishShutdown()713 void Server::MaybeFinishShutdown() {
714   if (!shutdown_flag_.load(std::memory_order_acquire) || shutdown_published_) {
715     return;
716   }
717   {
718     MutexLock lock(&mu_call_);
719     KillPendingWorkLocked(
720         GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
721   }
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);
727       gpr_log(GPR_DEBUG,
728               "Waiting for %" PRIuPTR " channels and %" PRIuPTR "/%" PRIuPTR
729               " listeners to be destroyed before shutting down server",
730               channels_.size(), listeners_.size() - listeners_destroyed_,
731               listeners_.size());
732     }
733     return;
734   }
735   shutdown_published_ = true;
736   for (auto& shutdown_tag : shutdown_tags_) {
737     Ref().release();
738     grpc_cq_end_op(shutdown_tag.cq, shutdown_tag.tag, GRPC_ERROR_NONE,
739                    DoneShutdownEvent, this, &shutdown_tag.completion);
740   }
741 }
742 
KillPendingWorkLocked(grpc_error * error)743 void Server::KillPendingWorkLocked(grpc_error* error) {
744   if (started_) {
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();
750     }
751   }
752   GRPC_ERROR_UNREF(error);
753 }
754 
GetChannelsLocked() const755 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");
761   }
762   return channels;
763 }
764 
ListenerDestroyDone(void * arg,grpc_error *)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();
770 }
771 
772 namespace {
773 
DonePublishedShutdown(void *,grpc_cq_completion * storage)774 void DonePublishedShutdown(void* /*done_arg*/, grpc_cq_completion* storage) {
775   delete storage;
776 }
777 
778 }  // namespace
779 
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().
783 //
784 // - Shuts down the listeners (i.e., the server will no longer listen on the
785 //   port for new incoming channels).
786 //
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.
ShutdownAndNotify(grpc_completion_queue * cq,void * tag)794 void Server::ShutdownAndNotify(grpc_completion_queue* cq, void* tag) {
795   ChannelBroadcaster broadcaster;
796   {
797     // Wait for startup to be finished.  Locks mu_global.
798     MutexLock lock(&mu_global_);
799     starting_cv_.WaitUntil(&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);
805       return;
806     }
807     shutdown_tags_.emplace_back(tag, cq);
808     if (shutdown_flag_.load(std::memory_order_acquire)) {
809       return;
810     }
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.
815     {
816       MutexLock lock(&mu_call_);
817       KillPendingWorkLocked(
818           GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
819     }
820     MaybeFinishShutdown();
821   }
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());
829     }
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();
834   }
835   broadcaster.BroadcastShutdown(/*send_goaway=*/true, GRPC_ERROR_NONE);
836 }
837 
CancelAllCalls()838 void Server::CancelAllCalls() {
839   ChannelBroadcaster broadcaster;
840   {
841     MutexLock lock(&mu_global_);
842     broadcaster.FillChannelsLocked(GetChannelsLocked());
843   }
844   broadcaster.BroadcastShutdown(
845       /*send_goaway=*/false,
846       GRPC_ERROR_CREATE_FROM_STATIC_STRING("Cancelling all calls"));
847 }
848 
Orphan()849 void Server::Orphan() {
850   {
851     MutexLock lock(&mu_global_);
852     GPR_ASSERT(shutdown_flag_.load(std::memory_order_acquire) ||
853                listeners_.empty());
854     GPR_ASSERT(listeners_destroyed_ == listeners_.size());
855   }
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_);
860   }
861   Unref();
862 }
863 
ValidateServerRequest(grpc_completion_queue * cq_for_notification,void * tag,grpc_byte_buffer ** optional_payload,RegisteredMethod * rm)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;
871   }
872   if (grpc_cq_begin_op(cq_for_notification, tag) == false) {
873     return GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN;
874   }
875   return GRPC_CALL_OK;
876 }
877 
ValidateServerRequestAndCq(size_t * cq_idx,grpc_completion_queue * cq_for_notification,void * tag,grpc_byte_buffer ** optional_payload,RegisteredMethod * rm)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) {
881   size_t idx;
882   for (idx = 0; idx < cqs_.size(); idx++) {
883     if (cqs_[idx] == cq_for_notification) {
884       break;
885     }
886   }
887   if (idx == cqs_.size()) {
888     return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
889   }
890   grpc_call_error error =
891       ValidateServerRequest(cq_for_notification, tag, optional_payload, rm);
892   if (error != GRPC_CALL_OK) {
893     return error;
894   }
895   *cq_idx = idx;
896   return GRPC_CALL_OK;
897 }
898 
QueueRequestedCall(size_t cq_idx,RequestedCall * rc)899 grpc_call_error Server::QueueRequestedCall(size_t cq_idx, RequestedCall* rc) {
900   if (shutdown_flag_.load(std::memory_order_acquire)) {
901     FailCall(cq_idx, rc,
902              GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
903     return GRPC_CALL_OK;
904   }
905   RequestMatcherInterface* rm;
906   switch (rc->type) {
907     case RequestedCall::Type::BATCH_CALL:
908       rm = unregistered_request_matcher_.get();
909       break;
910     case RequestedCall::Type::REGISTERED_CALL:
911       rm = rc->data.registered.method->matcher.get();
912       break;
913   }
914   rm->RequestCallWithPossiblePublish(cq_idx, rc);
915   return GRPC_CALL_OK;
916 }
917 
RequestCall(grpc_call ** call,grpc_call_details * details,grpc_metadata_array * request_metadata,grpc_completion_queue * cq_bound_to_call,grpc_completion_queue * cq_for_notification,void * tag)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,
923                                     void* tag) {
924   size_t cq_idx;
925   grpc_call_error error = ValidateServerRequestAndCq(
926       &cq_idx, cq_for_notification, tag, nullptr, nullptr);
927   if (error != GRPC_CALL_OK) {
928     return error;
929   }
930   RequestedCall* rc =
931       new RequestedCall(tag, cq_bound_to_call, call, request_metadata, details);
932   return QueueRequestedCall(cq_idx, rc);
933 }
934 
RequestRegisteredCall(RegisteredMethod * rm,grpc_call ** call,gpr_timespec * deadline,grpc_metadata_array * request_metadata,grpc_byte_buffer ** optional_payload,grpc_completion_queue * cq_bound_to_call,grpc_completion_queue * cq_for_notification,void * tag_new)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) {
940   size_t cq_idx;
941   grpc_call_error error = ValidateServerRequestAndCq(
942       &cq_idx, cq_for_notification, tag_new, optional_payload, rm);
943   if (error != GRPC_CALL_OK) {
944     return error;
945   }
946   RequestedCall* rc =
947       new RequestedCall(tag_new, cq_bound_to_call, call, request_metadata, rm,
948                         deadline, optional_payload);
949   return QueueRequestedCall(cq_idx, rc);
950 }
951 
952 //
953 // Server::ChannelData::ConnectivityWatcher
954 //
955 
956 class Server::ChannelData::ConnectivityWatcher
957     : public AsyncConnectivityStateWatcherInterface {
958  public:
ConnectivityWatcher(ChannelData * chand)959   explicit ConnectivityWatcher(ChannelData* chand) : chand_(chand) {
960     GRPC_CHANNEL_INTERNAL_REF(chand_->channel_, "connectivity");
961   }
962 
~ConnectivityWatcher()963   ~ConnectivityWatcher() override {
964     GRPC_CHANNEL_INTERNAL_UNREF(chand_->channel_, "connectivity");
965   }
966 
967  private:
OnConnectivityStateChange(grpc_connectivity_state new_state,const absl::Status &)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_);
974     chand_->Destroy();
975   }
976 
977   ChannelData* chand_;
978 };
979 
980 //
981 // Server::ChannelData
982 //
983 
~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);
990       if (crm.has_host) {
991         grpc_slice_unref_internal(crm.host);
992         GPR_DEBUG_ASSERT(crm.host.refcount == &kNoopRefcount ||
993                          crm.host.refcount == nullptr);
994       }
995     }
996     registered_methods_.reset();
997   }
998   if (server_ != nullptr) {
999     if (server_->channelz_node_ != nullptr && channelz_socket_uuid_ != 0) {
1000       server_->channelz_node_->RemoveChildSocket(channelz_socket_uuid_);
1001     }
1002     {
1003       MutexLock lock(&server_->mu_global_);
1004       if (list_position_.has_value()) {
1005         server_->channels_.erase(*list_position_);
1006         list_position_.reset();
1007       }
1008       server_->MaybeFinishShutdown();
1009     }
1010   }
1011 }
1012 
InitTransport(RefCountedPtr<Server> server,grpc_channel * channel,size_t cq_idx,grpc_transport * transport,intptr_t channelz_socket_uuid)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);
1018   channel_ = channel;
1019   cq_idx_ = cq_idx;
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();
1033       if (has_host) {
1034         host = ExternallyManagedSlice(rm->host.c_str());
1035       }
1036       uint32_t hash =
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;
1041            probes++) {
1042       }
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;
1049       if (has_host) {
1050         crm->host = host;
1051       }
1052       crm->method = method;
1053     }
1054     GPR_ASSERT(slots <= UINT32_MAX);
1055     registered_method_max_probes_ = max_probes;
1056   }
1057   // Publish channel.
1058   {
1059     MutexLock lock(&server_->mu_global_);
1060     server_->channels_.push_front(this);
1061     list_position_ = server_->channels_.begin();
1062   }
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");
1072   }
1073   grpc_transport_perform_op(transport, op);
1074 }
1075 
GetRegisteredMethod(const grpc_slice & host,const grpc_slice & path,bool is_idempotent)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) &&
1091         !is_idempotent) {
1092       continue;
1093     }
1094     return rm;
1095   }
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) &&
1105         !is_idempotent) {
1106       continue;
1107     }
1108     return rm;
1109   }
1110   return nullptr;
1111 }
1112 
AcceptStream(void * arg,grpc_transport *,const void * transport_server_data)1113 void Server::ChannelData::AcceptStream(void* arg, grpc_transport* /*transport*/,
1114                                        const void* transport_server_data) {
1115   auto* chand = static_cast<Server::ChannelData*>(arg);
1116   /* create a call */
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;
1122   args.cq = nullptr;
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;
1128   grpc_call* call;
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();
1136     return;
1137   }
1138   calld->Start(elem);
1139 }
1140 
FinishDestroy(void * arg,grpc_error *)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");
1145   server->Unref();
1146 }
1147 
Destroy()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");
1159   }
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),
1165       op);
1166 }
1167 
InitChannelElement(grpc_channel_element * elem,grpc_channel_element_args * args)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;
1174 }
1175 
DestroyChannelElement(grpc_channel_element * elem)1176 void Server::ChannelData::DestroyChannelElement(grpc_channel_element* elem) {
1177   auto* chand = static_cast<ChannelData*>(elem->channel_data);
1178   chand->~ChannelData();
1179 }
1180 
1181 //
1182 // Server::CallData
1183 //
1184 
CallData(grpc_call_element * elem,const grpc_call_element_args & args,RefCountedPtr<Server> server)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);
1195 }
1196 
~CallData()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_);
1202   }
1203   if (path_.has_value()) {
1204     grpc_slice_unref_internal(*path_);
1205   }
1206   grpc_metadata_array_destroy(&initial_metadata_);
1207   grpc_byte_buffer_destroy(payload_);
1208 }
1209 
SetState(CallState state)1210 void Server::CallData::SetState(CallState state) {
1211   state_.Store(state, MemoryOrder::RELAXED);
1212 }
1213 
MaybeActivate()1214 bool Server::CallData::MaybeActivate() {
1215   CallState expected = CallState::PENDING;
1216   return state_.CompareExchangeStrong(&expected, CallState::ACTIVATED,
1217                                       MemoryOrder::ACQ_REL,
1218                                       MemoryOrder::RELAXED);
1219 }
1220 
FailCallCreation()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)) {
1227     KillZombie();
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
1232     // queue... later.
1233   }
1234 }
1235 
Start(grpc_call_element * elem)1236 void Server::CallData::Start(grpc_call_element* elem) {
1237   grpc_op op;
1238   op.op = GRPC_OP_RECV_INITIAL_METADATA;
1239   op.flags = 0;
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_);
1247 }
1248 
Publish(size_t cq_idx,RequestedCall * rc)1249 void Server::CallData::Publish(size_t cq_idx, RequestedCall* rc) {
1250   grpc_call_set_completion_queue(call_, rc->cq_bound_to_call);
1251   *rc->call = call_;
1252   cq_new_ = server_->cqs_[cq_idx];
1253   GPR_SWAP(grpc_metadata_array, *rc->initial_metadata, initial_metadata_);
1254   switch (rc->type) {
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_;
1263       break;
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_;
1269         payload_ = nullptr;
1270       }
1271       break;
1272     default:
1273       GPR_UNREACHABLE_CODE(return );
1274   }
1275   grpc_cq_end_op(cq_new_, rc->tag, GRPC_ERROR_NONE, Server::DoneRequestEvent,
1276                  rc, &rc->completion, true);
1277 }
1278 
PublishNewRpc(void * arg,grpc_error * error)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();
1289     return;
1290   }
1291   rm->MatchOrQueue(chand->cq_idx(), calld);
1292 }
1293 
1294 namespace {
1295 
KillZombieClosure(void * call,grpc_error *)1296 void KillZombieClosure(void* call, grpc_error* /*error*/) {
1297   grpc_call_unref(static_cast<grpc_call*>(call));
1298 }
1299 
1300 }  // namespace
1301 
KillZombie()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);
1306 }
1307 
StartNewRpc(grpc_call_element * elem)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);
1312     KillZombie();
1313     return;
1314   }
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;
1327     }
1328   }
1329   // Start recv_message op if needed.
1330   switch (payload_handling) {
1331     case GRPC_SRM_PAYLOAD_NONE:
1332       PublishNewRpc(elem, GRPC_ERROR_NONE);
1333       break;
1334     case GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER: {
1335       grpc_op op;
1336       op.op = GRPC_OP_RECV_MESSAGE;
1337       op.flags = 0;
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_);
1343       break;
1344     }
1345   }
1346 }
1347 
RecvInitialMetadataBatchComplete(void * arg,grpc_error * error)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();
1354     return;
1355   }
1356   calld->StartNewRpc(elem);
1357 }
1358 
StartTransportStreamOpBatchImpl(grpc_call_element * elem,grpc_transport_stream_op_batch * batch)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_;
1371   }
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_;
1377   }
1378   grpc_call_next_op(elem, batch);
1379 }
1380 
RecvInitialMetadataReady(void * arg,grpc_error * error)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 !=
1388                      nullptr);
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);
1396   } else {
1397     GRPC_ERROR_REF(error);
1398   }
1399   op_deadline = calld->recv_initial_metadata_->deadline;
1400   if (op_deadline != GRPC_MILLIS_INF_FUTURE) {
1401     calld->deadline_ = op_deadline;
1402   }
1403   if (calld->host_.has_value() && calld->path_.has_value()) {
1404     /* do nothing */
1405   } else {
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);
1412   }
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");
1420   }
1421   Closure::Run(DEBUG_LOCATION, closure, error);
1422 }
1423 
RecvTrailingMetadataReady(void * arg,grpc_error * 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");
1436     return;
1437   }
1438   error =
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_,
1442                error);
1443 }
1444 
InitCallElement(grpc_call_element * elem,const grpc_call_element_args * args)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;
1450 }
1451 
DestroyCallElement(grpc_call_element * elem,const grpc_call_final_info *,grpc_closure *)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);
1456   calld->~CallData();
1457 }
1458 
StartTransportStreamOpBatch(grpc_call_element * elem,grpc_transport_stream_op_batch * batch)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);
1463 }
1464 
1465 }  // namespace grpc_core
1466 
1467 //
1468 // C-core API
1469 //
1470 
grpc_server_create(const grpc_channel_args * args,void * reserved)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);
1476   return c_server;
1477 }
1478 
grpc_server_register_completion_queue(grpc_server * server,grpc_completion_queue * cq,void * reserved)1479 void grpc_server_register_completion_queue(grpc_server* server,
1480                                            grpc_completion_queue* cq,
1481                                            void* reserved) {
1482   GRPC_API_TRACE(
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) {
1488     gpr_log(GPR_INFO,
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 */
1494   }
1495   server->core_server->RegisterCompletionQueue(cq);
1496 }
1497 
grpc_server_register_method(grpc_server * server,const char * method,const char * host,grpc_server_register_method_payload_handling payload_handling,uint32_t flags)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,
1501     uint32_t flags) {
1502   GRPC_API_TRACE(
1503       "grpc_server_register_method(server=%p, method=%s, host=%s, "
1504       "flags=0x%08x)",
1505       4, (server, method, host, flags));
1506   return server->core_server->RegisterMethod(method, host, payload_handling,
1507                                              flags);
1508 }
1509 
grpc_server_start(grpc_server * server)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();
1514 }
1515 
grpc_server_shutdown_and_notify(grpc_server * server,grpc_completion_queue * cq,void * tag)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,
1521                  (server, cq, tag));
1522   server->core_server->ShutdownAndNotify(cq, tag);
1523 }
1524 
grpc_server_cancel_all_calls(grpc_server * server)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();
1530 }
1531 
grpc_server_destroy(grpc_server * server)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));
1536   delete server;
1537 }
1538 
grpc_server_request_call(grpc_server * server,grpc_call ** call,grpc_call_details * details,grpc_metadata_array * request_metadata,grpc_completion_queue * cq_bound_to_call,grpc_completion_queue * cq_for_notification,void * tag)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();
1547   GRPC_API_TRACE(
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)",
1551       7,
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,
1556                                           tag);
1557 }
1558 
grpc_server_request_registered_call(grpc_server * server,void * registered_method,grpc_call ** call,gpr_timespec * deadline,grpc_metadata_array * request_metadata,grpc_byte_buffer ** optional_payload,grpc_completion_queue * cq_bound_to_call,grpc_completion_queue * cq_for_notification,void * tag_new)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();
1568   auto* rm =
1569       static_cast<grpc_core::Server::RegisteredMethod*>(registered_method);
1570   GRPC_API_TRACE(
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, "
1575       "tag=%p)",
1576       9,
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);
1582 }
1583 
grpc_server_set_config_fetcher(grpc_server * server,grpc_server_config_fetcher * server_config_fetcher)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));
1592 }
1593 
grpc_server_config_fetcher_destroy(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;
1601 }
1602