2 * Copyright 2014 Google Inc. All rights reserved.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
17 #ifndef FLATBUFFERS_GRPC_H_
18 #define FLATBUFFERS_GRPC_H_
20 // Helper functionality to glue FlatBuffers and GRPC.
22 #include "flatbuffers/flatbuffers.h"
23 #include "grpc++/support/byte_buffer.h"
24 #include "grpc/byte_buffer_reader.h"
26 namespace flatbuffers {
29 // Message is a typed wrapper around a buffer that manages the underlying
30 // `grpc_slice` and also provides flatbuffers-specific helpers such as `Verify`
31 // and `GetRoot`. Since it is backed by a `grpc_slice`, the underlying buffer
32 // is refcounted and ownership is be managed automatically.
33 template<class T> class Message {
35 Message() : slice_(grpc_empty_slice()) {}
37 Message(grpc_slice slice, bool add_ref)
38 : slice_(add_ref ? grpc_slice_ref(slice) : slice) {}
40 Message &operator=(const Message &other) = delete;
42 Message(Message &&other) : slice_(other.slice_) {
43 other.slice_ = grpc_empty_slice();
46 Message(const Message &other) = delete;
48 Message &operator=(Message &&other) {
49 grpc_slice_unref(slice_);
50 slice_ = other.slice_;
51 other.slice_ = grpc_empty_slice();
55 ~Message() { grpc_slice_unref(slice_); }
57 const uint8_t *mutable_data() const { return GRPC_SLICE_START_PTR(slice_); }
59 const uint8_t *data() const { return GRPC_SLICE_START_PTR(slice_); }
61 size_t size() const { return GRPC_SLICE_LENGTH(slice_); }
64 Verifier verifier(data(), size());
65 return verifier.VerifyBuffer<T>(nullptr);
68 T *GetMutableRoot() { return flatbuffers::GetMutableRoot<T>(mutable_data()); }
70 const T *GetRoot() const { return flatbuffers::GetRoot<T>(data()); }
72 // This is only intended for serializer use, or if you know what you're doing
73 const grpc_slice &BorrowSlice() const { return slice_; }
81 // SliceAllocator is a gRPC-specific allocator that uses the `grpc_slice`
82 // refcounted slices to manage memory ownership. This makes it easy and
83 // efficient to transfer buffers to gRPC.
84 class SliceAllocator : public Allocator {
86 SliceAllocator() : slice_(grpc_empty_slice()) {}
88 SliceAllocator(const SliceAllocator &other) = delete;
89 SliceAllocator &operator=(const SliceAllocator &other) = delete;
91 SliceAllocator(SliceAllocator &&other)
92 : slice_(grpc_empty_slice()) {
93 // default-construct and swap idiom
97 SliceAllocator &operator=(SliceAllocator &&other) {
98 // move-construct and swap idiom
99 SliceAllocator temp(std::move(other));
104 void swap(SliceAllocator &other) {
106 swap(slice_, other.slice_);
109 virtual ~SliceAllocator() { grpc_slice_unref(slice_); }
111 virtual uint8_t *allocate(size_t size) override {
112 FLATBUFFERS_ASSERT(GRPC_SLICE_IS_EMPTY(slice_));
113 slice_ = grpc_slice_malloc(size);
114 return GRPC_SLICE_START_PTR(slice_);
117 virtual void deallocate(uint8_t *p, size_t size) override {
118 FLATBUFFERS_ASSERT(p == GRPC_SLICE_START_PTR(slice_));
119 FLATBUFFERS_ASSERT(size == GRPC_SLICE_LENGTH(slice_));
120 grpc_slice_unref(slice_);
121 slice_ = grpc_empty_slice();
124 virtual uint8_t *reallocate_downward(uint8_t *old_p, size_t old_size,
125 size_t new_size, size_t in_use_back,
126 size_t in_use_front) override {
127 FLATBUFFERS_ASSERT(old_p == GRPC_SLICE_START_PTR(slice_));
128 FLATBUFFERS_ASSERT(old_size == GRPC_SLICE_LENGTH(slice_));
129 FLATBUFFERS_ASSERT(new_size > old_size);
130 grpc_slice old_slice = slice_;
131 grpc_slice new_slice = grpc_slice_malloc(new_size);
132 uint8_t *new_p = GRPC_SLICE_START_PTR(new_slice);
133 memcpy_downward(old_p, old_size, new_p, new_size, in_use_back,
136 grpc_slice_unref(old_slice);
141 grpc_slice &get_slice(uint8_t *p, size_t size) {
142 FLATBUFFERS_ASSERT(p == GRPC_SLICE_START_PTR(slice_));
143 FLATBUFFERS_ASSERT(size == GRPC_SLICE_LENGTH(slice_));
149 friend class MessageBuilder;
152 // SliceAllocatorMember is a hack to ensure that the MessageBuilder's
153 // slice_allocator_ member is constructed before the FlatBufferBuilder, since
154 // the allocator is used in the FlatBufferBuilder ctor.
156 struct SliceAllocatorMember {
157 SliceAllocator slice_allocator_;
159 } // namespace detail
161 // MessageBuilder is a gRPC-specific FlatBufferBuilder that uses SliceAllocator
162 // to allocate gRPC buffers.
163 class MessageBuilder : private detail::SliceAllocatorMember,
164 public FlatBufferBuilder {
166 explicit MessageBuilder(uoffset_t initial_size = 1024)
167 : FlatBufferBuilder(initial_size, &slice_allocator_, false) {}
169 MessageBuilder(const MessageBuilder &other) = delete;
170 MessageBuilder &operator=(const MessageBuilder &other) = delete;
172 MessageBuilder(MessageBuilder &&other)
173 : FlatBufferBuilder(1024, &slice_allocator_, false) {
174 // Default construct and swap idiom.
178 /// Create a MessageBuilder from a FlatBufferBuilder.
179 explicit MessageBuilder(FlatBufferBuilder &&src, void (*dealloc)(void*, size_t) = &DefaultAllocator::dealloc)
180 : FlatBufferBuilder(1024, &slice_allocator_, false) {
182 src.SwapBufAllocator(*this);
183 if (buf_.capacity()) {
184 uint8_t *buf = buf_.scratch_data(); // pointer to memory
185 size_t capacity = buf_.capacity(); // size of memory
186 slice_allocator_.slice_ = grpc_slice_new_with_len(buf, capacity, dealloc);
189 slice_allocator_.slice_ = grpc_empty_slice();
193 /// Move-assign a FlatBufferBuilder to a MessageBuilder.
194 /// Only FlatBufferBuilder with default allocator (basically, nullptr) is supported.
195 MessageBuilder &operator=(FlatBufferBuilder &&src) {
196 // Move construct a temporary and swap
197 MessageBuilder temp(std::move(src));
202 MessageBuilder &operator=(MessageBuilder &&other) {
203 // Move construct a temporary and swap
204 MessageBuilder temp(std::move(other));
209 void Swap(MessageBuilder &other) {
210 slice_allocator_.swap(other.slice_allocator_);
211 FlatBufferBuilder::Swap(other);
212 // After swapping the FlatBufferBuilder, we swap back the allocator, which restores
213 // the original allocator back in place. This is necessary because MessageBuilder's
214 // allocator is its own member (SliceAllocatorMember). The allocator passed to
215 // FlatBufferBuilder::vector_downward must point to this member.
216 buf_.swap_allocator(other.buf_);
219 // Releases the ownership of the buffer pointer.
220 // Returns the size, offset, and the original grpc_slice that
221 // allocated the buffer. Also see grpc_slice_unref().
222 uint8_t *ReleaseRaw(size_t &size, size_t &offset, grpc_slice &slice) {
223 uint8_t *buf = FlatBufferBuilder::ReleaseRaw(size, offset);
224 slice = slice_allocator_.slice_;
225 slice_allocator_.slice_ = grpc_empty_slice();
231 // GetMessage extracts the subslice of the buffer corresponding to the
232 // flatbuffers-encoded region and wraps it in a `Message<T>` to handle buffer
234 template<class T> Message<T> GetMessage() {
235 auto buf_data = buf_.scratch_data(); // pointer to memory
236 auto buf_size = buf_.capacity(); // size of memory
237 auto msg_data = buf_.data(); // pointer to msg
238 auto msg_size = buf_.size(); // size of msg
239 // Do some sanity checks on data/size
240 FLATBUFFERS_ASSERT(msg_data);
241 FLATBUFFERS_ASSERT(msg_size);
242 FLATBUFFERS_ASSERT(msg_data >= buf_data);
243 FLATBUFFERS_ASSERT(msg_data + msg_size <= buf_data + buf_size);
244 // Calculate offsets from the buffer start
245 auto begin = msg_data - buf_data;
246 auto end = begin + msg_size;
247 // Get the slice we are working with (no refcount change)
248 grpc_slice slice = slice_allocator_.get_slice(buf_data, buf_size);
249 // Extract a subslice of the existing slice (increment refcount)
250 grpc_slice subslice = grpc_slice_sub(slice, begin, end);
251 // Wrap the subslice in a `Message<T>`, but don't increment refcount
252 Message<T> msg(subslice, false);
256 template<class T> Message<T> ReleaseMessage() {
257 Message<T> msg = GetMessage<T>();
263 // SliceAllocator slice_allocator_; // part of SliceAllocatorMember
267 } // namespace flatbuffers
271 template<class T> class SerializationTraits<flatbuffers::grpc::Message<T>> {
273 static grpc::Status Serialize(const flatbuffers::grpc::Message<T> &msg,
274 grpc_byte_buffer **buffer, bool *own_buffer) {
275 // We are passed in a `Message<T>`, which is a wrapper around a
276 // `grpc_slice`. We extract it here using `BorrowSlice()`. The const cast
277 // is necessary because the `grpc_raw_byte_buffer_create` func expects
278 // non-const slices in order to increment their refcounts.
279 grpc_slice *slice = const_cast<grpc_slice *>(&msg.BorrowSlice());
280 // Now use `grpc_raw_byte_buffer_create` to package the single slice into a
281 // `grpc_byte_buffer`, incrementing the refcount in the process.
282 *buffer = grpc_raw_byte_buffer_create(slice, 1);
284 return grpc::Status::OK;
287 // Deserialize by pulling the
288 static grpc::Status Deserialize(grpc_byte_buffer *buffer,
289 flatbuffers::grpc::Message<T> *msg) {
291 return ::grpc::Status(::grpc::StatusCode::INTERNAL, "No payload");
293 // Check if this is a single uncompressed slice.
294 if ((buffer->type == GRPC_BB_RAW) &&
295 (buffer->data.raw.compression == GRPC_COMPRESS_NONE) &&
296 (buffer->data.raw.slice_buffer.count == 1)) {
297 // If it is, then we can reference the `grpc_slice` directly.
298 grpc_slice slice = buffer->data.raw.slice_buffer.slices[0];
299 // We wrap a `Message<T>` around the slice, incrementing the refcount.
300 *msg = flatbuffers::grpc::Message<T>(slice, true);
302 // Otherwise, we need to use `grpc_byte_buffer_reader_readall` to read
303 // `buffer` into a single contiguous `grpc_slice`. The gRPC reader gives
304 // us back a new slice with the refcount already incremented.
305 grpc_byte_buffer_reader reader;
306 grpc_byte_buffer_reader_init(&reader, buffer);
307 grpc_slice slice = grpc_byte_buffer_reader_readall(&reader);
308 grpc_byte_buffer_reader_destroy(&reader);
309 // We wrap a `Message<T>` around the slice, but don't increment refcount
310 *msg = flatbuffers::grpc::Message<T>(slice, false);
312 grpc_byte_buffer_destroy(buffer);
313 #if FLATBUFFERS_GRPC_DISABLE_AUTO_VERIFICATION
314 return ::grpc::Status::OK;
317 return ::grpc::Status::OK;
319 return ::grpc::Status(::grpc::StatusCode::INTERNAL,
320 "Message verification failed");
328 #endif // FLATBUFFERS_GRPC_H_