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 (FLAGS_quic_send_fec_packet_only_on_fec_alarm &&
 148       session_->config()->HasClientSentConnectionOption(kFSTR, perspective_)) {
 149     fec_policy_ = FEC_PROTECT_ALWAYS;
 150   }
 151 }
 152
 153 void ReliableQuicStream::OnStreamFrame(const QuicStreamFrame& frame) {
 154   if (read_side_closed_) {
 155     DVLOG(1) << ENDPOINT << "Ignoring frame " << frame.stream_id;
 156     // The subclass does not want read data:  blackhole the data.
 157     return;
 158   }
 159
 160   if (frame.stream_id != id_) {
 161     session_->connection()->SendConnectionClose(QUIC_INTERNAL_ERROR);
 162     return;
 163   }
 164
 165   if (frame.fin) {
 166     fin_received_ = true;
 167     if (fin_sent_) {
 168       session_->StreamDraining(id_);
 169     }
 170   }
 171
 172   // This count includes duplicate data received.
 173   size_t frame_payload_size = frame.data.size();
 174   stream_bytes_read_ += frame_payload_size;
 175
 176   // Flow control is interested in tracking highest received offset.
 177   if (MaybeIncreaseHighestReceivedOffset(frame.offset + frame_payload_size)) {
 178     // As the highest received offset has changed, check to see if this is a
 179     // violation of flow control.
 180     if (flow_controller_.FlowControlViolation() ||
 181         connection_flow_controller_->FlowControlViolation()) {
 182       session_->connection()->SendConnectionClose(
 183           QUIC_FLOW_CONTROL_RECEIVED_TOO_MUCH_DATA);
 184       return;
 185     }
 186   }
 187
 188   sequencer_.OnStreamFrame(frame);
 189 }
 190
 191 int ReliableQuicStream::num_frames_received() const {
 192   return sequencer_.num_frames_received();
 193 }
 194
 195 int ReliableQuicStream::num_early_frames_received() const {
 196   return sequencer_.num_early_frames_received();
 197 }
 198
 199 int ReliableQuicStream::num_duplicate_frames_received() const {
 200   return sequencer_.num_duplicate_frames_received();
 201 }
 202
 203 void ReliableQuicStream::OnStreamReset(const QuicRstStreamFrame& frame) {
 204   rst_received_ = true;
 205   MaybeIncreaseHighestReceivedOffset(frame.byte_offset);
 206
 207   stream_error_ = frame.error_code;
 208   CloseWriteSide();
 209   CloseReadSide();
 210 }
 211
 212 void ReliableQuicStream::OnConnectionClosed(QuicErrorCode error,
 213                                             bool from_peer) {
 214   if (read_side_closed_ && write_side_closed_) {
 215     return;
 216   }
 217   if (error != QUIC_NO_ERROR) {
 218     stream_error_ = QUIC_STREAM_CONNECTION_ERROR;
 219     connection_error_ = error;
 220   }
 221
 222   CloseWriteSide();
 223   CloseReadSide();
 224 }
 225
 226 void ReliableQuicStream::OnFinRead() {
 227   DCHECK(sequencer_.IsClosed());
 228   // OnFinRead can be called due to a FIN flag in a headers block, so there may
 229   // have been no OnStreamFrame call with a FIN in the frame.
 230   fin_received_ = true;
 231   // If fin_sent_ is true, then CloseWriteSide has already been called, and the
 232   // stream will be destroyed by CloseReadSide, so don't need to call
 233   // StreamDraining.
 234   CloseReadSide();
 235 }
 236
 237 void ReliableQuicStream::Reset(QuicRstStreamErrorCode error) {
 238   DCHECK_NE(QUIC_STREAM_NO_ERROR, error);
 239   stream_error_ = error;
 240   // Sending a RstStream results in calling CloseStream.
 241   session()->SendRstStream(id(), error, stream_bytes_written_);
 242   rst_sent_ = true;
 243 }
 244
 245 void ReliableQuicStream::CloseConnection(QuicErrorCode error) {
 246   session()->connection()->SendConnectionClose(error);
 247 }
 248
 249 void ReliableQuicStream::CloseConnectionWithDetails(QuicErrorCode error,
 250                                                     const string& details) {
 251   session()->connection()->SendConnectionCloseWithDetails(error, details);
 252 }
 253
 254 void ReliableQuicStream::WriteOrBufferData(
 255     StringPiece data,
 256     bool fin,
 257     QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
 258   if (data.empty() && !fin) {
 259     LOG(DFATAL) << "data.empty() && !fin";
 260     return;
 261   }
 262
 263   if (fin_buffered_) {
 264     LOG(DFATAL) << "Fin already buffered";
 265     return;
 266   }
 267   if (write_side_closed_) {
 268     DLOG(ERROR) << ENDPOINT << "Attempt to write when the write side is closed";
 269     return;
 270   }
 271
 272   scoped_refptr<ProxyAckNotifierDelegate> proxy_delegate;
 273   if (ack_notifier_delegate != nullptr) {
 274     proxy_delegate = new ProxyAckNotifierDelegate(ack_notifier_delegate);
 275   }
 276
 277   QuicConsumedData consumed_data(0, false);
 278   fin_buffered_ = fin;
 279
 280   if (queued_data_.empty()) {
 281     struct iovec iov(MakeIovec(data));
 282     consumed_data = WritevData(&iov, 1, fin, proxy_delegate.get());
 283     DCHECK_LE(consumed_data.bytes_consumed, data.length());
 284   }
 285
 286   bool write_completed;
 287   // If there's unconsumed data or an unconsumed fin, queue it.
 288   if (consumed_data.bytes_consumed < data.length() ||
 289       (fin && !consumed_data.fin_consumed)) {
 290     StringPiece remainder(data.substr(consumed_data.bytes_consumed));
 291     queued_data_.push_back(PendingData(remainder.as_string(), proxy_delegate));
 292     write_completed = false;
 293   } else {
 294     write_completed = true;
 295   }
 296
 297   if ((proxy_delegate.get() != nullptr) &&
 298       (consumed_data.bytes_consumed > 0 || consumed_data.fin_consumed)) {
 299     proxy_delegate->WroteData(write_completed);
 300   }
 301 }
 302
 303 void ReliableQuicStream::OnCanWrite() {
 304   bool fin = false;
 305   while (!queued_data_.empty()) {
 306     PendingData* pending_data = &queued_data_.front();
 307     ProxyAckNotifierDelegate* delegate = pending_data->delegate.get();
 308     if (queued_data_.size() == 1 && fin_buffered_) {
 309       fin = true;
 310     }
 311     if (pending_data->offset > 0 &&
 312         pending_data->offset >= pending_data->data.size()) {
 313       // This should be impossible because offset tracks the amount of
 314       // pending_data written thus far.
 315       LOG(DFATAL) << "Pending offset is beyond available data. offset: "
 316                   << pending_data->offset
 317                   << " vs: " << pending_data->data.size();
 318       return;
 319     }
 320     size_t remaining_len = pending_data->data.size() - pending_data->offset;
 321     struct iovec iov = {
 322         const_cast<char*>(pending_data->data.data()) + pending_data->offset,
 323         remaining_len};
 324     QuicConsumedData consumed_data = WritevData(&iov, 1, fin, delegate);
 325     if (consumed_data.bytes_consumed == remaining_len &&
 326         fin == consumed_data.fin_consumed) {
 327       queued_data_.pop_front();
 328       if (delegate != nullptr) {
 329         delegate->WroteData(true);
 330       }
 331     } else {
 332       if (consumed_data.bytes_consumed > 0) {
 333         pending_data->offset += consumed_data.bytes_consumed;
 334         if (delegate != nullptr) {
 335           delegate->WroteData(false);
 336         }
 337       }
 338       break;
 339     }
 340   }
 341 }
 342
 343 void ReliableQuicStream::MaybeSendBlocked() {
 344   flow_controller_.MaybeSendBlocked();
 345   if (!stream_contributes_to_connection_flow_control_) {
 346     return;
 347   }
 348   connection_flow_controller_->MaybeSendBlocked();
 349   // If the stream is blocked by connection-level flow control but not by
 350   // stream-level flow control, add the stream to the write blocked list so that
 351   // the stream will be given a chance to write when a connection-level
 352   // WINDOW_UPDATE arrives.
 353   if (connection_flow_controller_->IsBlocked() &&
 354       !flow_controller_.IsBlocked()) {
 355     session_->MarkConnectionLevelWriteBlocked(id(), EffectivePriority());
 356   }
 357 }
 358
 359 QuicConsumedData ReliableQuicStream::WritevData(
 360     const struct iovec* iov,
 361     int iov_count,
 362     bool fin,
 363     QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
 364   if (write_side_closed_) {
 365     DLOG(ERROR) << ENDPOINT << "Attempt to write when the write side is closed";
 366     return QuicConsumedData(0, false);
 367   }
 368
 369   // How much data was provided.
 370   size_t write_length = TotalIovecLength(iov, iov_count);
 371
 372   // A FIN with zero data payload should not be flow control blocked.
 373   bool fin_with_zero_data = (fin && write_length == 0);
 374
 375   // How much data flow control permits to be written.
 376   QuicByteCount send_window = flow_controller_.SendWindowSize();
 377   if (stream_contributes_to_connection_flow_control_) {
 378     send_window =
 379         min(send_window, connection_flow_controller_->SendWindowSize());
 380   }
 381
 382   if (send_window == 0 && !fin_with_zero_data) {
 383     // Quick return if nothing can be sent.
 384     MaybeSendBlocked();
 385     return QuicConsumedData(0, false);
 386   }
 387
 388   if (write_length > send_window) {
 389     // Don't send the FIN unless all the data will be sent.
 390     fin = false;
 391
 392     // Writing more data would be a violation of flow control.
 393     write_length = static_cast<size_t>(send_window);
 394   }
 395
 396   QuicConsumedData consumed_data = session()->WritevData(
 397       id(), QuicIOVector(iov, iov_count, write_length), stream_bytes_written_,
 398       fin, GetFecProtection(), ack_notifier_delegate);
 399   stream_bytes_written_ += consumed_data.bytes_consumed;
 400
 401   AddBytesSent(consumed_data.bytes_consumed);
 402
 403   if (consumed_data.bytes_consumed == write_length) {
 404     if (!fin_with_zero_data) {
 405       MaybeSendBlocked();
 406     }
 407     if (fin && consumed_data.fin_consumed) {
 408       fin_sent_ = true;
 409       if (fin_received_) {
 410         session_->StreamDraining(id_);
 411       }
 412       CloseWriteSide();
 413     } else if (fin && !consumed_data.fin_consumed) {
 414       session_->MarkConnectionLevelWriteBlocked(id(), EffectivePriority());
 415     }
 416   } else {
 417     session_->MarkConnectionLevelWriteBlocked(id(), EffectivePriority());
 418   }
 419   return consumed_data;
 420 }
 421
 422 FecProtection ReliableQuicStream::GetFecProtection() {
 423   return fec_policy_ == FEC_PROTECT_ALWAYS ? MUST_FEC_PROTECT : MAY_FEC_PROTECT;
 424 }
 425
 426 void ReliableQuicStream::CloseReadSide() {
 427   if (read_side_closed_) {
 428     return;
 429   }
 430   DVLOG(1) << ENDPOINT << "Done reading from stream " << id();
 431
 432   read_side_closed_ = true;
 433   if (write_side_closed_) {
 434     DVLOG(1) << ENDPOINT << "Closing stream: " << id();
 435     session_->CloseStream(id());
 436   }
 437 }
 438
 439 void ReliableQuicStream::CloseWriteSide() {
 440   if (write_side_closed_) {
 441     return;
 442   }
 443   DVLOG(1) << ENDPOINT << "Done writing to stream " << id();
 444
 445   write_side_closed_ = true;
 446   if (read_side_closed_) {
 447     DVLOG(1) << ENDPOINT << "Closing stream: " << id();
 448     session_->CloseStream(id());
 449   }
 450 }
 451
 452 bool ReliableQuicStream::HasBufferedData() const {
 453   return !queued_data_.empty();
 454 }
 455
 456 QuicVersion ReliableQuicStream::version() const {
 457   return session_->connection()->version();
 458 }
 459
 460 void ReliableQuicStream::OnClose() {
 461   CloseReadSide();
 462   CloseWriteSide();
 463
 464   if (!fin_sent_ && !rst_sent_) {
 465     // For flow control accounting, tell the peer how many bytes have been
 466     // written on this stream before termination. Done here if needed, using a
 467     // RST_STREAM frame.
 468     DVLOG(1) << ENDPOINT << "Sending RST_STREAM in OnClose: " << id();
 469     session_->SendRstStream(id(), QUIC_RST_ACKNOWLEDGEMENT,
 470                             stream_bytes_written_);
 471     rst_sent_ = true;
 472   }
 473
 474   // The stream is being closed and will not process any further incoming bytes.
 475   // As there may be more bytes in flight, to ensure that both endpoints have
 476   // the same connection level flow control state, mark all unreceived or
 477   // buffered bytes as consumed.
 478   QuicByteCount bytes_to_consume =
 479       flow_controller_.highest_received_byte_offset() -
 480       flow_controller_.bytes_consumed();
 481   AddBytesConsumed(bytes_to_consume);
 482 }
 483
 484 void ReliableQuicStream::OnWindowUpdateFrame(
 485     const QuicWindowUpdateFrame& frame) {
 486   if (flow_controller_.UpdateSendWindowOffset(frame.byte_offset)) {
 487     // Writing can be done again!
 488     // TODO(rjshade): This does not respect priorities (e.g. multiple
 489     //                outstanding POSTs are unblocked on arrival of
 490     //                SHLO with initial window).
 491     // As long as the connection is not flow control blocked, write on!
 492     OnCanWrite();
 493   }
 494 }
 495
 496 bool ReliableQuicStream::MaybeIncreaseHighestReceivedOffset(
 497     QuicStreamOffset new_offset) {
 498   uint64 increment =
 499       new_offset - flow_controller_.highest_received_byte_offset();
 500   if (!flow_controller_.UpdateHighestReceivedOffset(new_offset)) {
 501     return false;
 502   }
 503
 504   // If |new_offset| increased the stream flow controller's highest received
 505   // offset, increase the connection flow controller's value by the incremental
 506   // difference.
 507   if (stream_contributes_to_connection_flow_control_) {
 508     connection_flow_controller_->UpdateHighestReceivedOffset(
 509         connection_flow_controller_->highest_received_byte_offset() +
 510         increment);
 511   }
 512   return true;
 513 }
 514
 515 void ReliableQuicStream::AddBytesSent(QuicByteCount bytes) {
 516   flow_controller_.AddBytesSent(bytes);
 517   if (stream_contributes_to_connection_flow_control_) {
 518     connection_flow_controller_->AddBytesSent(bytes);
 519   }
 520 }
 521
 522 void ReliableQuicStream::AddBytesConsumed(QuicByteCount bytes) {
 523   // Only adjust stream level flow controller if still reading.
 524   if (!read_side_closed_) {
 525     flow_controller_.AddBytesConsumed(bytes);
 526   }
 527
 528   if (stream_contributes_to_connection_flow_control_) {
 529     connection_flow_controller_->AddBytesConsumed(bytes);
 530   }
 531 }
 532
 533 void ReliableQuicStream::UpdateSendWindowOffset(QuicStreamOffset new_window) {
 534   if (flow_controller_.UpdateSendWindowOffset(new_window)) {
 535     OnCanWrite();
 536   }
 537 }
 538
 539 }  // namespace net