Progress commit on fixing up C++

2 files changed, 26 insertions, 54 deletions

diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index dbd88c5b8c..c08506c97f 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -48,7 +48,6 @@
 #include <grpc++/time.h>
 
 #include "src/core/profiling/timers.h"
-#include "src/cpp/proto/proto_utils.h"
 
 namespace grpc {
 
@@ -68,10 +67,7 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
         in_flight_(false),
         has_request_payload_(method->method_type() == RpcMethod::NORMAL_RPC ||
                              method->method_type() ==
-                                 RpcMethod::SERVER_STREAMING),
-        has_response_payload_(method->method_type() == RpcMethod::NORMAL_RPC ||
-                              method->method_type() ==
-                                  RpcMethod::CLIENT_STREAMING) {
+                                 RpcMethod::SERVER_STREAMING) {
     grpc_metadata_array_init(&request_metadata_);
   }
 
@@ -116,7 +112,6 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
           ctx_(mrd->deadline_, mrd->request_metadata_.metadata,
                mrd->request_metadata_.count),
           has_request_payload_(mrd->has_request_payload_),
-          has_response_payload_(mrd->has_response_payload_),
           request_payload_(mrd->request_payload_),
           method_(mrd->method_) {
       ctx_.call_ = mrd->call_;
@@ -133,35 +128,9 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
     }
 
     void Run() {
-      std::unique_ptr<grpc::protobuf::Message> req;
-      std::unique_ptr<grpc::protobuf::Message> res;
-      if (has_request_payload_) {
-        GRPC_TIMER_BEGIN(GRPC_PTAG_PROTO_DESERIALIZE, call_.call());
-        req.reset(method_->AllocateRequestProto());
-        if (!DeserializeProto(request_payload_, req.get(),
-                              call_.max_message_size())) {
-          // FIXME(yangg) deal with deserialization failure
-          cq_.Shutdown();
-          return;
-        }
-        GRPC_TIMER_END(GRPC_PTAG_PROTO_DESERIALIZE, call_.call());
-      }
-      if (has_response_payload_) {
-        res.reset(method_->AllocateResponseProto());
-      }
       ctx_.BeginCompletionOp(&call_);
-      auto status = method_->handler()->RunHandler(
-          MethodHandler::HandlerParameter(&call_, &ctx_, req.get(), res.get()));
-      CallOpBuffer buf;
-      if (!ctx_.sent_initial_metadata_) {
-        buf.AddSendInitialMetadata(&ctx_.initial_metadata_);
-      }
-      if (has_response_payload_) {
-        buf.AddSendMessage(*res);
-      }
-      buf.AddServerSendStatus(&ctx_.trailing_metadata_, status);
-      call_.PerformOps(&buf);
-      cq_.Pluck(&buf);  /* status ignored */
+      method_->handler()->RunHandler(
+          MethodHandler::HandlerParameter(&call_, &ctx_, request_payload_, call_.max_message_size()));
       void* ignored_tag;
       bool ignored_ok;
       cq_.Shutdown();
@@ -173,7 +142,6 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
     Call call_;
     ServerContext ctx_;
     const bool has_request_payload_;
-    const bool has_response_payload_;
     grpc_byte_buffer* request_payload_;
     RpcServiceMethod* const method_;
   };
@@ -183,7 +151,6 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
   void* const tag_;
   bool in_flight_;
   const bool has_request_payload_;
-  const bool has_response_payload_;
   grpc_call* call_;
   gpr_timespec deadline_;
   grpc_metadata_array request_metadata_;
@@ -251,9 +218,9 @@ bool Server::RegisterService(RpcService* service) {
 }
 
 bool Server::RegisterAsyncService(AsynchronousService* service) {
-  GPR_ASSERT(service->dispatch_impl_ == nullptr &&
+  GPR_ASSERT(service->server_ == nullptr &&
              "Can only register an asynchronous service against one server.");
-  service->dispatch_impl_ = this;
+  service->server_ = this;
   service->request_args_ = new void*[service->method_count_];
   for (size_t i = 0; i < service->method_count_; ++i) {
     void* tag = grpc_server_register_method(server_, service->method_names_[i],
@@ -318,15 +285,16 @@ void Server::Wait() {
   }
 }
 
-void Server::PerformOpsOnCall(CallOpBuffer* buf, Call* call) {
+void Server::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) {
   static const size_t MAX_OPS = 8;
-  size_t nops = MAX_OPS;
-  grpc_op ops[MAX_OPS];
-  buf->FillOps(ops, &nops);
+  size_t nops = 0;
+  grpc_op cops[MAX_OPS];
+  ops->FillOps(cops, &nops);
   GPR_ASSERT(GRPC_CALL_OK ==
-             grpc_call_start_batch(call->call(), ops, nops, buf));
+             grpc_call_start_batch(call->call(), cops, nops, ops));
 }
 
+#if 0
 class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
  public:
   AsyncRequest(Server* server, void* registered_method, ServerContext* ctx,
@@ -352,9 +320,7 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
         notification_cq->cq(), this);
   }
 
-  AsyncRequest(Server* server, GenericServerContext* ctx,
-               ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
-               ServerCompletionQueue* notification_cq, void* tag)
+  AsyncRequest()
       : tag_(tag),
         request_(nullptr),
         stream_(stream),
@@ -454,6 +420,7 @@ void Server::RequestAsyncGenericCall(GenericServerContext* context,
                                      void* tag) {
   new AsyncRequest(this, context, stream, call_cq, notification_cq, tag);
 }
+#endif
 
 void Server::ScheduleCallback() {
   {
diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc
index 6b5e41d0a8..eea9645e37 100644
--- a/src/cpp/server/server_context.cc
+++ b/src/cpp/server/server_context.cc
@@ -43,12 +43,12 @@ namespace grpc {
 
 // CompletionOp
 
-class ServerContext::CompletionOp GRPC_FINAL : public CallOpBuffer {
+class ServerContext::CompletionOp GRPC_FINAL : public CallOpSetInterface {
  public:
   // initial refs: one in the server context, one in the cq
-  CompletionOp() : refs_(2), finalized_(false), cancelled_(false) {
-    AddServerRecvClose(&cancelled_);
-  }
+  CompletionOp() : refs_(2), finalized_(false), cancelled_(0) {}
+
+  void FillOps(grpc_op* ops, size_t* nops) GRPC_OVERRIDE;
   bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE;
 
   bool CheckCancelled(CompletionQueue* cq);
@@ -59,7 +59,7 @@ class ServerContext::CompletionOp GRPC_FINAL : public CallOpBuffer {
   grpc::mutex mu_;
   int refs_;
   bool finalized_;
-  bool cancelled_;
+  int cancelled_;
 };
 
 void ServerContext::CompletionOp::Unref() {
@@ -73,14 +73,19 @@ void ServerContext::CompletionOp::Unref() {
 bool ServerContext::CompletionOp::CheckCancelled(CompletionQueue* cq) {
   cq->TryPluck(this);
   grpc::lock_guard<grpc::mutex> g(mu_);
-  return finalized_ ? cancelled_ : false;
+  return finalized_ ? cancelled_ != 0 : false;
+}
+
+void ServerContext::CompletionOp::FillOps(grpc_op* ops, size_t* nops) {
+  ops->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  ops->data.recv_close_on_server.cancelled = &cancelled_;
+  *nops = 1;
 }
 
 bool ServerContext::CompletionOp::FinalizeResult(void** tag, bool* status) {
-  GPR_ASSERT(CallOpBuffer::FinalizeResult(tag, status));
   grpc::unique_lock<grpc::mutex> lock(mu_);
   finalized_ = true;
-  if (!*status) cancelled_ = true;
+  if (!*status) cancelled_ = 1;
   if (--refs_ == 0) {
     lock.unlock();
     delete this;