/*
*
* Copyright 2015 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
#include <grpcpp/server_context.h>
#include <algorithm>
#include <mutex>
#include <utility>
#include <grpc/compression.h>
#include <grpc/grpc.h>
#include <grpc/load_reporting.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpcpp/completion_queue.h>
#include <grpcpp/impl/call.h>
#include <grpcpp/support/time.h>
#include "src/core/lib/surface/call.h"
namespace grpc {
// CompletionOp
class ServerContext::CompletionOp final : public internal::CallOpSetInterface {
public:
// initial refs: one in the server context, one in the cq
CompletionOp()
: has_tag_(false),
tag_(nullptr),
refs_(2),
finalized_(false),
cancelled_(0) {}
void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) override;
bool FinalizeResult(void** tag, bool* status) override;
bool CheckCancelled(CompletionQueue* cq) {
cq->TryPluck(this);
return CheckCancelledNoPluck();
}
bool CheckCancelledAsync() { return CheckCancelledNoPluck(); }
void set_tag(void* tag) {
has_tag_ = true;
tag_ = tag;
}
/// TODO(vjpai): Allow override of cq_tag if appropriate for callback API
void* cq_tag() override { return this; }
void Unref();
private:
bool CheckCancelledNoPluck() {
std::lock_guard<std::mutex> g(mu_);
return finalized_ ? (cancelled_ != 0) : false;
}
bool has_tag_;
void* tag_;
std::mutex mu_;
int refs_;
bool finalized_;
int cancelled_;
};
void ServerContext::CompletionOp::Unref() {
std::unique_lock<std::mutex> lock(mu_);
if (--refs_ == 0) {
lock.unlock();
delete this;
}
}
void ServerContext::CompletionOp::FillOps(grpc_call* call, grpc_op* ops,
size_t* nops) {
ops->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
ops->data.recv_close_on_server.cancelled = &cancelled_;
ops->flags = 0;
ops->reserved = nullptr;
*nops = 1;
}
bool ServerContext::CompletionOp::FinalizeResult(void** tag, bool* status) {
std::unique_lock<std::mutex> lock(mu_);
finalized_ = true;
bool ret = false;
if (has_tag_) {
*tag = tag_;
ret = true;
}
if (!*status) cancelled_ = 1;
if (--refs_ == 0) {
lock.unlock();
delete this;
}
return ret;
}
// ServerContext body
ServerContext::ServerContext()
: completion_op_(nullptr),
has_notify_when_done_tag_(false),
async_notify_when_done_tag_(nullptr),
deadline_(gpr_inf_future(GPR_CLOCK_REALTIME)),
call_(nullptr),
cq_(nullptr),
sent_initial_metadata_(false),
compression_level_set_(false),
has_pending_ops_(false) {}
ServerContext::ServerContext(gpr_timespec deadline, grpc_metadata_array* arr)
: completion_op_(nullptr),
has_notify_when_done_tag_(false),
async_notify_when_done_tag_(nullptr),
deadline_(deadline),
call_(nullptr),
cq_(nullptr),
sent_initial_metadata_(false),
compression_level_set_(false),
has_pending_ops_(false) {
std::swap(*client_metadata_.arr(), *arr);
}
ServerContext::~ServerContext() {
if (call_) {
grpc_call_unref(call_);
}
if (completion_op_) {
completion_op_->Unref();
}
}
void ServerContext::BeginCompletionOp(internal::Call* call) {
GPR_ASSERT(!completion_op_);
completion_op_ = new CompletionOp();
if (has_notify_when_done_tag_) {
completion_op_->set_tag(async_notify_when_done_tag_);
}
call->PerformOps(completion_op_);
}
internal::CompletionQueueTag* ServerContext::GetCompletionOpTag() {
return static_cast<internal::CompletionQueueTag*>(completion_op_);
}
void ServerContext::AddInitialMetadata(const grpc::string& key,
const grpc::string& value) {
initial_metadata_.insert(std::make_pair(key, value));
}
void ServerContext::AddTrailingMetadata(const grpc::string& key,
const grpc::string& value) {
trailing_metadata_.insert(std::make_pair(key, value));
}
void ServerContext::TryCancel() const {
grpc_call_error err = grpc_call_cancel_with_status(
call_, GRPC_STATUS_CANCELLED, "Cancelled on the server side", nullptr);
if (err != GRPC_CALL_OK) {
gpr_log(GPR_ERROR, "TryCancel failed with: %d", err);
}
}
bool ServerContext::IsCancelled() const {
if (has_notify_when_done_tag_) {
// when using async API, but the result is only valid
// if the tag has already been delivered at the completion queue
return completion_op_ && completion_op_->CheckCancelledAsync();
} else {
// when using sync API
return completion_op_ && completion_op_->CheckCancelled(cq_);
}
}
void ServerContext::set_compression_algorithm(
grpc_compression_algorithm algorithm) {
compression_algorithm_ = algorithm;
const char* algorithm_name = nullptr;
if (!grpc_compression_algorithm_name(algorithm, &algorithm_name)) {
gpr_log(GPR_ERROR, "Name for compression algorithm '%d' unknown.",
algorithm);
abort();
}
GPR_ASSERT(algorithm_name != nullptr);
AddInitialMetadata(GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY, algorithm_name);
}
grpc::string ServerContext::peer() const {
grpc::string peer;
if (call_) {
char* c_peer = grpc_call_get_peer(call_);
peer = c_peer;
gpr_free(c_peer);
}
return peer;
}
const struct census_context* ServerContext::census_context() const {
return grpc_census_call_get_context(call_);
}
void ServerContext::SetLoadReportingCosts(
const std::vector<grpc::string>& cost_data) {
if (call_ == nullptr) return;
for (const auto& cost_datum : cost_data) {
AddTrailingMetadata(GRPC_LB_COST_MD_KEY, cost_datum);
}
}
} // namespace grpc