From efce6e1e5017e8a6c6a81f348a4adcc4825b757d Mon Sep 17 00:00:00 2001 From: Vijay Pai Date: Thu, 14 Sep 2017 09:07:50 -0700 Subject: Revert "Revert "Allow SerializationTraits to use grpc::ByteBuffer rather than only grpc_byte_buffer"" --- include/grpc++/impl/codegen/byte_buffer.h | 141 +++++++++++++++++++++ include/grpc++/impl/codegen/call.h | 114 +++++++++++++---- include/grpc++/impl/codegen/method_handler_impl.h | 9 +- include/grpc++/impl/codegen/rpc_service_method.h | 14 +- include/grpc++/impl/codegen/serialization_traits.h | 25 ++-- include/grpc++/impl/codegen/slice.h | 78 ++++++++++++ 6 files changed, 338 insertions(+), 43 deletions(-) create mode 100644 include/grpc++/impl/codegen/byte_buffer.h (limited to 'include/grpc++/impl') diff --git a/include/grpc++/impl/codegen/byte_buffer.h b/include/grpc++/impl/codegen/byte_buffer.h new file mode 100644 index 0000000000..87d390c688 --- /dev/null +++ b/include/grpc++/impl/codegen/byte_buffer.h @@ -0,0 +1,141 @@ +/* + * + * Copyright 2017 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. + * + */ + +#ifndef GRPCXX_IMPL_CODEGEN_BYTE_BUFFER_H +#define GRPCXX_IMPL_CODEGEN_BYTE_BUFFER_H + +#include + +#include +#include +#include +#include +#include + +#include + +namespace grpc { + +template +class CallOpRecvMessage; +class MethodHandler; +namespace internal { +template +class MessageDeserializer; +} + +/// A sequence of bytes. +class ByteBuffer final { + public: + /// Constuct an empty buffer. + ByteBuffer() : buffer_(nullptr) {} + + /// Construct buffer from \a slices, of which there are \a nslices. + ByteBuffer(const Slice* slices, size_t nslices); + + /// Constuct a byte buffer by referencing elements of existing buffer + /// \a buf. Wrapper of core function grpc_byte_buffer_copy + ByteBuffer(const ByteBuffer& buf); + + ~ByteBuffer() { + if (buffer_) { + g_core_codegen_interface->grpc_byte_buffer_destroy(buffer_); + } + } + + ByteBuffer& operator=(const ByteBuffer&); + + /// Dump (read) the buffer contents into \a slices. + Status Dump(std::vector* slices) const; + + /// Remove all data. + void Clear() { + if (buffer_) { + g_core_codegen_interface->grpc_byte_buffer_destroy(buffer_); + buffer_ = nullptr; + } + } + + /// Make a duplicate copy of the internals of this byte + /// buffer so that we have our own owned version of it. + /// bbuf.Duplicate(); is equivalent to bbuf=bbuf; but is actually readable + void Duplicate() { + buffer_ = g_core_codegen_interface->grpc_byte_buffer_copy(buffer_); + } + + /// Forget underlying byte buffer without destroying + /// Use this only for un-owned byte buffers + void Release() { buffer_ = nullptr; } + + /// Buffer size in bytes. + size_t Length() const; + + /// Swap the state of *this and *other. + void Swap(ByteBuffer* other); + + /// Is this ByteBuffer valid? + bool Valid() const { return (buffer_ != nullptr); } + + private: + friend class SerializationTraits; + friend class CallOpSendMessage; + template + friend class CallOpRecvMessage; + friend class CallOpGenericRecvMessage; + friend class MethodHandler; + template + friend class internal::MessageDeserializer; + + // takes ownership + void set_buffer(grpc_byte_buffer* buf) { + if (buffer_) { + Clear(); + } + buffer_ = buf; + } + + grpc_byte_buffer* c_buffer() { return buffer_; } + grpc_byte_buffer** c_buffer_ptr() { return &buffer_; } + + // DEPRECATED: Implicit conversion to transparently + // support deprecated SerializationTraits API + // No need to inline since deprecated + operator grpc_byte_buffer*(); + operator const grpc_byte_buffer*() const; + + grpc_byte_buffer* buffer_; +}; + +template <> +class SerializationTraits { + public: + static Status Deserialize(const ByteBuffer& byte_buffer, ByteBuffer* dest) { + dest->set_buffer(byte_buffer.buffer_); + return Status::OK; + } + static Status Serialize(const ByteBuffer& source, ByteBuffer* buffer, + bool* own_buffer) { + *buffer = source; + *own_buffer = true; + return Status::OK; + } +}; + +} // namespace grpc + +#endif // GRPCXX_IMPL_CODEGEN_BYTE_BUFFER_H diff --git a/include/grpc++/impl/codegen/call.h b/include/grpc++/impl/codegen/call.h index 74ed5cbfb9..3c418c7ae2 100644 --- a/include/grpc++/impl/codegen/call.h +++ b/include/grpc++/impl/codegen/call.h @@ -25,6 +25,7 @@ #include #include +#include #include #include #include @@ -39,8 +40,6 @@ #include #include -struct grpc_byte_buffer; - namespace grpc { class ByteBuffer; @@ -281,7 +280,7 @@ class CallOpSendInitialMetadata { class CallOpSendMessage { public: - CallOpSendMessage() : send_buf_(nullptr) {} + CallOpSendMessage() : send_buf_() {} /// Send \a message using \a options for the write. The \a options are cleared /// after use. @@ -294,33 +293,67 @@ class CallOpSendMessage { protected: void AddOp(grpc_op* ops, size_t* nops) { - if (send_buf_ == nullptr) return; + if (!send_buf_.Valid()) return; grpc_op* op = &ops[(*nops)++]; op->op = GRPC_OP_SEND_MESSAGE; op->flags = write_options_.flags(); op->reserved = NULL; - op->data.send_message.send_message = send_buf_; + op->data.send_message.send_message = send_buf_.c_buffer(); // Flags are per-message: clear them after use. write_options_.Clear(); } - void FinishOp(bool* status) { - g_core_codegen_interface->grpc_byte_buffer_destroy(send_buf_); - send_buf_ = nullptr; - } + void FinishOp(bool* status) { send_buf_.Clear(); } private: - grpc_byte_buffer* send_buf_; + template + class MessageSerializer; + + ByteBuffer send_buf_; WriteOptions write_options_; }; +namespace internal { +template +T Example(); +} // namespace internal + +template +class CallOpSendMessage::MessageSerializer< + M, typename std::enable_if::Serialize( + internal::Example(), + internal::Example(), + internal::Example()))>::value>::type> { + public: + static Status SendMessageInternal(const M& message, ByteBuffer* bbuf, + bool* own_buf) { + return SerializationTraits::Serialize(message, bbuf->c_buffer_ptr(), + own_buf); + } +}; + +template +class CallOpSendMessage::MessageSerializer< + M, typename std::enable_if::Serialize( + internal::Example(), + internal::Example<::grpc::ByteBuffer*>(), + internal::Example()))>::value>::type> { + public: + static Status SendMessageInternal(const M& message, ByteBuffer* bbuf, + bool* own_buf) { + return SerializationTraits::Serialize(message, bbuf, own_buf); + } +}; + template Status CallOpSendMessage::SendMessage(const M& message, WriteOptions options) { write_options_ = options; bool own_buf; Status result = - SerializationTraits::Serialize(message, &send_buf_, &own_buf); + MessageSerializer::SendMessageInternal(message, &send_buf_, &own_buf); if (!own_buf) { - send_buf_ = g_core_codegen_interface->grpc_byte_buffer_copy(send_buf_); + send_buf_.Duplicate(); } return result; } @@ -330,6 +363,36 @@ Status CallOpSendMessage::SendMessage(const M& message) { return SendMessage(message, WriteOptions()); } +namespace internal { +template +class MessageDeserializer; + +template +class MessageDeserializer< + M, typename std::enable_if::Deserialize( + internal::Example(), + internal::Example()))>::value>::type> { + public: + static Status Deserialize(const ByteBuffer& bbuf, M* message) { + return SerializationTraits::Deserialize(bbuf, message); + } +}; + +template +class MessageDeserializer< + M, typename std::enable_if::Deserialize( + internal::Example(), + internal::Example()))>::value>::type> { + public: + static Status Deserialize(const ByteBuffer& bbuf, M* message) { + return SerializationTraits::Deserialize( + const_cast(bbuf).c_buffer(), message); + } +}; +} // namespace internal + template class CallOpRecvMessage { public: @@ -352,18 +415,20 @@ class CallOpRecvMessage { op->op = GRPC_OP_RECV_MESSAGE; op->flags = 0; op->reserved = NULL; - op->data.recv_message.recv_message = &recv_buf_; + op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr(); } void FinishOp(bool* status) { if (message_ == nullptr) return; - if (recv_buf_) { + if (recv_buf_.Valid()) { if (*status) { got_message = *status = - SerializationTraits::Deserialize(recv_buf_, message_).ok(); + internal::MessageDeserializer::Deserialize(recv_buf_, message_) + .ok(); + recv_buf_.Release(); } else { got_message = false; - g_core_codegen_interface->grpc_byte_buffer_destroy(recv_buf_); + recv_buf_.Clear(); } } else { got_message = false; @@ -376,14 +441,14 @@ class CallOpRecvMessage { private: R* message_; - grpc_byte_buffer* recv_buf_; + ByteBuffer recv_buf_; bool allow_not_getting_message_; }; namespace CallOpGenericRecvMessageHelper { class DeserializeFunc { public: - virtual Status Deserialize(grpc_byte_buffer* buf) = 0; + virtual Status Deserialize(const ByteBuffer& buf) = 0; virtual ~DeserializeFunc() {} }; @@ -391,8 +456,8 @@ template class DeserializeFuncType final : public DeserializeFunc { public: DeserializeFuncType(R* message) : message_(message) {} - Status Deserialize(grpc_byte_buffer* buf) override { - return SerializationTraits::Deserialize(buf, message_); + Status Deserialize(const ByteBuffer& buf) override { + return grpc::internal::MessageDeserializer::Deserialize(buf, message_); } ~DeserializeFuncType() override {} @@ -428,18 +493,19 @@ class CallOpGenericRecvMessage { op->op = GRPC_OP_RECV_MESSAGE; op->flags = 0; op->reserved = NULL; - op->data.recv_message.recv_message = &recv_buf_; + op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr(); } void FinishOp(bool* status) { if (!deserialize_) return; - if (recv_buf_) { + if (recv_buf_.Valid()) { if (*status) { got_message = true; *status = deserialize_->Deserialize(recv_buf_).ok(); + recv_buf_.Release(); } else { got_message = false; - g_core_codegen_interface->grpc_byte_buffer_destroy(recv_buf_); + recv_buf_.Clear(); } } else { got_message = false; @@ -452,7 +518,7 @@ class CallOpGenericRecvMessage { private: std::unique_ptr deserialize_; - grpc_byte_buffer* recv_buf_; + ByteBuffer recv_buf_; bool allow_not_getting_message_; }; diff --git a/include/grpc++/impl/codegen/method_handler_impl.h b/include/grpc++/impl/codegen/method_handler_impl.h index 15e24bdcdc..8125e0a651 100644 --- a/include/grpc++/impl/codegen/method_handler_impl.h +++ b/include/grpc++/impl/codegen/method_handler_impl.h @@ -19,6 +19,7 @@ #ifndef GRPCXX_IMPL_CODEGEN_METHOD_HANDLER_IMPL_H #define GRPCXX_IMPL_CODEGEN_METHOD_HANDLER_IMPL_H +#include #include #include #include @@ -37,8 +38,8 @@ class RpcMethodHandler : public MethodHandler { void RunHandler(const HandlerParameter& param) final { RequestType req; - Status status = - SerializationTraits::Deserialize(param.request, &req); + Status status = internal::MessageDeserializer::Deserialize( + param.request, &req); ResponseType rsp; if (status.ok()) { status = func_(service_, param.server_context, &req, &rsp); @@ -123,8 +124,8 @@ class ServerStreamingHandler : public MethodHandler { void RunHandler(const HandlerParameter& param) final { RequestType req; - Status status = - SerializationTraits::Deserialize(param.request, &req); + Status status = internal::MessageDeserializer::Deserialize( + param.request, &req); if (status.ok()) { ServerWriter writer(param.call, param.server_context); diff --git a/include/grpc++/impl/codegen/rpc_service_method.h b/include/grpc++/impl/codegen/rpc_service_method.h index 7165774172..d356012ad6 100644 --- a/include/grpc++/impl/codegen/rpc_service_method.h +++ b/include/grpc++/impl/codegen/rpc_service_method.h @@ -25,14 +25,11 @@ #include #include +#include #include #include #include -extern "C" { -struct grpc_byte_buffer; -} - namespace grpc { class ServerContext; class StreamContextInterface; @@ -43,11 +40,14 @@ class MethodHandler { virtual ~MethodHandler() {} struct HandlerParameter { HandlerParameter(Call* c, ServerContext* context, grpc_byte_buffer* req) - : call(c), server_context(context), request(req) {} + : call(c), server_context(context) { + request.set_buffer(req); + } + ~HandlerParameter() { request.Release(); } Call* call; ServerContext* server_context; - // Handler required to grpc_byte_buffer_destroy this - grpc_byte_buffer* request; + // Handler required to destroy these contents + ByteBuffer request; }; virtual void RunHandler(const HandlerParameter& param) = 0; }; diff --git a/include/grpc++/impl/codegen/serialization_traits.h b/include/grpc++/impl/codegen/serialization_traits.h index b72d474126..7fc4e24172 100644 --- a/include/grpc++/impl/codegen/serialization_traits.h +++ b/include/grpc++/impl/codegen/serialization_traits.h @@ -26,15 +26,24 @@ namespace grpc { /// Used for hooking different message serialization API's into GRPC. /// Each SerializationTraits implementation must provide the following /// functions: -/// static Status Serialize(const Message& msg, -/// grpc_byte_buffer** buffer, -/// bool* own_buffer); -/// static Status Deserialize(grpc_byte_buffer* buffer, -/// Message* msg, -/// int max_receive_message_size); +/// 1. static Status Serialize(const Message& msg, +/// ByteBuffer* buffer, +/// bool* own_buffer); +/// AND/OR +/// static Status Serialize(const Message& msg, +/// grpc_byte_buffer** buffer, +/// bool* own_buffer); +/// The former is preferred; the latter is deprecated /// -/// Serialize is required to convert message to a grpc_byte_buffer, and -/// to store a pointer to that byte buffer at *buffer. *own_buffer should +/// 2. static Status Deserialize(const ByteBuffer& buffer, +/// Message* msg); +/// AND/OR +/// static Status Deserialize(grpc_byte_buffer* buffer, +/// Message* msg); +/// The former is preferred; the latter is deprecated +/// +/// Serialize is required to convert message to a ByteBuffer, and +/// return that byte buffer through *buffer. *own_buffer should /// be set to true if the caller owns said byte buffer, or false if /// ownership is retained elsewhere. /// diff --git a/include/grpc++/impl/codegen/slice.h b/include/grpc++/impl/codegen/slice.h index e682bdef6a..c185bf4fd0 100644 --- a/include/grpc++/impl/codegen/slice.h +++ b/include/grpc++/impl/codegen/slice.h @@ -19,11 +19,89 @@ #ifndef GRPCXX_IMPL_CODEGEN_SLICE_H #define GRPCXX_IMPL_CODEGEN_SLICE_H +#include #include #include +#include + namespace grpc { +/// A wrapper around \a grpc_slice. +/// +/// A slice represents a contiguous reference counted array of bytes. +/// It is cheap to take references to a slice, and it is cheap to create a +/// slice pointing to a subset of another slice. +class Slice final { + public: + /// Construct an empty slice. + Slice(); + /// Destructor - drops one reference. + ~Slice(); + + enum AddRef { ADD_REF }; + /// Construct a slice from \a slice, adding a reference. + Slice(grpc_slice slice, AddRef); + + enum StealRef { STEAL_REF }; + /// Construct a slice from \a slice, stealing a reference. + Slice(grpc_slice slice, StealRef); + + /// Allocate a slice of specified size + Slice(size_t len); + + /// Construct a slice from a copied buffer + Slice(const void* buf, size_t len); + + /// Construct a slice from a copied string + Slice(const grpc::string& str); + + enum StaticSlice { STATIC_SLICE }; + + /// Construct a slice from a static buffer + Slice(const void* buf, size_t len, StaticSlice); + + /// Copy constructor, adds a reference. + Slice(const Slice& other); + + /// Assignment, reference count is unchanged. + Slice& operator=(Slice other) { + std::swap(slice_, other.slice_); + return *this; + } + + /// Create a slice pointing at some data. Calls malloc to allocate a refcount + /// for the object, and arranges that destroy will be called with the + /// user data pointer passed in at destruction. Can be the same as buf or + /// different (e.g., if data is part of a larger structure that must be + /// destroyed when the data is no longer needed) + Slice(void* buf, size_t len, void (*destroy)(void*), void* user_data); + + /// Specialization of above for common case where buf == user_data + Slice(void* buf, size_t len, void (*destroy)(void*)) + : Slice(buf, len, destroy, buf) {} + + /// Similar to the above but has a destroy that also takes slice length + Slice(void* buf, size_t len, void (*destroy)(void*, size_t)); + + /// Byte size. + size_t size() const { return GRPC_SLICE_LENGTH(slice_); } + + /// Raw pointer to the beginning (first element) of the slice. + const uint8_t* begin() const { return GRPC_SLICE_START_PTR(slice_); } + + /// Raw pointer to the end (one byte \em past the last element) of the slice. + const uint8_t* end() const { return GRPC_SLICE_END_PTR(slice_); } + + /// Raw C slice. Caller needs to call grpc_slice_unref when done. + grpc_slice c_slice() const; + + private: + friend class ByteBuffer; + + grpc_slice slice_; +}; + inline grpc::string_ref StringRefFromSlice(const grpc_slice* slice) { return grpc::string_ref( reinterpret_cast(GRPC_SLICE_START_PTR(*slice)), -- cgit v1.2.3