-
Notifications
You must be signed in to change notification settings - Fork 15.3k
/
zero_copy_stream_impl_lite.h
517 lines (429 loc) · 21 KB
/
zero_copy_stream_impl_lite.h
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
// Protocol Buffers - Google's data interchange format
// Copyright 2008 Google Inc. All rights reserved.
//
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file or at
// https://developers.google.com/open-source/licenses/bsd
// Author: kenton@google.com (Kenton Varda)
// Based on original Protocol Buffers design by
// Sanjay Ghemawat, Jeff Dean, and others.
//
// This file contains common implementations of the interfaces defined in
// zero_copy_stream.h which are included in the "lite" protobuf library.
// These implementations cover I/O on raw arrays and strings, as well as
// adaptors which make it easy to implement streams based on traditional
// streams. Of course, many users will probably want to write their own
// implementations of these interfaces specific to the particular I/O
// abstractions they prefer to use, but these should cover the most common
// cases.
#ifndef GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_IMPL_LITE_H__
#define GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_IMPL_LITE_H__
#include <iosfwd>
#include <memory>
#include <string>
#include <utility>
#include "google/protobuf/stubs/callback.h"
#include "google/protobuf/stubs/common.h"
#include "absl/base/attributes.h"
#include "absl/strings/cord.h"
#include "absl/strings/cord_buffer.h"
#include "google/protobuf/io/zero_copy_stream.h"
#include "google/protobuf/port.h"
// Must be included last.
#include "google/protobuf/port_def.inc"
namespace google {
namespace protobuf {
namespace io {
// ===================================================================
// A ZeroCopyInputStream backed by an in-memory array of bytes.
class PROTOBUF_EXPORT ArrayInputStream final : public ZeroCopyInputStream {
public:
// Create an InputStream that returns the bytes pointed to by "data".
// "data" remains the property of the caller but must remain valid until
// the stream is destroyed. If a block_size is given, calls to Next()
// will return data blocks no larger than the given size. Otherwise, the
// first call to Next() returns the entire array. block_size is mainly
// useful for testing; in production you would probably never want to set
// it.
ArrayInputStream(const void* data, int size, int block_size = -1);
~ArrayInputStream() override = default;
// `ArrayInputStream` is neither copiable nor assignable
ArrayInputStream(const ArrayInputStream&) = delete;
ArrayInputStream& operator=(const ArrayInputStream&) = delete;
// implements ZeroCopyInputStream ----------------------------------
bool Next(const void** data, int* size) override;
void BackUp(int count) override;
bool Skip(int count) override;
int64_t ByteCount() const override;
private:
const uint8_t* const data_; // The byte array.
const int size_; // Total size of the array.
const int block_size_; // How many bytes to return at a time.
int position_;
int last_returned_size_; // How many bytes we returned last time Next()
// was called (used for error checking only).
};
// ===================================================================
// A ZeroCopyOutputStream backed by an in-memory array of bytes.
class PROTOBUF_EXPORT ArrayOutputStream final : public ZeroCopyOutputStream {
public:
// Create an OutputStream that writes to the bytes pointed to by "data".
// "data" remains the property of the caller but must remain valid until
// the stream is destroyed. If a block_size is given, calls to Next()
// will return data blocks no larger than the given size. Otherwise, the
// first call to Next() returns the entire array. block_size is mainly
// useful for testing; in production you would probably never want to set
// it.
ArrayOutputStream(void* data, int size, int block_size = -1);
~ArrayOutputStream() override = default;
// `ArrayOutputStream` is neither copiable nor assignable
ArrayOutputStream(const ArrayOutputStream&) = delete;
ArrayOutputStream& operator=(const ArrayOutputStream&) = delete;
// implements ZeroCopyOutputStream ---------------------------------
bool Next(void** data, int* size) override;
void BackUp(int count) override;
int64_t ByteCount() const override;
private:
uint8_t* const data_; // The byte array.
const int size_; // Total size of the array.
const int block_size_; // How many bytes to return at a time.
int position_;
int last_returned_size_; // How many bytes we returned last time Next()
// was called (used for error checking only).
};
// ===================================================================
// A ZeroCopyOutputStream which appends bytes to a string.
class PROTOBUF_EXPORT StringOutputStream final : public ZeroCopyOutputStream {
public:
// Create a StringOutputStream which appends bytes to the given string.
// The string remains property of the caller, but it is mutated in arbitrary
// ways and MUST NOT be accessed in any way until you're done with the
// stream. Either be sure there's no further usage, or (safest) destroy the
// stream before using the contents.
//
// Hint: If you call target->reserve(n) before creating the stream,
// the first call to Next() will return at least n bytes of buffer
// space.
explicit StringOutputStream(std::string* target);
~StringOutputStream() override = default;
// `StringOutputStream` is neither copiable nor assignable
StringOutputStream(const StringOutputStream&) = delete;
StringOutputStream& operator=(const StringOutputStream&) = delete;
// implements ZeroCopyOutputStream ---------------------------------
bool Next(void** data, int* size) override;
void BackUp(int count) override;
int64_t ByteCount() const override;
private:
static constexpr size_t kMinimumSize = 16;
std::string* target_;
};
// Note: There is no StringInputStream. Instead, just create an
// ArrayInputStream as follows:
// ArrayInputStream input(str.data(), str.size());
// ===================================================================
// A generic traditional input stream interface.
//
// Lots of traditional input streams (e.g. file descriptors, C stdio
// streams, and C++ iostreams) expose an interface where every read
// involves copying bytes into a buffer. If you want to take such an
// interface and make a ZeroCopyInputStream based on it, simply implement
// CopyingInputStream and then use CopyingInputStreamAdaptor.
//
// CopyingInputStream implementations should avoid buffering if possible.
// CopyingInputStreamAdaptor does its own buffering and will read data
// in large blocks.
class PROTOBUF_EXPORT CopyingInputStream {
public:
virtual ~CopyingInputStream() {}
// Reads up to "size" bytes into the given buffer. Returns the number of
// bytes read. Read() waits until at least one byte is available, or
// returns zero if no bytes will ever become available (EOF), or -1 if a
// permanent read error occurred.
virtual int Read(void* buffer, int size) = 0;
// Skips the next "count" bytes of input. Returns the number of bytes
// actually skipped. This will always be exactly equal to "count" unless
// EOF was reached or a permanent read error occurred.
//
// The default implementation just repeatedly calls Read() into a scratch
// buffer.
virtual int Skip(int count);
};
// A ZeroCopyInputStream which reads from a CopyingInputStream. This is
// useful for implementing ZeroCopyInputStreams that read from traditional
// streams. Note that this class is not really zero-copy.
//
// If you want to read from file descriptors or C++ istreams, this is
// already implemented for you: use FileInputStream or IstreamInputStream
// respectively.
class PROTOBUF_EXPORT CopyingInputStreamAdaptor : public ZeroCopyInputStream {
public:
// Creates a stream that reads from the given CopyingInputStream.
// If a block_size is given, it specifies the number of bytes that
// should be read and returned with each call to Next(). Otherwise,
// a reasonable default is used. The caller retains ownership of
// copying_stream unless SetOwnsCopyingStream(true) is called.
explicit CopyingInputStreamAdaptor(CopyingInputStream* copying_stream,
int block_size = -1);
~CopyingInputStreamAdaptor() override;
// `CopyingInputStreamAdaptor` is neither copiable nor assignable
CopyingInputStreamAdaptor(const CopyingInputStreamAdaptor&) = delete;
CopyingInputStreamAdaptor& operator=(const CopyingInputStreamAdaptor&) = delete;
// Call SetOwnsCopyingStream(true) to tell the CopyingInputStreamAdaptor to
// delete the underlying CopyingInputStream when it is destroyed.
void SetOwnsCopyingStream(bool value) { owns_copying_stream_ = value; }
// implements ZeroCopyInputStream ----------------------------------
bool Next(const void** data, int* size) override;
void BackUp(int count) override;
bool Skip(int count) override;
int64_t ByteCount() const override;
private:
// Insures that buffer_ is not NULL.
void AllocateBufferIfNeeded();
// Frees the buffer and resets buffer_used_.
void FreeBuffer();
// The underlying copying stream.
CopyingInputStream* copying_stream_;
bool owns_copying_stream_;
// True if we have seen a permanent error from the underlying stream.
bool failed_;
// The current position of copying_stream_, relative to the point where
// we started reading.
int64_t position_;
// Data is read into this buffer. It may be NULL if no buffer is currently
// in use. Otherwise, it points to an array of size buffer_size_.
std::unique_ptr<uint8_t[]> buffer_;
const int buffer_size_;
// Number of valid bytes currently in the buffer (i.e. the size last
// returned by Next()). 0 <= buffer_used_ <= buffer_size_.
int buffer_used_;
// Number of bytes in the buffer which were backed up over by a call to
// BackUp(). These need to be returned again.
// 0 <= backup_bytes_ <= buffer_used_
int backup_bytes_;
};
// ===================================================================
// A generic traditional output stream interface.
//
// Lots of traditional output streams (e.g. file descriptors, C stdio
// streams, and C++ iostreams) expose an interface where every write
// involves copying bytes from a buffer. If you want to take such an
// interface and make a ZeroCopyOutputStream based on it, simply implement
// CopyingOutputStream and then use CopyingOutputStreamAdaptor.
//
// CopyingOutputStream implementations should avoid buffering if possible.
// CopyingOutputStreamAdaptor does its own buffering and will write data
// in large blocks.
class PROTOBUF_EXPORT CopyingOutputStream {
public:
virtual ~CopyingOutputStream() {}
// Writes "size" bytes from the given buffer to the output. Returns true
// if successful, false on a write error.
virtual bool Write(const void* buffer, int size) = 0;
};
// A ZeroCopyOutputStream which writes to a CopyingOutputStream. This is
// useful for implementing ZeroCopyOutputStreams that write to traditional
// streams. Note that this class is not really zero-copy.
//
// If you want to write to file descriptors or C++ ostreams, this is
// already implemented for you: use FileOutputStream or OstreamOutputStream
// respectively.
class PROTOBUF_EXPORT CopyingOutputStreamAdaptor : public ZeroCopyOutputStream {
public:
// Creates a stream that writes to the given Unix file descriptor.
// If a block_size is given, it specifies the size of the buffers
// that should be returned by Next(). Otherwise, a reasonable default
// is used.
explicit CopyingOutputStreamAdaptor(CopyingOutputStream* copying_stream,
int block_size = -1);
~CopyingOutputStreamAdaptor() override;
// `CopyingOutputStreamAdaptor` is neither copiable nor assignable
CopyingOutputStreamAdaptor(const CopyingOutputStreamAdaptor&) = delete;
CopyingOutputStreamAdaptor& operator=(const CopyingOutputStreamAdaptor&) = delete;
// Writes all pending data to the underlying stream. Returns false if a
// write error occurred on the underlying stream. (The underlying
// stream itself is not necessarily flushed.)
bool Flush();
// Call SetOwnsCopyingStream(true) to tell the CopyingOutputStreamAdaptor to
// delete the underlying CopyingOutputStream when it is destroyed.
void SetOwnsCopyingStream(bool value) { owns_copying_stream_ = value; }
// implements ZeroCopyOutputStream ---------------------------------
bool Next(void** data, int* size) override;
void BackUp(int count) override;
int64_t ByteCount() const override;
bool WriteAliasedRaw(const void* data, int size) override;
bool AllowsAliasing() const override { return true; }
bool WriteCord(const absl::Cord& cord) override;
private:
// Write the current buffer, if it is present.
bool WriteBuffer();
// Insures that buffer_ is not NULL.
void AllocateBufferIfNeeded();
// Frees the buffer.
void FreeBuffer();
// The underlying copying stream.
CopyingOutputStream* copying_stream_;
bool owns_copying_stream_;
// True if we have seen a permanent error from the underlying stream.
bool failed_;
// The current position of copying_stream_, relative to the point where
// we started writing.
int64_t position_;
// Data is written from this buffer. It may be NULL if no buffer is
// currently in use. Otherwise, it points to an array of size buffer_size_.
std::unique_ptr<uint8_t[]> buffer_;
const int buffer_size_;
// Number of valid bytes currently in the buffer (i.e. the size last
// returned by Next()). When BackUp() is called, we just reduce this.
// 0 <= buffer_used_ <= buffer_size_.
int buffer_used_;
};
// ===================================================================
// A ZeroCopyInputStream which wraps some other stream and limits it to
// a particular byte count.
class PROTOBUF_EXPORT LimitingInputStream final : public ZeroCopyInputStream {
public:
LimitingInputStream(ZeroCopyInputStream* input, int64_t limit);
~LimitingInputStream() override;
// `LimitingInputStream` is neither copiable nor assignable
LimitingInputStream(const LimitingInputStream&) = delete;
LimitingInputStream& operator=(const LimitingInputStream&) = delete;
// implements ZeroCopyInputStream ----------------------------------
bool Next(const void** data, int* size) override;
void BackUp(int count) override;
bool Skip(int count) override;
int64_t ByteCount() const override;
bool ReadCord(absl::Cord* cord, int count) override;
private:
ZeroCopyInputStream* input_;
int64_t limit_; // Decreases as we go, becomes negative if we overshoot.
int64_t prior_bytes_read_; // Bytes read on underlying stream at construction
};
// ===================================================================
// A ZeroCopyInputStream backed by a Cord. This stream implements ReadCord()
// in a way that can share memory between the source and destination cords
// rather than copying.
class PROTOBUF_EXPORT CordInputStream final : public ZeroCopyInputStream {
public:
// Creates an InputStream that reads from the given Cord. `cord` must
// not be null and must outlive this CordInputStream instance. `cord` must
// not be modified while this instance is actively being used: any change
// to `cord` will lead to undefined behavior on any subsequent call into
// this instance.
explicit CordInputStream(
const absl::Cord* cord ABSL_ATTRIBUTE_LIFETIME_BOUND);
// `CordInputStream` is neither copiable nor assignable
CordInputStream(const CordInputStream&) = delete;
CordInputStream& operator=(const CordInputStream&) = delete;
// implements ZeroCopyInputStream ----------------------------------
bool Next(const void** data, int* size) override;
void BackUp(int count) override;
bool Skip(int count) override;
int64_t ByteCount() const override;
bool ReadCord(absl::Cord* cord, int count) override;
private:
// Moves `it_` to the next available chunk skipping `skip` extra bytes
// and updates the chunk data pointers.
bool NextChunk(size_t skip);
// Updates the current chunk data context `data_`, `size_` and `available_`.
// If `bytes_remaining_` is zero, sets `size_` and `available_` to zero.
// Returns true if more data is available, false otherwise.
bool LoadChunkData();
absl::Cord::CharIterator it_;
size_t length_;
size_t bytes_remaining_;
const char* data_;
size_t size_;
size_t available_;
};
// ===================================================================
// A ZeroCopyOutputStream that writes to a Cord. This stream implements
// WriteCord() in a way that can share memory between the source and
// destination cords rather than copying.
class PROTOBUF_EXPORT CordOutputStream final : public ZeroCopyOutputStream {
public:
// Creates an OutputStream streaming serialized data into a Cord. `size_hint`,
// if given, is the expected total size of the resulting Cord. This is a hint
// only, used for optimization. Callers can obtain the generated Cord value by
// invoking `Consume()`.
explicit CordOutputStream(size_t size_hint = 0);
// Creates an OutputStream with an initial Cord value. This constructor can be
// used by applications wanting to directly append serialization data to a
// given cord. In such cases, donating the existing value as in:
//
// CordOutputStream stream(std::move(cord));
// message.SerializeToZeroCopyStream(&stream);
// cord = std::move(stream.Consume());
//
// is more efficient then appending the serialized cord in application code:
//
// CordOutputStream stream;
// message.SerializeToZeroCopyStream(&stream);
// cord.Append(stream.Consume());
//
// The former allows `CordOutputStream` to utilize pre-existing privately
// owned Cord buffers from the donated cord where the latter does not, which
// may lead to more memory usage when serialuzing data into existing cords.
explicit CordOutputStream(absl::Cord cord, size_t size_hint = 0);
// Creates an OutputStream with an initial Cord value and initial buffer.
// This donates both the preexisting cord in `cord`, as well as any
// pre-existing data and additional capacity in `buffer`.
// This function is mainly intended to be used in internal serialization logic
// using eager buffer initialization in EpsCopyOutputStream.
// The donated buffer can be empty, partially empty or full: the outputstream
// will DTRT in all cases and preserve any pre-existing data.
explicit CordOutputStream(absl::Cord cord, absl::CordBuffer buffer,
size_t size_hint = 0);
// Creates an OutputStream with an initial buffer.
// This method is logically identical to, but more efficient than:
// `CordOutputStream(absl::Cord(), std::move(buffer), size_hint)`
explicit CordOutputStream(absl::CordBuffer buffer, size_t size_hint = 0);
// `CordOutputStream` is neither copiable nor assignable
CordOutputStream(const CordOutputStream&) = delete;
CordOutputStream& operator=(const CordOutputStream&) = delete;
// implements `ZeroCopyOutputStream` ---------------------------------
bool Next(void** data, int* size) final;
void BackUp(int count) final;
int64_t ByteCount() const final;
bool WriteCord(const absl::Cord& cord) final;
// Consumes the serialized data as a cord value. `Consume()` internally
// flushes any pending state 'as if' BackUp(0) was called. While a final call
// to BackUp() is generally required by the `ZeroCopyOutputStream` contract,
// applications using `CordOutputStream` directly can call `Consume()` without
// a preceding call to `BackUp()`.
//
// While it will rarely be useful in practice (and especially in the presence
// of size hints) an instance is safe to be used after a call to `Consume()`.
// The only logical change in state is that all serialized data is extracted,
// and any new serialization calls will serialize into new cord data.
absl::Cord Consume();
private:
// State of `buffer_` and 'cord_. As a default CordBuffer instance always has
// inlined capacity, we track state explicitly to avoid returning 'existing
// capacity' from the default or 'moved from' CordBuffer. 'kSteal' indicates
// we should (attempt to) steal the next buffer from the cord.
enum class State { kEmpty, kFull, kPartial, kSteal };
absl::Cord cord_;
size_t size_hint_;
State state_ = State::kEmpty;
absl::CordBuffer buffer_;
};
// ===================================================================
// Return a pointer to mutable characters underlying the given string. The
// return value is valid until the next time the string is resized. We
// trust the caller to treat the return value as an array of length s->size().
inline char* mutable_string_data(std::string* s) {
return &(*s)[0];
}
// as_string_data(s) is equivalent to
// ({ char* p = mutable_string_data(s); make_pair(p, p != NULL); })
// Sometimes it's faster: in some scenarios p cannot be NULL, and then the
// code can avoid that check.
inline std::pair<char*, bool> as_string_data(std::string* s) {
char* p = mutable_string_data(s);
return std::make_pair(p, true);
}
} // namespace io
} // namespace protobuf
} // namespace google
#include "google/protobuf/port_undef.inc"
#endif // GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_IMPL_LITE_H__