Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge branch 'maxmsgsz' into bm_msgsz

Browse Source
pull/9408/head
Craig Tiller 8 years ago
parent a2f61e3d33 1d77059e4b
commit 194769163f
8 changed files with 40 additions and 79 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 71
      include/grpc++/impl/codegen/call.h
  2. 8
      include/grpc++/impl/codegen/method_handler_impl.h
  3. 8
      include/grpc++/impl/codegen/proto_utils.h
  4. 9
      include/grpc++/impl/codegen/rpc_service_method.h
  5. 4
      include/grpc++/impl/codegen/server_interface.h
  6. 8
      include/grpc++/impl/codegen/sync_stream.h
  7. 3
      include/grpc++/support/byte_buffer.h
  8. 8
      src/cpp/server/server_cc.cc

71
include/grpc++/impl/codegen/call.h
Unescape View File

@ -175,7 +175,7 @@ template <int I>
class CallNoOp { class CallNoOp {
protected: protected:
void AddOp(grpc_op* ops, size_t* nops) {} void AddOp(grpc_op* ops, size_t* nops) {}
void FinishOp(bool* status, int max_receive_message_size) {} void FinishOp(bool* status) {}
}; };
class CallOpSendInitialMetadata { class CallOpSendInitialMetadata {
@ -213,7 +213,7 @@ class CallOpSendInitialMetadata {
op->data.send_initial_metadata.maybe_compression_level.level = op->data.send_initial_metadata.maybe_compression_level.level =
maybe_compression_level_.level; maybe_compression_level_.level;
} }
void FinishOp(bool* status, int max_receive_message_size) { void FinishOp(bool* status) {
if (!send_) return; if (!send_) return;
g_core_codegen_interface->gpr_free(initial_metadata_); g_core_codegen_interface->gpr_free(initial_metadata_);
send_ = false; send_ = false;
@ -253,7 +253,7 @@ class CallOpSendMessage {
// Flags are per-message: clear them after use. // Flags are per-message: clear them after use.
write_options_.Clear(); write_options_.Clear();
} }
void FinishOp(bool* status, int max_receive_message_size) { void FinishOp(bool* status) {
if (own_buf_) g_core_codegen_interface->grpc_byte_buffer_destroy(send_buf_); if (own_buf_) g_core_codegen_interface->grpc_byte_buffer_destroy(send_buf_);
send_buf_ = nullptr; send_buf_ = nullptr;
} }
@ -301,14 +301,12 @@ class CallOpRecvMessage {
op->data.recv_message = &recv_buf_; op->data.recv_message = &recv_buf_;
} }
void FinishOp(bool* status, int max_receive_message_size) { void FinishOp(bool* status) {
if (message_ == nullptr) return; if (message_ == nullptr) return;
if (recv_buf_) { if (recv_buf_) {
if (*status) { if (*status) {
got_message = *status = got_message = *status =
SerializationTraits<R>::Deserialize(recv_buf_, message_, SerializationTraits<R>::Deserialize(recv_buf_, message_).ok();
max_receive_message_size)
.ok();
} else { } else {
got_message = false; got_message = false;
g_core_codegen_interface->grpc_byte_buffer_destroy(recv_buf_); g_core_codegen_interface->grpc_byte_buffer_destroy(recv_buf_);
@ -331,8 +329,7 @@ class CallOpRecvMessage {
namespace CallOpGenericRecvMessageHelper { namespace CallOpGenericRecvMessageHelper {
class DeserializeFunc { class DeserializeFunc {
public: public:
virtual Status Deserialize(grpc_byte_buffer* buf, virtual Status Deserialize(grpc_byte_buffer* buf) = 0;
int max_receive_message_size) = 0;
virtual ~DeserializeFunc() {} virtual ~DeserializeFunc() {}
}; };
@ -340,10 +337,8 @@ template <class R>
class DeserializeFuncType final : public DeserializeFunc { class DeserializeFuncType final : public DeserializeFunc {
public: public:
DeserializeFuncType(R* message) : message_(message) {} DeserializeFuncType(R* message) : message_(message) {}
Status Deserialize(grpc_byte_buffer* buf, Status Deserialize(grpc_byte_buffer* buf) override {
int max_receive_message_size) override { return SerializationTraits<R>::Deserialize(buf, message_);
return SerializationTraits<R>::Deserialize(buf, message_,
max_receive_message_size);
} }
~DeserializeFuncType() override {} ~DeserializeFuncType() override {}
@ -382,13 +377,12 @@ class CallOpGenericRecvMessage {
op->data.recv_message = &recv_buf_; op->data.recv_message = &recv_buf_;
} }
void FinishOp(bool* status, int max_receive_message_size) { void FinishOp(bool* status) {
if (!deserialize_) return; if (!deserialize_) return;
if (recv_buf_) { if (recv_buf_) {
if (*status) { if (*status) {
got_message = true; got_message = true;
*status = *status = deserialize_->Deserialize(recv_buf_).ok();
deserialize_->Deserialize(recv_buf_, max_receive_message_size).ok();
} else { } else {
got_message = false; got_message = false;
g_core_codegen_interface->grpc_byte_buffer_destroy(recv_buf_); g_core_codegen_interface->grpc_byte_buffer_destroy(recv_buf_);
@ -422,7 +416,7 @@ class CallOpClientSendClose {
op->flags = 0; op->flags = 0;
op->reserved = NULL; op->reserved = NULL;
} }
void FinishOp(bool* status, int max_receive_message_size) { send_ = false; } void FinishOp(bool* status) { send_ = false; }
private: private:
bool send_; bool send_;
@ -457,7 +451,7 @@ class CallOpServerSendStatus {
op->reserved = NULL; op->reserved = NULL;
} }
void FinishOp(bool* status, int max_receive_message_size) { void FinishOp(bool* status) {
if (!send_status_available_) return; if (!send_status_available_) return;
g_core_codegen_interface->gpr_free(trailing_metadata_); g_core_codegen_interface->gpr_free(trailing_metadata_);
send_status_available_ = false; send_status_available_ = false;
@ -490,7 +484,7 @@ class CallOpRecvInitialMetadata {
op->flags = 0; op->flags = 0;
op->reserved = NULL; op->reserved = NULL;
} }
void FinishOp(bool* status, int max_receive_message_size) { void FinishOp(bool* status) {
if (recv_initial_metadata_ == nullptr) return; if (recv_initial_metadata_ == nullptr) return;
FillMetadataMap(&recv_initial_metadata_arr_, recv_initial_metadata_); FillMetadataMap(&recv_initial_metadata_arr_, recv_initial_metadata_);
recv_initial_metadata_ = nullptr; recv_initial_metadata_ = nullptr;
@ -529,7 +523,7 @@ class CallOpClientRecvStatus {
op->reserved = NULL; op->reserved = NULL;
} }
void FinishOp(bool* status, int max_receive_message_size) { void FinishOp(bool* status) {
if (recv_status_ == nullptr) return; if (recv_status_ == nullptr) return;
FillMetadataMap(&recv_trailing_metadata_arr_, recv_trailing_metadata_); FillMetadataMap(&recv_trailing_metadata_arr_, recv_trailing_metadata_);
*recv_status_ = Status( *recv_status_ = Status(
@ -566,22 +560,17 @@ class CallOpSetCollectionInterface
/// API. /// API.
class CallOpSetInterface : public CompletionQueueTag { class CallOpSetInterface : public CompletionQueueTag {
public: public:
CallOpSetInterface() : max_receive_message_size_(0) {} CallOpSetInterface() {}
/// Fills in grpc_op, starting from ops[*nops] and moving /// Fills in grpc_op, starting from ops[*nops] and moving
/// upwards. /// upwards.
virtual void FillOps(grpc_op* ops, size_t* nops) = 0; virtual void FillOps(grpc_op* ops, size_t* nops) = 0;
void set_max_receive_message_size(int max_receive_message_size) {
max_receive_message_size_ = max_receive_message_size;
}
/// Mark this as belonging to a collection if needed /// Mark this as belonging to a collection if needed
void SetCollection(std::shared_ptr<CallOpSetCollectionInterface> collection) { void SetCollection(std::shared_ptr<CallOpSetCollectionInterface> collection) {
collection_ = collection; collection_ = collection;
} }
protected: protected:
int max_receive_message_size_;
std::shared_ptr<CallOpSetCollectionInterface> collection_; std::shared_ptr<CallOpSetCollectionInterface> collection_;
}; };
@ -613,12 +602,12 @@ class CallOpSet : public CallOpSetInterface,
} }
bool FinalizeResult(void** tag, bool* status) override { bool FinalizeResult(void** tag, bool* status) override {
this->Op1::FinishOp(status, max_receive_message_size_); this->Op1::FinishOp(status);
this->Op2::FinishOp(status, max_receive_message_size_); this->Op2::FinishOp(status);
this->Op3::FinishOp(status, max_receive_message_size_); this->Op3::FinishOp(status);
this->Op4::FinishOp(status, max_receive_message_size_); this->Op4::FinishOp(status);
this->Op5::FinishOp(status, max_receive_message_size_); this->Op5::FinishOp(status);
this->Op6::FinishOp(status, max_receive_message_size_); this->Op6::FinishOp(status);
*tag = return_tag_; *tag = return_tag_;
collection_.reset(); // drop the ref at this point collection_.reset(); // drop the ref at this point
return true; return true;
@ -650,35 +639,19 @@ class Call final {
public: public:
/* call is owned by the caller */ /* call is owned by the caller */
Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq) Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq)
: call_hook_(call_hook), : call_hook_(call_hook), cq_(cq), call_(call) {}
cq_(cq),
call_(call),
max_receive_message_size_(-1) {}
Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq,
int max_receive_message_size)
: call_hook_(call_hook),
cq_(cq),
call_(call),
max_receive_message_size_(max_receive_message_size) {}
void PerformOps(CallOpSetInterface* ops) { void PerformOps(CallOpSetInterface* ops) {
if (max_receive_message_size_ > 0) {
ops->set_max_receive_message_size(max_receive_message_size_);
}
call_hook_->PerformOpsOnCall(ops, this); call_hook_->PerformOpsOnCall(ops, this);
} }
grpc_call* call() const { return call_; } grpc_call* call() const { return call_; }
CompletionQueue* cq() const { return cq_; } CompletionQueue* cq() const { return cq_; }
int max_receive_message_size() { return max_receive_message_size_; }
private: private:
CallHook* call_hook_; CallHook* call_hook_;
CompletionQueue* cq_; CompletionQueue* cq_;
grpc_call* call_; grpc_call* call_;
int max_receive_message_size_;
}; };
} // namespace grpc } // namespace grpc

8
include/grpc++/impl/codegen/method_handler_impl.h
Unescape View File

@ -52,8 +52,8 @@ class RpcMethodHandler : public MethodHandler {
void RunHandler(const HandlerParameter& param) final { void RunHandler(const HandlerParameter& param) final {
RequestType req; RequestType req;
Status status = SerializationTraits<RequestType>::Deserialize( Status status =
param.request, &req, param.max_receive_message_size); SerializationTraits<RequestType>::Deserialize(param.request, &req);
ResponseType rsp; ResponseType rsp;
if (status.ok()) { if (status.ok()) {
status = func_(service_, param.server_context, &req, &rsp); status = func_(service_, param.server_context, &req, &rsp);
@ -138,8 +138,8 @@ class ServerStreamingHandler : public MethodHandler {
void RunHandler(const HandlerParameter& param) final { void RunHandler(const HandlerParameter& param) final {
RequestType req; RequestType req;
Status status = SerializationTraits<RequestType>::Deserialize( Status status =
param.request, &req, param.max_receive_message_size); SerializationTraits<RequestType>::Deserialize(param.request, &req);
if (status.ok()) { if (status.ok()) {
ServerWriter<ResponseType> writer(param.call, param.server_context); ServerWriter<ResponseType> writer(param.call, param.server_context);

8
include/grpc++/impl/codegen/proto_utils.h
Unescape View File

@ -203,8 +203,7 @@ class SerializationTraits<T, typename std::enable_if<std::is_base_of<
} }
static Status Deserialize(grpc_byte_buffer* buffer, static Status Deserialize(grpc_byte_buffer* buffer,
grpc::protobuf::Message* msg, grpc::protobuf::Message* msg) {
int max_receive_message_size) {
if (buffer == nullptr) { if (buffer == nullptr) {
return Status(StatusCode::INTERNAL, "No payload"); return Status(StatusCode::INTERNAL, "No payload");
} }
@ -215,10 +214,7 @@ class SerializationTraits<T, typename std::enable_if<std::is_base_of<
return reader.status(); return reader.status();
} }
::grpc::protobuf::io::CodedInputStream decoder(&reader); ::grpc::protobuf::io::CodedInputStream decoder(&reader);
if (max_receive_message_size > 0) { decoder.SetTotalBytesLimit(INT_MAX, INT_MAX);
decoder.SetTotalBytesLimit(max_receive_message_size,
max_receive_message_size);
}
if (!msg->ParseFromCodedStream(&decoder)) { if (!msg->ParseFromCodedStream(&decoder)) {
result = Status(StatusCode::INTERNAL, msg->InitializationErrorString()); result = Status(StatusCode::INTERNAL, msg->InitializationErrorString());
} }

9
include/grpc++/impl/codegen/rpc_service_method.h
Unescape View File

@ -57,17 +57,12 @@ class MethodHandler {
public: public:
virtual ~MethodHandler() {} virtual ~MethodHandler() {}
struct HandlerParameter { struct HandlerParameter {
HandlerParameter(Call* c, ServerContext* context, grpc_byte_buffer* req, HandlerParameter(Call* c, ServerContext* context, grpc_byte_buffer* req)
int max_size) : call(c), server_context(context), request(req) {}
: call(c),
server_context(context),
request(req),
max_receive_message_size(max_size) {}
Call* call; Call* call;
ServerContext* server_context; ServerContext* server_context;
// Handler required to grpc_byte_buffer_destroy this // Handler required to grpc_byte_buffer_destroy this
grpc_byte_buffer* request; grpc_byte_buffer* request;
int max_receive_message_size;
}; };
virtual void RunHandler(const HandlerParameter& param) = 0; virtual void RunHandler(const HandlerParameter& param) = 0;
}; };

4
include/grpc++/impl/codegen/server_interface.h
Unescape View File

@ -199,9 +199,7 @@ class ServerInterface : public CallHook {
bool FinalizeResult(void** tag, bool* status) override { bool FinalizeResult(void** tag, bool* status) override {
bool serialization_status = bool serialization_status =
*status && payload_ && *status && payload_ &&
SerializationTraits<Message>::Deserialize( SerializationTraits<Message>::Deserialize(payload_, request_).ok();
payload_, request_, server_->max_receive_message_size())
.ok();
bool ret = RegisteredAsyncRequest::FinalizeResult(tag, status); bool ret = RegisteredAsyncRequest::FinalizeResult(tag, status);
*status = serialization_status && *status; *status = serialization_status && *status;
return ret; return ret;

8
include/grpc++/impl/codegen/sync_stream.h
Unescape View File

@ -160,7 +160,7 @@ class ClientReader final : public ClientReaderInterface<R> {
} }
bool NextMessageSize(uint32_t* sz) override { bool NextMessageSize(uint32_t* sz) override {
*sz = call_.max_receive_message_size(); *sz = INT_MAX;
return true; return true;
} }
@ -310,7 +310,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
} }
bool NextMessageSize(uint32_t* sz) override { bool NextMessageSize(uint32_t* sz) override {
*sz = call_.max_receive_message_size(); *sz = INT_MAX;
return true; return true;
} }
@ -382,7 +382,7 @@ class ServerReader final : public ServerReaderInterface<R> {
} }
bool NextMessageSize(uint32_t* sz) override { bool NextMessageSize(uint32_t* sz) override {
*sz = call_->max_receive_message_size(); *sz = INT_MAX;
return true; return true;
} }
@ -474,7 +474,7 @@ class ServerReaderWriterBody final {
} }
bool NextMessageSize(uint32_t* sz) { bool NextMessageSize(uint32_t* sz) {
*sz = call_->max_receive_message_size(); *sz = INT_MAX;
return true; return true;
} }

3
include/grpc++/support/byte_buffer.h
Unescape View File

@ -95,8 +95,7 @@ class ByteBuffer final {
template <> template <>
class SerializationTraits<ByteBuffer, void> { class SerializationTraits<ByteBuffer, void> {
public: public:
static Status Deserialize(grpc_byte_buffer* byte_buffer, ByteBuffer* dest, static Status Deserialize(grpc_byte_buffer* byte_buffer, ByteBuffer* dest) {
int max_receive_message_size) {
dest->set_buffer(byte_buffer); dest->set_buffer(byte_buffer);
return Status::OK; return Status::OK;
} }

8
src/cpp/server/server_cc.cc
Unescape View File

@ -186,7 +186,7 @@ class Server::SyncRequest final : public CompletionQueueTag {
public: public:
explicit CallData(Server* server, SyncRequest* mrd) explicit CallData(Server* server, SyncRequest* mrd)
: cq_(mrd->cq_), : cq_(mrd->cq_),
call_(mrd->call_, server, &cq_, server->max_receive_message_size_), call_(mrd->call_, server, &cq_),
ctx_(mrd->deadline_, mrd->request_metadata_.metadata, ctx_(mrd->deadline_, mrd->request_metadata_.metadata,
mrd->request_metadata_.count), mrd->request_metadata_.count),
has_request_payload_(mrd->has_request_payload_), has_request_payload_(mrd->has_request_payload_),
@ -208,8 +208,8 @@ class Server::SyncRequest final : public CompletionQueueTag {
void Run(std::shared_ptr<GlobalCallbacks> global_callbacks) { void Run(std::shared_ptr<GlobalCallbacks> global_callbacks) {
ctx_.BeginCompletionOp(&call_); ctx_.BeginCompletionOp(&call_);
global_callbacks->PreSynchronousRequest(&ctx_); global_callbacks->PreSynchronousRequest(&ctx_);
method_->handler()->RunHandler(MethodHandler::HandlerParameter( method_->handler()->RunHandler(
&call_, &ctx_, request_payload_, call_.max_receive_message_size())); MethodHandler::HandlerParameter(&call_, &ctx_, request_payload_));
global_callbacks->PostSynchronousRequest(&ctx_); global_callbacks->PostSynchronousRequest(&ctx_);
request_payload_ = nullptr; request_payload_ = nullptr;
void* ignored_tag; void* ignored_tag;
@ -598,7 +598,7 @@ bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag,
grpc_metadata_array_destroy(&initial_metadata_array_); grpc_metadata_array_destroy(&initial_metadata_array_);
context_->set_call(call_); context_->set_call(call_);
context_->cq_ = call_cq_; context_->cq_ = call_cq_;
Call call(call_, server_, call_cq_, server_->max_receive_message_size()); Call call(call_, server_, call_cq_);
if (*status && call_) { if (*status && call_) {
context_->BeginCompletionOp(&call); context_->BeginCompletionOp(&call);
} }

Write Preview
Loading…
Cancel
Save