aboutsummaryrefslogtreecommitdiffhomepage
path: root/include/grpc++/impl/codegen
diff options
context:
space:
mode:
Diffstat (limited to 'include/grpc++/impl/codegen')
-rw-r--r--include/grpc++/impl/codegen/async_unary_call.h91
-rw-r--r--include/grpc++/impl/codegen/call.h31
-rw-r--r--include/grpc++/impl/codegen/core_codegen.h3
-rw-r--r--include/grpc++/impl/codegen/core_codegen_interface.h3
4 files changed, 70 insertions, 58 deletions
diff --git a/include/grpc++/impl/codegen/async_unary_call.h b/include/grpc++/impl/codegen/async_unary_call.h
index b77a16b699..b120b37f1f 100644
--- a/include/grpc++/impl/codegen/async_unary_call.h
+++ b/include/grpc++/impl/codegen/async_unary_call.h
@@ -34,6 +34,7 @@
#ifndef GRPCXX_IMPL_CODEGEN_ASYNC_UNARY_CALL_H
#define GRPCXX_IMPL_CODEGEN_ASYNC_UNARY_CALL_H
+#include <assert.h>
#include <grpc++/impl/codegen/call.h>
#include <grpc++/impl/codegen/channel_interface.h>
#include <grpc++/impl/codegen/client_context.h>
@@ -41,6 +42,8 @@
#include <grpc++/impl/codegen/service_type.h>
#include <grpc++/impl/codegen/status.h>
+extern "C" void* grpc_call_arena_alloc(grpc_call* call, size_t size);
+
namespace grpc {
class CompletionQueue;
@@ -59,57 +62,67 @@ class ClientAsyncResponseReader final
: public ClientAsyncResponseReaderInterface<R> {
public:
template <class W>
- ClientAsyncResponseReader(ChannelInterface* channel, CompletionQueue* cq,
- const RpcMethod& method, ClientContext* context,
- const W& request)
- : context_(context),
- call_(channel->CreateCall(method, context, cq)),
- collection_(std::make_shared<CallOpSetCollection>()) {
- collection_->init_buf_.SetCollection(collection_);
- collection_->init_buf_.SendInitialMetadata(
- context->send_initial_metadata_, context->initial_metadata_flags());
+ static ClientAsyncResponseReader* Create(ChannelInterface* channel,
+ CompletionQueue* cq,
+ const RpcMethod& method,
+ ClientContext* context,
+ const W& request) {
+ Call call = channel->CreateCall(method, context, cq);
+ ClientAsyncResponseReader* reader =
+ new (grpc_call_arena_alloc(call.call(), sizeof(*reader)))
+ ClientAsyncResponseReader(call, context);
+
+ reader->init_buf_.SendInitialMetadata(context->send_initial_metadata_,
+ context->initial_metadata_flags());
// TODO(ctiller): don't assert
- GPR_CODEGEN_ASSERT(collection_->init_buf_.SendMessage(request).ok());
- collection_->init_buf_.ClientSendClose();
- call_.PerformOps(&collection_->init_buf_);
+ GPR_CODEGEN_ASSERT(reader->init_buf_.SendMessage(request).ok());
+ reader->init_buf_.ClientSendClose();
+ reader->call_.PerformOps(&reader->init_buf_);
+ return reader;
+ }
+
+ // always allocated against a call arena, no memory free required
+ static void operator delete(void* ptr, std::size_t size) {
+ assert(size == sizeof(ClientAsyncResponseReader));
}
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:
- ClientContext* context_;
+ ClientContext* const context_;
Call call_;
- class CallOpSetCollection : public CallOpSetCollectionInterface {
- public:
- SneakyCallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
- CallOpClientSendClose>
- init_buf_;
- CallOpSet<CallOpRecvInitialMetadata> meta_buf_;
- CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>,
- CallOpClientRecvStatus>
- finish_buf_;
- };
- std::shared_ptr<CallOpSetCollection> collection_;
+ ClientAsyncResponseReader(Call call, ClientContext* context)
+ : context_(context), call_(call) {}
+
+ // disable operator new
+ static void* operator new(std::size_t size);
+ static void* operator new(std::size_t size, void* p) { return p; };
+
+ SneakyCallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
+ CallOpClientSendClose>
+ init_buf_;
+ CallOpSet<CallOpRecvInitialMetadata> meta_buf_;
+ CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>,
+ CallOpClientRecvStatus>
+ finish_buf_;
};
template <class W>
@@ -179,4 +192,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 be6857c482..95cc063281 100644
--- a/include/grpc++/impl/codegen/call.h
+++ b/include/grpc++/impl/codegen/call.h
@@ -34,6 +34,7 @@
#ifndef GRPCXX_IMPL_CODEGEN_CALL_H
#define GRPCXX_IMPL_CODEGEN_CALL_H
+#include <assert.h>
#include <cstring>
#include <functional>
#include <map>
@@ -50,6 +51,7 @@
#include <grpc++/impl/codegen/status_helper.h>
#include <grpc++/impl/codegen/string_ref.h>
+#include <grpc/impl/codegen/atm.h>
#include <grpc/impl/codegen/compression_types.h>
#include <grpc/impl/codegen/grpc_types.h>
@@ -560,17 +562,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 have a shared_ptr back to the collection,
-/// as will the object that instantiates the collection, allowing
-/// for automatic ref-counting. In practice, 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 std::enable_shared_from_this<CallOpSetCollectionInterface> {};
-
/// 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
@@ -578,18 +569,9 @@ class CallOpSetCollectionInterface
/// API.
class CallOpSetInterface : public CompletionQueueTag {
public:
- CallOpSetInterface() {}
/// 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(std::shared_ptr<CallOpSetCollectionInterface> collection) {
- collection_ = collection;
- }
-
- protected:
- std::shared_ptr<CallOpSetCollectionInterface> collection_;
+ virtual void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) = 0;
};
/// Primary implementaiton of CallOpSetInterface.
@@ -610,13 +592,15 @@ 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);
+ g_core_codegen_interface->grpc_call_ref(call);
+ call_ = call;
}
bool FinalizeResult(void** tag, bool* status) override {
@@ -627,7 +611,7 @@ class CallOpSet : public CallOpSetInterface,
this->Op5::FinishOp(status);
this->Op6::FinishOp(status);
*tag = return_tag_;
- collection_.reset(); // drop the ref at this point
+ g_core_codegen_interface->grpc_call_unref(call_);
return true;
}
@@ -635,6 +619,7 @@ class CallOpSet : public CallOpSetInterface,
private:
void* return_tag_;
+ grpc_call* call_;
};
/// A CallOpSet that does not post completions to the completion queue.
diff --git a/include/grpc++/impl/codegen/core_codegen.h b/include/grpc++/impl/codegen/core_codegen.h
index 6bf95129b8..a5f762e21d 100644
--- a/include/grpc++/impl/codegen/core_codegen.h
+++ b/include/grpc++/impl/codegen/core_codegen.h
@@ -64,6 +64,9 @@ class CoreCodegen : public CoreCodegenInterface {
void gpr_cv_signal(gpr_cv* cv) override;
void gpr_cv_broadcast(gpr_cv* cv) override;
+ void grpc_call_ref(grpc_call* call) override;
+ void grpc_call_unref(grpc_call* call) override;
+
void grpc_byte_buffer_destroy(grpc_byte_buffer* bb) override;
int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader* reader,
diff --git a/include/grpc++/impl/codegen/core_codegen_interface.h b/include/grpc++/impl/codegen/core_codegen_interface.h
index e111d59364..a3df913c26 100644
--- a/include/grpc++/impl/codegen/core_codegen_interface.h
+++ b/include/grpc++/impl/codegen/core_codegen_interface.h
@@ -94,6 +94,9 @@ class CoreCodegenInterface {
virtual grpc_byte_buffer* grpc_raw_byte_buffer_create(grpc_slice* slice,
size_t nslices) = 0;
+ virtual void grpc_call_ref(grpc_call* call) = 0;
+ virtual void grpc_call_unref(grpc_call* call) = 0;
+
virtual grpc_slice grpc_empty_slice() = 0;
virtual grpc_slice grpc_slice_malloc(size_t length) = 0;
virtual void grpc_slice_unref(grpc_slice slice) = 0;
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-29 04:30:57 +0000