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