Remove old unclaimed streams pushed by a SPDY server.

BUG=113424
TEST=SpdySessionSpdy[2,3]Test.DeleteExpiredPushStreams

Review URL: https://chromiumcodereview.appspot.com/10702189

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@147321 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 232 insertions, 6 deletions

diff --git a/net/spdy/spdy_session.cc b/net/spdy/spdy_session.cc
index c127cdf..5528617 100644
--- a/net/spdy/spdy_session.cc
+++ b/net/spdy/spdy_session.cc
@@ -45,6 +45,9 @@ const int kReadBufferSize = 8 * 1024;
 const int kDefaultConnectionAtRiskOfLossSeconds = 10;
 const int kHungIntervalSeconds = 10;
 
+// Minimum seconds that unclaimed pushed streams will be kept in memory.
+const int kMinPushedStreamLifetimeSeconds = 300;
+
 Value* NetLogSpdySynCallback(const SpdyHeaderBlock* headers,
                              bool fin,
                              bool unidirectional,
@@ -182,6 +185,10 @@ size_t g_max_concurrent_stream_limit = 256;
 size_t g_default_initial_rcv_window_size = 10 * 1024 * 1024;  // 10MB
 bool g_enable_ping_based_connection_checking = true;
 
+typedef base::TimeTicks (*ExternalTimeFunc)(void);
+
+static ExternalTimeFunc g_time_func = base::TimeTicks::Now;
+
 }  // namespace
 
 // static
@@ -218,6 +225,23 @@ void SpdySession::ResetStaticSettingsToInit() {
   g_max_concurrent_stream_limit = 256;
   g_default_initial_rcv_window_size = kSpdyStreamInitialWindowSize;
   g_enable_ping_based_connection_checking = true;
+  g_time_func = base::TimeTicks::Now;
+}
+
+// Outside of tests, g_time_func will always be base::TimeTicks::Now.
+// When performing linker optimization for the main executable, the compiler
+// should be able to see that set_time_func() is an uncalled function, that
+// the static .cc variable never changes, and thus that the extra pointer
+// indirection can be removed.
+
+
+
+SpdySession::TimeFunc SpdySession::set_time_func(
+    SpdySession::TimeFunc time_func) {
+  SpdySession::TimeFunc old_time_func =
+      static_cast<SpdySession::TimeFunc>(g_time_func);
+  g_time_func = static_cast<ExternalTimeFunc>(time_func);
+  return old_time_func;
 }
 
 SpdySession::SpdySession(const HostPortProxyPair& host_port_proxy_pair,
@@ -270,6 +294,8 @@ SpdySession::SpdySession(const HostPortProxyPair& host_port_proxy_pair,
   net_log_.BeginEvent(
       NetLog::TYPE_SPDY_SESSION,
       base::Bind(&NetLogSpdySessionCallback, &host_port_proxy_pair_));
+  next_unclaimed_push_stream_sweep_time_ = g_time_func() +
+      base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
   // TODO(mbelshe): consider randomization of the stream_hi_water_mark.
 }
 
@@ -1118,7 +1144,7 @@ void SpdySession::DeleteStream(SpdyStreamId id, int status) {
     PushedStreamMap::iterator it;
     for (it = unclaimed_pushed_streams_.begin();
          it != unclaimed_pushed_streams_.end(); ++it) {
-      scoped_refptr<SpdyStream> curr = it->second;
+      scoped_refptr<SpdyStream> curr = it->second.first;
       if (id == curr->stream_id()) {
         unclaimed_pushed_streams_.erase(it);
         break;
@@ -1154,7 +1180,7 @@ scoped_refptr<SpdyStream> SpdySession::GetActivePushStream(
   PushedStreamMap::iterator it = unclaimed_pushed_streams_.find(path);
   if (it != unclaimed_pushed_streams_.end()) {
     net_log_.AddEvent(NetLog::TYPE_SPDY_STREAM_ADOPTED_PUSH_STREAM);
-    scoped_refptr<SpdyStream> stream = it->second;
+    scoped_refptr<SpdyStream> stream = it->second.first;
     unclaimed_pushed_streams_.erase(it);
     used_push_streams.Increment();
     return stream;
@@ -1363,7 +1389,11 @@ void SpdySession::OnSynStream(SpdyStreamId stream_id,
   stream->set_send_window_size(initial_send_window_size_);
   stream->set_recv_window_size(initial_recv_window_size_);
 
-  unclaimed_pushed_streams_[url] = stream;
+  DeleteExpiredPushedStreams();
+  unclaimed_pushed_streams_[url] =
+      std::pair<scoped_refptr<SpdyStream>, base::TimeTicks> (
+          stream, g_time_func());
+
 
   ActivateStream(stream);
   stream->set_response_received();
@@ -1376,6 +1406,38 @@ void SpdySession::OnSynStream(SpdyStreamId stream_id,
   push_requests.Increment();
 }
 
+void SpdySession::DeleteExpiredPushedStreams() {
+  if (unclaimed_pushed_streams_.empty())
+    return;
+
+  // Check that adequate time has elapsed since the last sweep.
+  if (g_time_func() < next_unclaimed_push_stream_sweep_time_)
+    return;
+
+  // Delete old streams.
+  base::TimeTicks minimum_freshness = g_time_func() -
+      base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
+  PushedStreamMap::iterator it;
+  for (it = unclaimed_pushed_streams_.begin();
+      it != unclaimed_pushed_streams_.end(); ) {
+    const scoped_refptr<SpdyStream>& stream = it->second.first;
+    base::TimeTicks creation_time = it->second.second;
+    // DeleteStream() will invalidate the current iterator, so move to next.
+    ++it;
+    if (minimum_freshness > creation_time) {
+      DeleteStream(stream->stream_id(), ERR_INVALID_SPDY_STREAM);
+      base::StatsCounter abandoned_push_streams(
+          "spdy.abandoned_push_streams");
+      base::StatsCounter abandoned_streams("spdy.abandoned_streams");
+      abandoned_push_streams.Increment();
+      abandoned_streams.Increment();
+      streams_abandoned_count_++;
+    }
+  }
+  next_unclaimed_push_stream_sweep_time_ = g_time_func() +
+      base::TimeDelta::FromSeconds(kMinPushedStreamLifetimeSeconds);
+}
+
 void SpdySession::OnSynReply(SpdyStreamId stream_id,
                              bool fin,
                              const SpdyHeaderBlock& headers) {
diff --git a/net/spdy/spdy_session.h b/net/spdy/spdy_session.h
index 9b4cf73..2975203 100644
--- a/net/spdy/spdy_session.h
+++ b/net/spdy/spdy_session.h
@@ -14,6 +14,7 @@
 #include "base/gtest_prod_util.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/weak_ptr.h"
+#include "base/time.h"
 #include "net/base/io_buffer.h"
 #include "net/base/load_states.h"
 #include "net/base/net_errors.h"
@@ -322,9 +323,11 @@ class NET_EXPORT SpdySession : public base::RefCounted<SpdySession>,
   FRIEND_TEST_ALL_PREFIXES(SpdySessionSpdy2Test, Ping);
   FRIEND_TEST_ALL_PREFIXES(SpdySessionSpdy2Test, FailedPing);
   FRIEND_TEST_ALL_PREFIXES(SpdySessionSpdy2Test, GetActivePushStream);
+  FRIEND_TEST_ALL_PREFIXES(SpdySessionSpdy2Test, DeleteExpiredPushStreams);
   FRIEND_TEST_ALL_PREFIXES(SpdySessionSpdy3Test, Ping);
   FRIEND_TEST_ALL_PREFIXES(SpdySessionSpdy3Test, FailedPing);
   FRIEND_TEST_ALL_PREFIXES(SpdySessionSpdy3Test, GetActivePushStream);
+  FRIEND_TEST_ALL_PREFIXES(SpdySessionSpdy3Test, DeleteExpiredPushStreams);
 
   struct PendingCreateStream {
     PendingCreateStream(const GURL& url, RequestPriority priority,
@@ -346,11 +349,11 @@ class NET_EXPORT SpdySession : public base::RefCounted<SpdySession>,
     const BoundNetLog* stream_net_log;
     CompletionCallback callback;
   };
-  typedef std::queue<PendingCreateStream, std::list< PendingCreateStream> >
+  typedef std::queue<PendingCreateStream, std::list<PendingCreateStream> >
       PendingCreateStreamQueue;
   typedef std::map<int, scoped_refptr<SpdyStream> > ActiveStreamMap;
-  // Only HTTP push a stream.
-  typedef std::map<std::string, scoped_refptr<SpdyStream> > PushedStreamMap;
+  typedef std::map<std::string,
+      std::pair<scoped_refptr<SpdyStream>, base::TimeTicks> > PushedStreamMap;
   typedef std::priority_queue<SpdyIOBuffer> OutputQueue;
 
   struct CallbackResultPair {
@@ -372,6 +375,8 @@ class NET_EXPORT SpdySession : public base::RefCounted<SpdySession>,
     CLOSED
   };
 
+  typedef base::TimeTicks (*TimeFunc)(void);
+
   virtual ~SpdySession();
 
   void ProcessPendingCreateStreams();
@@ -461,6 +466,9 @@ class NET_EXPORT SpdySession : public base::RefCounted<SpdySession>,
   // can be deferred to the MessageLoop, so we avoid re-entrancy problems.
   void InvokeUserStreamCreationCallback(scoped_refptr<SpdyStream>* stream);
 
+  // Remove old unclaimed pushed streams.
+  void DeleteExpiredPushedStreams();
+
   // BufferedSpdyFramerVisitorInterface:
   virtual void OnError(SpdyFramer::SpdyError error_code) OVERRIDE;
   virtual void OnStreamError(SpdyStreamId stream_id,
@@ -500,6 +508,9 @@ class NET_EXPORT SpdySession : public base::RefCounted<SpdySession>,
   // --------------------------
   // Helper methods for testing
   // --------------------------
+
+  static TimeFunc set_time_func(TimeFunc new_time_func);
+
   void set_connection_at_risk_of_loss_time(base::TimeDelta duration) {
     connection_at_risk_of_loss_time_ = duration;
   }
@@ -620,6 +631,10 @@ class NET_EXPORT SpdySession : public base::RefCounted<SpdySession>,
   // This is the last time we had activity in the session.
   base::TimeTicks last_activity_time_;
 
+  // This is the next time that unclaimed push streams should be checked for
+  // expirations.
+  base::TimeTicks next_unclaimed_push_stream_sweep_time_;
+
   // Indicate if we have already scheduled a delayed task to check the ping
   // status.
   bool check_ping_status_pending_;
diff --git a/net/spdy/spdy_session_spdy2_unittest.cc b/net/spdy/spdy_session_spdy2_unittest.cc
index a884854f..4171d9f 100644
--- a/net/spdy/spdy_session_spdy2_unittest.cc
+++ b/net/spdy/spdy_session_spdy2_unittest.cc
@@ -15,6 +15,15 @@
 
 using namespace net::test_spdy2;
 
+namespace {
+
+base::TimeTicks the_near_future() {
+  return base::TimeTicks::Now() +
+      base::TimeDelta::FromSeconds(301);
+}
+
+} // namespace
+
 namespace net {
 
 // TODO(cbentzel): Expose compression setter/getter in public SpdySession
@@ -251,6 +260,69 @@ TEST_F(SpdySessionSpdy2Test, Ping) {
   session = NULL;
 }
 
+TEST_F(SpdySessionSpdy2Test, DeleteExpiredPushStreams) {
+  SpdySessionDependencies session_deps;
+  session_deps.host_resolver->set_synchronous_mode(true);
+
+  SSLSocketDataProvider ssl(SYNCHRONOUS, OK);
+  session_deps.socket_factory->AddSSLSocketDataProvider(&ssl);
+
+  scoped_refptr<HttpNetworkSession> http_session(
+      SpdySessionDependencies::SpdyCreateSession(&session_deps));
+
+  const std::string kTestHost("www.google.com");
+  const int kTestPort = 80;
+  HostPortPair test_host_port_pair(kTestHost, kTestPort);
+  HostPortProxyPair pair(test_host_port_pair, ProxyServer::Direct());
+
+  SpdySessionPool* spdy_session_pool(http_session->spdy_session_pool());
+  EXPECT_FALSE(spdy_session_pool->HasSession(pair));
+  scoped_refptr<SpdySession> session =
+      spdy_session_pool->Get(pair, BoundNetLog());
+  EXPECT_TRUE(spdy_session_pool->HasSession(pair));
+
+  // Give the session a SPDY2 framer.
+  session->buffered_spdy_framer_.reset(new BufferedSpdyFramer(2));
+
+  // Create the associated stream and add to active streams.
+  scoped_ptr<SpdyHeaderBlock> request_headers(new SpdyHeaderBlock);
+  (*request_headers)["scheme"] = "http";
+  (*request_headers)["host"] = "www.google.com";
+  (*request_headers)["url"] = "/";
+
+  scoped_refptr<SpdyStream> stream(
+      new SpdyStream(session, 1, false, session->net_log_));
+  stream->set_spdy_headers(request_headers.Pass());
+  session->ActivateStream(stream);
+
+  SpdyHeaderBlock headers;
+  headers["url"] = "http://www.google.com/a.dat";
+  session->OnSynStream(2, 1, 0, 0, true, false, headers);
+
+  // Verify that there is one unclaimed push stream.
+  EXPECT_EQ(1u, session->num_unclaimed_pushed_streams());
+  SpdySession::PushedStreamMap::iterator  iter  =
+      session->unclaimed_pushed_streams_.find("http://www.google.com/a.dat");
+  EXPECT_TRUE(session->unclaimed_pushed_streams_.end() != iter);
+
+  // Shift time.
+  SpdySession::set_time_func(the_near_future);
+
+  headers["url"] = "http://www.google.com/b.dat";
+  session->OnSynStream(4, 1, 0, 0, true, false, headers);
+
+  // Verify that the second pushed stream evicted the first pushed stream.
+  EXPECT_EQ(1u, session->num_unclaimed_pushed_streams());
+  iter = session->unclaimed_pushed_streams_.find("http://www.google.com/b.dat");
+  EXPECT_TRUE(session->unclaimed_pushed_streams_.end() != iter);
+
+  SpdySession::ResetStaticSettingsToInit();
+
+  // Delete the session.
+  session = NULL;
+}
+
+
 TEST_F(SpdySessionSpdy2Test, FailedPing) {
   SpdySessionDependencies session_deps;
   session_deps.host_resolver->set_synchronous_mode(true);
diff --git a/net/spdy/spdy_session_spdy3_unittest.cc b/net/spdy/spdy_session_spdy3_unittest.cc
index 2d8d969..d8d526c 100644
--- a/net/spdy/spdy_session_spdy3_unittest.cc
+++ b/net/spdy/spdy_session_spdy3_unittest.cc
@@ -15,6 +15,15 @@
 
 using namespace net::test_spdy3;
 
+namespace {
+
+base::TimeTicks the_near_future() {
+  return base::TimeTicks::Now() +
+      base::TimeDelta::FromSeconds(301);
+}
+
+} // namespace
+
 namespace net {
 
 // TODO(cbentzel): Expose compression setter/getter in public SpdySession
@@ -251,6 +260,73 @@ TEST_F(SpdySessionSpdy3Test, Ping) {
   session = NULL;
 }
 
+TEST_F(SpdySessionSpdy3Test, DeleteExpiredPushStreams) {
+  SpdySessionDependencies session_deps;
+  session_deps.host_resolver->set_synchronous_mode(true);
+
+  SSLSocketDataProvider ssl(SYNCHRONOUS, OK);
+  session_deps.socket_factory->AddSSLSocketDataProvider(&ssl);
+
+  scoped_refptr<HttpNetworkSession> http_session(
+      SpdySessionDependencies::SpdyCreateSession(&session_deps));
+
+  const std::string kTestHost("www.google.com");
+  const int kTestPort = 80;
+  HostPortPair test_host_port_pair(kTestHost, kTestPort);
+  HostPortProxyPair pair(test_host_port_pair, ProxyServer::Direct());
+
+  SpdySessionPool* spdy_session_pool(http_session->spdy_session_pool());
+  EXPECT_FALSE(spdy_session_pool->HasSession(pair));
+  scoped_refptr<SpdySession> session =
+      spdy_session_pool->Get(pair, BoundNetLog());
+  EXPECT_TRUE(spdy_session_pool->HasSession(pair));
+
+  // Give the session a SPDY3 framer.
+  session->buffered_spdy_framer_.reset(new BufferedSpdyFramer(3));
+
+  // Create the associated stream and add to active streams.
+  scoped_ptr<SpdyHeaderBlock> request_headers(new SpdyHeaderBlock);
+  (*request_headers)[":scheme"] = "http";
+  (*request_headers)[":host"] = "www.google.com";
+  (*request_headers)[":path"] = "/";
+
+  scoped_refptr<SpdyStream> stream(
+      new SpdyStream(session, 1, false, session->net_log_));
+  stream->set_spdy_headers(request_headers.Pass());
+  session->ActivateStream(stream);
+
+  SpdyHeaderBlock headers;
+  headers[":scheme"] = "http";
+  headers[":host"] = "www.google.com";
+  headers[":path"] = "/a.dat";
+  session->OnSynStream(2, 1, 0, 0, true, false, headers);
+
+  // Verify that there is one unclaimed push stream.
+  EXPECT_EQ(1u, session->num_unclaimed_pushed_streams());
+  SpdySession::PushedStreamMap::iterator  iter  =
+      session->unclaimed_pushed_streams_.find("http://www.google.com/a.dat");
+  EXPECT_TRUE(session->unclaimed_pushed_streams_.end() != iter);
+
+  // Shift time.
+  SpdySession::set_time_func(the_near_future);
+
+  headers[":scheme"] = "http";
+  headers[":host"] = "www.google.com";
+  headers[":path"] = "/b.dat";
+  session->OnSynStream(4, 1, 0, 0, true, false, headers);
+
+  // Verify that the second pushed stream evicted the first pushed stream.
+  EXPECT_EQ(1u, session->num_unclaimed_pushed_streams());
+  iter = session->unclaimed_pushed_streams_.find("http://www.google.com/b.dat");
+  EXPECT_TRUE(session->unclaimed_pushed_streams_.end() != iter);
+
+  SpdySession::ResetStaticSettingsToInit();
+
+  // Delete the session.
+  session = NULL;
+}
+
+
 TEST_F(SpdySessionSpdy3Test, FailedPing) {
   SpdySessionDependencies session_deps;
   session_deps.host_resolver->set_synchronous_mode(true);
diff --git a/net/spdy/spdy_stream.h b/net/spdy/spdy_stream.h
index 1c6625b..40e436a 100644
--- a/net/spdy/spdy_stream.h
+++ b/net/spdy/spdy_stream.h
@@ -267,6 +267,7 @@ class NET_EXPORT_PRIVATE SpdyStream
   };
 
   friend class base::RefCounted<SpdyStream>;
+
   virtual ~SpdyStream();
 
   // If the stream is stalled and if |send_window_size_| is positive, then set