diff options
Diffstat (limited to 'include')
| -rw-r--r-- | include/grpc++/generic/generic_stub.h | 17 | ||||
| -rw-r--r-- | include/grpc++/impl/codegen/async_stream.h | 136 | ||||
| -rw-r--r-- | include/grpc++/impl/codegen/async_unary_call.h | 41 | ||||
| -rw-r--r-- | include/grpc++/impl/codegen/byte_buffer.h | 156 | ||||
| -rw-r--r-- | include/grpc++/impl/codegen/call.h | 65 | ||||
| -rw-r--r-- | include/grpc++/impl/codegen/method_handler_impl.h | 9 | ||||
| -rw-r--r-- | include/grpc++/impl/codegen/rpc_service_method.h | 14 | ||||
| -rw-r--r-- | include/grpc++/impl/codegen/serialization_traits.h | 29 | ||||
| -rw-r--r-- | include/grpc++/impl/codegen/slice.h | 78 | ||||
| -rw-r--r-- | include/grpc++/server_builder.h | 6 | ||||
| -rw-r--r-- | include/grpc++/support/byte_buffer.h | 68 | ||||
| -rw-r--r-- | include/grpc++/support/slice.h | 80 | ||||
| -rw-r--r-- | include/grpc/byte_buffer.h | 64 | ||||
| -rw-r--r-- | include/grpc/compression.h | 4 | ||||
| -rw-r--r-- | include/grpc/impl/codegen/byte_buffer.h | 86 | ||||
| -rw-r--r-- | include/grpc/impl/codegen/grpc_types.h | 20 | ||||
| -rw-r--r-- | include/grpc/impl/codegen/port_platform.h | 11 | ||||
| -rw-r--r-- | include/grpc/slice.h | 6 |
18 files changed, 573 insertions, 317 deletions
diff --git a/include/grpc++/generic/generic_stub.h b/include/grpc++/generic/generic_stub.h index 7b269d0887..d5064318cf 100644 --- a/include/grpc++/generic/generic_stub.h +++ b/include/grpc++/generic/generic_stub.h @@ -20,6 +20,7 @@ #define GRPCXX_GENERIC_GENERIC_STUB_H #include <grpc++/support/async_stream.h> +#include <grpc++/support/async_unary_call.h> #include <grpc++/support/byte_buffer.h> namespace grpc { @@ -27,6 +28,7 @@ namespace grpc { class CompletionQueue; typedef ClientAsyncReaderWriter<ByteBuffer, ByteBuffer> GenericClientAsyncReaderWriter; +typedef ClientAsyncResponseReader<ByteBuffer> GenericClientAsyncResponseReader; /// Generic stubs provide a type-unsafe interface to call gRPC methods /// by name. @@ -44,6 +46,21 @@ class GenericStub final { ClientContext* context, const grpc::string& method, CompletionQueue* cq, void* tag); + /// Setup a call to a named method \a method using \a context, but don't + /// start it. Let it be started explicitly with StartCall and a tag. + /// The return value only indicates whether or not registration of the call + /// succeeded (i.e. the call won't proceed if the return value is nullptr). + std::unique_ptr<GenericClientAsyncReaderWriter> PrepareCall( + ClientContext* context, const grpc::string& method, CompletionQueue* cq); + + /// Setup a unary call to a named method \a method using \a context, and don't + /// start it. Let it be started explicitly with StartCall. + /// The return value only indicates whether or not registration of the call + /// succeeded (i.e. the call won't proceed if the return value is nullptr). + std::unique_ptr<GenericClientAsyncResponseReader> PrepareUnaryCall( + ClientContext* context, const grpc::string& method, + const ByteBuffer& request, CompletionQueue* cq); + private: std::shared_ptr<ChannelInterface> channel_; }; diff --git a/include/grpc++/impl/codegen/async_stream.h b/include/grpc++/impl/codegen/async_stream.h index 9cf7ac30dd..e60572fc93 100644 --- a/include/grpc++/impl/codegen/async_stream.h +++ b/include/grpc++/impl/codegen/async_stream.h @@ -35,6 +35,11 @@ class ClientAsyncStreamingInterface { public: virtual ~ClientAsyncStreamingInterface() {} + /// Start the call that was set up by the constructor, but only if the + /// constructor was invoked through the "Prepare" API which doesn't actually + /// start the call + virtual void StartCall(void* tag) = 0; + /// Request notification of the reading of the initial metadata. Completion /// will be notified by \a tag on the associated completion queue. /// This call is optional, but if it is used, it cannot be used concurrently @@ -156,20 +161,22 @@ class ClientAsyncReaderInterface : public ClientAsyncStreamingInterface, template <class R> class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { public: - /// Create a stream and write the first request out. + /// Create a stream object. + /// Write the first request out if \a start is set. /// \a tag will be notified on \a cq when the call has been started and - /// \a request has been written out. + /// \a request has been written out. If \a start is not set, \a tag must be + /// nullptr and the actual call must be initiated by StartCall /// Note that \a context will be used to fill in custom initial metadata /// used to send to the server when starting the call. template <class W> static ClientAsyncReader* Create(ChannelInterface* channel, CompletionQueue* cq, const RpcMethod& method, ClientContext* context, const W& request, - void* tag) { + bool start, void* tag) { Call call = channel->CreateCall(method, context, cq); return new (g_core_codegen_interface->grpc_call_arena_alloc( call.call(), sizeof(ClientAsyncReader))) - ClientAsyncReader(call, context, request, tag); + ClientAsyncReader(call, context, request, start, tag); } // always allocated against a call arena, no memory free required @@ -177,6 +184,12 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { assert(size == sizeof(ClientAsyncReader)); } + void StartCall(void* tag) override { + assert(!started_); + started_ = true; + StartCallInternal(tag); + } + /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata /// method for semantics. /// @@ -186,6 +199,7 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { /// calling code can access the received metadata through the /// \a ClientContext. void ReadInitialMetadata(void* tag) override { + assert(started_); GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); meta_ops_.set_output_tag(tag); @@ -194,6 +208,7 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { } void Read(R* msg, void* tag) override { + assert(started_); read_ops_.set_output_tag(tag); if (!context_->initial_metadata_received_) { read_ops_.RecvInitialMetadata(context_); @@ -208,6 +223,7 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { /// - the \a ClientContext associated with this call is updated with /// possible initial and trailing metadata received from the server. void Finish(Status* status, void* tag) override { + assert(started_); finish_ops_.set_output_tag(tag); if (!context_->initial_metadata_received_) { finish_ops_.RecvInitialMetadata(context_); @@ -219,19 +235,28 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { private: template <class W> ClientAsyncReader(Call call, ClientContext* context, const W& request, - void* tag) - : context_(context), call_(call) { - init_ops_.set_output_tag(tag); - init_ops_.SendInitialMetadata(context->send_initial_metadata_, - context->initial_metadata_flags()); + bool start, void* tag) + : context_(context), call_(call), started_(start) { // TODO(ctiller): don't assert GPR_CODEGEN_ASSERT(init_ops_.SendMessage(request).ok()); init_ops_.ClientSendClose(); + if (start) { + StartCallInternal(tag); + } else { + assert(tag == nullptr); + } + } + + void StartCallInternal(void* tag) { + init_ops_.SendInitialMetadata(context_->send_initial_metadata_, + context_->initial_metadata_flags()); + init_ops_.set_output_tag(tag); call_.PerformOps(&init_ops_); } ClientContext* context_; Call call_; + bool started_; CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose> init_ops_; CallOpSet<CallOpRecvInitialMetadata> meta_ops_; @@ -257,9 +282,12 @@ class ClientAsyncWriterInterface : public ClientAsyncStreamingInterface, template <class W> class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { public: - /// Create a stream and write the first request out. + /// Create a stream object. + /// Start the RPC if \a start is set /// \a tag will be notified on \a cq when the call has been started (i.e. /// intitial metadata sent) and \a request has been written out. + /// If \a start is not set, \a tag must be nullptr and the actual call + /// must be initiated by StartCall /// Note that \a context will be used to fill in custom initial metadata /// used to send to the server when starting the call. /// \a response will be filled in with the single expected response @@ -269,11 +297,11 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { static ClientAsyncWriter* Create(ChannelInterface* channel, CompletionQueue* cq, const RpcMethod& method, ClientContext* context, R* response, - void* tag) { + bool start, void* tag) { Call call = channel->CreateCall(method, context, cq); return new (g_core_codegen_interface->grpc_call_arena_alloc( call.call(), sizeof(ClientAsyncWriter))) - ClientAsyncWriter(call, context, response, tag); + ClientAsyncWriter(call, context, response, start, tag); } // always allocated against a call arena, no memory free required @@ -281,6 +309,12 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { assert(size == sizeof(ClientAsyncWriter)); } + void StartCall(void* tag) override { + assert(!started_); + started_ = true; + StartCallInternal(tag); + } + /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method for /// semantics. /// @@ -289,6 +323,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { /// associated with this call is updated, and the calling code can access /// the received metadata through the \a ClientContext. void ReadInitialMetadata(void* tag) override { + assert(started_); GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); meta_ops_.set_output_tag(tag); @@ -297,6 +332,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { } void Write(const W& msg, void* tag) override { + assert(started_); write_ops_.set_output_tag(tag); // TODO(ctiller): don't assert GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok()); @@ -304,6 +340,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { } void Write(const W& msg, WriteOptions options, void* tag) override { + assert(started_); write_ops_.set_output_tag(tag); if (options.is_last_message()) { options.set_buffer_hint(); @@ -315,6 +352,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { } void WritesDone(void* tag) override { + assert(started_); write_ops_.set_output_tag(tag); write_ops_.ClientSendClose(); call_.PerformOps(&write_ops_); @@ -328,6 +366,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { /// - attempts to fill in the \a response parameter passed to this class's /// constructor with the server's response message. void Finish(Status* status, void* tag) override { + assert(started_); finish_ops_.set_output_tag(tag); if (!context_->initial_metadata_received_) { finish_ops_.RecvInitialMetadata(context_); @@ -338,25 +377,32 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { private: template <class R> - ClientAsyncWriter(Call call, ClientContext* context, R* response, void* tag) - : context_(context), call_(call) { + ClientAsyncWriter(Call call, ClientContext* context, R* response, bool start, + void* tag) + : context_(context), call_(call), started_(start) { finish_ops_.RecvMessage(response); finish_ops_.AllowNoMessage(); - // if corked bit is set in context, we buffer up the initial metadata to - // coalesce with later message to be sent. No op is performed. - if (context_->initial_metadata_corked_) { - write_ops_.SendInitialMetadata(context->send_initial_metadata_, - context->initial_metadata_flags()); + if (start) { + StartCallInternal(tag); } else { + assert(tag == nullptr); + } + } + + void StartCallInternal(void* tag) { + write_ops_.SendInitialMetadata(context_->send_initial_metadata_, + context_->initial_metadata_flags()); + // if corked bit is set in context, we just keep the initial metadata + // buffered up to coalesce with later message send. No op is performed. + if (!context_->initial_metadata_corked_) { write_ops_.set_output_tag(tag); - write_ops_.SendInitialMetadata(context->send_initial_metadata_, - context->initial_metadata_flags()); call_.PerformOps(&write_ops_); } } ClientContext* context_; Call call_; + bool started_; CallOpSet<CallOpRecvInitialMetadata> meta_ops_; CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose> write_ops_; @@ -388,20 +434,23 @@ template <class W, class R> class ClientAsyncReaderWriter final : public ClientAsyncReaderWriterInterface<W, R> { public: - /// Create a stream and write the first request out. + /// Create a stream object. + /// Start the RPC request if \a start is set. /// \a tag will be notified on \a cq when the call has been started (i.e. - /// intitial metadata sent). + /// intitial metadata sent). If \a start is not set, \a tag must be + /// nullptr and the actual call must be initiated by StartCall /// Note that \a context will be used to fill in custom initial metadata /// used to send to the server when starting the call. static ClientAsyncReaderWriter* Create(ChannelInterface* channel, CompletionQueue* cq, const RpcMethod& method, - ClientContext* context, void* tag) { + ClientContext* context, bool start, + void* tag) { Call call = channel->CreateCall(method, context, cq); return new (g_core_codegen_interface->grpc_call_arena_alloc( call.call(), sizeof(ClientAsyncReaderWriter))) - ClientAsyncReaderWriter(call, context, tag); + ClientAsyncReaderWriter(call, context, start, tag); } // always allocated against a call arena, no memory free required @@ -409,6 +458,12 @@ class ClientAsyncReaderWriter final assert(size == sizeof(ClientAsyncReaderWriter)); } + void StartCall(void* tag) override { + assert(!started_); + started_ = true; + StartCallInternal(tag); + } + /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method /// for semantics of this method. /// @@ -417,6 +472,7 @@ class ClientAsyncReaderWriter final /// is updated with it, and then the receiving initial metadata can /// be accessed through this \a ClientContext. void ReadInitialMetadata(void* tag) override { + assert(started_); GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); meta_ops_.set_output_tag(tag); @@ -425,6 +481,7 @@ class ClientAsyncReaderWriter final } void Read(R* msg, void* tag) override { + assert(started_); read_ops_.set_output_tag(tag); if (!context_->initial_metadata_received_) { read_ops_.RecvInitialMetadata(context_); @@ -434,6 +491,7 @@ class ClientAsyncReaderWriter final } void Write(const W& msg, void* tag) override { + assert(started_); write_ops_.set_output_tag(tag); // TODO(ctiller): don't assert GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok()); @@ -441,6 +499,7 @@ class ClientAsyncReaderWriter final } void Write(const W& msg, WriteOptions options, void* tag) override { + assert(started_); write_ops_.set_output_tag(tag); if (options.is_last_message()) { options.set_buffer_hint(); @@ -452,6 +511,7 @@ class ClientAsyncReaderWriter final } void WritesDone(void* tag) override { + assert(started_); write_ops_.set_output_tag(tag); write_ops_.ClientSendClose(); call_.PerformOps(&write_ops_); @@ -462,6 +522,7 @@ class ClientAsyncReaderWriter final /// - the \a ClientContext associated with this call is updated with /// possible initial and trailing metadata sent from the server. void Finish(Status* status, void* tag) override { + assert(started_); finish_ops_.set_output_tag(tag); if (!context_->initial_metadata_received_) { finish_ops_.RecvInitialMetadata(context_); @@ -471,23 +532,30 @@ class ClientAsyncReaderWriter final } private: - ClientAsyncReaderWriter(Call call, ClientContext* context, void* tag) - : context_(context), call_(call) { - if (context_->initial_metadata_corked_) { - // if corked bit is set in context, we buffer up the initial metadata to - // coalesce with later message to be sent. No op is performed. - write_ops_.SendInitialMetadata(context->send_initial_metadata_, - context->initial_metadata_flags()); + ClientAsyncReaderWriter(Call call, ClientContext* context, bool start, + void* tag) + : context_(context), call_(call), started_(start) { + if (start) { + StartCallInternal(tag); } else { + assert(tag == nullptr); + } + } + + void StartCallInternal(void* tag) { + write_ops_.SendInitialMetadata(context_->send_initial_metadata_, + context_->initial_metadata_flags()); + // if corked bit is set in context, we just keep the initial metadata + // buffered up to coalesce with later message send. No op is performed. + if (!context_->initial_metadata_corked_) { write_ops_.set_output_tag(tag); - write_ops_.SendInitialMetadata(context->send_initial_metadata_, - context->initial_metadata_flags()); call_.PerformOps(&write_ops_); } } ClientContext* context_; Call call_; + bool started_; CallOpSet<CallOpRecvInitialMetadata> meta_ops_; CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_; CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose> diff --git a/include/grpc++/impl/codegen/async_unary_call.h b/include/grpc++/impl/codegen/async_unary_call.h index f0f909686b..e472f04f56 100644 --- a/include/grpc++/impl/codegen/async_unary_call.h +++ b/include/grpc++/impl/codegen/async_unary_call.h @@ -32,13 +32,18 @@ namespace grpc { class CompletionQueue; extern CoreCodegenInterface* g_core_codegen_interface; -/// An interface relevant for async client side unary RPCS (which send +/// An interface relevant for async client side unary RPCs (which send /// one request message to a server and receive one response message). template <class R> class ClientAsyncResponseReaderInterface { public: virtual ~ClientAsyncResponseReaderInterface() {} + /// Start the call that was set up by the constructor, but only if the + /// constructor was invoked through the "Prepare" API which doesn't actually + /// start the call + virtual void StartCall() = 0; + /// Request notification of the reading of initial metadata. Completion /// will be notified by \a tag on the associated completion queue. /// This call is optional, but if it is used, it cannot be used concurrently @@ -70,9 +75,10 @@ template <class R> class ClientAsyncResponseReader final : public ClientAsyncResponseReaderInterface<R> { public: - /// Start a call and write the request out. + /// Start a call and write the request out if \a start is set. /// \a tag will be notified on \a cq when the call has been started (i.e. /// intitial metadata sent) and \a request has been written out. + /// If \a start is not set, the actual call must be initiated by StartCall /// Note that \a context will be used to fill in custom initial metadata /// used to send to the server when starting the call. template <class W> @@ -80,11 +86,11 @@ class ClientAsyncResponseReader final CompletionQueue* cq, const RpcMethod& method, ClientContext* context, - const W& request) { + const W& request, bool start) { Call call = channel->CreateCall(method, context, cq); return new (g_core_codegen_interface->grpc_call_arena_alloc( call.call(), sizeof(ClientAsyncResponseReader))) - ClientAsyncResponseReader(call, context, request); + ClientAsyncResponseReader(call, context, request, start); } // always allocated against a call arena, no memory free required @@ -92,13 +98,20 @@ class ClientAsyncResponseReader final assert(size == sizeof(ClientAsyncResponseReader)); } + void StartCall() override { + assert(!started_); + started_ = true; + StartCallInternal(); + } + /// See \a ClientAsyncResponseReaderInterface::ReadInitialMetadata for /// semantics. /// /// Side effect: /// - the \a ClientContext associated with this call is updated with /// possible initial and trailing metadata sent from the server. - void ReadInitialMetadata(void* tag) { + void ReadInitialMetadata(void* tag) override { + assert(started_); GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); meta_buf.set_output_tag(tag); @@ -111,7 +124,8 @@ class ClientAsyncResponseReader final /// Side effect: /// - the \a ClientContext associated with this call is updated with /// possible initial and trailing metadata sent from the server. - void Finish(R* msg, Status* status, void* tag) { + void Finish(R* msg, Status* status, void* tag) override { + assert(started_); finish_buf.set_output_tag(tag); if (!context_->initial_metadata_received_) { finish_buf.RecvInitialMetadata(context_); @@ -125,15 +139,22 @@ class ClientAsyncResponseReader final private: ClientContext* const context_; Call call_; + bool started_; template <class W> - ClientAsyncResponseReader(Call call, ClientContext* context, const W& request) - : context_(context), call_(call) { - init_buf.SendInitialMetadata(context->send_initial_metadata_, - context->initial_metadata_flags()); + ClientAsyncResponseReader(Call call, ClientContext* context, const W& request, + bool start) + : context_(context), call_(call), started_(start) { + // Bind the metadata at time of StartCallInternal but set up the rest here // TODO(ctiller): don't assert GPR_CODEGEN_ASSERT(init_buf.SendMessage(request).ok()); init_buf.ClientSendClose(); + if (start) StartCallInternal(); + } + + void StartCallInternal() { + init_buf.SendInitialMetadata(context_->send_initial_metadata_, + context_->initial_metadata_flags()); call_.PerformOps(&init_buf); } diff --git a/include/grpc++/impl/codegen/byte_buffer.h b/include/grpc++/impl/codegen/byte_buffer.h new file mode 100644 index 0000000000..57d731be18 --- /dev/null +++ b/include/grpc++/impl/codegen/byte_buffer.h @@ -0,0 +1,156 @@ +/* + * + * 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 <grpc/impl/codegen/byte_buffer.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/slice.h> +#include <grpc++/impl/codegen/status.h> + +#include <vector> + +namespace grpc { + +template <class R> +class CallOpRecvMessage; +class MethodHandler; +template <class ServiceType, class RequestType, class ResponseType> +class RpcMethodHandler; +template <class ServiceType, class RequestType, class ResponseType> +class ServerStreamingHandler; +namespace CallOpGenericRecvMessageHelper { +template <class R> +class DeserializeFuncType; +} // namespace CallOpGenericRecvMessageHelper + +/// 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<Slice>* 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<ByteBuffer, void>; + friend class CallOpSendMessage; + template <class R> + friend class CallOpRecvMessage; + friend class CallOpGenericRecvMessage; + friend class MethodHandler; + template <class ServiceType, class RequestType, class ResponseType> + friend class RpcMethodHandler; + template <class ServiceType, class RequestType, class ResponseType> + friend class ServerStreamingHandler; + template <class R> + friend class CallOpGenericRecvMessageHelper::DeserializeFuncType; + + grpc_byte_buffer* buffer_; + + // 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_; } + + class ByteBufferPointer { + public: + ByteBufferPointer(const ByteBuffer* b) + : bbuf_(const_cast<ByteBuffer*>(b)) {} + operator ByteBuffer*() { return bbuf_; } + operator grpc_byte_buffer*() { return bbuf_->buffer_; } + operator grpc_byte_buffer**() { return &bbuf_->buffer_; } + + private: + ByteBuffer* bbuf_; + }; + ByteBufferPointer bbuf_ptr() const { return ByteBufferPointer(this); } +}; + +template <> +class SerializationTraits<ByteBuffer, void> { + public: + static Status Deserialize(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 8e70225f86..d9988e51fc 100644 --- a/include/grpc++/impl/codegen/call.h +++ b/include/grpc++/impl/codegen/call.h @@ -25,6 +25,7 @@ #include <map> #include <memory> +#include <grpc++/impl/codegen/byte_buffer.h> #include <grpc++/impl/codegen/call_hook.h> #include <grpc++/impl/codegen/client_context.h> #include <grpc++/impl/codegen/completion_queue_tag.h> @@ -39,8 +40,6 @@ #include <grpc/impl/codegen/compression_types.h> #include <grpc/impl/codegen/grpc_types.h> -struct grpc_byte_buffer; - namespace grpc { class ByteBuffer; @@ -169,6 +168,15 @@ class WriteOptions { return *this; } + /// Guarantee that all bytes have been written to the wire before completing + /// this write (usually writes are completed when they pass flow control) + inline WriteOptions& set_write_through() { + SetBit(GRPC_WRITE_THROUGH); + return *this; + } + + inline bool is_write_through() const { return GetBit(GRPC_WRITE_THROUGH); } + /// Get value for the flag indicating that this is the last message, and /// should be coalesced with trailing metadata. /// @@ -272,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. @@ -285,33 +293,35 @@ 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_; + ByteBuffer send_buf_; WriteOptions write_options_; }; +namespace internal { +template <class T> +T Example(); +} // namespace internal + template <class M> Status CallOpSendMessage::SendMessage(const M& message, WriteOptions options) { write_options_ = options; bool own_buf; - Status result = - SerializationTraits<M>::Serialize(message, &send_buf_, &own_buf); + Status result = SerializationTraits<M>::Serialize( + message, send_buf_.bbuf_ptr(), &own_buf); if (!own_buf) { - send_buf_ = g_core_codegen_interface->grpc_byte_buffer_copy(send_buf_); + send_buf_.Duplicate(); } return result; } @@ -343,18 +353,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<R>::Deserialize(recv_buf_, message_).ok(); + SerializationTraits<R>::Deserialize(recv_buf_.bbuf_ptr(), 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; @@ -367,14 +379,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(ByteBuffer* buf) = 0; virtual ~DeserializeFunc() {} }; @@ -382,8 +394,8 @@ 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_); + Status Deserialize(ByteBuffer* buf) override { + return SerializationTraits<R>::Deserialize(buf->bbuf_ptr(), message_); } ~DeserializeFuncType() override {} @@ -419,18 +431,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(); + *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; @@ -443,7 +456,7 @@ class CallOpGenericRecvMessage { private: std::unique_ptr<CallOpGenericRecvMessageHelper::DeserializeFunc> 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..e14cb0e926 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 <grpc++/impl/codegen/byte_buffer.h> #include <grpc++/impl/codegen/core_codegen_interface.h> #include <grpc++/impl/codegen/rpc_service_method.h> #include <grpc++/impl/codegen/sync_stream.h> @@ -37,8 +38,8 @@ class RpcMethodHandler : public MethodHandler { void RunHandler(const HandlerParameter& param) final { RequestType req; - Status status = - SerializationTraits<RequestType>::Deserialize(param.request, &req); + Status status = SerializationTraits<RequestType>::Deserialize( + param.request.bbuf_ptr(), &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<RequestType>::Deserialize(param.request, &req); + Status status = SerializationTraits<RequestType>::Deserialize( + param.request.bbuf_ptr(), &req); if (status.ok()) { ServerWriter<ResponseType> 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 <memory> #include <vector> +#include <grpc++/impl/codegen/byte_buffer.h> #include <grpc++/impl/codegen/config.h> #include <grpc++/impl/codegen/rpc_method.h> #include <grpc++/impl/codegen/status.h> -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..4d91739cc4 100644 --- a/include/grpc++/impl/codegen/serialization_traits.h +++ b/include/grpc++/impl/codegen/serialization_traits.h @@ -24,17 +24,26 @@ namespace grpc { /// Defines how to serialize and deserialize some type. /// /// 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); +/// Each SerializationTraits<Message> implementation must provide the +/// following functions: +/// 1. static Status Serialize(const Message& msg, +/// ByteBuffer* buffer, +/// bool* own_buffer); +/// 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(ByteBuffer* buffer, +/// Message* msg); +/// 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 <grpc++/impl/codegen/config.h> #include <grpc++/impl/codegen/core_codegen_interface.h> #include <grpc++/impl/codegen/string_ref.h> +#include <grpc/impl/codegen/slice.h> + 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<const char*>(GRPC_SLICE_START_PTR(*slice)), diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h index 21ae70d13a..bbf45b3e74 100644 --- a/include/grpc++/server_builder.h +++ b/include/grpc++/server_builder.h @@ -136,8 +136,10 @@ class ServerBuilder { /// It can be invoked multiple times. /// /// \param addr_uri The address to try to bind to the server in URI form. If - /// the scheme name is omitted, "dns:///" is assumed. Valid values include - /// dns:///localhost:1234, / 192.168.1.1:31416, dns:///[::1]:27182, etc.). + /// the scheme name is omitted, "dns:///" is assumed. To bind to any address, + /// please use IPv6 any, i.e., [::]:<port>, which also accepts IPv4 + /// connections. Valid values include dns:///localhost:1234, / + /// 192.168.1.1:31416, dns:///[::1]:27182, etc.). /// \params creds The credentials associated with the server. /// \param selected_port[out] If not `nullptr`, gets populated with the port /// number bound to the \a grpc::Server for the corresponding endpoint after diff --git a/include/grpc++/support/byte_buffer.h b/include/grpc++/support/byte_buffer.h index df16c6a75e..81fa3b0a18 100644 --- a/include/grpc++/support/byte_buffer.h +++ b/include/grpc++/support/byte_buffer.h @@ -19,6 +19,7 @@ #ifndef GRPCXX_SUPPORT_BYTE_BUFFER_H #define GRPCXX_SUPPORT_BYTE_BUFFER_H +#include <grpc++/impl/codegen/byte_buffer.h> #include <grpc++/impl/serialization_traits.h> #include <grpc++/support/config.h> #include <grpc++/support/slice.h> @@ -27,71 +28,4 @@ #include <grpc/grpc.h> #include <grpc/support/log.h> -#include <vector> - -namespace grpc { - -/// 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(); - - ByteBuffer& operator=(const ByteBuffer&); - - /// Dump (read) the buffer contents into \a slices. - Status Dump(std::vector<Slice>* slices) const; - - /// Remove all data. - void Clear(); - - /// Buffer size in bytes. - size_t Length() const; - - /// Swap the state of *this and *other. - void Swap(ByteBuffer* other); - - private: - friend class SerializationTraits<ByteBuffer, void>; - - // takes ownership - void set_buffer(grpc_byte_buffer* buf) { - if (buffer_) { - Clear(); - } - buffer_ = buf; - } - - // For \a SerializationTraits's usage. - grpc_byte_buffer* buffer() const { return buffer_; } - - grpc_byte_buffer* buffer_; -}; - -template <> -class SerializationTraits<ByteBuffer, void> { - public: - static Status Deserialize(grpc_byte_buffer* byte_buffer, ByteBuffer* dest) { - dest->set_buffer(byte_buffer); - return Status::OK; - } - static Status Serialize(const ByteBuffer& source, grpc_byte_buffer** buffer, - bool* own_buffer) { - *buffer = grpc_byte_buffer_copy(source.buffer()); - *own_buffer = true; - return Status::OK; - } -}; - -} // namespace grpc - #endif // GRPCXX_SUPPORT_BYTE_BUFFER_H diff --git a/include/grpc++/support/slice.h b/include/grpc++/support/slice.h index bbf97f280e..10db10d79c 100644 --- a/include/grpc++/support/slice.h +++ b/include/grpc++/support/slice.h @@ -19,86 +19,8 @@ #ifndef GRPCXX_SUPPORT_SLICE_H #define GRPCXX_SUPPORT_SLICE_H +#include <grpc++/impl/codegen/slice.h> #include <grpc++/support/config.h> #include <grpc/slice.h> -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 { return grpc_slice_ref(slice_); } - - private: - friend class ByteBuffer; - - grpc_slice slice_; -}; - -} // namespace grpc - #endif // GRPCXX_SUPPORT_SLICE_H diff --git a/include/grpc/byte_buffer.h b/include/grpc/byte_buffer.h index 55e191da31..7669582af2 100644 --- a/include/grpc/byte_buffer.h +++ b/include/grpc/byte_buffer.h @@ -19,69 +19,7 @@ #ifndef GRPC_BYTE_BUFFER_H #define GRPC_BYTE_BUFFER_H -#include <grpc/impl/codegen/grpc_types.h> +#include <grpc/impl/codegen/byte_buffer.h> #include <grpc/slice_buffer.h> -#ifdef __cplusplus -extern "C" { -#endif - -/** Returns a RAW byte buffer instance over the given slices (up to \a nslices). - * - * Increases the reference count for all \a slices processed. The user is - * responsible for invoking grpc_byte_buffer_destroy on the returned instance.*/ -GRPCAPI grpc_byte_buffer *grpc_raw_byte_buffer_create(grpc_slice *slices, - size_t nslices); - -/** Returns a *compressed* RAW byte buffer instance over the given slices (up to - * \a nslices). The \a compression argument defines the compression algorithm - * used to generate the data in \a slices. - * - * Increases the reference count for all \a slices processed. The user is - * responsible for invoking grpc_byte_buffer_destroy on the returned instance.*/ -GRPCAPI grpc_byte_buffer *grpc_raw_compressed_byte_buffer_create( - grpc_slice *slices, size_t nslices, grpc_compression_algorithm compression); - -/** Copies input byte buffer \a bb. - * - * Increases the reference count of all the source slices. The user is - * responsible for calling grpc_byte_buffer_destroy over the returned copy. */ -GRPCAPI grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb); - -/** Returns the size of the given byte buffer, in bytes. */ -GRPCAPI size_t grpc_byte_buffer_length(grpc_byte_buffer *bb); - -/** Destroys \a byte_buffer deallocating all its memory. */ -GRPCAPI void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer); - -/** Reader for byte buffers. Iterates over slices in the byte buffer */ -struct grpc_byte_buffer_reader; -typedef struct grpc_byte_buffer_reader grpc_byte_buffer_reader; - -/** Initialize \a reader to read over \a buffer. - * Returns 1 upon success, 0 otherwise. */ -GRPCAPI int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader, - grpc_byte_buffer *buffer); - -/** Cleanup and destroy \a reader */ -GRPCAPI void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader); - -/** Updates \a slice with the next piece of data from from \a reader and returns - * 1. Returns 0 at the end of the stream. Caller is responsible for calling - * grpc_slice_unref on the result. */ -GRPCAPI int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader, - grpc_slice *slice); - -/** Merge all data from \a reader into single slice */ -GRPCAPI grpc_slice -grpc_byte_buffer_reader_readall(grpc_byte_buffer_reader *reader); - -/** Returns a RAW byte buffer instance from the output of \a reader. */ -GRPCAPI grpc_byte_buffer *grpc_raw_byte_buffer_from_reader( - grpc_byte_buffer_reader *reader); - -#ifdef __cplusplus -} -#endif - #endif /* GRPC_BYTE_BUFFER_H */ diff --git a/include/grpc/compression.h b/include/grpc/compression.h index 3a8de4b7b8..13a8dd66ad 100644 --- a/include/grpc/compression.h +++ b/include/grpc/compression.h @@ -44,13 +44,13 @@ int grpc_stream_compression_algorithm_parse( * algorithm. Note that \a name is statically allocated and must *not* be freed. * Returns 1 upon success, 0 otherwise. */ GRPCAPI int grpc_compression_algorithm_name( - grpc_compression_algorithm algorithm, char **name); + grpc_compression_algorithm algorithm, const char **name); /** Updates \a name with the encoding name corresponding to a valid \a * algorithm. Note that \a name is statically allocated and must *not* be freed. * Returns 1 upon success, 0 otherwise. */ GRPCAPI int grpc_stream_compression_algorithm_name( - grpc_stream_compression_algorithm algorithm, char **name); + grpc_stream_compression_algorithm algorithm, const char **name); /** Returns the compression algorithm corresponding to \a level for the * compression algorithms encoded in the \a accepted_encodings bitset. diff --git a/include/grpc/impl/codegen/byte_buffer.h b/include/grpc/impl/codegen/byte_buffer.h new file mode 100644 index 0000000000..fc33305713 --- /dev/null +++ b/include/grpc/impl/codegen/byte_buffer.h @@ -0,0 +1,86 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_IMPL_CODEGEN_BYTE_BUFFER_H +#define GRPC_IMPL_CODEGEN_BYTE_BUFFER_H + +#include <grpc/impl/codegen/grpc_types.h> + +#ifdef __cplusplus +extern "C" { +#endif + +/** Returns a RAW byte buffer instance over the given slices (up to \a nslices). + * + * Increases the reference count for all \a slices processed. The user is + * responsible for invoking grpc_byte_buffer_destroy on the returned instance.*/ +GRPCAPI grpc_byte_buffer *grpc_raw_byte_buffer_create(grpc_slice *slices, + size_t nslices); + +/** Returns a *compressed* RAW byte buffer instance over the given slices (up to + * \a nslices). The \a compression argument defines the compression algorithm + * used to generate the data in \a slices. + * + * Increases the reference count for all \a slices processed. The user is + * responsible for invoking grpc_byte_buffer_destroy on the returned instance.*/ +GRPCAPI grpc_byte_buffer *grpc_raw_compressed_byte_buffer_create( + grpc_slice *slices, size_t nslices, grpc_compression_algorithm compression); + +/** Copies input byte buffer \a bb. + * + * Increases the reference count of all the source slices. The user is + * responsible for calling grpc_byte_buffer_destroy over the returned copy. */ +GRPCAPI grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb); + +/** Returns the size of the given byte buffer, in bytes. */ +GRPCAPI size_t grpc_byte_buffer_length(grpc_byte_buffer *bb); + +/** Destroys \a byte_buffer deallocating all its memory. */ +GRPCAPI void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer); + +/** Reader for byte buffers. Iterates over slices in the byte buffer */ +struct grpc_byte_buffer_reader; +typedef struct grpc_byte_buffer_reader grpc_byte_buffer_reader; + +/** Initialize \a reader to read over \a buffer. + * Returns 1 upon success, 0 otherwise. */ +GRPCAPI int grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader, + grpc_byte_buffer *buffer); + +/** Cleanup and destroy \a reader */ +GRPCAPI void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader); + +/** Updates \a slice with the next piece of data from from \a reader and returns + * 1. Returns 0 at the end of the stream. Caller is responsible for calling + * grpc_slice_unref on the result. */ +GRPCAPI int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader, + grpc_slice *slice); + +/** Merge all data from \a reader into single slice */ +GRPCAPI grpc_slice +grpc_byte_buffer_reader_readall(grpc_byte_buffer_reader *reader); + +/** Returns a RAW byte buffer instance from the output of \a reader. */ +GRPCAPI grpc_byte_buffer *grpc_raw_byte_buffer_from_reader( + grpc_byte_buffer_reader *reader); + +#ifdef __cplusplus +} +#endif + +#endif /* GRPC_IMPL_CODEGEN_BYTE_BUFFER_H */ diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h index 59b90af03a..90f03f49a3 100644 --- a/include/grpc/impl/codegen/grpc_types.h +++ b/include/grpc/impl/codegen/grpc_types.h @@ -188,9 +188,14 @@ typedef struct { #define GRPC_ARG_HTTP2_MAX_FRAME_SIZE "grpc.http2.max_frame_size" /** Should BDP probing be performed? */ #define GRPC_ARG_HTTP2_BDP_PROBE "grpc.http2.bdp_probe" -/** Minimum time (in milliseconds) between successive ping frames being sent */ -#define GRPC_ARG_HTTP2_MIN_TIME_BETWEEN_PINGS_MS \ +/** Minimum time between sending successive ping frames without receiving any + data frame, Int valued, milliseconds. */ +#define GRPC_ARG_HTTP2_MIN_SENT_PING_INTERVAL_WITHOUT_DATA_MS \ "grpc.http2.min_time_between_pings_ms" +/** Minimum allowed time between receiving successive ping frames without + sending any data frame. Int valued, milliseconds */ +#define GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS \ + "grpc.http2.min_ping_interval_without_data_ms" /** Channel arg to override the http2 :scheme header */ #define GRPC_ARG_HTTP2_SCHEME "grpc.http2_scheme" /** How many pings can we send before needing to send a data frame or header @@ -202,10 +207,6 @@ typedef struct { closing the transport? (0 indicates that the server can bear an infinite number of misbehaving pings) */ #define GRPC_ARG_HTTP2_MAX_PING_STRIKES "grpc.http2.max_ping_strikes" -/** Minimum allowed time between two pings without sending any data frame. Int - valued, seconds */ -#define GRPC_ARG_HTTP2_MIN_PING_INTERVAL_WITHOUT_DATA_MS \ - "grpc.http2.min_ping_interval_without_data_ms" /** How much data are we willing to queue up per stream if GRPC_WRITE_BUFFER_HINT is set? This is an upper bound */ #define GRPC_ARG_HTTP2_WRITE_BUFFER_SIZE "grpc.http2.write_buffer_size" @@ -355,8 +356,11 @@ typedef enum grpc_call_error { /** Force compression to be disabled for a particular write (start_write/add_metadata). Illegal on invoke/accept. */ #define GRPC_WRITE_NO_COMPRESS (0x00000002u) +/** Force this message to be written to the socket before completing it */ +#define GRPC_WRITE_THROUGH (0x00000004u) /** Mask of all valid flags. */ -#define GRPC_WRITE_USED_MASK (GRPC_WRITE_BUFFER_HINT | GRPC_WRITE_NO_COMPRESS) +#define GRPC_WRITE_USED_MASK \ + (GRPC_WRITE_BUFFER_HINT | GRPC_WRITE_NO_COMPRESS | GRPC_WRITE_THROUGH) /** Initial metadata flags */ /** Signal that the call is idempotent */ @@ -377,7 +381,7 @@ typedef enum grpc_call_error { GRPC_INITIAL_METADATA_WAIT_FOR_READY | \ GRPC_INITIAL_METADATA_CACHEABLE_REQUEST | \ GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET | \ - GRPC_INITIAL_METADATA_CORKED) + GRPC_INITIAL_METADATA_CORKED | GRPC_WRITE_THROUGH) /** A single metadata element */ typedef struct grpc_metadata { diff --git a/include/grpc/impl/codegen/port_platform.h b/include/grpc/impl/codegen/port_platform.h index e84a75d295..1904c636f7 100644 --- a/include/grpc/impl/codegen/port_platform.h +++ b/include/grpc/impl/codegen/port_platform.h @@ -409,4 +409,15 @@ typedef unsigned __int64 uint64_t; #define CENSUSAPI GRPCAPI #endif +#ifndef GPR_ATTRIBUTE_NO_TSAN /* (1) */ +#if defined(__has_feature) +#if __has_feature(thread_sanitizer) +#define GPR_ATTRIBUTE_NO_TSAN __attribute__((no_sanitize("thread"))) +#endif /* __has_feature(thread_sanitizer) */ +#endif /* defined(__has_feature) */ +#ifndef GPR_ATTRIBUTE_NO_TSAN /* (2) */ +#define GPR_ATTRIBUTE_NO_TSAN +#endif /* GPR_ATTRIBUTE_NO_TSAN (2) */ +#endif /* GPR_ATTRIBUTE_NO_TSAN (1) */ + #endif /* GRPC_IMPL_CODEGEN_PORT_PLATFORM_H */ diff --git a/include/grpc/slice.h b/include/grpc/slice.h index 9d2a9fbc81..3f3cff1408 100644 --- a/include/grpc/slice.h +++ b/include/grpc/slice.h @@ -65,11 +65,7 @@ GPRAPI grpc_slice grpc_slice_new_with_len(void *p, size_t len, GPRAPI grpc_slice grpc_slice_malloc(size_t length); GPRAPI grpc_slice grpc_slice_malloc_large(size_t length); -#define GRPC_SLICE_MALLOC(len) \ - ((len) <= GRPC_SLICE_INLINED_SIZE \ - ? (grpc_slice){.refcount = NULL, \ - .data.inlined = {.length = (uint8_t)(len)}} \ - : grpc_slice_malloc_large((len))) +#define GRPC_SLICE_MALLOC(len) grpc_slice_malloc(len) /** Intern a slice: |