aboutsummaryrefslogtreecommitdiffhomepage
path: root/include
diff options
context:
space:
mode:
Diffstat (limited to 'include')
-rw-r--r--include/grpc++/impl/codegen/client_unary_call.h2
-rw-r--r--include/grpc++/impl/codegen/completion_queue.h40
-rw-r--r--include/grpc++/impl/codegen/core_codegen.h6
-rw-r--r--include/grpc++/impl/codegen/core_codegen_interface.h4
-rw-r--r--include/grpc++/impl/codegen/server_context.h3
-rw-r--r--include/grpc++/impl/codegen/server_interface.h4
-rw-r--r--include/grpc++/impl/codegen/sync_stream.h14
-rw-r--r--include/grpc++/server.h4
-rw-r--r--include/grpc/grpc.h22
-rw-r--r--include/grpc/impl/codegen/grpc_types.h7
10 files changed, 77 insertions, 29 deletions
diff --git a/include/grpc++/impl/codegen/client_unary_call.h b/include/grpc++/impl/codegen/client_unary_call.h
index 201e52ae07..a5a4f3d739 100644
--- a/include/grpc++/impl/codegen/client_unary_call.h
+++ b/include/grpc++/impl/codegen/client_unary_call.h
@@ -52,7 +52,7 @@ template <class InputMessage, class OutputMessage>
Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result) {
- CompletionQueue cq;
+ CompletionQueue cq(true); // Pluckable completion queue
Call call(channel->CreateCall(method, context, &cq));
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>,
diff --git a/include/grpc++/impl/codegen/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h
index 03cecdc21c..61617f2bdc 100644
--- a/include/grpc++/impl/codegen/completion_queue.h
+++ b/include/grpc++/impl/codegen/completion_queue.h
@@ -102,10 +102,7 @@ class CompletionQueue : private GrpcLibraryCodegen {
public:
/// Default constructor. Implicitly creates a \a grpc_completion_queue
/// instance.
- CompletionQueue() {
- cq_ = g_core_codegen_interface->grpc_completion_queue_create(nullptr);
- InitialAvalanching(); // reserve this for the future shutdown
- }
+ CompletionQueue() : CompletionQueue(false) {}
/// Wrap \a take, taking ownership of the instance.
///
@@ -218,6 +215,18 @@ class CompletionQueue : private GrpcLibraryCodegen {
const InputMessage& request,
OutputMessage* result);
+ /// Private constructor of CompletionQueue only visible to friend classes
+ CompletionQueue(bool is_pluck) {
+ if (is_pluck) {
+ cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_pluck(
+ nullptr);
+ } else {
+ cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_next(
+ nullptr);
+ }
+ InitialAvalanching(); // reserve this for the future shutdown
+ }
+
NextStatus AsyncNextInternal(void** tag, bool* ok, gpr_timespec deadline);
/// Wraps \a grpc_completion_queue_pluck.
@@ -237,6 +246,12 @@ class CompletionQueue : private GrpcLibraryCodegen {
/// Performs a single polling pluck on \a tag.
/// \warning Must not be mixed with calls to \a Next.
+ ///
+ /// TODO: sreek - This calls tag->FinalizeResult() even if the cq_ is already
+ /// shutdown. This is most likely a bug and if it is a bug, then change this
+ /// implementation to simple call the other TryPluck function with a zero
+ /// timeout. i.e:
+ /// TryPluck(tag, gpr_time_0(GPR_CLOCK_REALTIME))
void TryPluck(CompletionQueueTag* tag) {
auto deadline = g_core_codegen_interface->gpr_time_0(GPR_CLOCK_REALTIME);
auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
@@ -248,6 +263,23 @@ class CompletionQueue : private GrpcLibraryCodegen {
GPR_CODEGEN_ASSERT(!tag->FinalizeResult(&ignored, &ok));
}
+ /// Performs a single polling pluck on \a tag. Calls tag->FinalizeResult if
+ /// the pluck() was successful and returned the tag.
+ ///
+ /// This exects tag->FinalizeResult (if called) to return 'false' i.e expects
+ /// that the tag is internal not something that is returned to the user.
+ void TryPluck(CompletionQueueTag* tag, gpr_timespec deadline) {
+ auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
+ cq_, tag, deadline, nullptr);
+ if (ev.type == GRPC_QUEUE_TIMEOUT || ev.type == GRPC_QUEUE_SHUTDOWN) {
+ return;
+ }
+
+ bool ok = ev.success != 0;
+ void* ignored = tag;
+ GPR_CODEGEN_ASSERT(!tag->FinalizeResult(&ignored, &ok));
+ }
+
grpc_completion_queue* cq_; // owned
gpr_atm avalanches_in_flight_;
diff --git a/include/grpc++/impl/codegen/core_codegen.h b/include/grpc++/impl/codegen/core_codegen.h
index 6bf95129b8..65151590b2 100644
--- a/include/grpc++/impl/codegen/core_codegen.h
+++ b/include/grpc++/impl/codegen/core_codegen.h
@@ -38,6 +38,7 @@
#include <grpc++/impl/codegen/core_codegen_interface.h>
#include <grpc/byte_buffer.h>
+#include <grpc/grpc.h>
#include <grpc/impl/codegen/grpc_types.h>
namespace grpc {
@@ -45,7 +46,10 @@ namespace grpc {
/// Implementation of the core codegen interface.
class CoreCodegen : public CoreCodegenInterface {
private:
- grpc_completion_queue* grpc_completion_queue_create(void* reserved) override;
+ grpc_completion_queue* grpc_completion_queue_create_for_next(
+ void* reserved) override;
+ grpc_completion_queue* grpc_completion_queue_create_for_pluck(
+ void* reserved) override;
void grpc_completion_queue_destroy(grpc_completion_queue* cq) override;
grpc_event grpc_completion_queue_pluck(grpc_completion_queue* cq, void* tag,
gpr_timespec deadline,
diff --git a/include/grpc++/impl/codegen/core_codegen_interface.h b/include/grpc++/impl/codegen/core_codegen_interface.h
index e111d59364..529bef687b 100644
--- a/include/grpc++/impl/codegen/core_codegen_interface.h
+++ b/include/grpc++/impl/codegen/core_codegen_interface.h
@@ -59,7 +59,9 @@ class CoreCodegenInterface {
virtual void assert_fail(const char* failed_assertion, const char* file,
int line) = 0;
- virtual grpc_completion_queue* grpc_completion_queue_create(
+ virtual grpc_completion_queue* grpc_completion_queue_create_for_next(
+ void* reserved) = 0;
+ virtual grpc_completion_queue* grpc_completion_queue_create_for_pluck(
void* reserved) = 0;
virtual void grpc_completion_queue_destroy(grpc_completion_queue* cq) = 0;
virtual grpc_event grpc_completion_queue_pluck(grpc_completion_queue* cq,
diff --git a/include/grpc++/impl/codegen/server_context.h b/include/grpc++/impl/codegen/server_context.h
index 91f0be06e7..ada304d571 100644
--- a/include/grpc++/impl/codegen/server_context.h
+++ b/include/grpc++/impl/codegen/server_context.h
@@ -40,6 +40,7 @@
#include <grpc/impl/codegen/compression_types.h>
+#include <grpc++/impl/codegen/completion_queue_tag.h>
#include <grpc++/impl/codegen/config.h>
#include <grpc++/impl/codegen/create_auth_context.h>
#include <grpc++/impl/codegen/metadata_map.h>
@@ -211,6 +212,8 @@ class ServerContext {
class CompletionOp;
void BeginCompletionOp(Call* call);
+ // Return the tag queued by BeginCompletionOp()
+ CompletionQueueTag* GetCompletionOpTag();
ServerContext(gpr_timespec deadline, grpc_metadata_array* arr);
diff --git a/include/grpc++/impl/codegen/server_interface.h b/include/grpc++/impl/codegen/server_interface.h
index bd1b36e883..31d55529ef 100644
--- a/include/grpc++/impl/codegen/server_interface.h
+++ b/include/grpc++/impl/codegen/server_interface.h
@@ -122,9 +122,7 @@ class ServerInterface : public CallHook {
/// caller is required to keep all completion queues live until the server is
/// destroyed.
/// \param num_cqs How many completion queues does \a cqs hold.
- ///
- /// \return true on a successful shutdown.
- virtual bool Start(ServerCompletionQueue** cqs, size_t num_cqs) = 0;
+ virtual void Start(ServerCompletionQueue** cqs, size_t num_cqs) = 0;
virtual void ShutdownInternal(gpr_timespec deadline) = 0;
diff --git a/include/grpc++/impl/codegen/sync_stream.h b/include/grpc++/impl/codegen/sync_stream.h
index ae3b8e441d..328d5cb1e8 100644
--- a/include/grpc++/impl/codegen/sync_stream.h
+++ b/include/grpc++/impl/codegen/sync_stream.h
@@ -155,7 +155,9 @@ class ClientReader final : public ClientReaderInterface<R> {
template <class W>
ClientReader(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const W& request)
- : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+ : context_(context),
+ cq_(true), // Pluckable cq
+ call_(channel->CreateCall(method, context, &cq_)) {
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
ops;
@@ -227,7 +229,9 @@ class ClientWriter : public ClientWriterInterface<W> {
template <class R>
ClientWriter(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, R* response)
- : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+ : context_(context),
+ cq_(true), // Pluckable cq
+ call_(channel->CreateCall(method, context, &cq_)) {
finish_ops_.RecvMessage(response);
finish_ops_.AllowNoMessage();
@@ -325,7 +329,9 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
/// Blocking create a stream.
ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context)
- : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
+ : context_(context),
+ cq_(true), // Pluckable cq
+ call_(channel->CreateCall(method, context, &cq_)) {
if (!context_->initial_metadata_corked_) {
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(context->send_initial_metadata_,
@@ -562,7 +568,7 @@ class ServerReaderWriterBody final {
Call* const call_;
ServerContext* const ctx_;
};
-}
+} // namespace internal
// class to represent the user API for a bidirectional streaming call
template <class W, class R>
diff --git a/include/grpc++/server.h b/include/grpc++/server.h
index 489937712e..6d5f89d360 100644
--- a/include/grpc++/server.h
+++ b/include/grpc++/server.h
@@ -177,9 +177,7 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen {
/// caller is required to keep all completion queues live until the server is
/// destroyed.
/// \param num_cqs How many completion queues does \a cqs hold.
- ///
- /// \return true on a successful shutdown.
- bool Start(ServerCompletionQueue** cqs, size_t num_cqs) override;
+ void Start(ServerCompletionQueue** cqs, size_t num_cqs) override;
void PerformOpsOnCall(CallOpSetInterface* ops, Call* call) override;
diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index e088435d6c..42cf201da4 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -95,10 +95,10 @@ GRPCAPI const char *grpc_g_stands_for(void);
/** Specifies the type of APIs to use to pop events from the completion queue */
typedef enum {
- /* Events are popped out by calling grpc_completion_queue_next() API ONLY */
+ /** Events are popped out by calling grpc_completion_queue_next() API ONLY */
GRPC_CQ_NEXT = 1,
- /* Events are popped out by calling grpc_completion_queue_pluck() API ONLY */
+ /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/
GRPC_CQ_PLUCK
} grpc_cq_completion_type;
@@ -116,15 +116,15 @@ typedef enum {
restriction on the type of file descriptors the pollset may contain */
GRPC_CQ_DEFAULT_POLLING,
- /* Similar to GRPC_CQ_DEFAULT_POLLING except that the completion queues will
- not contain any 'listening file descriptors' (i.e file descriptors used to
- listen to incoming channels) */
+ /** Similar to GRPC_CQ_DEFAULT_POLLING except that the completion queues will
+ not contain any 'listening file descriptors' (i.e file descriptors used to
+ listen to incoming channels) */
GRPC_CQ_NON_LISTENING,
- /* The completion queue will not have an associated pollset. Note that
- grpc_completion_queue_next() or grpc_completion_queue_pluck() MUST still be
- called to pop events from the completion queue; it is not required to call
- them actively to make I/O progress */
+ /** The completion queue will not have an associated pollset. Note that
+ grpc_completion_queue_next() or grpc_completion_queue_pluck() MUST still
+ be called to pop events from the completion queue; it is not required to
+ call them actively to make I/O progress */
GRPC_CQ_NON_POLLING
} grpc_cq_polling_type;
@@ -159,7 +159,9 @@ GRPCAPI grpc_completion_queue *grpc_completion_queue_create_for_pluck(
void *reserved);
/** Create a completion queue */
-GRPCAPI grpc_completion_queue *grpc_completion_queue_create(void *reserved);
+GRPCAPI grpc_completion_queue *grpc_completion_queue_create(
+ const grpc_completion_queue_factory *factory,
+ const grpc_completion_queue_attributes *attributes, void *reserved);
/** Blocks until an event is available, the completion queue is being shut down,
or deadline is reached.
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h
index 1057d35877..06b67015e3 100644
--- a/include/grpc/impl/codegen/grpc_types.h
+++ b/include/grpc/impl/codegen/grpc_types.h
@@ -401,8 +401,11 @@ typedef enum grpc_completion_type {
typedef struct grpc_event {
/** The type of the completion. */
grpc_completion_type type;
- /** non-zero if the operation was successful, 0 upon failure.
- Only GRPC_OP_COMPLETE can succeed or fail. */
+ /** If the grpc_completion_type is GRPC_OP_COMPLETE, this field indicates
+ whether the operation was successful or not; 0 in case of failure and
+ non-zero in case of success.
+ If grpc_completion_type is GRPC_QUEUE_SHUTDOWN or GRPC_QUEUE_TIMEOUT, this
+ field is guaranteed to be 0 */
int success;
/** The tag passed to grpc_call_start_batch etc to start this operation.
Only GRPC_OP_COMPLETE has a tag. */
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-22 14:12:57 +0000