Call ref/unref, bugfixes

2 files changed, 33 insertions, 92 deletions

diff --git a/include/grpc++/impl/codegen/async_unary_call.h b/include/grpc++/impl/codegen/async_unary_call.h
index 50bb399df6..dd65cf89d6 100644
--- a/include/grpc++/impl/codegen/async_unary_call.h
+++ b/include/grpc++/impl/codegen/async_unary_call.h
@@ -61,11 +61,6 @@ template <class R>
 class ClientAsyncResponseReader final
     : public ClientAsyncResponseReaderInterface<R> {
  public:
-  ~ClientAsyncResponseReader() {
-    if (collection_ != nullptr && collection_->Unref()) {
-      delete collection_;
-    }
-  }
   template <class W>
   static ClientAsyncResponseReader* Create(ChannelInterface* channel,
                                            CompletionQueue* cq,
@@ -77,18 +72,13 @@ class ClientAsyncResponseReader final
         grpc_call_arena_alloc(call.call(), sizeof(*reader)));
     new (&reader->call_) Call(std::move(call));
     reader->context_ = context;
-    reader->collection_ =
-        new (grpc_call_arena_alloc(call.call(), sizeof(CallOpSetCollection)))
-            CallOpSetCollection();
 
-    reader->collection_->init_buf_.SetCollection(reader->collection_);
-    reader->collection_->init_buf_.SendInitialMetadata(
-        context->send_initial_metadata_, context->initial_metadata_flags());
+    reader->init_buf_.SendInitialMetadata(context->send_initial_metadata_,
+                                          context->initial_metadata_flags());
     // TODO(ctiller): don't assert
-    GPR_CODEGEN_ASSERT(
-        reader->collection_->init_buf_.SendMessage(request).ok());
-    reader->collection_->init_buf_.ClientSendClose();
-    reader->call_.PerformOps(&reader->collection_->init_buf_);
+    GPR_CODEGEN_ASSERT(reader->init_buf_.SendMessage(request).ok());
+    reader->init_buf_.ClientSendClose();
+    reader->call_.PerformOps(&reader->init_buf_);
     return reader;
   }
 
@@ -100,22 +90,20 @@ class ClientAsyncResponseReader final
   void ReadInitialMetadata(void* tag) {
     GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
 
-    collection_->meta_buf_.SetCollection(collection_);
-    collection_->meta_buf_.set_output_tag(tag);
-    collection_->meta_buf_.RecvInitialMetadata(context_);
-    call_.PerformOps(&collection_->meta_buf_);
+    meta_buf_.set_output_tag(tag);
+    meta_buf_.RecvInitialMetadata(context_);
+    call_.PerformOps(&meta_buf_);
   }
 
   void Finish(R* msg, Status* status, void* tag) {
-    collection_->finish_buf_.SetCollection(collection_);
-    collection_->finish_buf_.set_output_tag(tag);
+    finish_buf_.set_output_tag(tag);
     if (!context_->initial_metadata_received_) {
-      collection_->finish_buf_.RecvInitialMetadata(context_);
+      finish_buf_.RecvInitialMetadata(context_);
     }
-    collection_->finish_buf_.RecvMessage(msg);
-    collection_->finish_buf_.AllowNoMessage();
-    collection_->finish_buf_.ClientRecvStatus(context_, status);
-    call_.PerformOps(&collection_->finish_buf_);
+    finish_buf_.RecvMessage(msg);
+    finish_buf_.AllowNoMessage();
+    finish_buf_.ClientRecvStatus(context_, status);
+    call_.PerformOps(&finish_buf_);
   }
 
  private:
@@ -125,26 +113,13 @@ class ClientAsyncResponseReader final
   // disable operator new
   static void* operator new(std::size_t size);
 
-  class CallOpSetCollection final : public CallOpSetCollectionInterface {
-   public:
-    SneakyCallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
-                    CallOpClientSendClose>
-        init_buf_;
-    CallOpSet<CallOpRecvInitialMetadata> meta_buf_;
-    CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>,
-              CallOpClientRecvStatus>
-        finish_buf_;
-
-    static void* operator new(std::size_t size, void* p) { return p; }
-    static void operator delete(void* ptr, std::size_t size) {
-      assert(size == sizeof(CallOpSetCollection));
-    }
-
-   private:
-    // disable operator new
-    static void* operator new(std::size_t size);
-  };
-  CallOpSetCollection* collection_;
+  SneakyCallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
+                  CallOpClientSendClose>
+      init_buf_;
+  CallOpSet<CallOpRecvInitialMetadata> meta_buf_;
+  CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>,
+            CallOpClientRecvStatus>
+      finish_buf_;
 };
 
 template <class W>
@@ -214,4 +189,12 @@ class ServerAsyncResponseWriter final : public ServerAsyncStreamingInterface {
 
 }  // namespace grpc
 
+namespace std {
+template <class R>
+class default_delete<grpc::ClientAsyncResponseReader<R>> {
+ public:
+  void operator()(void* p) {}
+};
+}
+
 #endif  // GRPCXX_IMPL_CODEGEN_ASYNC_UNARY_CALL_H
diff --git a/include/grpc++/impl/codegen/call.h b/include/grpc++/impl/codegen/call.h
index 9c8611a116..4a52c2cbcf 100644
--- a/include/grpc++/impl/codegen/call.h
+++ b/include/grpc++/impl/codegen/call.h
@@ -561,32 +561,6 @@ class CallOpClientRecvStatus {
   grpc_slice status_details_;
 };
 
-/// An abstract collection of CallOpSet's, to be used whenever
-/// CallOpSet objects must be thought of as a group. Each member
-/// of the group should reference the collection, as will the object
-/// that instantiates the collection, allowing for ref-counting.
-/// Any actual use should derive from this base class. This is specifically
-/// necessary if some of the CallOpSet's in the collection are "Sneaky" and
-/// don't report back to the C++ layer CQ operations
-class CallOpSetCollectionInterface {
- public:
-  CallOpSetCollectionInterface() {
-    gpr_atm_rel_store(&refs_, static_cast<gpr_atm>(1));
-  }
-  // always allocated against a call arena, no memory free required
-  static void operator delete(void* ptr, std::size_t size) {
-  }
-  void Ref() { gpr_atm_no_barrier_fetch_add(&refs_, static_cast<gpr_atm>(1)); }
-  bool Unref() {
-    gpr_atm old =
-        gpr_atm_full_fetch_add(&refs_, static_cast<gpr_atm>(-1));
-    return (old == static_cast<gpr_atm>(1));
-  }
-
- private:
-  gpr_atm refs_;
-};
-
 /// An abstract collection of call ops, used to generate the
 /// grpc_call_op structure to pass down to the lower layers,
 /// and as it is-a CompletionQueueTag, also massages the final
@@ -594,26 +568,9 @@ class CallOpSetCollectionInterface {
 /// API.
 class CallOpSetInterface : public CompletionQueueTag {
  public:
-  CallOpSetInterface() : collection_(nullptr) {}
-  ~CallOpSetInterface() { ResetCollection(); }
   /// Fills in grpc_op, starting from ops[*nops] and moving
   /// upwards.
   virtual void FillOps(grpc_op* ops, size_t* nops) = 0;
-
-  /// Mark this as belonging to a collection if needed
-  void SetCollection(CallOpSetCollectionInterface* collection) {
-    collection_ = collection;
-    collection->Ref();
-  }
-  void ResetCollection() {
-    if (collection_ != nullptr && collection_->Unref()) {
-      delete collection_;
-    }
-    collection_ = nullptr;
-  }
-
- protected:
-  CallOpSetCollectionInterface* collection_;
 };
 
 /// Primary implementaiton of CallOpSetInterface.
@@ -634,16 +591,17 @@ class CallOpSet : public CallOpSetInterface,
                   public Op6 {
  public:
   CallOpSet() : return_tag_(this) {}
-  void FillOps(grpc_op* ops, size_t* nops) override {
+  void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) override {
     this->Op1::AddOp(ops, nops);
     this->Op2::AddOp(ops, nops);
     this->Op3::AddOp(ops, nops);
     this->Op4::AddOp(ops, nops);
     this->Op5::AddOp(ops, nops);
     this->Op6::AddOp(ops, nops);
+    grpc_call_ref(call);
   }
 
-  bool FinalizeResult(void** tag, bool* status) override {
+  bool FinalizeResult(grpc_call* call, void** tag, bool* status) override {
     this->Op1::FinishOp(status);
     this->Op2::FinishOp(status);
     this->Op3::FinishOp(status);
@@ -651,7 +609,7 @@ class CallOpSet : public CallOpSetInterface,
     this->Op5::FinishOp(status);
     this->Op6::FinishOp(status);
     *tag = return_tag_;
-    ResetCollection();  // drop the ref at this point
+    grpc_call_unref(call);
     return true;
   }