When an idle socket is added back to a socket pool, check for stalled jobs in lower pools

net/socket/client_socket_pool_base.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/socket/client_socket_pool_base.h"
 
 #include "base/compiler_specific.h"
 #include "base/format_macros.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
@@ skipping 143 matching lines @@
     : handle_(handle),
       callback_(callback),
       priority_(priority),
       ignore_limits_(ignore_limits),
       flags_(flags),
       net_log_(net_log) {}
 
 ClientSocketPoolBaseHelper::Request::~Request() {}
 
 ClientSocketPoolBaseHelper::ClientSocketPoolBaseHelper(
+    HigherLayeredPool* pool,
     int max_sockets,
     int max_sockets_per_group,
     base::TimeDelta unused_idle_socket_timeout,
     base::TimeDelta used_idle_socket_timeout,
     ConnectJobFactory* connect_job_factory)
     : idle_socket_count_(0),
       connecting_socket_count_(0),
       handed_out_socket_count_(0),
       max_sockets_(max_sockets),
       max_sockets_per_group_(max_sockets_per_group),
       use_cleanup_timer_(g_cleanup_timer_enabled),
       unused_idle_socket_timeout_(unused_idle_socket_timeout),
       used_idle_socket_timeout_(used_idle_socket_timeout),
       connect_job_factory_(connect_job_factory),
       connect_backup_jobs_enabled_(false),
       pool_generation_number_(0),
+      pool_(pool),
       weak_factory_(this) {
   DCHECK_LE(0, max_sockets_per_group);
   DCHECK_LE(max_sockets_per_group, max_sockets);
 
   NetworkChangeNotifier::AddIPAddressObserver(this);
 }
 
 ClientSocketPoolBaseHelper::~ClientSocketPoolBaseHelper() {
   // Clean up any idle sockets and pending connect jobs.  Assert that we have no
   // remaining active sockets or pending requests.  They should have all been
   // cleaned up prior to |this| being destroyed.
   FlushWithError(ERR_ABORTED);
   DCHECK(group_map_.empty());
   DCHECK(pending_callback_map_.empty());
   DCHECK_EQ(0, connecting_socket_count_);
-  CHECK(higher_layer_pools_.empty());
+  CHECK(higher_pools_.empty());
 
   NetworkChangeNotifier::RemoveIPAddressObserver(this);
+
+  // Remove from lower layer pools.
+  for (std::set<LowerLayeredPool*>::iterator it = lower_pools_.begin();
+       it != lower_pools_.end();
+       ++it) {
+    (*it)->RemoveHigherLayeredPool(pool_);
+  }
 }
 
 ClientSocketPoolBaseHelper::CallbackResultPair::CallbackResultPair()
     : result(OK) {
 }
 
 ClientSocketPoolBaseHelper::CallbackResultPair::CallbackResultPair(
     const CompletionCallback& callback_in, int result_in)
     : callback(callback_in),
       result(result_in) {
@@ skipping 19 matching lines @@
 ClientSocketPoolBaseHelper::RemoveRequestFromQueue(
     const RequestQueue::iterator& it, Group* group) {
   const Request* req = *it;
   group->mutable_pending_requests()->erase(it);
   // If there are no more requests, we kill the backup timer.
   if (group->pending_requests().empty())
     group->CleanupBackupJob();
   return req;
 }
 
-void ClientSocketPoolBaseHelper::AddLayeredPool(LayeredPool* pool) {
-  CHECK(pool);
-  CHECK(!ContainsKey(higher_layer_pools_, pool));
-  higher_layer_pools_.insert(pool);
+bool ClientSocketPoolBaseHelper::IsStalled() const {
+  // If a lower layer pool is stalled, consider |this| stalled as well.

[Ryan Hamilton, 2013/07/21 14:53:54]

This seems like mostly a good idea, but how does it work with SPDY sessions in
which case, the transport socket pool can be stalled but the SpdySession can
still vend new streams?

[mmenke, 2013/07/22 20:13:33]

SpdySessions can never be stalled.  They aren't considered LowerLayeredPools, so
they don't need to implement this function, conveniently enough.

Anything layered on top of a SpdySession works completely differently from a
socket pool - since it can always get streams from the session, it can't be
stalled (Unless there's a stream limit on the session, but that's beyond the
scope of this CL).

That having been said, SpdySessions can result in other requests (Possibly
requests needed to create other SpdySessions) can be stalled, of course, and so
if a session becomes idle, it will be closed if there are stalled lower level
pools.

[Ryan Hamilton, 2013/07/22 20:22:46]

Works for me.  FWIW, there is a streams/session limit which I think is typically
100.  Normally this isn't much of a problem.  Though in the case of a SPDY Proxy
it might potentially be one.  But no reason to worry about that in this CL, I
think.
 

[mmenke, 2013/07/24 19:05:23]

Yea, I don't think so - that would need to be dealt with just between the
SpdySession and and the things on top of it.  No need for it to dig down into
the socket pool, unless we wanted to use multiple SpdySessions at that point
(And I assume we wouldn't want to do that, since that defeats the point of the
stream limit in the first place).

+  for (std::set<LowerLayeredPool*>::const_iterator it = lower_pools_.begin();
+       it != lower_pools_.end();
+       ++it) {
+    if ((*it)->IsStalled())
+      return true;
+  }
+
+  // If we are not using |max_sockets_|, then clearly we are not stalled
+  if ((handed_out_socket_count_ + connecting_socket_count_) < max_sockets_)
+    return false;
+  // So in order to be stalled we need to be using |max_sockets_| AND
+  // we need to have a request that is actually stalled on the global
+  // socket limit.  To find such a request, we look for a group that
+  // a has more requests that jobs AND where the number of jobs is less
+  // than |max_sockets_per_group_|.  (If the number of jobs is equal to
+  // |max_sockets_per_group_|, then the request is stalled on the group,
+  // which does not count.)
+  for (GroupMap::const_iterator it = group_map_.begin();
+       it != group_map_.end(); ++it) {
+    if (it->second->IsStalledOnPoolMaxSockets(max_sockets_per_group_))
+      return true;
+  }
+  return false;
 }
 
-void ClientSocketPoolBaseHelper::RemoveLayeredPool(LayeredPool* pool) {
-  CHECK(pool);
-  CHECK(ContainsKey(higher_layer_pools_, pool));
-  higher_layer_pools_.erase(pool);
+void ClientSocketPoolBaseHelper::AddLowerLayeredPool(
+    LowerLayeredPool* lower_pool) {
+  DCHECK(pool_);
+  CHECK(!ContainsKey(lower_pools_, lower_pool));
+  lower_pools_.insert(lower_pool);
+  lower_pool->AddHigherLayeredPool(pool_);
+}
+
+void ClientSocketPoolBaseHelper::AddHigherLayeredPool(
+    HigherLayeredPool* higher_pool) {
+  CHECK(higher_pool);
+  CHECK(!ContainsKey(higher_pools_, higher_pool));
+  higher_pools_.insert(higher_pool);
+}
+
+void ClientSocketPoolBaseHelper::RemoveHigherLayeredPool(
+    HigherLayeredPool* higher_pool) {
+  CHECK(higher_pool);
+  CHECK(ContainsKey(higher_pools_, higher_pool));
+  higher_pools_.erase(higher_pool);
 }
 
 int ClientSocketPoolBaseHelper::RequestSocket(
     const std::string& group_name,
     const Request* request) {
   CHECK(!request->callback().is_null());
   CHECK(request->handle());
 
   // Cleanup any timed-out idle sockets if no timer is used.
   if (!use_cleanup_timer_)
@@ skipping 520 matching lines @@
     delete socket;
   }
 
   CheckForStalledSocketGroups();
 }
 
 void ClientSocketPoolBaseHelper::CheckForStalledSocketGroups() {
   // If we have idle sockets, see if we can give one to the top-stalled group.
   std::string top_group_name;
   Group* top_group = NULL;
-  if (!FindTopStalledGroup(&top_group, &top_group_name))
+  if (!FindTopStalledGroup(&top_group, &top_group_name)) {
+    // There may still be a stalled group in a lower level pool.
+    for (std::set<LowerLayeredPool*>::iterator it = lower_pools_.begin();
+         it != lower_pools_.end();
+         ++it) {
+       if ((*it)->IsStalled()) {
+         CloseOneIdleSocket();
+         break;
+       }
+    }
     return;
+  }
 
   if (ReachedMaxSocketsLimit()) {
     if (idle_socket_count() > 0) {
       CloseOneIdleSocket();
     } else {
       // We can't activate more sockets since we're already at our global
       // limit.
       return;
     }
   }
@@ skipping 109 matching lines @@
   FlushWithError(ERR_NETWORK_CHANGED);
 }
 
 void ClientSocketPoolBaseHelper::FlushWithError(int error) {
   pool_generation_number_++;
   CancelAllConnectJobs();
   CloseIdleSockets();
   CancelAllRequestsWithError(error);
 }
 
-bool ClientSocketPoolBaseHelper::IsStalled() const {
-  // If we are not using |max_sockets_|, then clearly we are not stalled
-  if ((handed_out_socket_count_ + connecting_socket_count_) < max_sockets_)
-    return false;
-  // So in order to be stalled we need to be using |max_sockets_| AND
-  // we need to have a request that is actually stalled on the global
-  // socket limit.  To find such a request, we look for a group that
-  // a has more requests that jobs AND where the number of jobs is less
-  // than |max_sockets_per_group_|.  (If the number of jobs is equal to
-  // |max_sockets_per_group_|, then the request is stalled on the group,
-  // which does not count.)
-  for (GroupMap::const_iterator it = group_map_.begin();
-       it != group_map_.end(); ++it) {
-    if (it->second->IsStalledOnPoolMaxSockets(max_sockets_per_group_))
-      return true;
-  }
-  return false;
-}
-
 void ClientSocketPoolBaseHelper::RemoveConnectJob(ConnectJob* job,
                                                   Group* group) {
   CHECK_GT(connecting_socket_count_, 0);
   connecting_socket_count_--;
 
   DCHECK(group);
   DCHECK(ContainsKey(group->jobs(), job));
   group->RemoveJob(job);
 
   // If we've got no more jobs for this group, then we no longer need a
@@ skipping 147 matching lines @@
       if (group->IsEmpty())
         RemoveGroup(i);
 
       return true;
     }
   }
 
   return false;
 }
 
-bool ClientSocketPoolBaseHelper::CloseOneIdleConnectionInLayeredPool() {
+bool ClientSocketPoolBaseHelper::CloseOneIdleConnectionInHigherLayeredPool() {
   // This pool doesn't have any idle sockets. It's possible that a pool at a
   // higher layer is holding one of this sockets active, but it's actually idle.
   // Query the higher layers.
-  for (std::set<LayeredPool*>::const_iterator it = higher_layer_pools_.begin();
-       it != higher_layer_pools_.end(); ++it) {
+  for (std::set<HigherLayeredPool*>::const_iterator it = higher_pools_.begin();
+       it != higher_pools_.end(); ++it) {
     if ((*it)->CloseOneIdleConnection())
       return true;
   }
   return false;
 }
 
 void ClientSocketPoolBaseHelper::InvokeUserCallbackLater(
     ClientSocketHandle* handle, const CompletionCallback& callback, int rv) {
   CHECK(!ContainsKey(pending_callback_map_, handle));
   pending_callback_map_[handle] = CallbackResultPair(callback, rv);
@@ skipping 15 matching lines @@
   CompletionCallback callback = it->second.callback;
   int result = it->second.result;
   pending_callback_map_.erase(it);
   callback.Run(result);
 }
 
 void ClientSocketPoolBaseHelper::TryToCloseSocketsInLayeredPools() {
   while (IsStalled()) {
     // Closing a socket will result in calling back into |this| to use the freed
     // socket slot, so nothing else is needed.
-    if (!CloseOneIdleConnectionInLayeredPool())
+    if (!CloseOneIdleConnectionInHigherLayeredPool())
       return;
   }
 }
 
 ClientSocketPoolBaseHelper::Group::Group()
     : unassigned_job_count_(0),
       active_socket_count_(0),
       weak_factory_(this) {}
 
 ClientSocketPoolBaseHelper::Group::~Group() {
@@ skipping 86 matching lines @@
   STLDeleteElements(&jobs_);
   unassigned_job_count_ = 0;
 
   // Cancel pending backup job.
   weak_factory_.InvalidateWeakPtrs();
 }
 
 }  // namespace internal
 
 }  // namespace net