net/quic/reliable_quic_stream.cc

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