diff options
author | ncteisen <ncteisen@gmail.com> | 2017-05-05 10:27:17 -0700 |
---|---|---|
committer | ncteisen <ncteisen@gmail.com> | 2017-05-05 10:27:17 -0700 |
commit | fde16afefa41510c6dc06afd88bfd6c3002467b5 (patch) | |
tree | 143deed3716f5b13f9340d7f57de27379a255e37 /include/grpc++/impl/codegen | |
parent | cc7e17b2d6d015d88b94c0d3b089a65e0450614b (diff) | |
parent | 6815e414a4dbca4d0d4dd62b5ec3c6faa60c9bb9 (diff) |
Merge branch 'master' of https://github.com/grpc/grpc into serialization-refactor
Diffstat (limited to 'include/grpc++/impl/codegen')
-rw-r--r-- | include/grpc++/impl/codegen/call.h | 37 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/client_unary_call.h | 4 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/completion_queue.h | 36 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/core_codegen.h | 9 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/core_codegen_interface.h | 6 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/proto_utils.h | 2 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/sync_stream.h | 12 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/thrift_serializer.h | 217 | ||||
-rw-r--r-- | include/grpc++/impl/codegen/thrift_utils.h | 83 |
9 files changed, 53 insertions, 353 deletions
diff --git a/include/grpc++/impl/codegen/call.h b/include/grpc++/impl/codegen/call.h index f334ba61d6..9fe2bbb75e 100644 --- a/include/grpc++/impl/codegen/call.h +++ b/include/grpc++/impl/codegen/call.h @@ -364,28 +364,6 @@ class CallOpRecvMessage { bool allow_not_getting_message_; }; -namespace CallOpGenericRecvMessageHelper { -class DeserializeFunc { - public: - virtual Status Deserialize(grpc_byte_buffer* buf) = 0; - virtual ~DeserializeFunc() {} -}; - -template <class R> -class DeserializeFuncType final : public DeserializeFunc { - public: - DeserializeFuncType(R* message) : message_(message) {} - Status Deserialize(grpc_byte_buffer* buf) override { - return SerializationTraits<R>::Deserialize(buf, message_); - } - - ~DeserializeFuncType() override {} - - private: - R* message_; // Not a managed pointer because management is external to this -}; -} // namespace CallOpGenericRecvMessageHelper - class CallOpGenericRecvMessage { public: CallOpGenericRecvMessage() @@ -393,11 +371,9 @@ class CallOpGenericRecvMessage { template <class R> void RecvMessage(R* message) { - // Use an explicit base class pointer to avoid resolution error in the - // following unique_ptr::reset for some old implementations. - CallOpGenericRecvMessageHelper::DeserializeFunc* func = - new CallOpGenericRecvMessageHelper::DeserializeFuncType<R>(message); - deserialize_.reset(func); + deserialize_ = [message](grpc_byte_buffer* buf) -> Status { + return SerializationTraits<R>::Deserialize(buf, message); + }; } // Do not change status if no message is received. @@ -420,7 +396,7 @@ class CallOpGenericRecvMessage { if (recv_buf_) { if (*status) { got_message = true; - *status = deserialize_->Deserialize(recv_buf_).ok(); + *status = deserialize_(recv_buf_).ok(); } else { got_message = false; g_core_codegen_interface->grpc_byte_buffer_destroy(recv_buf_); @@ -431,11 +407,12 @@ class CallOpGenericRecvMessage { *status = false; } } - deserialize_.reset(); + deserialize_ = DeserializeFunc(); } private: - std::unique_ptr<CallOpGenericRecvMessageHelper::DeserializeFunc> deserialize_; + typedef std::function<Status(grpc_byte_buffer*)> DeserializeFunc; + DeserializeFunc deserialize_; grpc_byte_buffer* recv_buf_; bool allow_not_getting_message_; }; diff --git a/include/grpc++/impl/codegen/client_unary_call.h b/include/grpc++/impl/codegen/client_unary_call.h index a5a4f3d739..4bf35ae778 100644 --- a/include/grpc++/impl/codegen/client_unary_call.h +++ b/include/grpc++/impl/codegen/client_unary_call.h @@ -52,7 +52,9 @@ template <class InputMessage, class OutputMessage> Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method, ClientContext* context, const InputMessage& request, OutputMessage* result) { - CompletionQueue cq(true); // Pluckable completion queue + CompletionQueue cq(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, + GRPC_CQ_DEFAULT_POLLING}); // 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 61617f2bdc..c8ab726b0f 100644 --- a/include/grpc++/impl/codegen/completion_queue.h +++ b/include/grpc++/impl/codegen/completion_queue.h @@ -102,7 +102,9 @@ class CompletionQueue : private GrpcLibraryCodegen { public: /// Default constructor. Implicitly creates a \a grpc_completion_queue /// instance. - CompletionQueue() : CompletionQueue(false) {} + CompletionQueue() + : CompletionQueue(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING}) {} /// Wrap \a take, taking ownership of the instance. /// @@ -182,6 +184,16 @@ class CompletionQueue : private GrpcLibraryCodegen { }; void CompleteAvalanching(); + protected: + /// Private constructor of CompletionQueue only visible to friend classes + CompletionQueue(const grpc_completion_queue_attributes& attributes) { + cq_ = g_core_codegen_interface->grpc_completion_queue_create( + g_core_codegen_interface->grpc_completion_queue_factory_lookup( + &attributes), + &attributes, NULL); + InitialAvalanching(); // reserve this for the future shutdown + } + private: // Friend synchronous wrappers so that they can access Pluck(), which is // a semi-private API geared towards the synchronous implementation. @@ -215,18 +227,6 @@ 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. @@ -289,17 +289,19 @@ class CompletionQueue : private GrpcLibraryCodegen { /// by servers. Instantiated by \a ServerBuilder. class ServerCompletionQueue : public CompletionQueue { public: - bool IsFrequentlyPolled() { return is_frequently_polled_; } + bool IsFrequentlyPolled() { return polling_type_ != GRPC_CQ_NON_LISTENING; } private: - bool is_frequently_polled_; + grpc_cq_polling_type polling_type_; friend class ServerBuilder; /// \param is_frequently_polled Informs the GRPC library about whether the /// server completion queue would be actively polled (by calling Next() or /// AsyncNext()). By default all server completion queues are assumed to be /// frequently polled. - ServerCompletionQueue(bool is_frequently_polled = true) - : is_frequently_polled_(is_frequently_polled) {} + ServerCompletionQueue(grpc_cq_polling_type polling_type) + : CompletionQueue(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type}), + polling_type_(polling_type) {} }; } // namespace grpc diff --git a/include/grpc++/impl/codegen/core_codegen.h b/include/grpc++/impl/codegen/core_codegen.h index 35dc9b859d..7a1236a716 100644 --- a/include/grpc++/impl/codegen/core_codegen.h +++ b/include/grpc++/impl/codegen/core_codegen.h @@ -44,8 +44,15 @@ namespace grpc { /// Implementation of the core codegen interface. -class CoreCodegen : public CoreCodegenInterface { +class CoreCodegen final : public CoreCodegenInterface { private: + virtual const grpc_completion_queue_factory* + grpc_completion_queue_factory_lookup( + const grpc_completion_queue_attributes* attributes) override; + virtual grpc_completion_queue* grpc_completion_queue_create( + const grpc_completion_queue_factory* factory, + const grpc_completion_queue_attributes* attributes, + void* reserved) override; grpc_completion_queue* grpc_completion_queue_create_for_next( void* reserved) override; grpc_completion_queue* grpc_completion_queue_create_for_pluck( diff --git a/include/grpc++/impl/codegen/core_codegen_interface.h b/include/grpc++/impl/codegen/core_codegen_interface.h index e30819e564..4ac37f5255 100644 --- a/include/grpc++/impl/codegen/core_codegen_interface.h +++ b/include/grpc++/impl/codegen/core_codegen_interface.h @@ -59,6 +59,12 @@ class CoreCodegenInterface { virtual void assert_fail(const char* failed_assertion, const char* file, int line) = 0; + virtual const grpc_completion_queue_factory* + grpc_completion_queue_factory_lookup( + const grpc_completion_queue_attributes* attributes) = 0; + virtual grpc_completion_queue* grpc_completion_queue_create( + const grpc_completion_queue_factory* factory, + const grpc_completion_queue_attributes* attributes, void* reserved) = 0; virtual grpc_completion_queue* grpc_completion_queue_create_for_next( void* reserved) = 0; virtual grpc_completion_queue* grpc_completion_queue_create_for_pluck( diff --git a/include/grpc++/impl/codegen/proto_utils.h b/include/grpc++/impl/codegen/proto_utils.h index 491b8eb943..a6e10cc8b6 100644 --- a/include/grpc++/impl/codegen/proto_utils.h +++ b/include/grpc++/impl/codegen/proto_utils.h @@ -52,7 +52,7 @@ namespace internal { class GrpcBufferWriterPeer; -const int kGrpcBufferWriterMaxBufferLength = 8192; +const int kGrpcBufferWriterMaxBufferLength = 1024 * 1024; class GrpcBufferWriter : public ::grpc::protobuf::io::ZeroCopyOutputStream { public: diff --git a/include/grpc++/impl/codegen/sync_stream.h b/include/grpc++/impl/codegen/sync_stream.h index 328d5cb1e8..a010924cef 100644 --- a/include/grpc++/impl/codegen/sync_stream.h +++ b/include/grpc++/impl/codegen/sync_stream.h @@ -156,7 +156,9 @@ class ClientReader final : public ClientReaderInterface<R> { ClientReader(ChannelInterface* channel, const RpcMethod& method, ClientContext* context, const W& request) : context_(context), - cq_(true), // Pluckable cq + cq_(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, + GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose> @@ -230,7 +232,9 @@ class ClientWriter : public ClientWriterInterface<W> { ClientWriter(ChannelInterface* channel, const RpcMethod& method, ClientContext* context, R* response) : context_(context), - cq_(true), // Pluckable cq + cq_(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, + GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { finish_ops_.RecvMessage(response); finish_ops_.AllowNoMessage(); @@ -330,7 +334,9 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> { ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method, ClientContext* context) : context_(context), - cq_(true), // Pluckable cq + cq_(grpc_completion_queue_attributes{ + GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, + GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq call_(channel->CreateCall(method, context, &cq_)) { if (!context_->initial_metadata_corked_) { CallOpSet<CallOpSendInitialMetadata> ops; diff --git a/include/grpc++/impl/codegen/thrift_serializer.h b/include/grpc++/impl/codegen/thrift_serializer.h deleted file mode 100644 index 86bc7105c0..0000000000 --- a/include/grpc++/impl/codegen/thrift_serializer.h +++ /dev/null @@ -1,217 +0,0 @@ -/* - * - * Copyright 2016, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ - -#ifndef GRPCXX_IMPL_CODEGEN_THRIFT_SERIALIZER_H -#define GRPCXX_IMPL_CODEGEN_THRIFT_SERIALIZER_H - -#include <grpc/impl/codegen/byte_buffer_reader.h> -#include <grpc/impl/codegen/slice.h> -#include <thrift/protocol/TBinaryProtocol.h> -#include <thrift/protocol/TCompactProtocol.h> -#include <thrift/protocol/TProtocolException.h> -#include <thrift/transport/TBufferTransports.h> -#include <thrift/transport/TTransportUtils.h> -#include <boost/make_shared.hpp> -#include <memory> -#include <stdexcept> -#include <string> - -namespace apache { -namespace thrift { -namespace util { - -using apache::thrift::protocol::TBinaryProtocolT; -using apache::thrift::protocol::TCompactProtocolT; -using apache::thrift::protocol::TMessageType; -using apache::thrift::protocol::TNetworkBigEndian; -using apache::thrift::transport::TMemoryBuffer; -using apache::thrift::transport::TBufferBase; -using apache::thrift::transport::TTransport; - -template <typename Dummy, typename Protocol> -class ThriftSerializer { - public: - ThriftSerializer() - : prepared_(false), - last_deserialized_(false), - serialize_version_(false) {} - - virtual ~ThriftSerializer() {} - - // Serialize the passed type into the internal buffer - // and returns a pointer to internal buffer and its size - template <typename T> - void Serialize(const T& fields, const uint8_t** serialized_buffer, - size_t* serialized_len) { - // prepare or reset buffer - if (!prepared_ || last_deserialized_) { - prepare(); - } else { - buffer_->resetBuffer(); - } - last_deserialized_ = false; - - // if required serialize protocol version - if (serialize_version_) { - protocol_->writeMessageBegin("", TMessageType(0), 0); - } - - // serialize fields into buffer - fields.write(protocol_.get()); - - // write the end of message - if (serialize_version_) { - protocol_->writeMessageEnd(); - } - - uint8_t* byte_buffer; - uint32_t byte_buffer_size; - buffer_->getBuffer(&byte_buffer, &byte_buffer_size); - *serialized_buffer = byte_buffer; - *serialized_len = byte_buffer_size; - } - - // Serialize the passed type into the byte buffer - template <typename T> - void Serialize(const T& fields, grpc_byte_buffer** bp) { - const uint8_t* byte_buffer; - size_t byte_buffer_size; - - Serialize(fields, &byte_buffer, &byte_buffer_size); - - grpc_slice slice = grpc_slice_from_copied_buffer( - reinterpret_cast<const char*>(byte_buffer), byte_buffer_size); - - *bp = grpc_raw_byte_buffer_create(&slice, 1); - - grpc_slice_unref(slice); - } - - // Deserialize the passed char array into the passed type, returns the number - // of bytes that have been consumed from the passed string. - template <typename T> - uint32_t Deserialize(uint8_t* serialized_buffer, size_t length, T* fields) { - // prepare buffer if necessary - if (!prepared_) { - prepare(); - } - last_deserialized_ = true; - - // reset buffer transport - buffer_->resetBuffer(serialized_buffer, length); - - // read the protocol version if necessary - if (serialize_version_) { - std::string name = ""; - TMessageType mt = static_cast<TMessageType>(0); - int32_t seq_id = 0; - protocol_->readMessageBegin(name, mt, seq_id); - } - - // deserialize buffer into fields - uint32_t len = fields->read(protocol_.get()); - - // read the end of message - if (serialize_version_) { - protocol_->readMessageEnd(); - } - - return len; - } - - // Deserialize the passed byte buffer to passed type, returns the number - // of bytes consumed from byte buffer - template <typename T> - uint32_t Deserialize(grpc_byte_buffer* buffer, T* msg) { - grpc_byte_buffer_reader reader; - grpc_byte_buffer_reader_init(&reader, buffer); - - grpc_slice slice = grpc_byte_buffer_reader_readall(&reader); - - uint32_t len = - Deserialize(GRPC_SLICE_START_PTR(slice), GRPC_SLICE_LENGTH(slice), msg); - - grpc_slice_unref(slice); - - grpc_byte_buffer_reader_destroy(&reader); - - return len; - } - - // set serialization version flag - void SetSerializeVersion(bool value) { serialize_version_ = value; } - - // Set the container size limit to deserialize - // This function should be called after buffer_ is initialized - void SetContainerSizeLimit(int32_t container_limit) { - if (!prepared_) { - prepare(); - } - protocol_->setContainerSizeLimit(container_limit); - } - - // Set the string size limit to deserialize - // This function should be called after buffer_ is initialized - void SetStringSizeLimit(int32_t string_limit) { - if (!prepared_) { - prepare(); - } - protocol_->setStringSizeLimit(string_limit); - } - - private: - bool prepared_; - bool last_deserialized_; - boost::shared_ptr<TMemoryBuffer> buffer_; - std::shared_ptr<Protocol> protocol_; - bool serialize_version_; - - void prepare() { - buffer_ = boost::make_shared<TMemoryBuffer>(); - // create a protocol for the memory buffer transport - protocol_ = std::make_shared<Protocol>(buffer_); - prepared_ = true; - } - -}; // ThriftSerializer - -typedef ThriftSerializer<void, TBinaryProtocolT<TBufferBase, TNetworkBigEndian>> - ThriftSerializerBinary; -typedef ThriftSerializer<void, TCompactProtocolT<TBufferBase>> - ThriftSerializerCompact; - -} // namespace util -} // namespace thrift -} // namespace apache - -#endif // GRPCXX_IMPL_CODEGEN_THRIFT_SERIALIZER_H diff --git a/include/grpc++/impl/codegen/thrift_utils.h b/include/grpc++/impl/codegen/thrift_utils.h deleted file mode 100644 index 742d739703..0000000000 --- a/include/grpc++/impl/codegen/thrift_utils.h +++ /dev/null @@ -1,83 +0,0 @@ -/* - * - * Copyright 2016, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ - -#ifndef GRPCXX_IMPL_CODEGEN_THRIFT_UTILS_H -#define GRPCXX_IMPL_CODEGEN_THRIFT_UTILS_H - -#include <grpc++/impl/codegen/config.h> -#include <grpc++/impl/codegen/core_codegen_interface.h> -#include <grpc++/impl/codegen/serialization_traits.h> -#include <grpc++/impl/codegen/status.h> -#include <grpc++/impl/codegen/status_code_enum.h> -#include <grpc++/impl/codegen/thrift_serializer.h> -#include <grpc/impl/codegen/byte_buffer_reader.h> -#include <grpc/impl/codegen/slice.h> -#include <cstdint> -#include <cstdlib> - -namespace grpc { - -using apache::thrift::util::ThriftSerializerCompact; - -template <class T> -class SerializationTraits<T, typename std::enable_if<std::is_base_of< - apache::thrift::TBase, T>::value>::type> { - public: - static Status Serialize(const T& msg, grpc_byte_buffer** bp, - bool* own_buffer) { - *own_buffer = true; - - ThriftSerializerCompact serializer; - serializer.Serialize(msg, bp); - - return Status(StatusCode::OK, "ok"); - } - - static Status Deserialize(grpc_byte_buffer* buffer, T* msg, - int max_receive_message_size) { - if (!buffer) { - return Status(StatusCode::INTERNAL, "No payload"); - } - - ThriftSerializerCompact deserializer; - deserializer.Deserialize(buffer, msg); - - grpc_byte_buffer_destroy(buffer); - - return Status(StatusCode::OK, "ok"); - } -}; - -} // namespace grpc - -#endif // GRPCXX_IMPL_CODEGEN_THRIFT_UTILS_H |