reliable_quic_stream.cc revision 5c02ac1a9c1b504631c0a3d2b6e737b5d738bae1 
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
28}  // namespace
29
30// Wrapper that aggregates OnAckNotifications for packets sent using
31// WriteOrBufferData and delivers them to the original
32// QuicAckNotifier::DelegateInterface after all bytes written using
33// WriteOrBufferData are acked.  This level of indirection is
34// necessary because the delegate interface provides no mechanism that
35// WriteOrBufferData can use to inform it that the write required
36// multiple WritevData calls or that only part of the data has been
37// sent out by the time ACKs start arriving.
38class ReliableQuicStream::ProxyAckNotifierDelegate
39    : public QuicAckNotifier::DelegateInterface {
40 public:
41  explicit ProxyAckNotifierDelegate(DelegateInterface* delegate)
42      : delegate_(delegate),
43        pending_acks_(0),
44        wrote_last_data_(false),
45        num_original_packets_(0),
46        num_original_bytes_(0),
47        num_retransmitted_packets_(0),
48        num_retransmitted_bytes_(0) {
49  }
50
51  virtual void OnAckNotification(int num_original_packets,
52                                 int num_original_bytes,
53                                 int num_retransmitted_packets,
54                                 int num_retransmitted_bytes,
55                                 QuicTime::Delta delta_largest_observed)
56      OVERRIDE {
57    DCHECK_LT(0, pending_acks_);
58    --pending_acks_;
59    num_original_packets_ += num_original_packets;
60    num_original_bytes_ += num_original_bytes;
61    num_retransmitted_packets_ += num_retransmitted_packets;
62    num_retransmitted_bytes_ += num_retransmitted_bytes;
63
64    if (wrote_last_data_ && pending_acks_ == 0) {
65      delegate_->OnAckNotification(num_original_packets_,
66                                   num_original_bytes_,
67                                   num_retransmitted_packets_,
68                                   num_retransmitted_bytes_,
69                                   delta_largest_observed);
70    }
71  }
72
73  void WroteData(bool last_data) {
74    DCHECK(!wrote_last_data_);
75    ++pending_acks_;
76    wrote_last_data_ = last_data;
77  }
78
79 protected:
80  // Delegates are ref counted.
81  virtual ~ProxyAckNotifierDelegate() {
82  }
83
84 private:
85  // Original delegate.  delegate_->OnAckNotification will be called when:
86  //   wrote_last_data_ == true and pending_acks_ == 0
87  scoped_refptr<DelegateInterface> delegate_;
88
89  // Number of outstanding acks.
90  int pending_acks_;
91
92  // True if no pending writes remain.
93  bool wrote_last_data_;
94
95  // Accumulators.
96  int num_original_packets_;
97  int num_original_bytes_;
98  int num_retransmitted_packets_;
99  int num_retransmitted_bytes_;
100
101  DISALLOW_COPY_AND_ASSIGN(ProxyAckNotifierDelegate);
102};
103
104ReliableQuicStream::PendingData::PendingData(
105    string data_in, scoped_refptr<ProxyAckNotifierDelegate> delegate_in)
106    : data(data_in), delegate(delegate_in) {
107}
108
109ReliableQuicStream::PendingData::~PendingData() {
110}
111
112ReliableQuicStream::ReliableQuicStream(QuicStreamId id, QuicSession* session)
113    : sequencer_(this),
114      id_(id),
115      session_(session),
116      stream_bytes_read_(0),
117      stream_bytes_written_(0),
118      stream_error_(QUIC_STREAM_NO_ERROR),
119      connection_error_(QUIC_NO_ERROR),
120      read_side_closed_(false),
121      write_side_closed_(false),
122      fin_buffered_(false),
123      fin_sent_(false),
124      rst_sent_(false),
125      is_server_(session_->is_server()),
126      flow_controller_(
127          session_->connection()->version(),
128          id_,
129          is_server_,
130          session_->config()->HasReceivedInitialFlowControlWindowBytes() ?
131              session_->config()->ReceivedInitialFlowControlWindowBytes() :
132              kDefaultFlowControlSendWindow,
133          session_->connection()->max_flow_control_receive_window_bytes(),
134          session_->connection()->max_flow_control_receive_window_bytes()) {
135}
136
137ReliableQuicStream::~ReliableQuicStream() {
138}
139
140bool ReliableQuicStream::OnStreamFrame(const QuicStreamFrame& frame) {
141  if (read_side_closed_) {
142    DVLOG(1) << ENDPOINT << "Ignoring frame " << frame.stream_id;
143    // We don't want to be reading: blackhole the data.
144    return true;
145  }
146
147  if (frame.stream_id != id_) {
148    LOG(ERROR) << "Error!";
149    return false;
150  }
151
152  // This count include duplicate data received.
153  stream_bytes_read_ += frame.data.TotalBufferSize();
154
155  bool accepted = sequencer_.OnStreamFrame(frame);
156
157  if (flow_controller_.FlowControlViolation()) {
158    session_->connection()->SendConnectionClose(QUIC_FLOW_CONTROL_ERROR);
159    return false;
160  }
161  MaybeSendWindowUpdate();
162
163  return accepted;
164}
165
166void ReliableQuicStream::MaybeSendWindowUpdate() {
167  flow_controller_.MaybeSendWindowUpdate(session()->connection());
168}
169
170int ReliableQuicStream::num_frames_received() {
171  return sequencer_.num_frames_received();
172}
173
174int ReliableQuicStream::num_duplicate_frames_received() {
175  return sequencer_.num_duplicate_frames_received();
176}
177
178void ReliableQuicStream::OnStreamReset(const QuicRstStreamFrame& frame) {
179  stream_error_ = frame.error_code;
180  CloseWriteSide();
181  CloseReadSide();
182}
183
184void ReliableQuicStream::OnConnectionClosed(QuicErrorCode error,
185                                            bool from_peer) {
186  if (read_side_closed_ && write_side_closed_) {
187    return;
188  }
189  if (error != QUIC_NO_ERROR) {
190    stream_error_ = QUIC_STREAM_CONNECTION_ERROR;
191    connection_error_ = error;
192  }
193
194  CloseWriteSide();
195  CloseReadSide();
196}
197
198void ReliableQuicStream::OnFinRead() {
199  DCHECK(sequencer_.IsClosed());
200  CloseReadSide();
201}
202
203void ReliableQuicStream::Reset(QuicRstStreamErrorCode error) {
204  DCHECK_NE(QUIC_STREAM_NO_ERROR, error);
205  stream_error_ = error;
206  // Sending a RstStream results in calling CloseStream.
207  session()->SendRstStream(id(), error, stream_bytes_written_);
208  rst_sent_ = true;
209}
210
211void ReliableQuicStream::CloseConnection(QuicErrorCode error) {
212  session()->connection()->SendConnectionClose(error);
213}
214
215void ReliableQuicStream::CloseConnectionWithDetails(QuicErrorCode error,
216                                                    const string& details) {
217  session()->connection()->SendConnectionCloseWithDetails(error, details);
218}
219
220QuicVersion ReliableQuicStream::version() const {
221  return session()->connection()->version();
222}
223
224void ReliableQuicStream::WriteOrBufferData(
225    StringPiece data,
226    bool fin,
227    QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
228  if (data.empty() && !fin) {
229    LOG(DFATAL) << "data.empty() && !fin";
230    return;
231  }
232
233  if (fin_buffered_) {
234    LOG(DFATAL) << "Fin already buffered";
235    return;
236  }
237
238  scoped_refptr<ProxyAckNotifierDelegate> proxy_delegate;
239  if (ack_notifier_delegate != NULL) {
240    proxy_delegate = new ProxyAckNotifierDelegate(ack_notifier_delegate);
241  }
242
243  QuicConsumedData consumed_data(0, false);
244  fin_buffered_ = fin;
245
246  if (queued_data_.empty()) {
247    struct iovec iov(MakeIovec(data));
248    consumed_data = WritevData(&iov, 1, fin, proxy_delegate.get());
249    DCHECK_LE(consumed_data.bytes_consumed, data.length());
250  }
251
252  bool write_completed;
253  // If there's unconsumed data or an unconsumed fin, queue it.
254  if (consumed_data.bytes_consumed < data.length() ||
255      (fin && !consumed_data.fin_consumed)) {
256    StringPiece remainder(data.substr(consumed_data.bytes_consumed));
257    queued_data_.push_back(PendingData(remainder.as_string(), proxy_delegate));
258    write_completed = false;
259  } else {
260    write_completed = true;
261  }
262
263  if ((proxy_delegate.get() != NULL) &&
264      (consumed_data.bytes_consumed > 0 || consumed_data.fin_consumed)) {
265    proxy_delegate->WroteData(write_completed);
266  }
267}
268
269void ReliableQuicStream::OnCanWrite() {
270  bool fin = false;
271  while (!queued_data_.empty()) {
272    PendingData* pending_data = &queued_data_.front();
273    ProxyAckNotifierDelegate* delegate = pending_data->delegate.get();
274    if (queued_data_.size() == 1 && fin_buffered_) {
275      fin = true;
276    }
277    struct iovec iov(MakeIovec(pending_data->data));
278    QuicConsumedData consumed_data = WritevData(&iov, 1, fin, delegate);
279    if (consumed_data.bytes_consumed == pending_data->data.size() &&
280        fin == consumed_data.fin_consumed) {
281      queued_data_.pop_front();
282      if (delegate != NULL) {
283        delegate->WroteData(true);
284      }
285    } else {
286      if (consumed_data.bytes_consumed > 0) {
287        pending_data->data.erase(0, consumed_data.bytes_consumed);
288        if (delegate != NULL) {
289          delegate->WroteData(false);
290        }
291      }
292      break;
293    }
294  }
295}
296
297QuicConsumedData ReliableQuicStream::WritevData(
298    const struct iovec* iov,
299    int iov_count,
300    bool fin,
301    QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
302  if (write_side_closed_) {
303    DLOG(ERROR) << ENDPOINT << "Attempt to write when the write side is closed";
304    return QuicConsumedData(0, false);
305  }
306
307  // How much data we want to write.
308  size_t write_length = TotalIovecLength(iov, iov_count);
309
310  // A FIN with zero data payload should not be flow control blocked.
311  bool fin_with_zero_data = (fin && write_length == 0);
312
313  if (flow_controller_.IsEnabled()) {
314    // How much data we are allowed to write from flow control.
315    size_t send_window = flow_controller_.SendWindowSize();
316
317    if (send_window == 0 && !fin_with_zero_data) {
318      // Quick return if we can't send anything.
319      flow_controller_.MaybeSendBlocked(session()->connection());
320      return QuicConsumedData(0, false);
321    }
322
323    if (write_length > send_window) {
324      // Don't send the FIN if we aren't going to send all the data.
325      fin = false;
326
327      // Writing more data would be a violation of flow control.
328      write_length = send_window;
329    }
330  }
331
332  // Fill an IOVector with bytes from the iovec.
333  IOVector data;
334  data.AppendIovecAtMostBytes(iov, iov_count, write_length);
335
336  QuicConsumedData consumed_data = session()->WritevData(
337      id(), data, stream_bytes_written_, fin, ack_notifier_delegate);
338  stream_bytes_written_ += consumed_data.bytes_consumed;
339
340  flow_controller_.AddBytesSent(consumed_data.bytes_consumed);
341
342  if (consumed_data.bytes_consumed == write_length) {
343    if (!fin_with_zero_data) {
344      flow_controller_.MaybeSendBlocked(session()->connection());
345    }
346    if (fin && consumed_data.fin_consumed) {
347      fin_sent_ = true;
348      CloseWriteSide();
349    } else if (fin && !consumed_data.fin_consumed) {
350      session_->MarkWriteBlocked(id(), EffectivePriority());
351    }
352  } else {
353    session_->MarkWriteBlocked(id(), EffectivePriority());
354  }
355  return consumed_data;
356}
357
358void ReliableQuicStream::CloseReadSide() {
359  if (read_side_closed_) {
360    return;
361  }
362  DVLOG(1) << ENDPOINT << "Done reading from stream " << id();
363
364  read_side_closed_ = true;
365  if (write_side_closed_) {
366    DVLOG(1) << ENDPOINT << "Closing stream: " << id();
367    session_->CloseStream(id());
368  }
369}
370
371void ReliableQuicStream::CloseWriteSide() {
372  if (write_side_closed_) {
373    return;
374  }
375  DVLOG(1) << ENDPOINT << "Done writing to stream " << id();
376
377  write_side_closed_ = true;
378  if (read_side_closed_) {
379    DVLOG(1) << ENDPOINT << "Closing stream: " << id();
380    session_->CloseStream(id());
381  }
382}
383
384bool ReliableQuicStream::HasBufferedData() {
385  return !queued_data_.empty();
386}
387
388void ReliableQuicStream::OnClose() {
389  CloseReadSide();
390  CloseWriteSide();
391
392  if (!fin_sent_ && !rst_sent_) {
393    // For flow control accounting, we must tell the peer how many bytes we have
394    // written on this stream before termination. Done here if needed, using a
395    // RST frame.
396    DVLOG(1) << ENDPOINT << "Sending RST in OnClose: " << id();
397    session_->SendRstStream(id(), QUIC_RST_FLOW_CONTROL_ACCOUNTING,
398                            stream_bytes_written_);
399    rst_sent_ = true;
400  }
401}
402
403void ReliableQuicStream::OnWindowUpdateFrame(
404    const QuicWindowUpdateFrame& frame) {
405  if (!flow_controller_.IsEnabled()) {
406    DLOG(DFATAL) << "Flow control not enabled! " << version();
407    return;
408  }
409
410  if (flow_controller_.UpdateSendWindowOffset(frame.byte_offset)) {
411    // We can write again!
412    // TODO(rjshade): This does not respect priorities (e.g. multiple
413    //                outstanding POSTs are unblocked on arrival of
414    //                SHLO with initial window).
415    OnCanWrite();
416  }
417}
418
419}  // namespace net
420