reliable_quic_stream.cc revision 6d86b77056ed63eb6871182f42a9fd5f07550f90 
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.
4
5#include "net/quic/reliable_quic_stream.h"
6
7#include "base/logging.h"
8#include "net/quic/iovector.h"
9#include "net/quic/quic_flow_controller.h"
10#include "net/quic/quic_session.h"
11#include "net/quic/quic_write_blocked_list.h"
12
13using base::StringPiece;
14using std::min;
15
16namespace net {
17
18#define ENDPOINT (is_server_ ? "Server: " : " Client: ")
19
20namespace {
21
22struct iovec MakeIovec(StringPiece data) {
23  struct iovec iov = {const_cast<char*>(data.data()),
24                      static_cast<size_t>(data.size())};
25  return iov;
26}
27
28size_t GetInitialStreamFlowControlWindowToSend(QuicSession* session) {
29  QuicVersion version = session->connection()->version();
30  if (version <= QUIC_VERSION_19) {
31    return session->config()->GetInitialFlowControlWindowToSend();
32  }
33
34  return session->config()->GetInitialStreamFlowControlWindowToSend();
35}
36
37size_t GetReceivedFlowControlWindow(QuicSession* session) {
38  QuicVersion version = session->connection()->version();
39  if (version <= QUIC_VERSION_19) {
40    if (session->config()->HasReceivedInitialFlowControlWindowBytes()) {
41      return session->config()->ReceivedInitialFlowControlWindowBytes();
42    }
43
44    return kDefaultFlowControlSendWindow;
45  }
46
47  // Version must be >= QUIC_VERSION_20, so we check for stream specific flow
48  // control window.
49  if (session->config()->HasReceivedInitialStreamFlowControlWindowBytes()) {
50    return session->config()->ReceivedInitialStreamFlowControlWindowBytes();
51  }
52
53  return kDefaultFlowControlSendWindow;
54}
55
56}  // namespace
57
58// Wrapper that aggregates OnAckNotifications for packets sent using
59// WriteOrBufferData and delivers them to the original
60// QuicAckNotifier::DelegateInterface after all bytes written using
61// WriteOrBufferData are acked.  This level of indirection is
62// necessary because the delegate interface provides no mechanism that
63// WriteOrBufferData can use to inform it that the write required
64// multiple WritevData calls or that only part of the data has been
65// sent out by the time ACKs start arriving.
66class ReliableQuicStream::ProxyAckNotifierDelegate
67    : public QuicAckNotifier::DelegateInterface {
68 public:
69  explicit ProxyAckNotifierDelegate(DelegateInterface* delegate)
70      : delegate_(delegate),
71        pending_acks_(0),
72        wrote_last_data_(false),
73        num_original_packets_(0),
74        num_original_bytes_(0),
75        num_retransmitted_packets_(0),
76        num_retransmitted_bytes_(0) {
77  }
78
79  virtual void OnAckNotification(int num_original_packets,
80                                 int num_original_bytes,
81                                 int num_retransmitted_packets,
82                                 int num_retransmitted_bytes,
83                                 QuicTime::Delta delta_largest_observed)
84      OVERRIDE {
85    DCHECK_LT(0, pending_acks_);
86    --pending_acks_;
87    num_original_packets_ += num_original_packets;
88    num_original_bytes_ += num_original_bytes;
89    num_retransmitted_packets_ += num_retransmitted_packets;
90    num_retransmitted_bytes_ += num_retransmitted_bytes;
91
92    if (wrote_last_data_ && pending_acks_ == 0) {
93      delegate_->OnAckNotification(num_original_packets_,
94                                   num_original_bytes_,
95                                   num_retransmitted_packets_,
96                                   num_retransmitted_bytes_,
97                                   delta_largest_observed);
98    }
99  }
100
101  void WroteData(bool last_data) {
102    DCHECK(!wrote_last_data_);
103    ++pending_acks_;
104    wrote_last_data_ = last_data;
105  }
106
107 protected:
108  // Delegates are ref counted.
109  virtual ~ProxyAckNotifierDelegate() OVERRIDE {
110  }
111
112 private:
113  // Original delegate.  delegate_->OnAckNotification will be called when:
114  //   wrote_last_data_ == true and pending_acks_ == 0
115  scoped_refptr<DelegateInterface> delegate_;
116
117  // Number of outstanding acks.
118  int pending_acks_;
119
120  // True if no pending writes remain.
121  bool wrote_last_data_;
122
123  // Accumulators.
124  int num_original_packets_;
125  int num_original_bytes_;
126  int num_retransmitted_packets_;
127  int num_retransmitted_bytes_;
128
129  DISALLOW_COPY_AND_ASSIGN(ProxyAckNotifierDelegate);
130};
131
132ReliableQuicStream::PendingData::PendingData(
133    string data_in, scoped_refptr<ProxyAckNotifierDelegate> delegate_in)
134    : data(data_in), delegate(delegate_in) {
135}
136
137ReliableQuicStream::PendingData::~PendingData() {
138}
139
140ReliableQuicStream::ReliableQuicStream(QuicStreamId id, QuicSession* session)
141    : sequencer_(this),
142      id_(id),
143      session_(session),
144      stream_bytes_read_(0),
145      stream_bytes_written_(0),
146      stream_error_(QUIC_STREAM_NO_ERROR),
147      connection_error_(QUIC_NO_ERROR),
148      read_side_closed_(false),
149      write_side_closed_(false),
150      fin_buffered_(false),
151      fin_sent_(false),
152      fin_received_(false),
153      rst_sent_(false),
154      rst_received_(false),
155      fec_policy_(FEC_PROTECT_OPTIONAL),
156      is_server_(session_->is_server()),
157      flow_controller_(
158          session_->connection(), id_, is_server_,
159          GetReceivedFlowControlWindow(session),
160          GetInitialStreamFlowControlWindowToSend(session),
161          GetInitialStreamFlowControlWindowToSend(session)),
162      connection_flow_controller_(session_->flow_controller()) {
163}
164
165ReliableQuicStream::~ReliableQuicStream() {
166}
167
168bool ReliableQuicStream::OnStreamFrame(const QuicStreamFrame& frame) {
169  if (read_side_closed_) {
170    DVLOG(1) << ENDPOINT << "Ignoring frame " << frame.stream_id;
171    // We don't want to be reading: blackhole the data.
172    return true;
173  }
174
175  if (frame.stream_id != id_) {
176    LOG(ERROR) << "Error!";
177    return false;
178  }
179
180  if (frame.fin) {
181    fin_received_ = true;
182  }
183
184  // This count include duplicate data received.
185  size_t frame_payload_size = frame.data.TotalBufferSize();
186  stream_bytes_read_ += frame_payload_size;
187
188  // Flow control is interested in tracking highest received offset.
189  if (MaybeIncreaseHighestReceivedOffset(frame.offset + frame_payload_size)) {
190    // As the highest received offset has changed, we should check to see if
191    // this is a violation of flow control.
192    if (flow_controller_.FlowControlViolation() ||
193        connection_flow_controller_->FlowControlViolation()) {
194      session_->connection()->SendConnectionClose(
195          QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA);
196      return false;
197    }
198  }
199
200  return sequencer_.OnStreamFrame(frame);
201}
202
203int ReliableQuicStream::num_frames_received() const {
204  return sequencer_.num_frames_received();
205}
206
207int ReliableQuicStream::num_duplicate_frames_received() const {
208  return sequencer_.num_duplicate_frames_received();
209}
210
211void ReliableQuicStream::OnStreamReset(const QuicRstStreamFrame& frame) {
212  rst_received_ = true;
213  MaybeIncreaseHighestReceivedOffset(frame.byte_offset);
214
215  stream_error_ = frame.error_code;
216  CloseWriteSide();
217  CloseReadSide();
218}
219
220void ReliableQuicStream::OnConnectionClosed(QuicErrorCode error,
221                                            bool from_peer) {
222  if (read_side_closed_ && write_side_closed_) {
223    return;
224  }
225  if (error != QUIC_NO_ERROR) {
226    stream_error_ = QUIC_STREAM_CONNECTION_ERROR;
227    connection_error_ = error;
228  }
229
230  CloseWriteSide();
231  CloseReadSide();
232}
233
234void ReliableQuicStream::OnFinRead() {
235  DCHECK(sequencer_.IsClosed());
236  CloseReadSide();
237}
238
239void ReliableQuicStream::Reset(QuicRstStreamErrorCode error) {
240  DCHECK_NE(QUIC_STREAM_NO_ERROR, error);
241  stream_error_ = error;
242  // Sending a RstStream results in calling CloseStream.
243  session()->SendRstStream(id(), error, stream_bytes_written_);
244  rst_sent_ = true;
245}
246
247void ReliableQuicStream::CloseConnection(QuicErrorCode error) {
248  session()->connection()->SendConnectionClose(error);
249}
250
251void ReliableQuicStream::CloseConnectionWithDetails(QuicErrorCode error,
252                                                    const string& details) {
253  session()->connection()->SendConnectionCloseWithDetails(error, details);
254}
255
256QuicVersion ReliableQuicStream::version() const {
257  return session()->connection()->version();
258}
259
260void ReliableQuicStream::WriteOrBufferData(
261    StringPiece data,
262    bool fin,
263    QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
264  if (data.empty() && !fin) {
265    LOG(DFATAL) << "data.empty() && !fin";
266    return;
267  }
268
269  if (fin_buffered_) {
270    LOG(DFATAL) << "Fin already buffered";
271    return;
272  }
273
274  scoped_refptr<ProxyAckNotifierDelegate> proxy_delegate;
275  if (ack_notifier_delegate != NULL) {
276    proxy_delegate = new ProxyAckNotifierDelegate(ack_notifier_delegate);
277  }
278
279  QuicConsumedData consumed_data(0, false);
280  fin_buffered_ = fin;
281
282  if (queued_data_.empty()) {
283    struct iovec iov(MakeIovec(data));
284    consumed_data = WritevData(&iov, 1, fin, proxy_delegate.get());
285    DCHECK_LE(consumed_data.bytes_consumed, data.length());
286  }
287
288  bool write_completed;
289  // If there's unconsumed data or an unconsumed fin, queue it.
290  if (consumed_data.bytes_consumed < data.length() ||
291      (fin && !consumed_data.fin_consumed)) {
292    StringPiece remainder(data.substr(consumed_data.bytes_consumed));
293    queued_data_.push_back(PendingData(remainder.as_string(), proxy_delegate));
294    write_completed = false;
295  } else {
296    write_completed = true;
297  }
298
299  if ((proxy_delegate.get() != NULL) &&
300      (consumed_data.bytes_consumed > 0 || consumed_data.fin_consumed)) {
301    proxy_delegate->WroteData(write_completed);
302  }
303}
304
305void ReliableQuicStream::OnCanWrite() {
306  bool fin = false;
307  while (!queued_data_.empty()) {
308    PendingData* pending_data = &queued_data_.front();
309    ProxyAckNotifierDelegate* delegate = pending_data->delegate.get();
310    if (queued_data_.size() == 1 && fin_buffered_) {
311      fin = true;
312    }
313    struct iovec iov(MakeIovec(pending_data->data));
314    QuicConsumedData consumed_data = WritevData(&iov, 1, fin, delegate);
315    if (consumed_data.bytes_consumed == pending_data->data.size() &&
316        fin == consumed_data.fin_consumed) {
317      queued_data_.pop_front();
318      if (delegate != NULL) {
319        delegate->WroteData(true);
320      }
321    } else {
322      if (consumed_data.bytes_consumed > 0) {
323        pending_data->data.erase(0, consumed_data.bytes_consumed);
324        if (delegate != NULL) {
325          delegate->WroteData(false);
326        }
327      }
328      break;
329    }
330  }
331}
332
333void ReliableQuicStream::MaybeSendBlocked() {
334  flow_controller_.MaybeSendBlocked();
335  connection_flow_controller_->MaybeSendBlocked();
336  // If we are connection level flow control blocked, then add the stream
337  // to the write blocked list. It will be given a chance to write when a
338  // connection level WINDOW_UPDATE arrives.
339  if (connection_flow_controller_->IsBlocked() &&
340      !flow_controller_.IsBlocked()) {
341    session_->MarkWriteBlocked(id(), EffectivePriority());
342  }
343}
344
345QuicConsumedData ReliableQuicStream::WritevData(
346    const struct iovec* iov,
347    int iov_count,
348    bool fin,
349    QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
350  if (write_side_closed_) {
351    DLOG(ERROR) << ENDPOINT << "Attempt to write when the write side is closed";
352    return QuicConsumedData(0, false);
353  }
354
355  // How much data we want to write.
356  size_t write_length = TotalIovecLength(iov, iov_count);
357
358  // A FIN with zero data payload should not be flow control blocked.
359  bool fin_with_zero_data = (fin && write_length == 0);
360
361  if (flow_controller_.IsEnabled()) {
362    // How much data we are allowed to write from flow control.
363    uint64 send_window = flow_controller_.SendWindowSize();
364    if (connection_flow_controller_->IsEnabled()) {
365      send_window =
366          min(send_window, connection_flow_controller_->SendWindowSize());
367    }
368
369    if (send_window == 0 && !fin_with_zero_data) {
370      // Quick return if we can't send anything.
371      MaybeSendBlocked();
372      return QuicConsumedData(0, false);
373    }
374
375    if (write_length > send_window) {
376      // Don't send the FIN if we aren't going to send all the data.
377      fin = false;
378
379      // Writing more data would be a violation of flow control.
380      write_length = send_window;
381    }
382  }
383
384  // Fill an IOVector with bytes from the iovec.
385  IOVector data;
386  data.AppendIovecAtMostBytes(iov, iov_count, write_length);
387
388  QuicConsumedData consumed_data = session()->WritevData(
389      id(), data, stream_bytes_written_, fin, GetFecProtection(),
390      ack_notifier_delegate);
391  stream_bytes_written_ += consumed_data.bytes_consumed;
392
393  AddBytesSent(consumed_data.bytes_consumed);
394
395  if (consumed_data.bytes_consumed == write_length) {
396    if (!fin_with_zero_data) {
397      MaybeSendBlocked();
398    }
399    if (fin && consumed_data.fin_consumed) {
400      fin_sent_ = true;
401      CloseWriteSide();
402    } else if (fin && !consumed_data.fin_consumed) {
403      session_->MarkWriteBlocked(id(), EffectivePriority());
404    }
405  } else {
406    session_->MarkWriteBlocked(id(), EffectivePriority());
407  }
408  return consumed_data;
409}
410
411FecProtection ReliableQuicStream::GetFecProtection() {
412  return fec_policy_ == FEC_PROTECT_ALWAYS ? MUST_FEC_PROTECT : MAY_FEC_PROTECT;
413}
414
415void ReliableQuicStream::CloseReadSide() {
416  if (read_side_closed_) {
417    return;
418  }
419  DVLOG(1) << ENDPOINT << "Done reading from stream " << id();
420
421  read_side_closed_ = true;
422  if (write_side_closed_) {
423    DVLOG(1) << ENDPOINT << "Closing stream: " << id();
424    session_->CloseStream(id());
425  }
426}
427
428void ReliableQuicStream::CloseWriteSide() {
429  if (write_side_closed_) {
430    return;
431  }
432  DVLOG(1) << ENDPOINT << "Done writing to stream " << id();
433
434  write_side_closed_ = true;
435  if (read_side_closed_) {
436    DVLOG(1) << ENDPOINT << "Closing stream: " << id();
437    session_->CloseStream(id());
438  }
439}
440
441bool ReliableQuicStream::HasBufferedData() const {
442  return !queued_data_.empty();
443}
444
445void ReliableQuicStream::OnClose() {
446  CloseReadSide();
447  CloseWriteSide();
448
449  if (!fin_sent_ && !rst_sent_) {
450    // For flow control accounting, we must tell the peer how many bytes we have
451    // written on this stream before termination. Done here if needed, using a
452    // RST frame.
453    DVLOG(1) << ENDPOINT << "Sending RST in OnClose: " << id();
454    session_->SendRstStream(id(), QUIC_RST_FLOW_CONTROL_ACCOUNTING,
455                            stream_bytes_written_);
456    rst_sent_ = true;
457  }
458
459  // We are closing the stream and will not process any further incoming bytes.
460  // As there may be more bytes in flight and we need to ensure that both
461  // endpoints have the same connection level flow control state, mark all
462  // unreceived or buffered bytes as consumed.
463  uint64 bytes_to_consume = flow_controller_.highest_received_byte_offset() -
464      flow_controller_.bytes_consumed();
465  AddBytesConsumed(bytes_to_consume);
466}
467
468void ReliableQuicStream::OnWindowUpdateFrame(
469    const QuicWindowUpdateFrame& frame) {
470  if (!flow_controller_.IsEnabled()) {
471    DLOG(DFATAL) << "Flow control not enabled! " << version();
472    return;
473  }
474
475  if (flow_controller_.UpdateSendWindowOffset(frame.byte_offset)) {
476    // We can write again!
477    // TODO(rjshade): This does not respect priorities (e.g. multiple
478    //                outstanding POSTs are unblocked on arrival of
479    //                SHLO with initial window).
480    // As long as the connection is not flow control blocked, we can write!
481    OnCanWrite();
482  }
483}
484
485bool ReliableQuicStream::MaybeIncreaseHighestReceivedOffset(uint64 new_offset) {
486  if (flow_controller_.IsEnabled()) {
487    uint64 increment =
488        new_offset - flow_controller_.highest_received_byte_offset();
489    if (flow_controller_.UpdateHighestReceivedOffset(new_offset)) {
490      // If |new_offset| increased the stream flow controller's highest received
491      // offset, then we need to increase the connection flow controller's value
492      // by the incremental difference.
493      connection_flow_controller_->UpdateHighestReceivedOffset(
494          connection_flow_controller_->highest_received_byte_offset() +
495          increment);
496      return true;
497    }
498  }
499  return false;
500}
501
502void ReliableQuicStream::AddBytesSent(uint64 bytes) {
503  if (flow_controller_.IsEnabled()) {
504    flow_controller_.AddBytesSent(bytes);
505    connection_flow_controller_->AddBytesSent(bytes);
506  }
507}
508
509void ReliableQuicStream::AddBytesConsumed(uint64 bytes) {
510  if (flow_controller_.IsEnabled()) {
511    // Only adjust stream level flow controller if we are still reading.
512    if (!read_side_closed_) {
513      flow_controller_.AddBytesConsumed(bytes);
514    }
515
516    connection_flow_controller_->AddBytesConsumed(bytes);
517  }
518}
519
520bool ReliableQuicStream::IsFlowControlBlocked() {
521  return flow_controller_.IsBlocked() ||
522         connection_flow_controller_->IsBlocked();
523}
524
525}  // namespace net
526