src/core/lib/surface/server.cc

   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
  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),
  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
  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),
  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 {
 119   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:
 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_handle 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:
 196   explicit RealRequestMatcher(Server* server)
 197       : server_(server), requests_per_cq_(server->cqs_.size()) {}
 198
 199   ~RealRequestMatcher() override {
 200     for (LockedMultiProducerSingleConsumerQueue& queue : requests_per_cq_) {
 201       GPR_ASSERT(queue.Pop() == nullptr);
 202     }
 203   }
 204
 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
 214   void KillRequests(grpc_error_handle 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
 225   size_t request_queue_count() const override {
 226     return requests_per_cq_.size();
 227   }
 228
 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
 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
 309   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). They are also
 322 // used for the sync API.
 323 class Server::AllocatingRequestMatcherBase : public RequestMatcherInterface {
 324  public:
 325   AllocatingRequestMatcherBase(Server* server, grpc_completion_queue* cq)
 326       : server_(server), cq_(cq) {
 327     size_t idx;
 328     for (idx = 0; idx < server->cqs_.size(); idx++) {
 329       if (server->cqs_[idx] == cq) {
 330         break;
 331       }
 332     }
 333     GPR_ASSERT(idx < server->cqs_.size());
 334     cq_idx_ = idx;
 335   }
 336
 337   void ZombifyPending() override {}
 338
 339   void KillRequests(grpc_error_handle error) override {
 340     GRPC_ERROR_UNREF(error);
 341   }
 342
 343   size_t request_queue_count() const override { return 0; }
 344
 345   void RequestCallWithPossiblePublish(size_t /*request_queue_index*/,
 346                                       RequestedCall* /*call*/) final {
 347     GPR_ASSERT(false);
 348   }
 349
 350   Server* server() const override { return server_; }
 351
 352   // Supply the completion queue related to this request matcher
 353   grpc_completion_queue* cq() const { return cq_; }
 354
 355   // Supply the completion queue's index relative to the server.
 356   size_t cq_idx() const { return cq_idx_; }
 357
 358  private:
 359   Server* const server_;
 360   grpc_completion_queue* const cq_;
 361   size_t cq_idx_;
 362 };
 363
 364 // An allocating request matcher for non-registered methods (used for generic
 365 // API and unimplemented RPCs).
 366 class Server::AllocatingRequestMatcherBatch
 367     : public AllocatingRequestMatcherBase {
 368  public:
 369   AllocatingRequestMatcherBatch(Server* server, grpc_completion_queue* cq,
 370                                 std::function<BatchCallAllocation()> allocator)
 371       : AllocatingRequestMatcherBase(server, cq),
 372         allocator_(std::move(allocator)) {}
 373
 374   void MatchOrQueue(size_t /*start_request_queue_index*/,
 375                     CallData* calld) override {
 376     if (server()->ShutdownRefOnRequest()) {
 377       BatchCallAllocation call_info = allocator_();
 378       GPR_ASSERT(server()->ValidateServerRequest(
 379                      cq(), static_cast<void*>(call_info.tag), nullptr,
 380                      nullptr) == GRPC_CALL_OK);
 381       RequestedCall* rc = new RequestedCall(
 382           static_cast<void*>(call_info.tag), call_info.cq, call_info.call,
 383           call_info.initial_metadata, call_info.details);
 384       calld->SetState(CallData::CallState::ACTIVATED);
 385       calld->Publish(cq_idx(), rc);
 386     } else {
 387       calld->FailCallCreation();
 388     }
 389     server()->ShutdownUnrefOnRequest();
 390   }
 391
 392  private:
 393   std::function<BatchCallAllocation()> allocator_;
 394 };
 395
 396 // An allocating request matcher for registered methods.
 397 class Server::AllocatingRequestMatcherRegistered
 398     : public AllocatingRequestMatcherBase {
 399  public:
 400   AllocatingRequestMatcherRegistered(
 401       Server* server, grpc_completion_queue* cq, RegisteredMethod* rm,
 402       std::function<RegisteredCallAllocation()> allocator)
 403       : AllocatingRequestMatcherBase(server, cq),
 404         registered_method_(rm),
 405         allocator_(std::move(allocator)) {}
 406
 407   void MatchOrQueue(size_t /*start_request_queue_index*/,
 408                     CallData* calld) override {
 409     if (server()->ShutdownRefOnRequest()) {
 410       RegisteredCallAllocation call_info = allocator_();
 411       GPR_ASSERT(server()->ValidateServerRequest(
 412                      cq(), call_info.tag, call_info.optional_payload,
 413                      registered_method_) == GRPC_CALL_OK);
 414       RequestedCall* rc =
 415           new RequestedCall(call_info.tag, call_info.cq, call_info.call,
 416                             call_info.initial_metadata, registered_method_,
 417                             call_info.deadline, call_info.optional_payload);
 418       calld->SetState(CallData::CallState::ACTIVATED);
 419       calld->Publish(cq_idx(), rc);
 420     } else {
 421       calld->FailCallCreation();
 422     }
 423     server()->ShutdownUnrefOnRequest();
 424   }
 425
 426  private:
 427   RegisteredMethod* const registered_method_;
 428   std::function<RegisteredCallAllocation()> allocator_;
 429 };
 430
 431 //
 432 // ChannelBroadcaster
 433 //
 434
 435 namespace {
 436
 437 class ChannelBroadcaster {
 438  public:
 439   // This can have an empty constructor and destructor since we want to control
 440   // when the actual setup and shutdown broadcast take place.
 441
 442   // Copies over the channels from the locked server.
 443   void FillChannelsLocked(std::vector<grpc_channel*> channels) {
 444     GPR_DEBUG_ASSERT(channels_.empty());
 445     channels_ = std::move(channels);
 446   }
 447
 448   // Broadcasts a shutdown on each channel.
 449   void BroadcastShutdown(bool send_goaway, grpc_error_handle force_disconnect) {
 450     for (grpc_channel* channel : channels_) {
 451       SendShutdown(channel, send_goaway, GRPC_ERROR_REF(force_disconnect));
 452       GRPC_CHANNEL_INTERNAL_UNREF(channel, "broadcast");
 453     }
 454     channels_.clear();  // just for safety against double broadcast
 455     GRPC_ERROR_UNREF(force_disconnect);
 456   }
 457
 458  private:
 459   struct ShutdownCleanupArgs {
 460     grpc_closure closure;
 461     grpc_slice slice;
 462   };
 463
 464   static void ShutdownCleanup(void* arg, grpc_error_handle /*error*/) {
 465     ShutdownCleanupArgs* a = static_cast<ShutdownCleanupArgs*>(arg);
 466     grpc_slice_unref_internal(a->slice);
 467     delete a;
 468   }
 469
 470   static void SendShutdown(grpc_channel* channel, bool send_goaway,
 471                            grpc_error_handle send_disconnect) {
 472     ShutdownCleanupArgs* sc = new ShutdownCleanupArgs;
 473     GRPC_CLOSURE_INIT(&sc->closure, ShutdownCleanup, sc,
 474                       grpc_schedule_on_exec_ctx);
 475     grpc_transport_op* op = grpc_make_transport_op(&sc->closure);
 476     grpc_channel_element* elem;
 477     op->goaway_error =
 478         send_goaway
 479             ? grpc_error_set_int(
 480                   GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server shutdown"),
 481                   GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_OK)
 482             : GRPC_ERROR_NONE;
 483     op->set_accept_stream = true;
 484     sc->slice = grpc_slice_from_copied_string("Server shutdown");
 485     op->disconnect_with_error = send_disconnect;
 486     elem =
 487         grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
 488     elem->filter->start_transport_op(elem, op);
 489   }
 490
 491   std::vector<grpc_channel*> channels_;
 492 };
 493
 494 }  // namespace
 495
 496 //
 497 // Server
 498 //
 499
 500 const grpc_channel_filter Server::kServerTopFilter = {
 501     Server::CallData::StartTransportStreamOpBatch,
 502     grpc_channel_next_op,
 503     sizeof(Server::CallData),
 504     Server::CallData::InitCallElement,
 505     grpc_call_stack_ignore_set_pollset_or_pollset_set,
 506     Server::CallData::DestroyCallElement,
 507     sizeof(Server::ChannelData),
 508     Server::ChannelData::InitChannelElement,
 509     Server::ChannelData::DestroyChannelElement,
 510     grpc_channel_next_get_info,
 511     "server",
 512 };
 513
 514 namespace {
 515
 516 RefCountedPtr<channelz::ServerNode> CreateChannelzNode(
 517     const grpc_channel_args* args) {
 518   RefCountedPtr<channelz::ServerNode> channelz_node;
 519   if (grpc_channel_args_find_bool(args, GRPC_ARG_ENABLE_CHANNELZ,
 520                                   GRPC_ENABLE_CHANNELZ_DEFAULT)) {
 521     size_t channel_tracer_max_memory = grpc_channel_args_find_integer(
 522         args, GRPC_ARG_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE,
 523         {GRPC_MAX_CHANNEL_TRACE_EVENT_MEMORY_PER_NODE_DEFAULT, 0, INT_MAX});
 524     channelz_node =
 525         MakeRefCounted<channelz::ServerNode>(channel_tracer_max_memory);
 526     channelz_node->AddTraceEvent(
 527         channelz::ChannelTrace::Severity::Info,
 528         grpc_slice_from_static_string("Server created"));
 529   }
 530   return channelz_node;
 531 }
 532
 533 }  // namespace
 534
 535 Server::Server(const grpc_channel_args* args)
 536     : channel_args_(grpc_channel_args_copy(args)),
 537       channelz_node_(CreateChannelzNode(args)) {}
 538
 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
 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
 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
 600 grpc_error_handle 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, size_t preallocated_bytes) {
 605   // Create channel.
 606   grpc_error_handle error = GRPC_ERROR_NONE;
 607   grpc_channel* channel =
 608       grpc_channel_create(nullptr, args, GRPC_SERVER_CHANNEL, transport,
 609                           resource_user, preallocated_bytes, &error);
 610   if (channel == nullptr) {
 611     return error;
 612   }
 613   ChannelData* chand = static_cast<ChannelData*>(
 614       grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0)
 615           ->channel_data);
 616   // Set up CQs.
 617   size_t cq_idx;
 618   for (cq_idx = 0; cq_idx < cqs_.size(); cq_idx++) {
 619     if (grpc_cq_pollset(cqs_[cq_idx]) == accepting_pollset) break;
 620   }
 621   if (cq_idx == cqs_.size()) {
 622     // Completion queue not found.  Pick a random one to publish new calls to.
 623     cq_idx = static_cast<size_t>(rand()) % cqs_.size();
 624   }
 625   // Set up channelz node.
 626   intptr_t channelz_socket_uuid = 0;
 627   if (socket_node != nullptr) {
 628     channelz_socket_uuid = socket_node->uuid();
 629     channelz_node_->AddChildSocket(socket_node);
 630   }
 631   // Initialize chand.
 632   chand->InitTransport(Ref(), channel, cq_idx, transport, channelz_socket_uuid);
 633   return GRPC_ERROR_NONE;
 634 }
 635
 636 bool Server::HasOpenConnections() {
 637   MutexLock lock(&mu_global_);
 638   return !channels_.empty();
 639 }
 640
 641 void Server::SetRegisteredMethodAllocator(
 642     grpc_completion_queue* cq, void* method_tag,
 643     std::function<RegisteredCallAllocation()> allocator) {
 644   RegisteredMethod* rm = static_cast<RegisteredMethod*>(method_tag);
 645   rm->matcher = absl::make_unique<AllocatingRequestMatcherRegistered>(
 646       this, cq, rm, std::move(allocator));
 647 }
 648
 649 void Server::SetBatchMethodAllocator(
 650     grpc_completion_queue* cq, std::function<BatchCallAllocation()> allocator) {
 651   GPR_DEBUG_ASSERT(unregistered_request_matcher_ == nullptr);
 652   unregistered_request_matcher_ =
 653       absl::make_unique<AllocatingRequestMatcherBatch>(this, cq,
 654                                                        std::move(allocator));
 655 }
 656
 657 void Server::RegisterCompletionQueue(grpc_completion_queue* cq) {
 658   for (grpc_completion_queue* queue : cqs_) {
 659     if (queue == cq) return;
 660   }
 661   GRPC_CQ_INTERNAL_REF(cq, "server");
 662   cqs_.push_back(cq);
 663 }
 664
 665 namespace {
 666
 667 bool streq(const std::string& a, const char* b) {
 668   return (a.empty() && b == nullptr) ||
 669          ((b != nullptr) && !strcmp(a.c_str(), b));
 670 }
 671
 672 }  // namespace
 673
 674 Server::RegisteredMethod* Server::RegisterMethod(
 675     const char* method, const char* host,
 676     grpc_server_register_method_payload_handling payload_handling,
 677     uint32_t flags) {
 678   if (!method) {
 679     gpr_log(GPR_ERROR,
 680             "grpc_server_register_method method string cannot be NULL");
 681     return nullptr;
 682   }
 683   for (std::unique_ptr<RegisteredMethod>& m : registered_methods_) {
 684     if (streq(m->method, method) && streq(m->host, host)) {
 685       gpr_log(GPR_ERROR, "duplicate registration for %s@%s", method,
 686               host ? host : "*");
 687       return nullptr;
 688     }
 689   }
 690   if ((flags & ~GRPC_INITIAL_METADATA_USED_MASK) != 0) {
 691     gpr_log(GPR_ERROR, "grpc_server_register_method invalid flags 0x%08x",
 692             flags);
 693     return nullptr;
 694   }
 695   registered_methods_.emplace_back(absl::make_unique<RegisteredMethod>(
 696       method, host, payload_handling, flags));
 697   return registered_methods_.back().get();
 698 }
 699
 700 void Server::DoneRequestEvent(void* req, grpc_cq_completion* /*c*/) {
 701   delete static_cast<RequestedCall*>(req);
 702 }
 703
 704 void Server::FailCall(size_t cq_idx, RequestedCall* rc,
 705                       grpc_error_handle error) {
 706   *rc->call = nullptr;
 707   rc->initial_metadata->count = 0;
 708   GPR_ASSERT(error != GRPC_ERROR_NONE);
 709   grpc_cq_end_op(cqs_[cq_idx], rc->tag, error, DoneRequestEvent, rc,
 710                  &rc->completion);
 711 }
 712
 713 // Before calling MaybeFinishShutdown(), we must hold mu_global_ and not
 714 // hold mu_call_.
 715 void Server::MaybeFinishShutdown() {
 716   if (!ShutdownReady() || shutdown_published_) {
 717     return;
 718   }
 719   {
 720     MutexLock lock(&mu_call_);
 721     KillPendingWorkLocked(
 722         GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
 723   }
 724   if (!channels_.empty() || listeners_destroyed_ < listeners_.size()) {
 725     if (gpr_time_cmp(gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME),
 726                                   last_shutdown_message_time_),
 727                      gpr_time_from_seconds(1, GPR_TIMESPAN)) >= 0) {
 728       last_shutdown_message_time_ = gpr_now(GPR_CLOCK_REALTIME);
 729       gpr_log(GPR_DEBUG,
 730               "Waiting for %" PRIuPTR " channels and %" PRIuPTR "/%" PRIuPTR
 731               " listeners to be destroyed before shutting down server",
 732               channels_.size(), listeners_.size() - listeners_destroyed_,
 733               listeners_.size());
 734     }
 735     return;
 736   }
 737   shutdown_published_ = true;
 738   for (auto& shutdown_tag : shutdown_tags_) {
 739     Ref().release();
 740     grpc_cq_end_op(shutdown_tag.cq, shutdown_tag.tag, GRPC_ERROR_NONE,
 741                    DoneShutdownEvent, this, &shutdown_tag.completion);
 742   }
 743 }
 744
 745 void Server::KillPendingWorkLocked(grpc_error_handle error) {
 746   if (started_) {
 747     unregistered_request_matcher_->KillRequests(GRPC_ERROR_REF(error));
 748     unregistered_request_matcher_->ZombifyPending();
 749     for (std::unique_ptr<RegisteredMethod>& rm : registered_methods_) {
 750       rm->matcher->KillRequests(GRPC_ERROR_REF(error));
 751       rm->matcher->ZombifyPending();
 752     }
 753   }
 754   GRPC_ERROR_UNREF(error);
 755 }
 756
 757 std::vector<grpc_channel*> Server::GetChannelsLocked() const {
 758   std::vector<grpc_channel*> channels;
 759   channels.reserve(channels_.size());
 760   for (const ChannelData* chand : channels_) {
 761     channels.push_back(chand->channel());
 762     GRPC_CHANNEL_INTERNAL_REF(chand->channel(), "broadcast");
 763   }
 764   return channels;
 765 }
 766
 767 void Server::ListenerDestroyDone(void* arg, grpc_error_handle /*error*/) {
 768   Server* server = static_cast<Server*>(arg);
 769   MutexLock lock(&server->mu_global_);
 770   server->listeners_destroyed_++;
 771   server->MaybeFinishShutdown();
 772 }
 773
 774 namespace {
 775
 776 void DonePublishedShutdown(void* /*done_arg*/, grpc_cq_completion* storage) {
 777   delete storage;
 778 }
 779
 780 }  // namespace
 781
 782 // - Kills all pending requests-for-incoming-RPC-calls (i.e., the requests made
 783 //   via grpc_server_request_call() and grpc_server_request_registered_call()
 784 //   will now be cancelled). See KillPendingWorkLocked().
 785 //
 786 // - Shuts down the listeners (i.e., the server will no longer listen on the
 787 //   port for new incoming channels).
 788 //
 789 // - Iterates through all channels on the server and sends shutdown msg (see
 790 //   ChannelBroadcaster::BroadcastShutdown() for details) to the clients via
 791 //   the transport layer. The transport layer then guarantees the following:
 792 //    -- Sends shutdown to the client (e.g., HTTP2 transport sends GOAWAY).
 793 //    -- If the server has outstanding calls that are in the process, the
 794 //       connection is NOT closed until the server is done with all those calls.
 795 //    -- Once there are no more calls in progress, the channel is closed.
 796 void Server::ShutdownAndNotify(grpc_completion_queue* cq, void* tag) {
 797   ChannelBroadcaster broadcaster;
 798   {
 799     // Wait for startup to be finished.  Locks mu_global.
 800     MutexLock lock(&mu_global_);
 801     while (starting_) {
 802       starting_cv_.Wait(&mu_global_);
 803     }
 804     // Stay locked, and gather up some stuff to do.
 805     GPR_ASSERT(grpc_cq_begin_op(cq, tag));
 806     if (shutdown_published_) {
 807       grpc_cq_end_op(cq, tag, GRPC_ERROR_NONE, DonePublishedShutdown, nullptr,
 808                      new grpc_cq_completion);
 809       return;
 810     }
 811     shutdown_tags_.emplace_back(tag, cq);
 812     if (ShutdownCalled()) {
 813       return;
 814     }
 815     last_shutdown_message_time_ = gpr_now(GPR_CLOCK_REALTIME);
 816     broadcaster.FillChannelsLocked(GetChannelsLocked());
 817     // Collect all unregistered then registered calls.
 818     {
 819       MutexLock lock(&mu_call_);
 820       KillPendingWorkLocked(
 821           GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
 822     }
 823     ShutdownUnrefOnShutdownCall();
 824   }
 825   // Shutdown listeners.
 826   for (auto& listener : listeners_) {
 827     channelz::ListenSocketNode* channelz_listen_socket_node =
 828         listener.listener->channelz_listen_socket_node();
 829     if (channelz_node_ != nullptr && channelz_listen_socket_node != nullptr) {
 830       channelz_node_->RemoveChildListenSocket(
 831           channelz_listen_socket_node->uuid());
 832     }
 833     GRPC_CLOSURE_INIT(&listener.destroy_done, ListenerDestroyDone, this,
 834                       grpc_schedule_on_exec_ctx);
 835     listener.listener->SetOnDestroyDone(&listener.destroy_done);
 836     listener.listener.reset();
 837   }
 838   broadcaster.BroadcastShutdown(/*send_goaway=*/true, GRPC_ERROR_NONE);
 839 }
 840
 841 void Server::CancelAllCalls() {
 842   ChannelBroadcaster broadcaster;
 843   {
 844     MutexLock lock(&mu_global_);
 845     broadcaster.FillChannelsLocked(GetChannelsLocked());
 846   }
 847   broadcaster.BroadcastShutdown(
 848       /*send_goaway=*/false,
 849       GRPC_ERROR_CREATE_FROM_STATIC_STRING("Cancelling all calls"));
 850 }
 851
 852 void Server::Orphan() {
 853   {
 854     MutexLock lock(&mu_global_);
 855     GPR_ASSERT(ShutdownCalled() || listeners_.empty());
 856     GPR_ASSERT(listeners_destroyed_ == listeners_.size());
 857   }
 858   Unref();
 859 }
 860
 861 grpc_call_error Server::ValidateServerRequest(
 862     grpc_completion_queue* cq_for_notification, void* tag,
 863     grpc_byte_buffer** optional_payload, RegisteredMethod* rm) {
 864   if ((rm == nullptr && optional_payload != nullptr) ||
 865       ((rm != nullptr) && ((optional_payload == nullptr) !=
 866                            (rm->payload_handling == GRPC_SRM_PAYLOAD_NONE)))) {
 867     return GRPC_CALL_ERROR_PAYLOAD_TYPE_MISMATCH;
 868   }
 869   if (!grpc_cq_begin_op(cq_for_notification, tag)) {
 870     return GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN;
 871   }
 872   return GRPC_CALL_OK;
 873 }
 874
 875 grpc_call_error Server::ValidateServerRequestAndCq(
 876     size_t* cq_idx, grpc_completion_queue* cq_for_notification, void* tag,
 877     grpc_byte_buffer** optional_payload, RegisteredMethod* rm) {
 878   size_t idx;
 879   for (idx = 0; idx < cqs_.size(); idx++) {
 880     if (cqs_[idx] == cq_for_notification) {
 881       break;
 882     }
 883   }
 884   if (idx == cqs_.size()) {
 885     return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
 886   }
 887   grpc_call_error error =
 888       ValidateServerRequest(cq_for_notification, tag, optional_payload, rm);
 889   if (error != GRPC_CALL_OK) {
 890     return error;
 891   }
 892   *cq_idx = idx;
 893   return GRPC_CALL_OK;
 894 }
 895
 896 grpc_call_error Server::QueueRequestedCall(size_t cq_idx, RequestedCall* rc) {
 897   if (ShutdownCalled()) {
 898     FailCall(cq_idx, rc,
 899              GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server Shutdown"));
 900     return GRPC_CALL_OK;
 901   }
 902   RequestMatcherInterface* rm;
 903   switch (rc->type) {
 904     case RequestedCall::Type::BATCH_CALL:
 905       rm = unregistered_request_matcher_.get();
 906       break;
 907     case RequestedCall::Type::REGISTERED_CALL:
 908       rm = rc->data.registered.method->matcher.get();
 909       break;
 910   }
 911   rm->RequestCallWithPossiblePublish(cq_idx, rc);
 912   return GRPC_CALL_OK;
 913 }
 914
 915 grpc_call_error Server::RequestCall(grpc_call** call,
 916                                     grpc_call_details* details,
 917                                     grpc_metadata_array* request_metadata,
 918                                     grpc_completion_queue* cq_bound_to_call,
 919                                     grpc_completion_queue* cq_for_notification,
 920                                     void* tag) {
 921   size_t cq_idx;
 922   grpc_call_error error = ValidateServerRequestAndCq(
 923       &cq_idx, cq_for_notification, tag, nullptr, nullptr);
 924   if (error != GRPC_CALL_OK) {
 925     return error;
 926   }
 927   RequestedCall* rc =
 928       new RequestedCall(tag, cq_bound_to_call, call, request_metadata, details);
 929   return QueueRequestedCall(cq_idx, rc);
 930 }
 931
 932 grpc_call_error Server::RequestRegisteredCall(
 933     RegisteredMethod* rm, grpc_call** call, gpr_timespec* deadline,
 934     grpc_metadata_array* request_metadata, grpc_byte_buffer** optional_payload,
 935     grpc_completion_queue* cq_bound_to_call,
 936     grpc_completion_queue* cq_for_notification, void* tag_new) {
 937   size_t cq_idx;
 938   grpc_call_error error = ValidateServerRequestAndCq(
 939       &cq_idx, cq_for_notification, tag_new, optional_payload, rm);
 940   if (error != GRPC_CALL_OK) {
 941     return error;
 942   }
 943   RequestedCall* rc =
 944       new RequestedCall(tag_new, cq_bound_to_call, call, request_metadata, rm,
 945                         deadline, optional_payload);
 946   return QueueRequestedCall(cq_idx, rc);
 947 }
 948
 949 //
 950 // Server::ChannelData::ConnectivityWatcher
 951 //
 952
 953 class Server::ChannelData::ConnectivityWatcher
 954     : public AsyncConnectivityStateWatcherInterface {
 955  public:
 956   explicit ConnectivityWatcher(ChannelData* chand) : chand_(chand) {
 957     GRPC_CHANNEL_INTERNAL_REF(chand_->channel_, "connectivity");
 958   }
 959
 960   ~ConnectivityWatcher() override {
 961     GRPC_CHANNEL_INTERNAL_UNREF(chand_->channel_, "connectivity");
 962   }
 963
 964  private:
 965   void OnConnectivityStateChange(grpc_connectivity_state new_state,
 966                                  const absl::Status& /*status*/) override {
 967     // Don't do anything until we are being shut down.
 968     if (new_state != GRPC_CHANNEL_SHUTDOWN) return;
 969     // Shut down channel.
 970     MutexLock lock(&chand_->server_->mu_global_);
 971     chand_->Destroy();
 972   }
 973
 974   ChannelData* chand_;
 975 };
 976
 977 //
 978 // Server::ChannelData
 979 //
 980
 981 Server::ChannelData::~ChannelData() {
 982   if (registered_methods_ != nullptr) {
 983     for (const ChannelRegisteredMethod& crm : *registered_methods_) {
 984       grpc_slice_unref_internal(crm.method);
 985       GPR_DEBUG_ASSERT(crm.method.refcount == &kNoopRefcount ||
 986                        crm.method.refcount == nullptr);
 987       if (crm.has_host) {
 988         grpc_slice_unref_internal(crm.host);
 989         GPR_DEBUG_ASSERT(crm.host.refcount == &kNoopRefcount ||
 990                          crm.host.refcount == nullptr);
 991       }
 992     }
 993     registered_methods_.reset();
 994   }
 995   if (server_ != nullptr) {
 996     if (server_->channelz_node_ != nullptr && channelz_socket_uuid_ != 0) {
 997       server_->channelz_node_->RemoveChildSocket(channelz_socket_uuid_);
 998     }
 999     {
1000       MutexLock lock(&server_->mu_global_);
1001       if (list_position_.has_value()) {
1002         server_->channels_.erase(*list_position_);
1003         list_position_.reset();
1004       }
1005       server_->MaybeFinishShutdown();
1006     }
1007   }
1008 }
1009
1010 void Server::ChannelData::InitTransport(RefCountedPtr<Server> server,
1011                                         grpc_channel* channel, size_t cq_idx,
1012                                         grpc_transport* transport,
1013                                         intptr_t channelz_socket_uuid) {
1014   server_ = std::move(server);
1015   channel_ = channel;
1016   cq_idx_ = cq_idx;
1017   channelz_socket_uuid_ = channelz_socket_uuid;
1018   // Build a lookup table phrased in terms of mdstr's in this channels context
1019   // to quickly find registered methods.
1020   size_t num_registered_methods = server_->registered_methods_.size();
1021   if (num_registered_methods > 0) {
1022     uint32_t max_probes = 0;
1023     size_t slots = 2 * num_registered_methods;
1024     registered_methods_ =
1025         absl::make_unique<std::vector<ChannelRegisteredMethod>>(slots);
1026     for (std::unique_ptr<RegisteredMethod>& rm : server_->registered_methods_) {
1027       ExternallyManagedSlice host;
1028       ExternallyManagedSlice method(rm->method.c_str());
1029       const bool has_host = !rm->host.empty();
1030       if (has_host) {
1031         host = ExternallyManagedSlice(rm->host.c_str());
1032       }
1033       uint32_t hash =
1034           GRPC_MDSTR_KV_HASH(has_host ? host.Hash() : 0, method.Hash());
1035       uint32_t probes = 0;
1036       for (probes = 0; (*registered_methods_)[(hash + probes) % slots]
1037                            .server_registered_method != nullptr;
1038            probes++) {
1039       }
1040       if (probes > max_probes) max_probes = probes;
1041       ChannelRegisteredMethod* crm =
1042           &(*registered_methods_)[(hash + probes) % slots];
1043       crm->server_registered_method = rm.get();
1044       crm->flags = rm->flags;
1045       crm->has_host = has_host;
1046       if (has_host) {
1047         crm->host = host;
1048       }
1049       crm->method = method;
1050     }
1051     GPR_ASSERT(slots <= UINT32_MAX);
1052     registered_method_max_probes_ = max_probes;
1053   }
1054   // Publish channel.
1055   {
1056     MutexLock lock(&server_->mu_global_);
1057     server_->channels_.push_front(this);
1058     list_position_ = server_->channels_.begin();
1059   }
1060   // Start accept_stream transport op.
1061   grpc_transport_op* op = grpc_make_transport_op(nullptr);
1062   op->set_accept_stream = true;
1063   op->set_accept_stream_fn = AcceptStream;
1064   op->set_accept_stream_user_data = this;
1065   op->start_connectivity_watch = MakeOrphanable<ConnectivityWatcher>(this);
1066   if (server_->ShutdownCalled()) {
1067     op->disconnect_with_error =
1068         GRPC_ERROR_CREATE_FROM_STATIC_STRING("Server shutdown");
1069   }
1070   grpc_transport_perform_op(transport, op);
1071 }
1072
1073 Server::ChannelRegisteredMethod* Server::ChannelData::GetRegisteredMethod(
1074     const grpc_slice& host, const grpc_slice& path, bool is_idempotent) {
1075   if (registered_methods_ == nullptr) return nullptr;
1076   /* TODO(ctiller): unify these two searches */
1077   /* check for an exact match with host */
1078   uint32_t hash = GRPC_MDSTR_KV_HASH(grpc_slice_hash_internal(host),
1079                                      grpc_slice_hash_internal(path));
1080   for (size_t i = 0; i <= registered_method_max_probes_; i++) {
1081     ChannelRegisteredMethod* rm =
1082         &(*registered_methods_)[(hash + i) % registered_methods_->size()];
1083     if (rm->server_registered_method == nullptr) break;
1084     if (!rm->has_host) continue;
1085     if (rm->host != host) continue;
1086     if (rm->method != path) continue;
1087     if ((rm->flags & GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) &&
1088         !is_idempotent) {
1089       continue;
1090     }
1091     return rm;
1092   }
1093   /* check for a wildcard method definition (no host set) */
1094   hash = GRPC_MDSTR_KV_HASH(0, grpc_slice_hash_internal(path));
1095   for (size_t i = 0; i <= registered_method_max_probes_; i++) {
1096     ChannelRegisteredMethod* rm =
1097         &(*registered_methods_)[(hash + i) % registered_methods_->size()];
1098     if (rm->server_registered_method == nullptr) break;
1099     if (rm->has_host) continue;
1100     if (rm->method != path) continue;
1101     if ((rm->flags & GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) &&
1102         !is_idempotent) {
1103       continue;
1104     }
1105     return rm;
1106   }
1107   return nullptr;
1108 }
1109
1110 void Server::ChannelData::AcceptStream(void* arg, grpc_transport* /*transport*/,
1111                                        const void* transport_server_data) {
1112   auto* chand = static_cast<Server::ChannelData*>(arg);
1113   /* create a call */
1114   grpc_call_create_args args;
1115   args.channel = chand->channel_;
1116   args.server = chand->server_.get();
1117   args.parent = nullptr;
1118   args.propagation_mask = 0;
1119   args.cq = nullptr;
1120   args.pollset_set_alternative = nullptr;
1121   args.server_transport_data = transport_server_data;
1122   args.add_initial_metadata = nullptr;
1123   args.add_initial_metadata_count = 0;
1124   args.send_deadline = GRPC_MILLIS_INF_FUTURE;
1125   grpc_call* call;
1126   grpc_error_handle error = grpc_call_create(&args, &call);
1127   grpc_call_element* elem =
1128       grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
1129   auto* calld = static_cast<Server::CallData*>(elem->call_data);
1130   if (error != GRPC_ERROR_NONE) {
1131     GRPC_ERROR_UNREF(error);
1132     calld->FailCallCreation();
1133     return;
1134   }
1135   calld->Start(elem);
1136 }
1137
1138 void Server::ChannelData::FinishDestroy(void* arg,
1139                                         grpc_error_handle /*error*/) {
1140   auto* chand = static_cast<Server::ChannelData*>(arg);
1141   Server* server = chand->server_.get();
1142   GRPC_CHANNEL_INTERNAL_UNREF(chand->channel_, "server");
1143   server->Unref();
1144 }
1145
1146 void Server::ChannelData::Destroy() {
1147   if (!list_position_.has_value()) return;
1148   GPR_ASSERT(server_ != nullptr);
1149   server_->channels_.erase(*list_position_);
1150   list_position_.reset();
1151   server_->Ref().release();
1152   server_->MaybeFinishShutdown();
1153   GRPC_CLOSURE_INIT(&finish_destroy_channel_closure_, FinishDestroy, this,
1154                     grpc_schedule_on_exec_ctx);
1155   if (GRPC_TRACE_FLAG_ENABLED(grpc_server_channel_trace)) {
1156     gpr_log(GPR_INFO, "Disconnected client");
1157   }
1158   grpc_transport_op* op =
1159       grpc_make_transport_op(&finish_destroy_channel_closure_);
1160   op->set_accept_stream = true;
1161   grpc_channel_next_op(
1162       grpc_channel_stack_element(grpc_channel_get_channel_stack(channel_), 0),
1163       op);
1164 }
1165
1166 grpc_error_handle Server::ChannelData::InitChannelElement(
1167     grpc_channel_element* elem, grpc_channel_element_args* args) {
1168   GPR_ASSERT(args->is_first);
1169   GPR_ASSERT(!args->is_last);
1170   new (elem->channel_data) ChannelData();
1171   return GRPC_ERROR_NONE;
1172 }
1173
1174 void Server::ChannelData::DestroyChannelElement(grpc_channel_element* elem) {
1175   auto* chand = static_cast<ChannelData*>(elem->channel_data);
1176   chand->~ChannelData();
1177 }
1178
1179 //
1180 // Server::CallData
1181 //
1182
1183 Server::CallData::CallData(grpc_call_element* elem,
1184                            const grpc_call_element_args& args,
1185                            RefCountedPtr<Server> server)
1186     : server_(std::move(server)),
1187       call_(grpc_call_from_top_element(elem)),
1188       call_combiner_(args.call_combiner) {
1189   GRPC_CLOSURE_INIT(&recv_initial_metadata_ready_, RecvInitialMetadataReady,
1190                     elem, grpc_schedule_on_exec_ctx);
1191   GRPC_CLOSURE_INIT(&recv_trailing_metadata_ready_, RecvTrailingMetadataReady,
1192                     elem, grpc_schedule_on_exec_ctx);
1193 }
1194
1195 Server::CallData::~CallData() {
1196   GPR_ASSERT(state_.load(std::memory_order_relaxed) != CallState::PENDING);
1197   GRPC_ERROR_UNREF(recv_initial_metadata_error_);
1198   if (host_.has_value()) {
1199     grpc_slice_unref_internal(*host_);
1200   }
1201   if (path_.has_value()) {
1202     grpc_slice_unref_internal(*path_);
1203   }
1204   grpc_metadata_array_destroy(&initial_metadata_);
1205   grpc_byte_buffer_destroy(payload_);
1206 }
1207
1208 void Server::CallData::SetState(CallState state) {
1209   state_.store(state, std::memory_order_relaxed);
1210 }
1211
1212 bool Server::CallData::MaybeActivate() {
1213   CallState expected = CallState::PENDING;
1214   return state_.compare_exchange_strong(expected, CallState::ACTIVATED,
1215                                         std::memory_order_acq_rel,
1216                                         std::memory_order_relaxed);
1217 }
1218
1219 void Server::CallData::FailCallCreation() {
1220   CallState expected_not_started = CallState::NOT_STARTED;
1221   CallState expected_pending = CallState::PENDING;
1222   if (state_.compare_exchange_strong(expected_not_started, CallState::ZOMBIED,
1223                                      std::memory_order_acq_rel,
1224                                      std::memory_order_acquire)) {
1225     KillZombie();
1226   } else if (state_.compare_exchange_strong(
1227                  expected_pending, CallState::ZOMBIED,
1228                  std::memory_order_acq_rel, std::memory_order_relaxed)) {
1229     // Zombied call will be destroyed when it's removed from the pending
1230     // queue... later.
1231   }
1232 }
1233
1234 void Server::CallData::Start(grpc_call_element* elem) {
1235   grpc_op op;
1236   op.op = GRPC_OP_RECV_INITIAL_METADATA;
1237   op.flags = 0;
1238   op.reserved = nullptr;
1239   op.data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_;
1240   GRPC_CLOSURE_INIT(&recv_initial_metadata_batch_complete_,
1241                     RecvInitialMetadataBatchComplete, elem,
1242                     grpc_schedule_on_exec_ctx);
1243   grpc_call_start_batch_and_execute(call_, &op, 1,
1244                                     &recv_initial_metadata_batch_complete_);
1245 }
1246
1247 void Server::CallData::Publish(size_t cq_idx, RequestedCall* rc) {
1248   grpc_call_set_completion_queue(call_, rc->cq_bound_to_call);
1249   *rc->call = call_;
1250   cq_new_ = server_->cqs_[cq_idx];
1251   GPR_SWAP(grpc_metadata_array, *rc->initial_metadata, initial_metadata_);
1252   switch (rc->type) {
1253     case RequestedCall::Type::BATCH_CALL:
1254       GPR_ASSERT(host_.has_value());
1255       GPR_ASSERT(path_.has_value());
1256       rc->data.batch.details->host = grpc_slice_ref_internal(*host_);
1257       rc->data.batch.details->method = grpc_slice_ref_internal(*path_);
1258       rc->data.batch.details->deadline =
1259           grpc_millis_to_timespec(deadline_, GPR_CLOCK_MONOTONIC);
1260       rc->data.batch.details->flags = recv_initial_metadata_flags_;
1261       break;
1262     case RequestedCall::Type::REGISTERED_CALL:
1263       *rc->data.registered.deadline =
1264           grpc_millis_to_timespec(deadline_, GPR_CLOCK_MONOTONIC);
1265       if (rc->data.registered.optional_payload != nullptr) {
1266         *rc->data.registered.optional_payload = payload_;
1267         payload_ = nullptr;
1268       }
1269       break;
1270     default:
1271       GPR_UNREACHABLE_CODE(return );
1272   }
1273   grpc_cq_end_op(cq_new_, rc->tag, GRPC_ERROR_NONE, Server::DoneRequestEvent,
1274                  rc, &rc->completion, true);
1275 }
1276
1277 void Server::CallData::PublishNewRpc(void* arg, grpc_error_handle error) {
1278   grpc_call_element* call_elem = static_cast<grpc_call_element*>(arg);
1279   auto* calld = static_cast<Server::CallData*>(call_elem->call_data);
1280   auto* chand = static_cast<Server::ChannelData*>(call_elem->channel_data);
1281   RequestMatcherInterface* rm = calld->matcher_;
1282   Server* server = rm->server();
1283   if (error != GRPC_ERROR_NONE || server->ShutdownCalled()) {
1284     calld->state_.store(CallState::ZOMBIED, std::memory_order_relaxed);
1285     calld->KillZombie();
1286     return;
1287   }
1288   rm->MatchOrQueue(chand->cq_idx(), calld);
1289 }
1290
1291 namespace {
1292
1293 void KillZombieClosure(void* call, grpc_error_handle /*error*/) {
1294   grpc_call_unref(static_cast<grpc_call*>(call));
1295 }
1296
1297 }  // namespace
1298
1299 void Server::CallData::KillZombie() {
1300   GRPC_CLOSURE_INIT(&kill_zombie_closure_, KillZombieClosure, call_,
1301                     grpc_schedule_on_exec_ctx);
1302   ExecCtx::Run(DEBUG_LOCATION, &kill_zombie_closure_, GRPC_ERROR_NONE);
1303 }
1304
1305 void Server::CallData::StartNewRpc(grpc_call_element* elem) {
1306   auto* chand = static_cast<ChannelData*>(elem->channel_data);
1307   if (server_->ShutdownCalled()) {
1308     state_.store(CallState::ZOMBIED, std::memory_order_relaxed);
1309     KillZombie();
1310     return;
1311   }
1312   // Find request matcher.
1313   matcher_ = server_->unregistered_request_matcher_.get();
1314   grpc_server_register_method_payload_handling payload_handling =
1315       GRPC_SRM_PAYLOAD_NONE;
1316   if (path_.has_value() && host_.has_value()) {
1317     ChannelRegisteredMethod* rm =
1318         chand->GetRegisteredMethod(*host_, *path_,
1319                                    (recv_initial_metadata_flags_ &
1320                                     GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST));
1321     if (rm != nullptr) {
1322       matcher_ = rm->server_registered_method->matcher.get();
1323       payload_handling = rm->server_registered_method->payload_handling;
1324     }
1325   }
1326   // Start recv_message op if needed.
1327   switch (payload_handling) {
1328     case GRPC_SRM_PAYLOAD_NONE:
1329       PublishNewRpc(elem, GRPC_ERROR_NONE);
1330       break;
1331     case GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER: {
1332       grpc_op op;
1333       op.op = GRPC_OP_RECV_MESSAGE;
1334       op.flags = 0;
1335       op.reserved = nullptr;
1336       op.data.recv_message.recv_message = &payload_;
1337       GRPC_CLOSURE_INIT(&publish_, PublishNewRpc, elem,
1338                         grpc_schedule_on_exec_ctx);
1339       grpc_call_start_batch_and_execute(call_, &op, 1, &publish_);
1340       break;
1341     }
1342   }
1343 }
1344
1345 void Server::CallData::RecvInitialMetadataBatchComplete(
1346     void* arg, grpc_error_handle error) {
1347   grpc_call_element* elem = static_cast<grpc_call_element*>(arg);
1348   auto* calld = static_cast<Server::CallData*>(elem->call_data);
1349   if (error != GRPC_ERROR_NONE) {
1350     calld->FailCallCreation();
1351     return;
1352   }
1353   calld->StartNewRpc(elem);
1354 }
1355
1356 void Server::CallData::StartTransportStreamOpBatchImpl(
1357     grpc_call_element* elem, grpc_transport_stream_op_batch* batch) {
1358   if (batch->recv_initial_metadata) {
1359     GPR_ASSERT(batch->payload->recv_initial_metadata.recv_flags == nullptr);
1360     recv_initial_metadata_ =
1361         batch->payload->recv_initial_metadata.recv_initial_metadata;
1362     original_recv_initial_metadata_ready_ =
1363         batch->payload->recv_initial_metadata.recv_initial_metadata_ready;
1364     batch->payload->recv_initial_metadata.recv_initial_metadata_ready =
1365         &recv_initial_metadata_ready_;
1366     batch->payload->recv_initial_metadata.recv_flags =
1367         &recv_initial_metadata_flags_;
1368   }
1369   if (batch->recv_trailing_metadata) {
1370     original_recv_trailing_metadata_ready_ =
1371         batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready;
1372     batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready =
1373         &recv_trailing_metadata_ready_;
1374   }
1375   grpc_call_next_op(elem, batch);
1376 }
1377
1378 void Server::CallData::RecvInitialMetadataReady(void* arg,
1379                                                 grpc_error_handle error) {
1380   grpc_call_element* elem = static_cast<grpc_call_element*>(arg);
1381   CallData* calld = static_cast<CallData*>(elem->call_data);
1382   grpc_millis op_deadline;
1383   if (error == GRPC_ERROR_NONE) {
1384     GPR_DEBUG_ASSERT(calld->recv_initial_metadata_->idx.named.path != nullptr);
1385     GPR_DEBUG_ASSERT(calld->recv_initial_metadata_->idx.named.authority !=
1386                      nullptr);
1387     calld->path_.emplace(grpc_slice_ref_internal(
1388         GRPC_MDVALUE(calld->recv_initial_metadata_->idx.named.path->md)));
1389     calld->host_.emplace(grpc_slice_ref_internal(
1390         GRPC_MDVALUE(calld->recv_initial_metadata_->idx.named.authority->md)));
1391     grpc_metadata_batch_remove(calld->recv_initial_metadata_, GRPC_BATCH_PATH);
1392     grpc_metadata_batch_remove(calld->recv_initial_metadata_,
1393                                GRPC_BATCH_AUTHORITY);
1394   } else {
1395     GRPC_ERROR_REF(error);
1396   }
1397   op_deadline = calld->recv_initial_metadata_->deadline;
1398   if (op_deadline != GRPC_MILLIS_INF_FUTURE) {
1399     calld->deadline_ = op_deadline;
1400   }
1401   if (calld->host_.has_value() && calld->path_.has_value()) {
1402     /* do nothing */
1403   } else {
1404     /* Pass the error reference to calld->recv_initial_metadata_error */
1405     grpc_error_handle src_error = error;
1406     error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
1407         "Missing :authority or :path", &src_error, 1);
1408     GRPC_ERROR_UNREF(src_error);
1409     calld->recv_initial_metadata_error_ = GRPC_ERROR_REF(error);
1410   }
1411   grpc_closure* closure = calld->original_recv_initial_metadata_ready_;
1412   calld->original_recv_initial_metadata_ready_ = nullptr;
1413   if (calld->seen_recv_trailing_metadata_ready_) {
1414     GRPC_CALL_COMBINER_START(calld->call_combiner_,
1415                              &calld->recv_trailing_metadata_ready_,
1416                              calld->recv_trailing_metadata_error_,
1417                              "continue server recv_trailing_metadata_ready");
1418   }
1419   Closure::Run(DEBUG_LOCATION, closure, error);
1420 }
1421
1422 void Server::CallData::RecvTrailingMetadataReady(void* arg,
1423                                                  grpc_error_handle error) {
1424   grpc_call_element* elem = static_cast<grpc_call_element*>(arg);
1425   CallData* calld = static_cast<CallData*>(elem->call_data);
1426   if (calld->original_recv_initial_metadata_ready_ != nullptr) {
1427     calld->recv_trailing_metadata_error_ = GRPC_ERROR_REF(error);
1428     calld->seen_recv_trailing_metadata_ready_ = true;
1429     GRPC_CLOSURE_INIT(&calld->recv_trailing_metadata_ready_,
1430                       RecvTrailingMetadataReady, elem,
1431                       grpc_schedule_on_exec_ctx);
1432     GRPC_CALL_COMBINER_STOP(calld->call_combiner_,
1433                             "deferring server recv_trailing_metadata_ready "
1434                             "until after recv_initial_metadata_ready");
1435     return;
1436   }
1437   error =
1438       grpc_error_add_child(GRPC_ERROR_REF(error),
1439                            GRPC_ERROR_REF(calld->recv_initial_metadata_error_));
1440   Closure::Run(DEBUG_LOCATION, calld->original_recv_trailing_metadata_ready_,
1441                error);
1442 }
1443
1444 grpc_error_handle Server::CallData::InitCallElement(
1445     grpc_call_element* elem, const grpc_call_element_args* args) {
1446   auto* chand = static_cast<ChannelData*>(elem->channel_data);
1447   new (elem->call_data) Server::CallData(elem, *args, chand->server());
1448   return GRPC_ERROR_NONE;
1449 }
1450
1451 void Server::CallData::DestroyCallElement(
1452     grpc_call_element* elem, const grpc_call_final_info* /*final_info*/,
1453     grpc_closure* /*ignored*/) {
1454   auto* calld = static_cast<CallData*>(elem->call_data);
1455   calld->~CallData();
1456 }
1457
1458 void Server::CallData::StartTransportStreamOpBatch(
1459     grpc_call_element* elem, grpc_transport_stream_op_batch* batch) {
1460   auto* calld = static_cast<CallData*>(elem->call_data);
1461   calld->StartTransportStreamOpBatchImpl(elem, batch);
1462 }
1463
1464 }  // namespace grpc_core
1465
1466 //
1467 // C-core API
1468 //
1469
1470 grpc_server* grpc_server_create(const grpc_channel_args* args, void* reserved) {
1471   grpc_core::ExecCtx exec_ctx;
1472   GRPC_API_TRACE("grpc_server_create(%p, %p)", 2, (args, reserved));
1473   grpc_server* c_server = new grpc_server;
1474   c_server->core_server = grpc_core::MakeOrphanable<grpc_core::Server>(args);
1475   return c_server;
1476 }
1477
1478 void grpc_server_register_completion_queue(grpc_server* server,
1479                                            grpc_completion_queue* cq,
1480                                            void* reserved) {
1481   GRPC_API_TRACE(
1482       "grpc_server_register_completion_queue(server=%p, cq=%p, reserved=%p)", 3,
1483       (server, cq, reserved));
1484   GPR_ASSERT(!reserved);
1485   auto cq_type = grpc_get_cq_completion_type(cq);
1486   if (cq_type != GRPC_CQ_NEXT && cq_type != GRPC_CQ_CALLBACK) {
1487     gpr_log(GPR_INFO,
1488             "Completion queue of type %d is being registered as a "
1489             "server-completion-queue",
1490             static_cast<int>(cq_type));
1491     /* Ideally we should log an error and abort but ruby-wrapped-language API
1492        calls grpc_completion_queue_pluck() on server completion queues */
1493   }
1494   server->core_server->RegisterCompletionQueue(cq);
1495 }
1496
1497 void* grpc_server_register_method(
1498     grpc_server* server, const char* method, const char* host,
1499     grpc_server_register_method_payload_handling payload_handling,
1500     uint32_t flags) {
1501   GRPC_API_TRACE(
1502       "grpc_server_register_method(server=%p, method=%s, host=%s, "
1503       "flags=0x%08x)",
1504       4, (server, method, host, flags));
1505   return server->core_server->RegisterMethod(method, host, payload_handling,
1506                                              flags);
1507 }
1508
1509 void grpc_server_start(grpc_server* server) {
1510   grpc_core::ExecCtx exec_ctx;
1511   GRPC_API_TRACE("grpc_server_start(server=%p)", 1, (server));
1512   server->core_server->Start();
1513 }
1514
1515 void grpc_server_shutdown_and_notify(grpc_server* server,
1516                                      grpc_completion_queue* cq, void* tag) {
1517   grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1518   grpc_core::ExecCtx exec_ctx;
1519   GRPC_API_TRACE("grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", 3,
1520                  (server, cq, tag));
1521   server->core_server->ShutdownAndNotify(cq, tag);
1522 }
1523
1524 void grpc_server_cancel_all_calls(grpc_server* server) {
1525   grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1526   grpc_core::ExecCtx exec_ctx;
1527   GRPC_API_TRACE("grpc_server_cancel_all_calls(server=%p)", 1, (server));
1528   server->core_server->CancelAllCalls();
1529 }
1530
1531 void grpc_server_destroy(grpc_server* server) {
1532   grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1533   grpc_core::ExecCtx exec_ctx;
1534   GRPC_API_TRACE("grpc_server_destroy(server=%p)", 1, (server));
1535   delete server;
1536 }
1537
1538 grpc_call_error grpc_server_request_call(
1539     grpc_server* server, grpc_call** call, grpc_call_details* details,
1540     grpc_metadata_array* request_metadata,
1541     grpc_completion_queue* cq_bound_to_call,
1542     grpc_completion_queue* cq_for_notification, void* tag) {
1543   grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1544   grpc_core::ExecCtx exec_ctx;
1545   GRPC_STATS_INC_SERVER_REQUESTED_CALLS();
1546   GRPC_API_TRACE(
1547       "grpc_server_request_call("
1548       "server=%p, call=%p, details=%p, initial_metadata=%p, "
1549       "cq_bound_to_call=%p, cq_for_notification=%p, tag=%p)",
1550       7,
1551       (server, call, details, request_metadata, cq_bound_to_call,
1552        cq_for_notification, tag));
1553   return server->core_server->RequestCall(call, details, request_metadata,
1554                                           cq_bound_to_call, cq_for_notification,
1555                                           tag);
1556 }
1557
1558 grpc_call_error grpc_server_request_registered_call(
1559     grpc_server* server, void* registered_method, grpc_call** call,
1560     gpr_timespec* deadline, grpc_metadata_array* request_metadata,
1561     grpc_byte_buffer** optional_payload,
1562     grpc_completion_queue* cq_bound_to_call,
1563     grpc_completion_queue* cq_for_notification, void* tag_new) {
1564   grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1565   grpc_core::ExecCtx exec_ctx;
1566   GRPC_STATS_INC_SERVER_REQUESTED_CALLS();
1567   auto* rm =
1568       static_cast<grpc_core::Server::RegisteredMethod*>(registered_method);
1569   GRPC_API_TRACE(
1570       "grpc_server_request_registered_call("
1571       "server=%p, registered_method=%p, call=%p, deadline=%p, "
1572       "request_metadata=%p, "
1573       "optional_payload=%p, cq_bound_to_call=%p, cq_for_notification=%p, "
1574       "tag=%p)",
1575       9,
1576       (server, registered_method, call, deadline, request_metadata,
1577        optional_payload, cq_bound_to_call, cq_for_notification, tag_new));
1578   return server->core_server->RequestRegisteredCall(
1579       rm, call, deadline, request_metadata, optional_payload, cq_bound_to_call,
1580       cq_for_notification, tag_new);
1581 }
1582
1583 void grpc_server_set_config_fetcher(
1584     grpc_server* server, grpc_server_config_fetcher* server_config_fetcher) {
1585   grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1586   grpc_core::ExecCtx exec_ctx;
1587   GRPC_API_TRACE("grpc_server_set_config_fetcher(server=%p, config_fetcher=%p)",
1588                  2, (server, server_config_fetcher));
1589   server->core_server->set_config_fetcher(
1590       std::unique_ptr<grpc_server_config_fetcher>(server_config_fetcher));
1591 }
1592
1593 void grpc_server_config_fetcher_destroy(
1594     grpc_server_config_fetcher* server_config_fetcher) {
1595   grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
1596   grpc_core::ExecCtx exec_ctx;
1597   GRPC_API_TRACE("grpc_server_config_fetcher_destroy(config_fetcher=%p)", 1,
1598                  (server_config_fetcher));
1599   delete server_config_fetcher;
1600 }