1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
5 #include "net/quic/reliable_quic_stream.h"
7 #include "base/logging.h"
8 #include "base/profiler/scoped_tracker.h"
9 #include "net/quic/iovector.h"
10 #include "net/quic/quic_flow_controller.h"
11 #include "net/quic/quic_session.h"
12 #include "net/quic/quic_write_blocked_list.h"
14 using base::StringPiece
;
20 #define ENDPOINT (is_server_ ? "Server: " : " Client: ")
24 struct iovec
MakeIovec(StringPiece data
) {
25 struct iovec iov
= {const_cast<char*>(data
.data()),
26 static_cast<size_t>(data
.size())};
30 size_t GetInitialStreamFlowControlWindowToSend(QuicSession
* session
) {
31 return session
->config()->GetInitialStreamFlowControlWindowToSend();
34 size_t GetReceivedFlowControlWindow(QuicSession
* session
) {
35 if (session
->config()->HasReceivedInitialStreamFlowControlWindowBytes()) {
36 return session
->config()->ReceivedInitialStreamFlowControlWindowBytes();
39 return kMinimumFlowControlSendWindow
;
44 // Wrapper that aggregates OnAckNotifications for packets sent using
45 // WriteOrBufferData and delivers them to the original
46 // QuicAckNotifier::DelegateInterface after all bytes written using
47 // WriteOrBufferData are acked. This level of indirection is
48 // necessary because the delegate interface provides no mechanism that
49 // WriteOrBufferData can use to inform it that the write required
50 // multiple WritevData calls or that only part of the data has been
51 // sent out by the time ACKs start arriving.
52 class ReliableQuicStream::ProxyAckNotifierDelegate
53 : public QuicAckNotifier::DelegateInterface
{
55 explicit ProxyAckNotifierDelegate(DelegateInterface
* delegate
)
56 : delegate_(delegate
),
58 wrote_last_data_(false),
59 num_retransmitted_packets_(0),
60 num_retransmitted_bytes_(0) {
63 void OnAckNotification(int num_retransmitted_packets
,
64 int num_retransmitted_bytes
,
65 QuicTime::Delta delta_largest_observed
) override
{
66 DCHECK_LT(0, pending_acks_
);
68 num_retransmitted_packets_
+= num_retransmitted_packets
;
69 num_retransmitted_bytes_
+= num_retransmitted_bytes
;
71 if (wrote_last_data_
&& pending_acks_
== 0) {
72 delegate_
->OnAckNotification(num_retransmitted_packets_
,
73 num_retransmitted_bytes_
,
74 delta_largest_observed
);
78 void WroteData(bool last_data
) {
79 DCHECK(!wrote_last_data_
);
81 wrote_last_data_
= last_data
;
85 // Delegates are ref counted.
86 ~ProxyAckNotifierDelegate() override
{}
89 // Original delegate. delegate_->OnAckNotification will be called when:
90 // wrote_last_data_ == true and pending_acks_ == 0
91 scoped_refptr
<DelegateInterface
> delegate_
;
93 // Number of outstanding acks.
96 // True if no pending writes remain.
97 bool wrote_last_data_
;
100 int num_original_packets_
;
101 int num_original_bytes_
;
102 int num_retransmitted_packets_
;
103 int num_retransmitted_bytes_
;
105 DISALLOW_COPY_AND_ASSIGN(ProxyAckNotifierDelegate
);
108 ReliableQuicStream::PendingData::PendingData(
109 string data_in
, scoped_refptr
<ProxyAckNotifierDelegate
> delegate_in
)
110 : data(data_in
), delegate(delegate_in
) {
113 ReliableQuicStream::PendingData::~PendingData() {
116 ReliableQuicStream::ReliableQuicStream(QuicStreamId id
, QuicSession
* session
)
120 stream_bytes_read_(0),
121 stream_bytes_written_(0),
122 stream_error_(QUIC_STREAM_NO_ERROR
),
123 connection_error_(QUIC_NO_ERROR
),
124 read_side_closed_(false),
125 write_side_closed_(false),
126 fin_buffered_(false),
128 fin_received_(false),
130 rst_received_(false),
131 fec_policy_(FEC_PROTECT_OPTIONAL
),
132 is_server_(session_
->is_server()),
134 session_
->connection(), id_
, is_server_
,
135 GetReceivedFlowControlWindow(session
),
136 GetInitialStreamFlowControlWindowToSend(session
),
137 GetInitialStreamFlowControlWindowToSend(session
)),
138 connection_flow_controller_(session_
->flow_controller()),
139 stream_contributes_to_connection_flow_control_(true) {
142 ReliableQuicStream::~ReliableQuicStream() {
145 void ReliableQuicStream::OnStreamFrame(const QuicStreamFrame
& frame
) {
146 if (read_side_closed_
) {
147 DVLOG(1) << ENDPOINT
<< "Ignoring frame " << frame
.stream_id
;
148 // We don't want to be reading: blackhole the data.
152 if (frame
.stream_id
!= id_
) {
153 session_
->connection()->SendConnectionClose(QUIC_INTERNAL_ERROR
);
158 fin_received_
= true;
161 // This count include duplicate data received.
162 size_t frame_payload_size
= frame
.data
.TotalBufferSize();
163 stream_bytes_read_
+= frame_payload_size
;
165 // Flow control is interested in tracking highest received offset.
166 if (MaybeIncreaseHighestReceivedOffset(frame
.offset
+ frame_payload_size
)) {
167 // As the highest received offset has changed, we should check to see if
168 // this is a violation of flow control.
169 if (flow_controller_
.FlowControlViolation() ||
170 connection_flow_controller_
->FlowControlViolation()) {
171 session_
->connection()->SendConnectionClose(
172 QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA
);
177 sequencer_
.OnStreamFrame(frame
);
180 int ReliableQuicStream::num_frames_received() const {
181 return sequencer_
.num_frames_received();
184 int ReliableQuicStream::num_duplicate_frames_received() const {
185 return sequencer_
.num_duplicate_frames_received();
188 void ReliableQuicStream::OnStreamReset(const QuicRstStreamFrame
& frame
) {
189 rst_received_
= true;
190 MaybeIncreaseHighestReceivedOffset(frame
.byte_offset
);
192 stream_error_
= frame
.error_code
;
197 void ReliableQuicStream::OnConnectionClosed(QuicErrorCode error
,
199 if (read_side_closed_
&& write_side_closed_
) {
202 if (error
!= QUIC_NO_ERROR
) {
203 stream_error_
= QUIC_STREAM_CONNECTION_ERROR
;
204 connection_error_
= error
;
211 void ReliableQuicStream::OnFinRead() {
212 DCHECK(sequencer_
.IsClosed());
213 fin_received_
= true;
217 void ReliableQuicStream::Reset(QuicRstStreamErrorCode error
) {
218 DCHECK_NE(QUIC_STREAM_NO_ERROR
, error
);
219 stream_error_
= error
;
220 // Sending a RstStream results in calling CloseStream.
221 session()->SendRstStream(id(), error
, stream_bytes_written_
);
225 void ReliableQuicStream::CloseConnection(QuicErrorCode error
) {
226 // TODO(vadimt): Remove ScopedTracker below once crbug.com/422516 is fixed.
227 tracked_objects::ScopedTracker
tracking_profile(
228 FROM_HERE_WITH_EXPLICIT_FUNCTION(
229 "422516 ReliableQuicStream::CloseConnection"));
231 session()->connection()->SendConnectionClose(error
);
234 void ReliableQuicStream::CloseConnectionWithDetails(QuicErrorCode error
,
235 const string
& details
) {
236 session()->connection()->SendConnectionCloseWithDetails(error
, details
);
239 QuicVersion
ReliableQuicStream::version() const {
240 return session()->connection()->version();
243 void ReliableQuicStream::WriteOrBufferData(
246 QuicAckNotifier::DelegateInterface
* ack_notifier_delegate
) {
247 if (data
.empty() && !fin
) {
248 LOG(DFATAL
) << "data.empty() && !fin";
253 LOG(DFATAL
) << "Fin already buffered";
257 scoped_refptr
<ProxyAckNotifierDelegate
> proxy_delegate
;
258 if (ack_notifier_delegate
!= nullptr) {
259 proxy_delegate
= new ProxyAckNotifierDelegate(ack_notifier_delegate
);
262 QuicConsumedData
consumed_data(0, false);
265 if (queued_data_
.empty()) {
266 struct iovec
iov(MakeIovec(data
));
267 consumed_data
= WritevData(&iov
, 1, fin
, proxy_delegate
.get());
268 DCHECK_LE(consumed_data
.bytes_consumed
, data
.length());
271 bool write_completed
;
272 // If there's unconsumed data or an unconsumed fin, queue it.
273 if (consumed_data
.bytes_consumed
< data
.length() ||
274 (fin
&& !consumed_data
.fin_consumed
)) {
275 StringPiece
remainder(data
.substr(consumed_data
.bytes_consumed
));
276 queued_data_
.push_back(PendingData(remainder
.as_string(), proxy_delegate
));
277 write_completed
= false;
279 write_completed
= true;
282 if ((proxy_delegate
.get() != nullptr) &&
283 (consumed_data
.bytes_consumed
> 0 || consumed_data
.fin_consumed
)) {
284 proxy_delegate
->WroteData(write_completed
);
288 void ReliableQuicStream::OnCanWrite() {
290 while (!queued_data_
.empty()) {
291 PendingData
* pending_data
= &queued_data_
.front();
292 ProxyAckNotifierDelegate
* delegate
= pending_data
->delegate
.get();
293 if (queued_data_
.size() == 1 && fin_buffered_
) {
296 struct iovec
iov(MakeIovec(pending_data
->data
));
297 QuicConsumedData consumed_data
= WritevData(&iov
, 1, fin
, delegate
);
298 if (consumed_data
.bytes_consumed
== pending_data
->data
.size() &&
299 fin
== consumed_data
.fin_consumed
) {
300 queued_data_
.pop_front();
301 if (delegate
!= nullptr) {
302 delegate
->WroteData(true);
305 if (consumed_data
.bytes_consumed
> 0) {
306 pending_data
->data
.erase(0, consumed_data
.bytes_consumed
);
307 if (delegate
!= nullptr) {
308 delegate
->WroteData(false);
316 void ReliableQuicStream::MaybeSendBlocked() {
317 flow_controller_
.MaybeSendBlocked();
318 if (!stream_contributes_to_connection_flow_control_
) {
321 connection_flow_controller_
->MaybeSendBlocked();
322 // If we are connection level flow control blocked, then add the stream
323 // to the write blocked list. It will be given a chance to write when a
324 // connection level WINDOW_UPDATE arrives.
325 if (connection_flow_controller_
->IsBlocked() &&
326 !flow_controller_
.IsBlocked()) {
327 session_
->MarkWriteBlocked(id(), EffectivePriority());
331 QuicConsumedData
ReliableQuicStream::WritevData(
332 const struct iovec
* iov
,
335 QuicAckNotifier::DelegateInterface
* ack_notifier_delegate
) {
336 if (write_side_closed_
) {
337 DLOG(ERROR
) << ENDPOINT
<< "Attempt to write when the write side is closed";
338 return QuicConsumedData(0, false);
341 // How much data we want to write.
342 size_t write_length
= TotalIovecLength(iov
, iov_count
);
344 // A FIN with zero data payload should not be flow control blocked.
345 bool fin_with_zero_data
= (fin
&& write_length
== 0);
347 if (flow_controller_
.IsEnabled()) {
348 // How much data we are allowed to write from flow control.
349 QuicByteCount send_window
= flow_controller_
.SendWindowSize();
350 if (stream_contributes_to_connection_flow_control_
) {
352 min(send_window
, connection_flow_controller_
->SendWindowSize());
355 if (send_window
== 0 && !fin_with_zero_data
) {
356 // Quick return if we can't send anything.
358 return QuicConsumedData(0, false);
361 if (write_length
> send_window
) {
362 // Don't send the FIN if we aren't going to send all the data.
365 // Writing more data would be a violation of flow control.
366 write_length
= static_cast<size_t>(send_window
);
370 // Fill an IOVector with bytes from the iovec.
372 data
.AppendIovecAtMostBytes(iov
, iov_count
, write_length
);
374 QuicConsumedData consumed_data
= session()->WritevData(
375 id(), data
, stream_bytes_written_
, fin
, GetFecProtection(),
376 ack_notifier_delegate
);
377 stream_bytes_written_
+= consumed_data
.bytes_consumed
;
379 AddBytesSent(consumed_data
.bytes_consumed
);
381 if (consumed_data
.bytes_consumed
== write_length
) {
382 if (!fin_with_zero_data
) {
385 if (fin
&& consumed_data
.fin_consumed
) {
388 } else if (fin
&& !consumed_data
.fin_consumed
) {
389 session_
->MarkWriteBlocked(id(), EffectivePriority());
392 session_
->MarkWriteBlocked(id(), EffectivePriority());
394 return consumed_data
;
397 FecProtection
ReliableQuicStream::GetFecProtection() {
398 return fec_policy_
== FEC_PROTECT_ALWAYS
? MUST_FEC_PROTECT
: MAY_FEC_PROTECT
;
401 void ReliableQuicStream::CloseReadSide() {
402 if (read_side_closed_
) {
405 DVLOG(1) << ENDPOINT
<< "Done reading from stream " << id();
407 read_side_closed_
= true;
408 if (write_side_closed_
) {
409 DVLOG(1) << ENDPOINT
<< "Closing stream: " << id();
410 session_
->CloseStream(id());
414 void ReliableQuicStream::CloseWriteSide() {
415 if (write_side_closed_
) {
418 DVLOG(1) << ENDPOINT
<< "Done writing to stream " << id();
420 write_side_closed_
= true;
421 if (read_side_closed_
) {
422 DVLOG(1) << ENDPOINT
<< "Closing stream: " << id();
423 session_
->CloseStream(id());
427 bool ReliableQuicStream::HasBufferedData() const {
428 return !queued_data_
.empty();
431 void ReliableQuicStream::OnClose() {
435 if (!fin_sent_
&& !rst_sent_
) {
436 // For flow control accounting, we must tell the peer how many bytes we have
437 // written on this stream before termination. Done here if needed, using a
439 DVLOG(1) << ENDPOINT
<< "Sending RST in OnClose: " << id();
440 session_
->SendRstStream(id(), QUIC_RST_ACKNOWLEDGEMENT
,
441 stream_bytes_written_
);
445 // We are closing the stream and will not process any further incoming bytes.
446 // As there may be more bytes in flight and we need to ensure that both
447 // endpoints have the same connection level flow control state, mark all
448 // unreceived or buffered bytes as consumed.
449 QuicByteCount bytes_to_consume
=
450 flow_controller_
.highest_received_byte_offset() -
451 flow_controller_
.bytes_consumed();
452 AddBytesConsumed(bytes_to_consume
);
455 void ReliableQuicStream::OnWindowUpdateFrame(
456 const QuicWindowUpdateFrame
& frame
) {
457 if (!flow_controller_
.IsEnabled()) {
458 DLOG(DFATAL
) << "Flow control not enabled! " << version();
461 if (flow_controller_
.UpdateSendWindowOffset(frame
.byte_offset
)) {
462 // We can write again!
463 // TODO(rjshade): This does not respect priorities (e.g. multiple
464 // outstanding POSTs are unblocked on arrival of
465 // SHLO with initial window).
466 // As long as the connection is not flow control blocked, we can write!
471 bool ReliableQuicStream::MaybeIncreaseHighestReceivedOffset(
472 QuicStreamOffset new_offset
) {
473 if (!flow_controller_
.IsEnabled()) {
477 new_offset
- flow_controller_
.highest_received_byte_offset();
478 if (!flow_controller_
.UpdateHighestReceivedOffset(new_offset
)) {
482 // If |new_offset| increased the stream flow controller's highest received
483 // offset, then we need to increase the connection flow controller's value
484 // by the incremental difference.
485 if (stream_contributes_to_connection_flow_control_
) {
486 connection_flow_controller_
->UpdateHighestReceivedOffset(
487 connection_flow_controller_
->highest_received_byte_offset() +
493 void ReliableQuicStream::AddBytesSent(QuicByteCount bytes
) {
494 if (flow_controller_
.IsEnabled()) {
495 flow_controller_
.AddBytesSent(bytes
);
496 if (stream_contributes_to_connection_flow_control_
) {
497 connection_flow_controller_
->AddBytesSent(bytes
);
502 void ReliableQuicStream::AddBytesConsumed(QuicByteCount bytes
) {
503 if (flow_controller_
.IsEnabled()) {
504 // Only adjust stream level flow controller if we are still reading.
505 if (!read_side_closed_
) {
506 flow_controller_
.AddBytesConsumed(bytes
);
509 if (stream_contributes_to_connection_flow_control_
) {
510 connection_flow_controller_
->AddBytesConsumed(bytes
);
515 void ReliableQuicStream::UpdateSendWindowOffset(QuicStreamOffset new_window
) {
516 if (flow_controller_
.UpdateSendWindowOffset(new_window
)) {
521 bool ReliableQuicStream::IsFlowControlBlocked() {
522 if (flow_controller_
.IsBlocked()) {
525 return stream_contributes_to_connection_flow_control_
&&
526 connection_flow_controller_
->IsBlocked();