diff options
Diffstat (limited to 'src/cpp')
-rw-r--r-- | src/cpp/client/client_unary_call.cc | 5 | ||||
-rw-r--r-- | src/cpp/common/call.cc | 3 | ||||
-rw-r--r-- | src/cpp/common/completion_queue.cc | 52 | ||||
-rw-r--r-- | src/cpp/server/server.cc | 17 | ||||
-rw-r--r-- | src/cpp/server/server_context.cc | 68 |
5 files changed, 117 insertions, 28 deletions
diff --git a/src/cpp/client/client_unary_call.cc b/src/cpp/client/client_unary_call.cc index 8fdd483474..a5ef989946 100644 --- a/src/cpp/client/client_unary_call.cc +++ b/src/cpp/client/client_unary_call.cc @@ -63,9 +63,10 @@ Status BlockingUnaryCall(ChannelInterface *channel, const RpcMethod &method, class ClientAsyncRequest final : public CallOpBuffer { public: - void FinalizeResult(void **tag, bool *status) override { - CallOpBuffer::FinalizeResult(tag, status); + bool FinalizeResult(void **tag, bool *status) override { + bool r = CallOpBuffer::FinalizeResult(tag, status); delete this; + return r; } }; diff --git a/src/cpp/common/call.cc b/src/cpp/common/call.cc index e29c6a053d..e6a20a252d 100644 --- a/src/cpp/common/call.cc +++ b/src/cpp/common/call.cc @@ -231,7 +231,7 @@ void CallOpBuffer::FillOps(grpc_op* ops, size_t* nops) { } } -void CallOpBuffer::FinalizeResult(void** tag, bool* status) { +bool CallOpBuffer::FinalizeResult(void** tag, bool* status) { // Release send buffers. if (send_message_buf_) { grpc_byte_buffer_destroy(send_message_buf_); @@ -274,6 +274,7 @@ void CallOpBuffer::FinalizeResult(void** tag, bool* status) { if (recv_closed_) { *recv_closed_ = cancelled_buf_ != 0; } + return true; } Call::Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq) diff --git a/src/cpp/common/completion_queue.cc b/src/cpp/common/completion_queue.cc index c7d883d5b0..5d186e06cc 100644 --- a/src/cpp/common/completion_queue.cc +++ b/src/cpp/common/completion_queue.cc @@ -43,7 +43,7 @@ namespace grpc { CompletionQueue::CompletionQueue() { cq_ = grpc_completion_queue_create(); } -CompletionQueue::CompletionQueue(grpc_completion_queue *take) : cq_(take) {} +CompletionQueue::CompletionQueue(grpc_completion_queue* take) : cq_(take) {} CompletionQueue::~CompletionQueue() { grpc_completion_queue_destroy(cq_); } @@ -52,34 +52,52 @@ void CompletionQueue::Shutdown() { grpc_completion_queue_shutdown(cq_); } // Helper class so we can declare a unique_ptr with grpc_event class EventDeleter { public: - void operator()(grpc_event *ev) { + void operator()(grpc_event* ev) { if (ev) grpc_event_finish(ev); } }; -bool CompletionQueue::Next(void **tag, bool *ok) { +bool CompletionQueue::Next(void** tag, bool* ok) { std::unique_ptr<grpc_event, EventDeleter> ev; - ev.reset(grpc_completion_queue_next(cq_, gpr_inf_future)); - if (ev->type == GRPC_QUEUE_SHUTDOWN) { - return false; + for (;;) { + ev.reset(grpc_completion_queue_next(cq_, gpr_inf_future)); + if (ev->type == GRPC_QUEUE_SHUTDOWN) { + return false; + } + auto cq_tag = static_cast<CompletionQueueTag*>(ev->tag); + *ok = ev->data.op_complete == GRPC_OP_OK; + *tag = cq_tag; + if (cq_tag->FinalizeResult(tag, ok)) { + return true; + } } - auto cq_tag = static_cast<CompletionQueueTag *>(ev->tag); - *ok = ev->data.op_complete == GRPC_OP_OK; - *tag = cq_tag; - cq_tag->FinalizeResult(tag, ok); - return true; } -bool CompletionQueue::Pluck(CompletionQueueTag *tag) { +bool CompletionQueue::Pluck(CompletionQueueTag* tag) { std::unique_ptr<grpc_event, EventDeleter> ev; - ev.reset(grpc_completion_queue_pluck(cq_, tag, gpr_inf_future)); + for (;;) { + ev.reset(grpc_completion_queue_pluck(cq_, tag, gpr_inf_future)); + bool ok = ev->data.op_complete == GRPC_OP_OK; + void* ignored = tag; + if (tag->FinalizeResult(&ignored, &ok)) { + GPR_ASSERT(ignored == tag); + return ok; + } + } +} + +void CompletionQueue::TryPluck(CompletionQueueTag* tag, bool forever) { + std::unique_ptr<grpc_event, EventDeleter> ev; + + ev.reset(grpc_completion_queue_pluck( + cq_, tag, forever ? gpr_inf_future : gpr_inf_past)); + if (!ev) return; bool ok = ev->data.op_complete == GRPC_OP_OK; - void *ignored = tag; - tag->FinalizeResult(&ignored, &ok); - GPR_ASSERT(ignored == tag); - return ok; + void* ignored = tag; + // the tag must be swallowed if using TryPluck + GPR_ASSERT(!tag->FinalizeResult(&ignored, &ok)); } } // namespace grpc diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc index da98cf5ce0..7cccc58afd 100644 --- a/src/cpp/server/server.cc +++ b/src/cpp/server/server.cc @@ -163,10 +163,11 @@ class Server::SyncRequest final : public CompletionQueueTag { this)); } - void FinalizeResult(void** tag, bool* status) override { + bool FinalizeResult(void** tag, bool* status) override { if (!*status) { grpc_completion_queue_destroy(cq_); } + return true; } class CallData final { @@ -204,6 +205,7 @@ class Server::SyncRequest final : public CompletionQueueTag { 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; @@ -214,10 +216,12 @@ class Server::SyncRequest final : public CompletionQueueTag { buf.AddSendMessage(*res); } buf.AddServerSendStatus(&ctx_.trailing_metadata_, status); - bool cancelled; - buf.AddServerRecvClose(&cancelled); call_.PerformOps(&buf); GPR_ASSERT(cq_.Pluck(&buf)); + void* ignored_tag; + bool ignored_ok; + cq_.Shutdown(); + GPR_ASSERT(cq_.Next(&ignored_tag, &ignored_ok) == false); } private: @@ -310,11 +314,11 @@ class Server::AsyncRequest final : public CompletionQueueTag { grpc_metadata_array_destroy(&array_); } - void FinalizeResult(void** tag, bool* status) override { + bool FinalizeResult(void** tag, bool* status) override { *tag = tag_; if (*status && request_) { if (payload_) { - *status = *status && DeserializeProto(payload_, request_); + *status = DeserializeProto(payload_, request_); } else { *status = false; } @@ -331,8 +335,11 @@ class Server::AsyncRequest final : public CompletionQueueTag { } ctx_->call_ = call_; Call call(call_, server_, cq_); + ctx_->BeginCompletionOp(&call); + // just the pointers inside call are copied here stream_->BindCall(&call); delete this; + return true; } private: diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index 10cce450d7..71e3b464d9 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -34,13 +34,62 @@ #include <grpc++/server_context.h> #include <grpc++/impl/call.h> #include <grpc/grpc.h> +#include <grpc/support/log.h> #include "src/cpp/util/time.h" namespace grpc { +// CompletionOp + +class ServerContext::CompletionOp final : public CallOpBuffer { + public: + CompletionOp(); + bool FinalizeResult(void** tag, bool* status) override; + + bool CheckCancelled(CompletionQueue* cq); + + void Unref(); + + private: + std::mutex mu_; + int refs_ = 2; // initial refs: one in the server context, one in the cq + bool finalized_ = false; + bool cancelled_ = false; +}; + +ServerContext::CompletionOp::CompletionOp() { AddServerRecvClose(&cancelled_); } + +void ServerContext::CompletionOp::Unref() { + std::unique_lock<std::mutex> lock(mu_); + if (--refs_ == 0) { + lock.unlock(); + delete this; + } +} + +bool ServerContext::CompletionOp::CheckCancelled(CompletionQueue* cq) { + cq->TryPluck(this, false); + std::lock_guard<std::mutex> g(mu_); + return finalized_ ? cancelled_ : false; +} + +bool ServerContext::CompletionOp::FinalizeResult(void** tag, bool* status) { + GPR_ASSERT(CallOpBuffer::FinalizeResult(tag, status)); + std::unique_lock<std::mutex> lock(mu_); + finalized_ = true; + if (!*status) cancelled_ = true; + if (--refs_ == 0) { + lock.unlock(); + delete this; + } + return false; +} + +// ServerContext body + ServerContext::ServerContext() {} -ServerContext::ServerContext(gpr_timespec deadline, grpc_metadata *metadata, +ServerContext::ServerContext(gpr_timespec deadline, grpc_metadata* metadata, size_t metadata_count) : deadline_(Timespec2Timepoint(deadline)) { for (size_t i = 0; i < metadata_count; i++) { @@ -55,16 +104,29 @@ ServerContext::~ServerContext() { if (call_) { grpc_call_destroy(call_); } + if (completion_op_) { + completion_op_->Unref(); + } +} + +void ServerContext::BeginCompletionOp(Call* call) { + GPR_ASSERT(!completion_op_); + completion_op_ = new CompletionOp(); + call->PerformOps(completion_op_); } void ServerContext::AddInitialMetadata(const grpc::string& key, - const grpc::string& value) { + const grpc::string& value) { initial_metadata_.insert(std::make_pair(key, value)); } void ServerContext::AddTrailingMetadata(const grpc::string& key, - const grpc::string& value) { + const grpc::string& value) { trailing_metadata_.insert(std::make_pair(key, value)); } +bool ServerContext::IsCancelled() { + return completion_op_ && completion_op_->CheckCancelled(cq_); +} + } // namespace grpc |