Pull out flow control functionality from ReliableQuicStream into a new

net/quic/reliable_quic_stream.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/quic/reliable_quic_stream.h"
 
 #include "base/logging.h"
 #include "net/quic/iovector.h"
+#include "net/quic/quic_flow_controller.h"
 #include "net/quic/quic_session.h"
 #include "net/quic/quic_write_blocked_list.h"
 
 using base::StringPiece;
 using std::min;
 
 namespace net {
 
 #define ENDPOINT (is_server_ ? "Server: " : " Client: ")
 
@@ skipping 90 matching lines @@
 }
 
 ReliableQuicStream::ReliableQuicStream(QuicStreamId id, QuicSession* session)
     : sequencer_(this),
       id_(id),
       session_(session),
       stream_bytes_read_(0),
       stream_bytes_written_(0),
       stream_error_(QUIC_STREAM_NO_ERROR),
       connection_error_(QUIC_NO_ERROR),
-      flow_control_send_limit_(
-          session_->config()->peer_initial_flow_control_window_bytes()),
-      max_flow_control_receive_window_bytes_(
-          session_->connection()->max_flow_control_receive_window_bytes()),
-      flow_control_receive_window_offset_bytes_(
-          session_->connection()->max_flow_control_receive_window_bytes()),
       read_side_closed_(false),
       write_side_closed_(false),
       fin_buffered_(false),
       fin_sent_(false),
       rst_sent_(false),
-      is_server_(session_->is_server()) {
-  DVLOG(1) << ENDPOINT << "Created stream " << id_
-           << ", setting initial receive window to: "
-           << flow_control_receive_window_offset_bytes_
-           << ", setting send window to: " << flow_control_send_limit_;
+      is_server_(session_->is_server()),
+      flow_controller_(
+          id_,
+          is_server_,
+          session_->config()->peer_initial_flow_control_window_bytes(),
+          session_->connection()->max_flow_control_receive_window_bytes(),
+          session_->connection()->max_flow_control_receive_window_bytes()) {
 }
 
 ReliableQuicStream::~ReliableQuicStream() {
 }
 
 bool ReliableQuicStream::WillAcceptStreamFrame(
     const QuicStreamFrame& frame) const {
   if (read_side_closed_) {
     return true;
   }
@@ skipping 10 matching lines @@
     DVLOG(1) << ENDPOINT << "Ignoring frame " << frame.stream_id;
     // We don't want to be reading: blackhole the data.
     return true;
   }
 
   // This count include duplicate data received.
   stream_bytes_read_ += frame.data.TotalBufferSize();
 
   bool accepted = sequencer_.OnStreamFrame(frame);
 
-  if (IsFlowControlEnabled()) {
-    if (flow_control_receive_window_offset_bytes_ < TotalReceivedBytes()) {
-      // TODO(rjshade): Lower severity from DFATAL once we have established that
-      //                flow control is working correctly.
-      LOG(DFATAL)
-          << ENDPOINT << "Flow control violation on stream: " << id()
-          << ", our receive offset is: "
-          << flow_control_receive_window_offset_bytes_
-          << ", we have consumed: " << sequencer_.num_bytes_consumed()
-          << ", we have buffered: " << sequencer_.num_bytes_buffered()
-          << ", total: " << TotalReceivedBytes();
+  if (version() >= QUIC_VERSION_17) {
+    if (flow_controller_.FlowControlViolation()) {
       session_->connection()->SendConnectionClose(QUIC_FLOW_CONTROL_ERROR);
       return false;
     }
     MaybeSendWindowUpdate();
   }
 
   return accepted;
 }
 
 void ReliableQuicStream::MaybeSendWindowUpdate() {
-  if (!IsFlowControlEnabled()) {
-    return;
-  }
-
-  // Send WindowUpdate to increase receive window if
-  // (receive window offset - consumed bytes) < (max window / 2).
-  // This is behaviour copied from SPDY.
-  size_t consumed_window = flow_control_receive_window_offset_bytes_ -
-                           sequencer_.num_bytes_consumed();
-  size_t threshold = (max_flow_control_receive_window_bytes_ / 2);
-  if (consumed_window < threshold) {
-    // Update our receive window.
-    flow_control_receive_window_offset_bytes_ +=
-        (max_flow_control_receive_window_bytes_ - consumed_window);
-    DVLOG(1) << ENDPOINT << "Stream: " << id()
-             << ", sending WindowUpdate frame. "
-             << "Consumed bytes: " << sequencer_.num_bytes_consumed()
-             << ", Receive window offset: "
-             << flow_control_receive_window_offset_bytes_
-             << ", Consumed window: " << consumed_window
-             << ", and threshold: " << threshold
-             << ". New receive window offset is: "
-             << flow_control_receive_window_offset_bytes_;
-
-    // Inform the peer of our new receive window.
-    session()->connection()->SendWindowUpdate(
-        id(), flow_control_receive_window_offset_bytes_);
+  if (version() >= QUIC_VERSION_17) {
+    flow_controller_.MaybeSendWindowUpdate(session()->connection());
   }
 }
 
 int ReliableQuicStream::num_frames_received() {
   return sequencer_.num_frames_received();
 }
 
 int ReliableQuicStream::num_duplicate_frames_received() {
   return sequencer_.num_duplicate_frames_received();
 }
@@ skipping 124 matching lines @@
     QuicAckNotifier::DelegateInterface* ack_notifier_delegate) {
   if (write_side_closed_) {
     DLOG(ERROR) << ENDPOINT << "Attempt to write when the write side is closed";
     return QuicConsumedData(0, false);
   }
 
   // How much data we want to write.
   size_t write_length = TotalIovecLength(iov, iov_count);
 
   // How much data we are allowed to write from flow control.
-  size_t send_window = SendWindowSize();
+  size_t send_window = flow_controller_.SendWindowSize();
 
   // A FIN with zero data payload should not be flow control blocked.
   bool fin_with_zero_data = (fin && write_length == 0);
 
-  if (IsFlowControlEnabled()) {
+  if (version() >= QUIC_VERSION_17 && flow_controller_.IsEnabled()) {
     if (send_window == 0 && !fin_with_zero_data) {
       // Quick return if we can't send anything.
-      session()->connection()->SendBlocked(id());
+      flow_controller_.MaybeSendBlocked(session()->connection());
       return QuicConsumedData(0, false);
     }
 
     if (write_length > send_window) {
       // Don't send the FIN if we aren't going to send all the data.
       fin = false;
 
       // Writing more data would be a violation of flow control.
       write_length = send_window;
     }
   }
 
   // Fill an IOVector with bytes from the iovec.
   IOVector data;
   data.AppendIovecAtMostBytes(iov, iov_count, write_length);
 
   QuicConsumedData consumed_data = session()->WritevData(
       id(), data, stream_bytes_written_, fin, ack_notifier_delegate);
   stream_bytes_written_ += consumed_data.bytes_consumed;
 
+  if (version() >= QUIC_VERSION_17 && flow_controller_.IsEnabled()) {
+    flow_controller_.AddBytesSent(consumed_data.bytes_consumed);
+  }
+
   if (consumed_data.bytes_consumed == write_length) {
-    if (IsFlowControlEnabled() && write_length == send_window &&
-        !fin_with_zero_data) {
-      DVLOG(1) << ENDPOINT << "Stream " << id()
-               << " is flow control blocked. "
-               << "Send window: " << send_window
-               << ", stream_bytes_written: " << stream_bytes_written_
-               << ", flow_control_send_limit: "
-               << flow_control_send_limit_;
-      // The entire send_window has been consumed, we are now flow control
-      // blocked.
-      session()->connection()->SendBlocked(id());
+    if (!fin_with_zero_data) {
+      if (version() >= QUIC_VERSION_17) {
+        flow_controller_.MaybeSendBlocked(session()->connection());
+      }
     }
     if (fin && consumed_data.fin_consumed) {
       fin_sent_ = true;
       CloseWriteSide();
     } else if (fin && !consumed_data.fin_consumed) {
       session_->MarkWriteBlocked(id(), EffectivePriority());
     }
   } else {
     session_->MarkWriteBlocked(id(), EffectivePriority());
   }
@@ skipping 40 matching lines @@
     // written on this stream before termination. Done here if needed, using a
     // RST frame.
     DVLOG(1) << ENDPOINT << "Sending RST in OnClose: " << id();
     session_->SendRstStream(id(), QUIC_STREAM_NO_ERROR, stream_bytes_written_);
     rst_sent_ = true;
   }
 }
 
 void ReliableQuicStream::OnWindowUpdateFrame(
     const QuicWindowUpdateFrame& frame) {
-  if (!IsFlowControlEnabled()) {
+  if (!flow_controller_.IsEnabled()) {
     DLOG(DFATAL) << "Flow control not enabled! " << version();
     return;
   }
 
-  DVLOG(1) << ENDPOINT
-           << "OnWindowUpdateFrame for stream " << id()
-           << " with byte offset " << frame.byte_offset
-           << " , current offset: " << flow_control_send_limit_ << ").";
-
-  UpdateFlowControlSendLimit(frame.byte_offset);
-}
-
-void ReliableQuicStream::UpdateFlowControlSendLimit(QuicStreamOffset offset) {
-  if (offset <= flow_control_send_limit_) {
-    DVLOG(1) << ENDPOINT << "Stream " << id()
-             << ", not changing window, current: " << flow_control_send_limit_
-             << " new: " << offset;
-    // No change to our send window.
-    return;
+  if (flow_controller_.UpdateSendWindowOffset(frame.byte_offset)) {
+    // We can write again!
+    // TODO(rjshade): This does not respect priorities (e.g. multiple
+    //                outstanding POSTs are unblocked on arrival of
+    //                SHLO with initial window).
+    OnCanWrite();
   }
-
-  DVLOG(1) << ENDPOINT << "Stream " << id()
-           << ", changing window, current: " << flow_control_send_limit_
-           << " new: " << offset;
-  // Send window has increased.
-  flow_control_send_limit_ = offset;
-
-  // We can write again!
-  // TODO(rjshade): This does not respect priorities (e.g. multiple outstanding
-  //                POSTs are unblocked on arrival of SHLO with initial window).
-  OnCanWrite();
-}
-
-bool ReliableQuicStream::IsFlowControlBlocked() const {
-  return IsFlowControlEnabled() && SendWindowSize() == 0;
-}
-
-uint64 ReliableQuicStream::SendWindowSize() const {
-  return flow_control_send_limit_ - stream_bytes_written();
-}
-
-uint64 ReliableQuicStream::TotalReceivedBytes() const {
-  return sequencer_.num_bytes_consumed() + sequencer_.num_bytes_buffered();
 }
 
 }  // namespace net