Close idle connections / SPDY sessions when needed.

net/socket/client_socket_pool_base.cc

 // Copyright (c) 2011 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 <math.h>
 #include "base/compiler_specific.h"
 #include "base/format_macros.h"
 #include "base/logging.h"
@@ skipping 184 matching lines @@
 }
 
 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.
   Flush();
   DCHECK(group_map_.empty());
   DCHECK(pending_callback_map_.empty());
   DCHECK_EQ(0, connecting_socket_count_);
+  DCHECK(higher_layer_pools_.empty());
 
   NetworkChangeNotifier::RemoveIPAddressObserver(this);
 }
 
 // InsertRequestIntoQueue inserts the request into the queue based on
 // priority.  Highest priorities are closest to the front.  Older requests are
 // prioritized over requests of equal priority.
 //
 // static
 void ClientSocketPoolBaseHelper::InsertRequestIntoQueue(
     const Request* r, RequestQueue* pending_requests) {
   RequestQueue::iterator it = pending_requests->begin();
   while (it != pending_requests->end() && r->priority() >= (*it)->priority())
     ++it;
   pending_requests->insert(it, r);
 }
 
 // static
 const ClientSocketPoolBaseHelper::Request*
 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);
+}
+
+void ClientSocketPoolBaseHelper::RemoveLayeredPool(LayeredPool* pool) {
+  CHECK(pool);
+  CHECK(ContainsKey(higher_layer_pools_, pool));
+  higher_layer_pools_.erase(pool);
+}
+
 int ClientSocketPoolBaseHelper::RequestSocket(
     const std::string& group_name,
     const Request* request) {
   CHECK(request->callback());
   CHECK(request->handle());
 
   request->net_log().BeginEvent(NetLog::TYPE_SOCKET_POOL, NULL);
   Group* group = GetOrCreateGroup(group_name);
 
   int rv = RequestSocketInternal(group_name, request);
@@ skipping 70 matching lines @@
     if (AssignIdleSocketToGroup(request, group))
       return OK;
   }
 
   if (!preconnecting && group->TryToUsePreconnectConnectJob())
     return ERR_IO_PENDING;
 
   // Can we make another active socket now?
   if (!group->HasAvailableSocketSlot(max_sockets_per_group_) &&
       !request->ignore_limits()) {
+    // TODO(willchan): Consider whether or not we need to close a socket in a
+    // higher layered group. I don't think this makes sense since we would just
+    // reuse that socket then if we needed one and wouldn't make it down to this
+    // layer.
     request->net_log().AddEvent(
         NetLog::TYPE_SOCKET_POOL_STALLED_MAX_SOCKETS_PER_GROUP, NULL);
     return ERR_IO_PENDING;
   }
 
   if (ReachedMaxSocketsLimit() && !request->ignore_limits()) {
     if (idle_socket_count() > 0) {
+      // There's an idle socket in this pool. Either that's because there's
+      // still one in this group, but we got here due to preconnecting bypassing
+      // idle sockets, or because there's an idle socket in another group.
       bool closed = CloseOneIdleSocketExceptInGroup(group);
       if (preconnecting && !closed)
         return ERR_PRECONNECT_MAX_SOCKET_LIMIT;
-    } else {
+    } else if (!CloseOneIdleConnectionInLayeredPool()) {

[mmenke, 2011/10/27 18:52:37]

Is this guaranteed to result in ReachedMaxSocketsLimit() returning false?  Seems
to me that if we had a socket with |ignore_limits| set to true, maybe not.

May be an obscure enough case that we can ignore it here, though, as things
should still work.

[willchan no longer on Chromium, 2011/11/08 22:26:13]

Sorry, I don't grok. Which path do you mean? Also note that we're within a code
block where !request->ignore_limits().

[mmenke, 2011/11/08 22:41:04]

So you have a request with ignore_limits() set to true.  We then exceed the
limit by one.  A new request then reaches this code.  We close an idle socket. 
We're now at the limit.  We create a new socket, and end up exceeding the limit
once more.

Not exactly a critical problem, but does manage to bypass the limit in a rather
non-transparent manner.

[willchan no longer on Chromium, 2011/11/15 18:44:06]

GREAT catch. I'm impressed you caught this given your relative unfamiliarity
with this case. I'll see if I can write a test to catch this behavior.

       // We could check if we really have a stalled group here, but it requires
       // a scan of all groups, so just flip a flag here, and do the check later.
       request->net_log().AddEvent(
           NetLog::TYPE_SOCKET_POOL_STALLED_MAX_SOCKETS, NULL);
       return ERR_IO_PENDING;
     }
   }
 
-  // We couldn't find a socket to reuse, so allocate and connect a new one.
+  // We couldn't find a socket to reuse, and there's space to allocate one,
+  // so allocate and connect a new one.
   scoped_ptr<ConnectJob> connect_job(
       connect_job_factory_->NewConnectJob(group_name, *request, this));
 
   connect_job->Initialize(preconnecting);
   int rv = connect_job->Connect();
   if (rv == OK) {
     LogBoundConnectJobToRequest(connect_job->net_log().source(), request);
     if (!preconnecting) {
       HandOutSocket(connect_job->ReleaseSocket(), false /* not reused */,
                     handle, base::TimeDelta(), group, request->net_log());
@@ skipping 400 matching lines @@
   //        its limit, may be left with other stalled groups that could be
   //        woken.  This isn't optimal, but there is no starvation, so to avoid
   //        the looping we leave it at this.
   OnAvailableSocketSlot(top_group_name, top_group);
 }
 
 // Search for the highest priority pending request, amongst the groups that
 // are not at the |max_sockets_per_group_| limit. Note: for requests with
 // the same priority, the winner is based on group hash ordering (and not
 // insertion order).
-bool ClientSocketPoolBaseHelper::FindTopStalledGroup(Group** group,
-                                                     std::string* group_name) {
+bool ClientSocketPoolBaseHelper::FindTopStalledGroup(
+    Group** group,
+    std::string* group_name) const {
+  CHECK((group && group_name) || (!group && !group_name));
   Group* top_group = NULL;
   const std::string* top_group_name = NULL;
   bool has_stalled_group = false;
-  for (GroupMap::iterator i = group_map_.begin();
+  for (GroupMap::const_iterator i = group_map_.begin();
        i != group_map_.end(); ++i) {
     Group* curr_group = i->second;
     const RequestQueue& queue = curr_group->pending_requests();
     if (queue.empty())
       continue;
     if (curr_group->IsStalled(max_sockets_per_group_)) {
+      if (!group)
+        return true;
       has_stalled_group = true;
       bool has_higher_priority = !top_group ||
           curr_group->TopPendingPriority() < top_group->TopPendingPriority();
       if (has_higher_priority) {
         top_group = curr_group;
         top_group_name = &i->first;
       }
     }
   }
 
   if (top_group) {
+    CHECK(group);
     *group = top_group;
     *group_name = *top_group_name;
+  } else {
+    CHECK(!has_stalled_group);
   }
   return has_stalled_group;
 }
 
 void ClientSocketPoolBaseHelper::OnConnectJobComplete(
     int result, ConnectJob* job) {
   DCHECK_NE(ERR_IO_PENDING, result);
   const std::string group_name = job->group_name();
   GroupMap::iterator group_it = group_map_.find(group_name);
   CHECK(group_it != group_map_.end());
@@ skipping 52 matching lines @@
   Flush();
 }
 
 void ClientSocketPoolBaseHelper::Flush() {
   pool_generation_number_++;
   CancelAllConnectJobs();
   CloseIdleSockets();
   AbortAllRequests();
 }
 
+bool ClientSocketPoolBaseHelper::IsStalled() const {
+  CHECK_LE(handed_out_socket_count_ + connecting_socket_count_, max_sockets_);

[mmenke, 2011/10/27 18:52:37]

Given that requests have an "ignore_limits" flag, I think this CHECK may result
in random crashing whenever it's set.

[willchan no longer on Chromium, 2011/11/08 22:26:13]

Good point!

+  if ((handed_out_socket_count_ + connecting_socket_count_) != max_sockets_)
+    return false;
+  for (GroupMap::const_iterator it = group_map_.begin();
+       it != group_map_.end(); it++) {
+    if (it->second->IsStalled(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 117 matching lines @@
   // Each connecting socket will eventually connect and be handed out.
   int total = handed_out_socket_count_ + connecting_socket_count_ +
       idle_socket_count();
   // There can be more sockets than the limit since some requests can ignore
   // the limit
   if (total < max_sockets_)
     return false;
   return true;
 }
 
-void ClientSocketPoolBaseHelper::CloseOneIdleSocket() {
-  CloseOneIdleSocketExceptInGroup(NULL);
+bool ClientSocketPoolBaseHelper::CloseOneIdleSocket() {
+  if (idle_socket_count() == 0)
+    return false;
+  return CloseOneIdleSocketExceptInGroup(NULL);
 }
 
 bool ClientSocketPoolBaseHelper::CloseOneIdleSocketExceptInGroup(
     const Group* exception_group) {
   CHECK_GT(idle_socket_count(), 0);
 
   for (GroupMap::iterator i = group_map_.begin(); i != group_map_.end(); ++i) {
     Group* group = i->second;
     if (exception_group == group)
       continue;
     std::list<IdleSocket>* idle_sockets = group->mutable_idle_sockets();
 
     if (!idle_sockets->empty()) {
       delete idle_sockets->front().socket;
       idle_sockets->pop_front();
       DecrementIdleCount();
       if (group->IsEmpty())
         RemoveGroup(i);
 
       return true;
     }
   }
 
-  if (!exception_group)
-    LOG(DFATAL) << "No idle socket found to close!.";
-
   return false;
 }
 
+bool ClientSocketPoolBaseHelper::CloseOneIdleConnectionInLayeredPool() {
+  // 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) {

[mmenke, 2011/10/27 18:52:37]

nit:  Looks to me like higher_layer_pools_.begin(); will just fit on the
previous line, making for more readable code.

[willchan no longer on Chromium, 2011/11/08 22:26:13]

Done.

+    if ((*it)->CloseOneIdleConnection())
+      return true;
+  }
+  return false;
+}
+
 void ClientSocketPoolBaseHelper::InvokeUserCallbackLater(
     ClientSocketHandle* handle, OldCompletionCallback* callback, int rv) {
   CHECK(!ContainsKey(pending_callback_map_, handle));
   pending_callback_map_[handle] = CallbackResultPair(callback, rv);
   MessageLoop::current()->PostTask(
       FROM_HERE,
       method_factory_.NewRunnableMethod(
           &ClientSocketPoolBaseHelper::InvokeUserCallback,
           handle));
 }
@@ skipping 84 matching lines @@
   // Delete active jobs.
   STLDeleteElements(&jobs_);
 
   // Cancel pending backup job.
   method_factory_.RevokeAll();
 }
 
 }  // namespace internal
 
 }  // namespace net