include/grpcpp/impl/codegen/client_callback.h

   1 /*
   2  *
   3  * Copyright 2018 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #ifndef GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
  20 #define GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H
  21
  22 #include <functional>
  23
  24 #include <grpcpp/impl/codegen/call.h>
  25 #include <grpcpp/impl/codegen/call_op_set.h>
  26 #include <grpcpp/impl/codegen/callback_common.h>
  27 #include <grpcpp/impl/codegen/channel_interface.h>
  28 #include <grpcpp/impl/codegen/config.h>
  29 #include <grpcpp/impl/codegen/core_codegen_interface.h>
  30 #include <grpcpp/impl/codegen/status.h>
  31
  32 namespace grpc_impl {
  33 class Channel;
  34 }
  35
  36 namespace grpc {
  37
  38 class ClientContext;
  39
  40 namespace internal {
  41 class RpcMethod;
  42
  43 /// Perform a callback-based unary call
  44 /// TODO(vjpai): Combine as much as possible with the blocking unary call code
  45 template <class InputMessage, class OutputMessage>
  46 void CallbackUnaryCall(ChannelInterface* channel, const RpcMethod& method,
  47                        ClientContext* context, const InputMessage* request,
  48                        OutputMessage* result,
  49                        std::function<void(Status)> on_completion) {
  50   CallbackUnaryCallImpl<InputMessage, OutputMessage> x(
  51       channel, method, context, request, result, on_completion);
  52 }
  53
  54 template <class InputMessage, class OutputMessage>
  55 class CallbackUnaryCallImpl {
  56  public:
  57   CallbackUnaryCallImpl(ChannelInterface* channel, const RpcMethod& method,
  58                         ClientContext* context, const InputMessage* request,
  59                         OutputMessage* result,
  60                         std::function<void(Status)> on_completion) {
  61     CompletionQueue* cq = channel->CallbackCQ();
  62     GPR_CODEGEN_ASSERT(cq != nullptr);
  63     Call call(channel->CreateCall(method, context, cq));
  64
  65     using FullCallOpSet =
  66         CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
  67                   CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>,
  68                   CallOpClientSendClose, CallOpClientRecvStatus>;
  69
  70     auto* ops = new (g_core_codegen_interface->grpc_call_arena_alloc(
  71         call.call(), sizeof(FullCallOpSet))) FullCallOpSet;
  72
  73     auto* tag = new (g_core_codegen_interface->grpc_call_arena_alloc(
  74         call.call(), sizeof(CallbackWithStatusTag)))
  75         CallbackWithStatusTag(call.call(), on_completion, ops);
  76
  77     // TODO(vjpai): Unify code with sync API as much as possible
  78     Status s = ops->SendMessagePtr(request);
  79     if (!s.ok()) {
  80       tag->force_run(s);
  81       return;
  82     }
  83     ops->SendInitialMetadata(&context->send_initial_metadata_,
  84                              context->initial_metadata_flags());
  85     ops->RecvInitialMetadata(context);
  86     ops->RecvMessage(result);
  87     ops->AllowNoMessage();
  88     ops->ClientSendClose();
  89     ops->ClientRecvStatus(context, tag->status_ptr());
  90     ops->set_core_cq_tag(tag);
  91     call.PerformOps(ops);
  92   }
  93 };
  94 }  // namespace internal
  95
  96 namespace experimental {
  97
  98 // Forward declarations
  99 template <class Request, class Response>
 100 class ClientBidiReactor;
 101 template <class Response>
 102 class ClientReadReactor;
 103 template <class Request>
 104 class ClientWriteReactor;
 105 class ClientUnaryReactor;
 106
 107 // NOTE: The streaming objects are not actually implemented in the public API.
 108 //       These interfaces are provided for mocking only. Typical applications
 109 //       will interact exclusively with the reactors that they define.
 110 template <class Request, class Response>
 111 class ClientCallbackReaderWriter {
 112  public:
 113   virtual ~ClientCallbackReaderWriter() {}
 114   virtual void StartCall() = 0;
 115   virtual void Write(const Request* req, WriteOptions options) = 0;
 116   virtual void WritesDone() = 0;
 117   virtual void Read(Response* resp) = 0;
 118   virtual void AddHold(int holds) = 0;
 119   virtual void RemoveHold() = 0;
 120
 121  protected:
 122   void BindReactor(ClientBidiReactor<Request, Response>* reactor) {
 123     reactor->BindStream(this);
 124   }
 125 };
 126
 127 template <class Response>
 128 class ClientCallbackReader {
 129  public:
 130   virtual ~ClientCallbackReader() {}
 131   virtual void StartCall() = 0;
 132   virtual void Read(Response* resp) = 0;
 133   virtual void AddHold(int holds) = 0;
 134   virtual void RemoveHold() = 0;
 135
 136  protected:
 137   void BindReactor(ClientReadReactor<Response>* reactor) {
 138     reactor->BindReader(this);
 139   }
 140 };
 141
 142 template <class Request>
 143 class ClientCallbackWriter {
 144  public:
 145   virtual ~ClientCallbackWriter() {}
 146   virtual void StartCall() = 0;
 147   void Write(const Request* req) { Write(req, WriteOptions()); }
 148   virtual void Write(const Request* req, WriteOptions options) = 0;
 149   void WriteLast(const Request* req, WriteOptions options) {
 150     Write(req, options.set_last_message());
 151   }
 152   virtual void WritesDone() = 0;
 153
 154   virtual void AddHold(int holds) = 0;
 155   virtual void RemoveHold() = 0;
 156
 157  protected:
 158   void BindReactor(ClientWriteReactor<Request>* reactor) {
 159     reactor->BindWriter(this);
 160   }
 161 };
 162
 163 class ClientCallbackUnary {
 164  public:
 165   virtual ~ClientCallbackUnary() {}
 166   virtual void StartCall() = 0;
 167
 168  protected:
 169   void BindReactor(ClientUnaryReactor* reactor);
 170 };
 171
 172 // The following classes are the reactor interfaces that are to be implemented
 173 // by the user. They are passed in to the library as an argument to a call on a
 174 // stub (either a codegen-ed call or a generic call). The streaming RPC is
 175 // activated by calling StartCall, possibly after initiating StartRead,
 176 // StartWrite, or AddHold operations on the streaming object. Note that none of
 177 // the classes are pure; all reactions have a default empty reaction so that the
 178 // user class only needs to override those classes that it cares about.
 179 // The reactor must be passed to the stub invocation before any of the below
 180 // operations can be called.
 181
 182 /// \a ClientBidiReactor is the interface for a bidirectional streaming RPC.
 183 template <class Request, class Response>
 184 class ClientBidiReactor {
 185  public:
 186   virtual ~ClientBidiReactor() {}
 187
 188   /// Activate the RPC and initiate any reads or writes that have been Start'ed
 189   /// before this call. All streaming RPCs issued by the client MUST have
 190   /// StartCall invoked on them (even if they are canceled) as this call is the
 191   /// activation of their lifecycle.
 192   void StartCall() { stream_->StartCall(); }
 193
 194   /// Initiate a read operation (or post it for later initiation if StartCall
 195   /// has not yet been invoked).
 196   ///
 197   /// \param[out] resp Where to eventually store the read message. Valid when
 198   ///                  the library calls OnReadDone
 199   void StartRead(Response* resp) { stream_->Read(resp); }
 200
 201   /// Initiate a write operation (or post it for later initiation if StartCall
 202   /// has not yet been invoked).
 203   ///
 204   /// \param[in] req The message to be written. The library takes temporary
 205   ///                ownership until OnWriteDone, at which point the application
 206   ///                regains ownership of msg.
 207   void StartWrite(const Request* req) { StartWrite(req, WriteOptions()); }
 208
 209   /// Initiate/post a write operation with specified options.
 210   ///
 211   /// \param[in] req The message to be written. The library takes temporary
 212   ///                ownership until OnWriteDone, at which point the application
 213   ///                regains ownership of msg.
 214   /// \param[in] options The WriteOptions to use for writing this message
 215   void StartWrite(const Request* req, WriteOptions options) {
 216     stream_->Write(req, std::move(options));
 217   }
 218
 219   /// Initiate/post a write operation with specified options and an indication
 220   /// that this is the last write (like StartWrite and StartWritesDone, merged).
 221   /// Note that calling this means that no more calls to StartWrite,
 222   /// StartWriteLast, or StartWritesDone are allowed.
 223   ///
 224   /// \param[in] req The message to be written. The library takes temporary
 225   ///                ownership until OnWriteDone, at which point the application
 226   ///                regains ownership of msg.
 227   /// \param[in] options The WriteOptions to use for writing this message
 228   void StartWriteLast(const Request* req, WriteOptions options) {
 229     StartWrite(req, std::move(options.set_last_message()));
 230   }
 231
 232   /// Indicate that the RPC will have no more write operations. This can only be
 233   /// issued once for a given RPC. This is not required or allowed if
 234   /// StartWriteLast is used since that already has the same implication.
 235   /// Note that calling this means that no more calls to StartWrite,
 236   /// StartWriteLast, or StartWritesDone are allowed.
 237   void StartWritesDone() { stream_->WritesDone(); }
 238
 239   /// Holds are needed if (and only if) this stream has operations that take
 240   /// place on it after StartCall but from outside one of the reactions
 241   /// (OnReadDone, etc). This is _not_ a common use of the streaming API.
 242   ///
 243   /// Holds must be added before calling StartCall. If a stream still has a hold
 244   /// in place, its resources will not be destroyed even if the status has
 245   /// already come in from the wire and there are currently no active callbacks
 246   /// outstanding. Similarly, the stream will not call OnDone if there are still
 247   /// holds on it.
 248   ///
 249   /// For example, if a StartRead or StartWrite operation is going to be
 250   /// initiated from elsewhere in the application, the application should call
 251   /// AddHold or AddMultipleHolds before StartCall.  If there is going to be,
 252   /// for example, a read-flow and a write-flow taking place outside the
 253   /// reactions, then call AddMultipleHolds(2) before StartCall. When the
 254   /// application knows that it won't issue any more read operations (such as
 255   /// when a read comes back as not ok), it should issue a RemoveHold(). It
 256   /// should also call RemoveHold() again after it does StartWriteLast or
 257   /// StartWritesDone that indicates that there will be no more write ops.
 258   /// The number of RemoveHold calls must match the total number of AddHold
 259   /// calls plus the number of holds added by AddMultipleHolds.
 260   void AddHold() { AddMultipleHolds(1); }
 261   void AddMultipleHolds(int holds) { stream_->AddHold(holds); }
 262   void RemoveHold() { stream_->RemoveHold(); }
 263
 264   /// Notifies the application that all operations associated with this RPC
 265   /// have completed and provides the RPC status outcome.
 266   ///
 267   /// \param[in] s The status outcome of this RPC
 268   virtual void OnDone(const Status& s) {}
 269
 270   /// Notifies the application that a read of initial metadata from the
 271   /// server is done. If the application chooses not to implement this method,
 272   /// it can assume that the initial metadata has been read before the first
 273   /// call of OnReadDone or OnDone.
 274   ///
 275   /// \param[in] ok Was the initial metadata read successfully? If false, no
 276   ///               further read-side operation will succeed.
 277   virtual void OnReadInitialMetadataDone(bool ok) {}
 278
 279   /// Notifies the application that a StartRead operation completed.
 280   ///
 281   /// \param[in] ok Was it successful? If false, no further read-side operation
 282   ///               will succeed.
 283   virtual void OnReadDone(bool ok) {}
 284
 285   /// Notifies the application that a StartWrite operation completed.
 286   ///
 287   /// \param[in] ok Was it successful? If false, no further write-side operation
 288   ///               will succeed.
 289   virtual void OnWriteDone(bool ok) {}
 290
 291   /// Notifies the application that a StartWritesDone operation completed. Note
 292   /// that this is only used on explicit StartWritesDone operations and not for
 293   /// those that are implicitly invoked as part of a StartWriteLast.
 294   ///
 295   /// \param[in] ok Was it successful? If false, the application will later see
 296   ///               the failure reflected as a bad status in OnDone.
 297   virtual void OnWritesDoneDone(bool ok) {}
 298
 299  private:
 300   friend class ClientCallbackReaderWriter<Request, Response>;
 301   void BindStream(ClientCallbackReaderWriter<Request, Response>* stream) {
 302     stream_ = stream;
 303   }
 304   ClientCallbackReaderWriter<Request, Response>* stream_;
 305 };
 306
 307 /// \a ClientReadReactor is the interface for a server-streaming RPC.
 308 /// All public methods behave as in ClientBidiReactor.
 309 template <class Response>
 310 class ClientReadReactor {
 311  public:
 312   virtual ~ClientReadReactor() {}
 313
 314   void StartCall() { reader_->StartCall(); }
 315   void StartRead(Response* resp) { reader_->Read(resp); }
 316
 317   void AddHold() { AddMultipleHolds(1); }
 318   void AddMultipleHolds(int holds) { reader_->AddHold(holds); }
 319   void RemoveHold() { reader_->RemoveHold(); }
 320
 321   virtual void OnDone(const Status& s) {}
 322   virtual void OnReadInitialMetadataDone(bool ok) {}
 323   virtual void OnReadDone(bool ok) {}
 324
 325  private:
 326   friend class ClientCallbackReader<Response>;
 327   void BindReader(ClientCallbackReader<Response>* reader) { reader_ = reader; }
 328   ClientCallbackReader<Response>* reader_;
 329 };
 330
 331 /// \a ClientWriteReactor is the interface for a client-streaming RPC.
 332 /// All public methods behave as in ClientBidiReactor.
 333 template <class Request>
 334 class ClientWriteReactor {
 335  public:
 336   virtual ~ClientWriteReactor() {}
 337
 338   void StartCall() { writer_->StartCall(); }
 339   void StartWrite(const Request* req) { StartWrite(req, WriteOptions()); }
 340   void StartWrite(const Request* req, WriteOptions options) {
 341     writer_->Write(req, std::move(options));
 342   }
 343   void StartWriteLast(const Request* req, WriteOptions options) {
 344     StartWrite(req, std::move(options.set_last_message()));
 345   }
 346   void StartWritesDone() { writer_->WritesDone(); }
 347
 348   void AddHold() { AddMultipleHolds(1); }
 349   void AddMultipleHolds(int holds) { writer_->AddHold(holds); }
 350   void RemoveHold() { writer_->RemoveHold(); }
 351
 352   virtual void OnDone(const Status& s) {}
 353   virtual void OnReadInitialMetadataDone(bool ok) {}
 354   virtual void OnWriteDone(bool ok) {}
 355   virtual void OnWritesDoneDone(bool ok) {}
 356
 357  private:
 358   friend class ClientCallbackWriter<Request>;
 359   void BindWriter(ClientCallbackWriter<Request>* writer) { writer_ = writer; }
 360   ClientCallbackWriter<Request>* writer_;
 361 };
 362
 363 /// \a ClientUnaryReactor is a reactor-style interface for a unary RPC.
 364 /// This is _not_ a common way of invoking a unary RPC. In practice, this
 365 /// option should be used only if the unary RPC wants to receive initial
 366 /// metadata without waiting for the response to complete. Most deployments of
 367 /// RPC systems do not use this option, but it is needed for generality.
 368 /// All public methods behave as in ClientBidiReactor.
 369 /// StartCall is included for consistency with the other reactor flavors: even
 370 /// though there are no StartRead or StartWrite operations to queue before the
 371 /// call (that is part of the unary call itself) and there is no reactor object
 372 /// being created as a result of this call, we keep a consistent 2-phase
 373 /// initiation API among all the reactor flavors.
 374 class ClientUnaryReactor {
 375  public:
 376   virtual ~ClientUnaryReactor() {}
 377
 378   void StartCall() { call_->StartCall(); }
 379   virtual void OnDone(const Status& s) {}
 380   virtual void OnReadInitialMetadataDone(bool ok) {}
 381
 382  private:
 383   friend class ClientCallbackUnary;
 384   void BindCall(ClientCallbackUnary* call) { call_ = call; }
 385   ClientCallbackUnary* call_;
 386 };
 387
 388 // Define function out-of-line from class to avoid forward declaration issue
 389 inline void ClientCallbackUnary::BindReactor(ClientUnaryReactor* reactor) {
 390   reactor->BindCall(this);
 391 }
 392
 393 }  // namespace experimental
 394
 395 namespace internal {
 396
 397 // Forward declare factory classes for friendship
 398 template <class Request, class Response>
 399 class ClientCallbackReaderWriterFactory;
 400 template <class Response>
 401 class ClientCallbackReaderFactory;
 402 template <class Request>
 403 class ClientCallbackWriterFactory;
 404
 405 template <class Request, class Response>
 406 class ClientCallbackReaderWriterImpl
 407     : public ::grpc::experimental::ClientCallbackReaderWriter<Request,
 408                                                               Response> {
 409  public:
 410   // always allocated against a call arena, no memory free required
 411   static void operator delete(void* ptr, std::size_t size) {
 412     assert(size == sizeof(ClientCallbackReaderWriterImpl));
 413   }
 414
 415   // This operator should never be called as the memory should be freed as part
 416   // of the arena destruction. It only exists to provide a matching operator
 417   // delete to the operator new so that some compilers will not complain (see
 418   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
 419   // there are no tests catching the compiler warning.
 420   static void operator delete(void*, void*) { assert(0); }
 421
 422   void MaybeFinish() {
 423     if (--callbacks_outstanding_ == 0) {
 424       Status s = std::move(finish_status_);
 425       auto* reactor = reactor_;
 426       auto* call = call_.call();
 427       this->~ClientCallbackReaderWriterImpl();
 428       g_core_codegen_interface->grpc_call_unref(call);
 429       reactor->OnDone(s);
 430     }
 431   }
 432
 433   void StartCall() override {
 434     // This call initiates two batches, plus any backlog, each with a callback
 435     // 1. Send initial metadata (unless corked) + recv initial metadata
 436     // 2. Any read backlog
 437     // 3. Any write backlog
 438     // 4. Recv trailing metadata, on_completion callback
 439     started_ = true;
 440
 441     start_tag_.Set(call_.call(),
 442                    [this](bool ok) {
 443                      reactor_->OnReadInitialMetadataDone(ok);
 444                      MaybeFinish();
 445                    },
 446                    &start_ops_);
 447     if (!start_corked_) {
 448       start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 449                                      context_->initial_metadata_flags());
 450     }
 451     start_ops_.RecvInitialMetadata(context_);
 452     start_ops_.set_core_cq_tag(&start_tag_);
 453     call_.PerformOps(&start_ops_);
 454
 455     // Also set up the read and write tags so that they don't have to be set up
 456     // each time
 457     write_tag_.Set(call_.call(),
 458                    [this](bool ok) {
 459                      reactor_->OnWriteDone(ok);
 460                      MaybeFinish();
 461                    },
 462                    &write_ops_);
 463     write_ops_.set_core_cq_tag(&write_tag_);
 464
 465     read_tag_.Set(call_.call(),
 466                   [this](bool ok) {
 467                     reactor_->OnReadDone(ok);
 468                     MaybeFinish();
 469                   },
 470                   &read_ops_);
 471     read_ops_.set_core_cq_tag(&read_tag_);
 472     if (read_ops_at_start_) {
 473       call_.PerformOps(&read_ops_);
 474     }
 475
 476     if (write_ops_at_start_) {
 477       call_.PerformOps(&write_ops_);
 478     }
 479
 480     if (writes_done_ops_at_start_) {
 481       call_.PerformOps(&writes_done_ops_);
 482     }
 483
 484     finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); },
 485                     &finish_ops_);
 486     finish_ops_.ClientRecvStatus(context_, &finish_status_);
 487     finish_ops_.set_core_cq_tag(&finish_tag_);
 488     call_.PerformOps(&finish_ops_);
 489   }
 490
 491   void Read(Response* msg) override {
 492     read_ops_.RecvMessage(msg);
 493     callbacks_outstanding_++;
 494     if (started_) {
 495       call_.PerformOps(&read_ops_);
 496     } else {
 497       read_ops_at_start_ = true;
 498     }
 499   }
 500
 501   void Write(const Request* msg, WriteOptions options) override {
 502     if (start_corked_) {
 503       write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 504                                      context_->initial_metadata_flags());
 505       start_corked_ = false;
 506     }
 507
 508     if (options.is_last_message()) {
 509       options.set_buffer_hint();
 510       write_ops_.ClientSendClose();
 511     }
 512     // TODO(vjpai): don't assert
 513     GPR_CODEGEN_ASSERT(write_ops_.SendMessagePtr(msg, options).ok());
 514     callbacks_outstanding_++;
 515     if (started_) {
 516       call_.PerformOps(&write_ops_);
 517     } else {
 518       write_ops_at_start_ = true;
 519     }
 520   }
 521   void WritesDone() override {
 522     if (start_corked_) {
 523       writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 524                                            context_->initial_metadata_flags());
 525       start_corked_ = false;
 526     }
 527     writes_done_ops_.ClientSendClose();
 528     writes_done_tag_.Set(call_.call(),
 529                          [this](bool ok) {
 530                            reactor_->OnWritesDoneDone(ok);
 531                            MaybeFinish();
 532                          },
 533                          &writes_done_ops_);
 534     writes_done_ops_.set_core_cq_tag(&writes_done_tag_);
 535     callbacks_outstanding_++;
 536     if (started_) {
 537       call_.PerformOps(&writes_done_ops_);
 538     } else {
 539       writes_done_ops_at_start_ = true;
 540     }
 541   }
 542
 543   virtual void AddHold(int holds) override { callbacks_outstanding_ += holds; }
 544   virtual void RemoveHold() override { MaybeFinish(); }
 545
 546  private:
 547   friend class ClientCallbackReaderWriterFactory<Request, Response>;
 548
 549   ClientCallbackReaderWriterImpl(
 550       Call call, ClientContext* context,
 551       ::grpc::experimental::ClientBidiReactor<Request, Response>* reactor)
 552       : context_(context),
 553         call_(call),
 554         reactor_(reactor),
 555         start_corked_(context_->initial_metadata_corked_) {
 556     this->BindReactor(reactor);
 557   }
 558
 559   ClientContext* const context_;
 560   Call call_;
 561   ::grpc::experimental::ClientBidiReactor<Request, Response>* const reactor_;
 562
 563   CallOpSet<CallOpSendInitialMetadata, CallOpRecvInitialMetadata> start_ops_;
 564   CallbackWithSuccessTag start_tag_;
 565   bool start_corked_;
 566
 567   CallOpSet<CallOpClientRecvStatus> finish_ops_;
 568   CallbackWithSuccessTag finish_tag_;
 569   Status finish_status_;
 570
 571   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
 572       write_ops_;
 573   CallbackWithSuccessTag write_tag_;
 574   bool write_ops_at_start_{false};
 575
 576   CallOpSet<CallOpSendInitialMetadata, CallOpClientSendClose> writes_done_ops_;
 577   CallbackWithSuccessTag writes_done_tag_;
 578   bool writes_done_ops_at_start_{false};
 579
 580   CallOpSet<CallOpRecvMessage<Response>> read_ops_;
 581   CallbackWithSuccessTag read_tag_;
 582   bool read_ops_at_start_{false};
 583
 584   // Minimum of 2 callbacks to pre-register for start and finish
 585   std::atomic_int callbacks_outstanding_{2};
 586   bool started_{false};
 587 };
 588
 589 template <class Request, class Response>
 590 class ClientCallbackReaderWriterFactory {
 591  public:
 592   static void Create(
 593       ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
 594       ClientContext* context,
 595       ::grpc::experimental::ClientBidiReactor<Request, Response>* reactor) {
 596     Call call = channel->CreateCall(method, context, channel->CallbackCQ());
 597
 598     g_core_codegen_interface->grpc_call_ref(call.call());
 599     new (g_core_codegen_interface->grpc_call_arena_alloc(
 600         call.call(), sizeof(ClientCallbackReaderWriterImpl<Request, Response>)))
 601         ClientCallbackReaderWriterImpl<Request, Response>(call, context,
 602                                                           reactor);
 603   }
 604 };
 605
 606 template <class Response>
 607 class ClientCallbackReaderImpl
 608     : public ::grpc::experimental::ClientCallbackReader<Response> {
 609  public:
 610   // always allocated against a call arena, no memory free required
 611   static void operator delete(void* ptr, std::size_t size) {
 612     assert(size == sizeof(ClientCallbackReaderImpl));
 613   }
 614
 615   // This operator should never be called as the memory should be freed as part
 616   // of the arena destruction. It only exists to provide a matching operator
 617   // delete to the operator new so that some compilers will not complain (see
 618   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
 619   // there are no tests catching the compiler warning.
 620   static void operator delete(void*, void*) { assert(0); }
 621
 622   void MaybeFinish() {
 623     if (--callbacks_outstanding_ == 0) {
 624       Status s = std::move(finish_status_);
 625       auto* reactor = reactor_;
 626       auto* call = call_.call();
 627       this->~ClientCallbackReaderImpl();
 628       g_core_codegen_interface->grpc_call_unref(call);
 629       reactor->OnDone(s);
 630     }
 631   }
 632
 633   void StartCall() override {
 634     // This call initiates two batches, plus any backlog, each with a callback
 635     // 1. Send initial metadata (unless corked) + recv initial metadata
 636     // 2. Any backlog
 637     // 3. Recv trailing metadata, on_completion callback
 638     started_ = true;
 639
 640     start_tag_.Set(call_.call(),
 641                    [this](bool ok) {
 642                      reactor_->OnReadInitialMetadataDone(ok);
 643                      MaybeFinish();
 644                    },
 645                    &start_ops_);
 646     start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 647                                    context_->initial_metadata_flags());
 648     start_ops_.RecvInitialMetadata(context_);
 649     start_ops_.set_core_cq_tag(&start_tag_);
 650     call_.PerformOps(&start_ops_);
 651
 652     // Also set up the read tag so it doesn't have to be set up each time
 653     read_tag_.Set(call_.call(),
 654                   [this](bool ok) {
 655                     reactor_->OnReadDone(ok);
 656                     MaybeFinish();
 657                   },
 658                   &read_ops_);
 659     read_ops_.set_core_cq_tag(&read_tag_);
 660     if (read_ops_at_start_) {
 661       call_.PerformOps(&read_ops_);
 662     }
 663
 664     finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); },
 665                     &finish_ops_);
 666     finish_ops_.ClientRecvStatus(context_, &finish_status_);
 667     finish_ops_.set_core_cq_tag(&finish_tag_);
 668     call_.PerformOps(&finish_ops_);
 669   }
 670
 671   void Read(Response* msg) override {
 672     read_ops_.RecvMessage(msg);
 673     callbacks_outstanding_++;
 674     if (started_) {
 675       call_.PerformOps(&read_ops_);
 676     } else {
 677       read_ops_at_start_ = true;
 678     }
 679   }
 680
 681   virtual void AddHold(int holds) override { callbacks_outstanding_ += holds; }
 682   virtual void RemoveHold() override { MaybeFinish(); }
 683
 684  private:
 685   friend class ClientCallbackReaderFactory<Response>;
 686
 687   template <class Request>
 688   ClientCallbackReaderImpl(
 689       Call call, ClientContext* context, Request* request,
 690       ::grpc::experimental::ClientReadReactor<Response>* reactor)
 691       : context_(context), call_(call), reactor_(reactor) {
 692     this->BindReactor(reactor);
 693     // TODO(vjpai): don't assert
 694     GPR_CODEGEN_ASSERT(start_ops_.SendMessagePtr(request).ok());
 695     start_ops_.ClientSendClose();
 696   }
 697
 698   ClientContext* const context_;
 699   Call call_;
 700   ::grpc::experimental::ClientReadReactor<Response>* const reactor_;
 701
 702   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose,
 703             CallOpRecvInitialMetadata>
 704       start_ops_;
 705   CallbackWithSuccessTag start_tag_;
 706
 707   CallOpSet<CallOpClientRecvStatus> finish_ops_;
 708   CallbackWithSuccessTag finish_tag_;
 709   Status finish_status_;
 710
 711   CallOpSet<CallOpRecvMessage<Response>> read_ops_;
 712   CallbackWithSuccessTag read_tag_;
 713   bool read_ops_at_start_{false};
 714
 715   // Minimum of 2 callbacks to pre-register for start and finish
 716   std::atomic_int callbacks_outstanding_{2};
 717   bool started_{false};
 718 };
 719
 720 template <class Response>
 721 class ClientCallbackReaderFactory {
 722  public:
 723   template <class Request>
 724   static void Create(
 725       ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
 726       ClientContext* context, const Request* request,
 727       ::grpc::experimental::ClientReadReactor<Response>* reactor) {
 728     Call call = channel->CreateCall(method, context, channel->CallbackCQ());
 729
 730     g_core_codegen_interface->grpc_call_ref(call.call());
 731     new (g_core_codegen_interface->grpc_call_arena_alloc(
 732         call.call(), sizeof(ClientCallbackReaderImpl<Response>)))
 733         ClientCallbackReaderImpl<Response>(call, context, request, reactor);
 734   }
 735 };
 736
 737 template <class Request>
 738 class ClientCallbackWriterImpl
 739     : public ::grpc::experimental::ClientCallbackWriter<Request> {
 740  public:
 741   // always allocated against a call arena, no memory free required
 742   static void operator delete(void* ptr, std::size_t size) {
 743     assert(size == sizeof(ClientCallbackWriterImpl));
 744   }
 745
 746   // This operator should never be called as the memory should be freed as part
 747   // of the arena destruction. It only exists to provide a matching operator
 748   // delete to the operator new so that some compilers will not complain (see
 749   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
 750   // there are no tests catching the compiler warning.
 751   static void operator delete(void*, void*) { assert(0); }
 752
 753   void MaybeFinish() {
 754     if (--callbacks_outstanding_ == 0) {
 755       Status s = std::move(finish_status_);
 756       auto* reactor = reactor_;
 757       auto* call = call_.call();
 758       this->~ClientCallbackWriterImpl();
 759       g_core_codegen_interface->grpc_call_unref(call);
 760       reactor->OnDone(s);
 761     }
 762   }
 763
 764   void StartCall() override {
 765     // This call initiates two batches, plus any backlog, each with a callback
 766     // 1. Send initial metadata (unless corked) + recv initial metadata
 767     // 2. Any backlog
 768     // 3. Recv trailing metadata, on_completion callback
 769     started_ = true;
 770
 771     start_tag_.Set(call_.call(),
 772                    [this](bool ok) {
 773                      reactor_->OnReadInitialMetadataDone(ok);
 774                      MaybeFinish();
 775                    },
 776                    &start_ops_);
 777     if (!start_corked_) {
 778       start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 779                                      context_->initial_metadata_flags());
 780     }
 781     start_ops_.RecvInitialMetadata(context_);
 782     start_ops_.set_core_cq_tag(&start_tag_);
 783     call_.PerformOps(&start_ops_);
 784
 785     // Also set up the read and write tags so that they don't have to be set up
 786     // each time
 787     write_tag_.Set(call_.call(),
 788                    [this](bool ok) {
 789                      reactor_->OnWriteDone(ok);
 790                      MaybeFinish();
 791                    },
 792                    &write_ops_);
 793     write_ops_.set_core_cq_tag(&write_tag_);
 794
 795     if (write_ops_at_start_) {
 796       call_.PerformOps(&write_ops_);
 797     }
 798
 799     if (writes_done_ops_at_start_) {
 800       call_.PerformOps(&writes_done_ops_);
 801     }
 802
 803     finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); },
 804                     &finish_ops_);
 805     finish_ops_.ClientRecvStatus(context_, &finish_status_);
 806     finish_ops_.set_core_cq_tag(&finish_tag_);
 807     call_.PerformOps(&finish_ops_);
 808   }
 809
 810   void Write(const Request* msg, WriteOptions options) override {
 811     if (start_corked_) {
 812       write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 813                                      context_->initial_metadata_flags());
 814       start_corked_ = false;
 815     }
 816
 817     if (options.is_last_message()) {
 818       options.set_buffer_hint();
 819       write_ops_.ClientSendClose();
 820     }
 821     // TODO(vjpai): don't assert
 822     GPR_CODEGEN_ASSERT(write_ops_.SendMessagePtr(msg, options).ok());
 823     callbacks_outstanding_++;
 824     if (started_) {
 825       call_.PerformOps(&write_ops_);
 826     } else {
 827       write_ops_at_start_ = true;
 828     }
 829   }
 830   void WritesDone() override {
 831     if (start_corked_) {
 832       writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 833                                            context_->initial_metadata_flags());
 834       start_corked_ = false;
 835     }
 836     writes_done_ops_.ClientSendClose();
 837     writes_done_tag_.Set(call_.call(),
 838                          [this](bool ok) {
 839                            reactor_->OnWritesDoneDone(ok);
 840                            MaybeFinish();
 841                          },
 842                          &writes_done_ops_);
 843     writes_done_ops_.set_core_cq_tag(&writes_done_tag_);
 844     callbacks_outstanding_++;
 845     if (started_) {
 846       call_.PerformOps(&writes_done_ops_);
 847     } else {
 848       writes_done_ops_at_start_ = true;
 849     }
 850   }
 851
 852   virtual void AddHold(int holds) override { callbacks_outstanding_ += holds; }
 853   virtual void RemoveHold() override { MaybeFinish(); }
 854
 855  private:
 856   friend class ClientCallbackWriterFactory<Request>;
 857
 858   template <class Response>
 859   ClientCallbackWriterImpl(
 860       Call call, ClientContext* context, Response* response,
 861       ::grpc::experimental::ClientWriteReactor<Request>* reactor)
 862       : context_(context),
 863         call_(call),
 864         reactor_(reactor),
 865         start_corked_(context_->initial_metadata_corked_) {
 866     this->BindReactor(reactor);
 867     finish_ops_.RecvMessage(response);
 868     finish_ops_.AllowNoMessage();
 869   }
 870
 871   ClientContext* const context_;
 872   Call call_;
 873   ::grpc::experimental::ClientWriteReactor<Request>* const reactor_;
 874
 875   CallOpSet<CallOpSendInitialMetadata, CallOpRecvInitialMetadata> start_ops_;
 876   CallbackWithSuccessTag start_tag_;
 877   bool start_corked_;
 878
 879   CallOpSet<CallOpGenericRecvMessage, CallOpClientRecvStatus> finish_ops_;
 880   CallbackWithSuccessTag finish_tag_;
 881   Status finish_status_;
 882
 883   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
 884       write_ops_;
 885   CallbackWithSuccessTag write_tag_;
 886   bool write_ops_at_start_{false};
 887
 888   CallOpSet<CallOpSendInitialMetadata, CallOpClientSendClose> writes_done_ops_;
 889   CallbackWithSuccessTag writes_done_tag_;
 890   bool writes_done_ops_at_start_{false};
 891
 892   // Minimum of 2 callbacks to pre-register for start and finish
 893   std::atomic_int callbacks_outstanding_{2};
 894   bool started_{false};
 895 };
 896
 897 template <class Request>
 898 class ClientCallbackWriterFactory {
 899  public:
 900   template <class Response>
 901   static void Create(
 902       ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
 903       ClientContext* context, Response* response,
 904       ::grpc::experimental::ClientWriteReactor<Request>* reactor) {
 905     Call call = channel->CreateCall(method, context, channel->CallbackCQ());
 906
 907     g_core_codegen_interface->grpc_call_ref(call.call());
 908     new (g_core_codegen_interface->grpc_call_arena_alloc(
 909         call.call(), sizeof(ClientCallbackWriterImpl<Request>)))
 910         ClientCallbackWriterImpl<Request>(call, context, response, reactor);
 911   }
 912 };
 913
 914 class ClientCallbackUnaryImpl final
 915     : public ::grpc::experimental::ClientCallbackUnary {
 916  public:
 917   // always allocated against a call arena, no memory free required
 918   static void operator delete(void* ptr, std::size_t size) {
 919     assert(size == sizeof(ClientCallbackUnaryImpl));
 920   }
 921
 922   // This operator should never be called as the memory should be freed as part
 923   // of the arena destruction. It only exists to provide a matching operator
 924   // delete to the operator new so that some compilers will not complain (see
 925   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
 926   // there are no tests catching the compiler warning.
 927   static void operator delete(void*, void*) { assert(0); }
 928
 929   void StartCall() override {
 930     // This call initiates two batches, each with a callback
 931     // 1. Send initial metadata + write + writes done + recv initial metadata
 932     // 2. Read message, recv trailing metadata
 933     started_ = true;
 934
 935     start_tag_.Set(call_.call(),
 936                    [this](bool ok) {
 937                      reactor_->OnReadInitialMetadataDone(ok);
 938                      MaybeFinish();
 939                    },
 940                    &start_ops_);
 941     start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 942                                    context_->initial_metadata_flags());
 943     start_ops_.RecvInitialMetadata(context_);
 944     start_ops_.set_core_cq_tag(&start_tag_);
 945     call_.PerformOps(&start_ops_);
 946
 947     finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); },
 948                     &finish_ops_);
 949     finish_ops_.ClientRecvStatus(context_, &finish_status_);
 950     finish_ops_.set_core_cq_tag(&finish_tag_);
 951     call_.PerformOps(&finish_ops_);
 952   }
 953
 954   void MaybeFinish() {
 955     if (--callbacks_outstanding_ == 0) {
 956       Status s = std::move(finish_status_);
 957       auto* reactor = reactor_;
 958       auto* call = call_.call();
 959       this->~ClientCallbackUnaryImpl();
 960       g_core_codegen_interface->grpc_call_unref(call);
 961       reactor->OnDone(s);
 962     }
 963   }
 964
 965  private:
 966   friend class ClientCallbackUnaryFactory;
 967
 968   template <class Request, class Response>
 969   ClientCallbackUnaryImpl(Call call, ClientContext* context, Request* request,
 970                           Response* response,
 971                           ::grpc::experimental::ClientUnaryReactor* reactor)
 972       : context_(context), call_(call), reactor_(reactor) {
 973     this->BindReactor(reactor);
 974     // TODO(vjpai): don't assert
 975     GPR_CODEGEN_ASSERT(start_ops_.SendMessagePtr(request).ok());
 976     start_ops_.ClientSendClose();
 977     finish_ops_.RecvMessage(response);
 978     finish_ops_.AllowNoMessage();
 979   }
 980
 981   ClientContext* const context_;
 982   Call call_;
 983   ::grpc::experimental::ClientUnaryReactor* const reactor_;
 984
 985   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose,
 986             CallOpRecvInitialMetadata>
 987       start_ops_;
 988   CallbackWithSuccessTag start_tag_;
 989
 990   CallOpSet<CallOpGenericRecvMessage, CallOpClientRecvStatus> finish_ops_;
 991   CallbackWithSuccessTag finish_tag_;
 992   Status finish_status_;
 993
 994   // This call will have 2 callbacks: start and finish
 995   std::atomic_int callbacks_outstanding_{2};
 996   bool started_{false};
 997 };
 998
 999 class ClientCallbackUnaryFactory {
1000  public:
1001   template <class Request, class Response>
1002   static void Create(ChannelInterface* channel,
1003                      const ::grpc::internal::RpcMethod& method,
1004                      ClientContext* context, const Request* request,
1005                      Response* response,
1006                      ::grpc::experimental::ClientUnaryReactor* reactor) {
1007     Call call = channel->CreateCall(method, context, channel->CallbackCQ());
1008
1009     g_core_codegen_interface->grpc_call_ref(call.call());
1010
1011     new (g_core_codegen_interface->grpc_call_arena_alloc(
1012         call.call(), sizeof(ClientCallbackUnaryImpl)))
1013         ClientCallbackUnaryImpl(call, context, request, response, reactor);
1014   }
1015 };
1016
1017 }  // namespace internal
1018 }  // namespace grpc
1019
1020 #endif  // GRPCPP_IMPL_CODEGEN_CLIENT_CALLBACK_H