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
|
// Copyright (c) 2015 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#ifndef NET_QUIC_STREAM_SEQUENCER_BUFFER_H_
#define NET_QUIC_STREAM_SEQUENCER_BUFFER_H_
// StreamSequencerBuffer implements QuicStreamSequencerBufferInterface.
// It is a circular stream buffer with random write and
// in-sequence read. It consists of an std::vector of pointers pointing
// to memory blocks created as needed and a std::list of Gaps to indicate
// the missing data between the data already written into the buffer.
// - Data are written in with offset indicating where it should be in the
// stream, and the buffer grown as needed (up to the maximum buffer capacity),
// without expensive copying (extra blocks are allocated).
// - Data can be read from the buffer if there is no gap before it,
// and the buffer shrinks as the data are consumed.
// - An upper limit on the number of blocks in the buffer provides an upper
// bound on memory use.
//
// This class is thread-unsafe.
//
// StreamSequencerBuffer maintains a concept of the readable region, which
// contains all written data that has not been read.
// It promises stability of the underlying memory addresses in the readable
// region, so pointers into it can be maintained, and the offset of a pointer
// from the start of the read region can be calculated.
//
// Expected Use:
// StreamSequencerBuffer buffer(2.5 * 8 * 1024);
// std::string source(1024, 'a');
// base::StringPiece std::string_piece(source.data(), source.size());
// size_t written = 0;
// buffer.OnStreamData(800, std::string_piece, GetEpollClockNow(), &written);
// source = std::string{800, 'b'};
// base::StringPiece std::string_piece1(source.data(), 800);
// // Try to write to [1, 801), but should fail due to overlapping,
// // res should be QUIC_INVALID_STREAM_DATA
// auto res = buffer.OnStreamData(1, std::string_piece1, &written));
// // write to [0, 800), res should be QUIC_NO_ERROR
// auto res = buffer.OnStreamData(0, std::string_piece1, GetEpollClockNow(),
// &written);
//
// // Read into a iovec array with total capacity of 120 bytes.
// char dest[120];
// iovec iovecs[3]{iovec{dest, 40}, iovec{dest + 40, 40},
// iovec{dest + 80, 40}};
// size_t read = buffer.Readv(iovecs, 3);
//
// // Get single readable region with timestamp.
// QuicTime t;
// iovec iov;
// buffer.GetReadableRegion(iov, &t);
//
// // Get readable regions from [256, 1024) and consume some of it.
// iovec iovs[2];
// int iov_count = buffer.GetReadableRegions(iovs, 2);
// // Consume some bytes in iovs, returning number of bytes having been
// consumed.
// size_t consumed = consume_iovs(iovs, iov_count);
// buffer.MarkConsumed(consumed);
#include <functional>
#include <list>
#include <memory>
#include "base/macros.h"
#include "net/quic/quic_protocol.h"
#include "net/quic/quic_stream_sequencer_buffer_interface.h"
namespace net {
namespace test {
class StreamSequencerBufferPeer;
} // namespace test
class NET_EXPORT_PRIVATE StreamSequencerBuffer
: public QuicStreamSequencerBufferInterface {
public:
// A Gap indicates a missing chunk of bytes between
// [begin_offset, end_offset) in the stream
struct Gap {
Gap(QuicStreamOffset begin_offset, QuicStreamOffset end_offset);
QuicStreamOffset begin_offset;
QuicStreamOffset end_offset;
};
// A FrameInfo stores the length of a frame and the time it arrived.
struct NET_EXPORT_PRIVATE FrameInfo {
FrameInfo();
FrameInfo(size_t length, QuicTime timestamp);
size_t length;
QuicTime timestamp;
};
// Size of blocks used by this buffer.
// Choose 8K to make block large enough to hold multiple frames, each of
// which could be up to 1.5 KB.
static const size_t kBlockSizeBytes = 8 * 1024; // 8KB
// The basic storage block used by this buffer.
struct BufferBlock {
char buffer[kBlockSizeBytes];
};
explicit StreamSequencerBuffer(size_t max_capacity_bytes);
~StreamSequencerBuffer() override;
// QuicStreamSequencerBufferInterface implementation.
void Clear() override;
bool Empty() const override;
QuicErrorCode OnStreamData(QuicStreamOffset offset,
base::StringPiece data,
QuicTime timestamp,
size_t* bytes_buffered) override;
size_t Readv(const struct iovec* dest_iov, size_t dest_count) override;
int GetReadableRegions(struct iovec* iov, int iov_len) const override;
bool GetReadableRegion(iovec* iov, QuicTime* timestamp) const override;
bool MarkConsumed(size_t bytes_buffered) override;
size_t FlushBufferedFrames() override;
bool HasBytesToRead() const override;
QuicStreamOffset BytesConsumed() const override;
size_t BytesBuffered() const override;
private:
friend class test::StreamSequencerBufferPeer;
// Dispose the given buffer block.
// After calling this method, blocks_[index] is set to nullptr
// in order to indicate that no memory set is allocated for that block.
void RetireBlock(size_t index);
// Should only be called after the indexed block is read till the end of the
// block or a gap has been reached.
// If the block at |block_index| contains no buffered data, then the block is
// retired.
void RetireBlockIfEmpty(size_t block_index);
// Called within OnStreamData() to update the gap OnStreamData() writes into
// (remove, split or change begin/end offset).
void UpdateGapList(std::list<Gap>::iterator gap_with_new_data_written,
QuicStreamOffset start_offset,
size_t bytes_written);
// Calculate the capacity of block at specified index.
// Return value should be either kBlockSizeBytes for non-trailing blocks and
// max_buffer_capacity % kBlockSizeBytes for trailing block.
size_t GetBlockCapacity(size_t index) const;
// Does not check if offset is within reasonable range.
size_t GetBlockIndex(QuicStreamOffset offset) const;
// Given an offset in the stream, return the offset from the beginning of the
// block which contains this data.
size_t GetInBlockOffset(QuicStreamOffset offset) const;
// Get offset relative to index 0 in logical 1st block to start next read.
size_t ReadOffset() const;
// Get the index of the logical 1st block to start next read.
size_t NextBlockToRead() const;
// Returns number of bytes available to be read out.
size_t ReadableBytes() const;
// Called after Readv() and MarkConsumed() to keep frame_arrival_time_map_
// up to date.
// |offset| is the byte next read should start from. All frames before it
// should be removed from the map.
void UpdateFrameArrivalMap(QuicStreamOffset offset);
// The maximum total capacity of this buffer in byte, as constructed.
const size_t max_buffer_capacity_bytes_;
// How many blocks this buffer would need when it reaches full capacity.
const size_t blocks_count_;
// Number of bytes read out of buffer.
QuicStreamOffset total_bytes_read_;
// Contains Gaps which represents currently missing data.
std::list<Gap> gaps_;
// An ordered, variable-length std::list of blocks, with the length limited
// such that the number of blocks never exceeds blocks_count_.
// Each std::list entry can hold up to kBlockSizeBytes bytes.
std::vector<BufferBlock*> blocks_;
// Number of bytes in buffer.
size_t num_bytes_buffered_;
// Stores all the buffered frames' start offset, length and arrival time.
std::map<QuicStreamOffset, FrameInfo> frame_arrival_time_map_;
DISALLOW_COPY_AND_ASSIGN(StreamSequencerBuffer);
};
} // namespace net
#endif // NET_QUIC_STREAM_SEQUENCER_BUFFER_H_
|
