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
|
// Copyright (c) 2012 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.
#include "net/quic/quic_stream_sequencer.h"
#include <algorithm>
#include <limits>
#include "base/logging.h"
#include "net/quic/reliable_quic_stream.h"
using std::min;
using std::numeric_limits;
namespace net {
QuicStreamSequencer::QuicStreamSequencer(ReliableQuicStream* quic_stream)
: stream_(quic_stream),
num_bytes_consumed_(0),
max_frame_memory_(numeric_limits<size_t>::max()),
close_offset_(numeric_limits<QuicStreamOffset>::max()),
half_close_(true) {
}
QuicStreamSequencer::QuicStreamSequencer(size_t max_frame_memory,
ReliableQuicStream* quic_stream)
: stream_(quic_stream),
num_bytes_consumed_(0),
max_frame_memory_(max_frame_memory),
close_offset_(numeric_limits<QuicStreamOffset>::max()),
half_close_(true) {
if (max_frame_memory < kMaxPacketSize) {
LOG(DFATAL) << "Setting max frame memory to " << max_frame_memory
<< ". Some frames will be impossible to handle.";
}
}
QuicStreamSequencer::~QuicStreamSequencer() {
}
bool QuicStreamSequencer::WillAcceptStreamFrame(
const QuicStreamFrame& frame) const {
size_t data_len = frame.data.size();
DCHECK_LE(data_len, max_frame_memory_);
if (IsDuplicate(frame)) {
return true;
}
QuicStreamOffset byte_offset = frame.offset;
if (data_len > max_frame_memory_) {
// We're never going to buffer this frame and we can't pass it up.
// The stream might only consume part of it and we'd need a partial ack.
//
// Ideally this should never happen, as we check that
// max_frame_memory_ > kMaxPacketSize and lower levels should reject
// frames larger than that.
return false;
}
if (byte_offset + data_len - num_bytes_consumed_ > max_frame_memory_) {
// We can buffer this but not right now. Toss it.
// It might be worth trying an experiment where we try best-effort buffering
return false;
}
return true;
}
bool QuicStreamSequencer::OnStreamFrame(const QuicStreamFrame& frame) {
if (!WillAcceptStreamFrame(frame)) {
// This should not happen, as WillAcceptFrame should be called before
// OnStreamFrame. Error handling should be done by the caller.
return false;
}
if (IsDuplicate(frame)) {
// Silently ignore duplicates.
return true;
}
QuicStreamOffset byte_offset = frame.offset;
const char* data = frame.data.data();
size_t data_len = frame.data.size();
if (byte_offset == num_bytes_consumed_) {
DVLOG(1) << "Processing byte offset " << byte_offset;
size_t bytes_consumed = stream_->ProcessData(data, data_len);
num_bytes_consumed_ += bytes_consumed;
if (MaybeCloseStream()) {
return true;
}
if (bytes_consumed > data_len) {
stream_->Close(QUIC_SERVER_ERROR_PROCESSING_STREAM);
return false;
} else if (bytes_consumed == data_len) {
FlushBufferedFrames();
return true; // it's safe to ack this frame.
} else {
// Set ourselves up to buffer what's left
data_len -= bytes_consumed;
data += bytes_consumed;
byte_offset += bytes_consumed;
}
}
DVLOG(1) << "Buffering packet at offset " << byte_offset;
frames_.insert(make_pair(byte_offset, string(data, data_len)));
return true;
}
void QuicStreamSequencer::CloseStreamAtOffset(QuicStreamOffset offset,
bool half_close) {
const QuicStreamOffset kMaxOffset = numeric_limits<QuicStreamOffset>::max();
// If we have a scheduled termination or close, any new offset should match
// it.
if (close_offset_ != kMaxOffset && offset != close_offset_) {
stream_->Close(QUIC_MULTIPLE_TERMINATION_OFFSETS);
return;
}
close_offset_ = offset;
// Full close overrides half close.
if (half_close == false) {
half_close_ = false;
}
MaybeCloseStream();
}
bool QuicStreamSequencer::MaybeCloseStream() {
if (IsHalfClosed()) {
DVLOG(1) << "Passing up termination, as we've processed "
<< num_bytes_consumed_ << " of " << close_offset_
<< " bytes.";
// Technically it's an error if num_bytes_consumed isn't exactly
// equal, but error handling seems silly at this point.
stream_->TerminateFromPeer(half_close_);
return true;
}
return false;
}
bool QuicStreamSequencer::HasBytesToRead() const {
FrameMap::const_iterator it = frames_.begin();
return it != frames_.end() && it->first == num_bytes_consumed_;
}
bool QuicStreamSequencer::IsHalfClosed() const {
return num_bytes_consumed_ >= close_offset_;
}
bool QuicStreamSequencer::IsClosed() const {
return num_bytes_consumed_ >= close_offset_ && half_close_ == false;
}
bool QuicStreamSequencer::IsDuplicate(const QuicStreamFrame& frame) const {
// A frame is duplicate if the frame offset is smaller than our bytes consumed
// or we have stored the frame in our map.
// TODO(pwestin): Is it possible that a new frame contain more data even if
// the offset is the same?
return (frame.offset < num_bytes_consumed_ ||
frames_.find(frame.offset) != frames_.end());
}
void QuicStreamSequencer::FlushBufferedFrames() {
FrameMap::iterator it = frames_.find(num_bytes_consumed_);
while (it != frames_.end()) {
DVLOG(1) << "Flushing buffered packet at offset " << it->first;
string* data = &it->second;
size_t bytes_consumed = stream_->ProcessData(data->c_str(), data->size());
num_bytes_consumed_ += bytes_consumed;
if (MaybeCloseStream()) {
return;
}
if (bytes_consumed > data->size()) {
stream_->Close(QUIC_SERVER_ERROR_PROCESSING_STREAM); // Programming error
return;
} else if (bytes_consumed == data->size()) {
frames_.erase(it);
it = frames_.find(num_bytes_consumed_);
} else {
string new_data = it->second.substr(bytes_consumed);
frames_.erase(it);
frames_.insert(make_pair(num_bytes_consumed_, new_data));
return;
}
}
}
} // namespace net
|
