1 // Protocol Buffers - Google's data interchange format
2 // Copyright 2008 Google Inc. All rights reserved.
3 //
4 // Use of this source code is governed by a BSD-style
5 // license that can be found in the LICENSE file or at
6 // https://developers.google.com/open-source/licenses/bsd
7
8 #include "google/protobuf/json/internal/zero_copy_buffered_stream.h"
9
10 #include <algorithm>
11 #include <iterator>
12 #include <string>
13 #include <utility>
14
15 #include "absl/algorithm/container.h"
16 #include "absl/strings/str_format.h"
17 #include "absl/strings/string_view.h"
18 #include "google/protobuf/stubs/status_macros.h"
19
20 // Must be included last.
21 #include "google/protobuf/port_def.inc"
22
23 namespace google {
24 namespace protobuf {
25 namespace json_internal {
Advance(size_t bytes)26 absl::Status ZeroCopyBufferedStream::Advance(size_t bytes) {
27 while (bytes != 0) {
28 if (Unread().empty() && !ReadChunk()) {
29 return absl::InvalidArgumentError("unexpected EOF");
30 }
31 size_t to_skip = std::min(bytes, Unread().size());
32 cursor_ += to_skip;
33 bytes -= to_skip;
34 }
35
36 if (using_buf_) {
37 ABSL_DCHECK_LE(cursor_, buffer_start_ + buf_.size());
38 } else {
39 ABSL_DCHECK_LE(cursor_, last_chunk_.size());
40 }
41
42 return absl::OkStatus();
43 }
44
BufferAtLeast(size_t bytes)45 absl::StatusOr<BufferingGuard> ZeroCopyBufferedStream::BufferAtLeast(
46 size_t bytes) {
47 // This MUST be an empty guard before the first call to ReadChunk();
48 // otherwise we risk unconditional buffering.
49 BufferingGuard guard;
50 while (Unread().size() < bytes) {
51 if (!Unread().empty()) {
52 // We must buffer before reading if Unread() is nonempty; otherwise we
53 // risk discarding part of the unread buffer. However, we must NOT
54 // buffer before calling ReadChunk if it *is* empty, because then we
55 // would buffer unconditionally.
56 //
57 // There are tests to verify both of these cases.
58 guard = BufferingGuard(this);
59 }
60 if (!ReadChunk()) {
61 return absl::InvalidArgumentError("unexpected EOF");
62 }
63 guard = BufferingGuard(this);
64 }
65 ABSL_DCHECK_GE(Unread().size(), bytes);
66 return BufferingGuard(this);
67 }
68
DownRefBuffer()69 void ZeroCopyBufferedStream::DownRefBuffer() {
70 ABSL_DCHECK_GT(outstanding_buffer_borrows_, 0);
71
72 --outstanding_buffer_borrows_;
73 if (outstanding_buffer_borrows_ > 0 || !using_buf_) {
74 return;
75 }
76
77 // If we have hit EOF then that means we might be buffering one or more
78 // chunks of data that we have not yet logically advanced through. We need to
79 // leave the buffer in place to ensure that we do not inadvertently drop such
80 // chunks.
81 if (eof_) {
82 return;
83 }
84
85 // The "virtual length" is the size of the buffer cursor_ indexes into, which
86 // is bigger than buf_.
87 size_t virtual_buf_len = buf_.size() + buffer_start_;
88 size_t last_chunk_in_buf = virtual_buf_len - last_chunk_.size();
89 // If we are inside of `last_chunk_`, set the cursor there; otherwise, we have
90 // a dangling reference somewhere.
91 ABSL_DCHECK_LE(last_chunk_in_buf, virtual_buf_len) << absl::StrFormat(
92 "%d, %d, %d", buf_.size(), last_chunk_.size(), buffer_start_);
93 if (cursor_ <= last_chunk_in_buf) {
94 cursor_ = 0;
95 } else {
96 cursor_ -= last_chunk_in_buf;
97 }
98 buf_.clear();
99 using_buf_ = false;
100 }
101
ReadChunk()102 bool ZeroCopyBufferedStream::ReadChunk() {
103 if (eof_) {
104 return false;
105 }
106 // We are buffering a second chunk, so we need to put the current chunk
107 // into the buffer.
108 if (outstanding_buffer_borrows_ > 0 && !using_buf_) {
109 absl::c_copy(RawBuffer(buffer_start_), std::back_inserter(buf_));
110 using_buf_ = true;
111 }
112
113 const void* data;
114 int len;
115 if (!stream_->Next(&data, &len)) {
116 eof_ = true;
117 return false;
118 }
119
120 last_chunk_ = absl::string_view(static_cast<const char*>(data),
121 static_cast<size_t>(len));
122 if (using_buf_) {
123 absl::c_copy(last_chunk_, std::back_inserter(buf_));
124 // Cursor does not need to move, because it is still inside of `buf_`.
125 } else {
126 cursor_ = 0;
127 buffer_start_ = 0;
128 }
129 return true;
130 }
131 } // namespace json_internal
132 } // namespace protobuf
133 } // namespace google
134
135 #include "google/protobuf/port_undef.inc"
136