[SPDY] Avoid leaking bytes from the session flow control receive window

net/spdy/spdy_session.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/spdy/spdy_session.h"
 
 #include <algorithm>
 #include <map>
 
 #include "base/basictypes.h"
@@ skipping 1384 matching lines @@
         NetLog::TYPE_SPDY_SESSION_RECV_DATA,
         base::Bind(&NetLogSpdyDataCallback, stream_id, len, fin));
   }
 
   ActiveStreamMap::const_iterator it = active_streams_.find(stream_id);
 
   // By the time data comes in, the stream may already be inactive.
   if (it == active_streams_.end())
     return;
 
-  // Only decrease the window size for data for active streams.
-  if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION && len > 0)
-    DecreaseRecvWindowSize(static_cast<int32>(len));
-
   scoped_ptr<SpdyBuffer> buffer;
   if (data) {
     DCHECK_GT(len, 0u);
     buffer.reset(new SpdyBuffer(data, len));
+
+    if (flow_control_state_ == FLOW_CONTROL_STREAM_AND_SESSION) {
+      DecreaseRecvWindowSize(static_cast<int32>(len));
+      buffer->AddConsumeCallback(
+          base::Bind(&SpdySession::IncreaseRecvWindowSize,
+                     weak_factory_.GetWeakPtr()));
+    }
   } else {
     DCHECK_EQ(len, 0u);
   }
   it->second->OnDataReceived(buffer.Pass());
 }
 
 void SpdySession::OnSetting(SpdySettingsIds id,
                             uint8 flags,
                             uint32 value) {
   HandleSetting(id, value);
@@ skipping 375 matching lines @@
 
 void SpdySession::SendStreamWindowUpdate(SpdyStreamId stream_id,
                                          uint32 delta_window_size) {
   CHECK_GE(flow_control_state_, FLOW_CONTROL_STREAM);
   CHECK(IsStreamActive(stream_id));
   scoped_refptr<SpdyStream> stream = active_streams_[stream_id];
   CHECK_EQ(stream->stream_id(), stream_id);
   SendWindowUpdateFrame(stream_id, delta_window_size, stream->priority());
 }
 
-void SpdySession::IncreaseRecvWindowSize(int32 delta_window_size) {
-  if (flow_control_state_ < FLOW_CONTROL_STREAM_AND_SESSION)
-    return;
-
-  DCHECK_GE(session_unacked_recv_window_bytes_, 0);
-  DCHECK_GE(session_recv_window_size_, session_unacked_recv_window_bytes_);
-  DCHECK_GE(delta_window_size, 1);
-  // Check for overflow.
-  DCHECK_LE(delta_window_size, kint32max - session_recv_window_size_);
-
-  session_recv_window_size_ += delta_window_size;
-  net_log_.AddEvent(
-      NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW,
-      base::Bind(&NetLogSpdySessionWindowUpdateCallback,
-                 delta_window_size, session_recv_window_size_));
-
-  session_unacked_recv_window_bytes_ += delta_window_size;
-  if (session_unacked_recv_window_bytes_ > kSpdySessionInitialWindowSize / 2) {
-    SendWindowUpdateFrame(kSessionFlowControlStreamId,
-                          session_unacked_recv_window_bytes_,
-                          HIGHEST);
-    session_unacked_recv_window_bytes_ = 0;
-  }
-}
-
 // Given a cwnd that we would have sent to the server, modify it based on the
 // field trial policy.
 uint32 ApplyCwndFieldTrialPolicy(int cwnd) {
   base::FieldTrial* trial = base::FieldTrialList::Find("SpdyCwnd");
   if (!trial) {
       LOG(WARNING) << "Could not find \"SpdyCwnd\" in FieldTrialList";
       return cwnd;
   }
   if (trial->group_name() == "cwnd10")
     return 10;
@@ skipping 430 matching lines @@
   DCHECK_GE(session_send_window_size_, delta_window_size);
 
   session_send_window_size_ -= delta_window_size;
 
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_UPDATE_SEND_WINDOW,
       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
                  -delta_window_size, session_send_window_size_));
 }
 
+void SpdySession::IncreaseRecvWindowSize(size_t delta_window_size) {
+  if (flow_control_state_ < FLOW_CONTROL_STREAM_AND_SESSION)
+    return;
+
+  DCHECK_GE(session_unacked_recv_window_bytes_, 0);
+  DCHECK_GE(session_recv_window_size_, session_unacked_recv_window_bytes_);
+  DCHECK_GE(delta_window_size, 1u);
+  // Check for overflow.
+  DCHECK_LE(delta_window_size,
+            static_cast<size_t>(kint32max - session_recv_window_size_));
+
+  session_recv_window_size_ += delta_window_size;
+  net_log_.AddEvent(
+      NetLog::TYPE_SPDY_STREAM_UPDATE_RECV_WINDOW,
+      base::Bind(&NetLogSpdySessionWindowUpdateCallback,
+                 delta_window_size, session_recv_window_size_));
+
+  session_unacked_recv_window_bytes_ += delta_window_size;
+  if (session_unacked_recv_window_bytes_ > kSpdySessionInitialWindowSize / 2) {
+    SendWindowUpdateFrame(kSessionFlowControlStreamId,
+                          session_unacked_recv_window_bytes_,
+                          HIGHEST);
+    session_unacked_recv_window_bytes_ = 0;
+  }
+}
+
 void SpdySession::DecreaseRecvWindowSize(int32 delta_window_size) {
   DCHECK_EQ(flow_control_state_, FLOW_CONTROL_STREAM_AND_SESSION);
   DCHECK_GE(delta_window_size, 1);
 
   // |delta_window_size| should never cause
   // |session_recv_window_size_| to go negative. If we do, it's a
   // client-side bug.
   if (delta_window_size > session_recv_window_size_) {
     NOTREACHED() << "Received session WINDOW_UPDATE with an "
                  << "invalid delta_window_size " << delta_window_size;
     // TODO(akalin): Figure out whether we should instead send a
     // GOAWAY and close the connection here.
     return;
   }
 
   session_recv_window_size_ -= delta_window_size;
   net_log_.AddEvent(
       NetLog::TYPE_SPDY_SESSION_UPDATE_RECV_WINDOW,
       base::Bind(&NetLogSpdySessionWindowUpdateCallback,
                  -delta_window_size, session_recv_window_size_));
 }
 
 }  // namespace net