Add a bound on the total number of active sockets in ClientSocketPool....

net/base/client_socket_pool.cc

 // Copyright (c) 2006-2008 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/base/client_socket_pool.h"
 
 #include "base/message_loop.h"
 #include "net/base/client_socket.h"
 #include "net/base/client_socket_handle.h"
 #include "net/base/net_errors.h"
@@ skipping 10 matching lines @@
 // some conditions.  See http://crbug.com/4606.
 const int kCleanupInterval = 10;  // DO NOT INCREASE THIS TIMEOUT.
 
 // The maximum duration, in seconds, to keep idle persistent sockets alive.
 const int kIdleTimeout = 300;  // 5 minutes.
 
 }  // namespace
 
 namespace net {
 
-ClientSocketPool::ClientSocketPool(int max_sockets_per_group)
+ClientSocketPool::ClientSocketPool(int max_sockets_per_group, int max_sockets)
     : idle_socket_count_(0),
-      max_sockets_per_group_(max_sockets_per_group) {
+      active_socket_count_(0),
+      max_sockets_per_group_(max_sockets_per_group),
+      max_sockets_(max_sockets),
+      may_have_stalled_group_(false) {
+  DCHECK(max_sockets_per_group <= max_sockets);
 }
 
 ClientSocketPool::~ClientSocketPool() {
   // Clean up any idle sockets.  Assert that we have no remaining active
   // sockets or pending requests.  They should have all been cleaned up prior
   // to the manager being destroyed.
   CloseIdleSockets();
   DCHECK(group_map_.empty());
 }
 
 // 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 ClientSocketPool::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);
 }
 
 int ClientSocketPool::RequestSocket(ClientSocketHandle* handle,
                                     int priority,
                                     CompletionCallback* callback) {
   Group& group = group_map_[handle->group_name_];
 
   // Can we make another active socket now?
-  if (group.active_socket_count == max_sockets_per_group_) {
+  bool has_max_sockets = (active_socket_count_ == max_sockets_);
+  if (group.active_socket_count == max_sockets_per_group_ || has_max_sockets) {
+    may_have_stalled_group_ |= has_max_sockets;
     Request r;
     r.handle = handle;
     DCHECK(callback);
     r.callback = callback;
     r.priority = priority;
     InsertRequestIntoQueue(r, &group.pending_requests);
     return ERR_IO_PENDING;
   }
 
   // OK, we are going to activate one.
   group.active_socket_count++;
+  active_socket_count_++;
 
   // Use idle sockets in LIFO order because they're more likely to be
   // still reusable.
   while (!group.idle_sockets.empty()) {
     IdleSocket idle_socket = group.idle_sockets.back();
     group.idle_sockets.pop_back();
     DecrementIdleCount();
     if ((*idle_socket.ptr)->IsConnectedAndIdle()) {
       // We found one we can reuse!
       handle->socket_ = idle_socket.ptr;
       return OK;
     }
     delete idle_socket.ptr;
   }
 
   handle->socket_ = new ClientSocketPtr();
   return OK;
 }
 
 void ClientSocketPool::CancelRequest(ClientSocketHandle* handle) {
-  Group& group = group_map_[handle->group_name_];
-
-  // In order for us to be canceling a pending request, we must have active
-  // sockets equaling the limit.  NOTE: The correctness of the code doesn't
-  // require this assertion.
-  DCHECK(group.active_socket_count == max_sockets_per_group_);
+  GroupMap::iterator i = group_map_.find(handle->group_name_);
+  DCHECK(i != group_map_.end());
+  Group& group = i->second;
 
   // Search pending_requests for matching handle.
   std::deque<Request>::iterator it = group.pending_requests.begin();
   for (; it != group.pending_requests.end(); ++it) {
     if (it->handle == handle) {
       group.pending_requests.erase(it);
       break;
     }
   }
+
+  // Delete group if no longer needed.
+  if (group.empty())
+    group_map_.erase(i);
 }
 
 void ClientSocketPool::ReleaseSocket(ClientSocketHandle* handle) {
   // Run this asynchronously to allow the caller to finish before we let
   // another to begin doing work.  This also avoids nasty recursion issues.
   // NOTE: We cannot refer to the handle argument after this method returns.
   MessageLoop::current()->PostTask(FROM_HERE, NewRunnableMethod(
       this, &ClientSocketPool::DoReleaseSocket, handle->group_name_,
       handle->socket_));
 }
@@ skipping 25 matching lines @@
       if (force || j->ShouldCleanup(now)) {
         delete j->ptr;
         j = group.idle_sockets.erase(j);
         DecrementIdleCount();
       } else {
         ++j;
       }
     }
 
     // Delete group if no longer needed.
-    if (group.active_socket_count == 0 && group.idle_sockets.empty()) {
-      DCHECK(group.pending_requests.empty());
+    if (group.empty()) {
       group_map_.erase(i++);
     } else {
       ++i;
     }
   }
 }
 
 void ClientSocketPool::IncrementIdleCount() {
   if (++idle_socket_count_ == 1)
     timer_.Start(TimeDelta::FromSeconds(kCleanupInterval), this,
                  &ClientSocketPool::OnCleanupTimerFired);
 }
 
 void ClientSocketPool::DecrementIdleCount() {
   if (--idle_socket_count_ == 0)
     timer_.Stop();
 }
 
 void ClientSocketPool::DoReleaseSocket(const std::string& group_name,
                                        ClientSocketPtr* ptr) {
   GroupMap::iterator i = group_map_.find(group_name);
   DCHECK(i != group_map_.end());
 
   Group& group = i->second;
 
   DCHECK(group.active_socket_count > 0);
+  DCHECK(active_socket_count_ > 0);
+
   group.active_socket_count--;
+  active_socket_count_--;
 
   bool can_reuse = ptr->get() && (*ptr)->IsConnectedAndIdle();
   if (can_reuse) {
     IdleSocket idle_socket;
     idle_socket.ptr = ptr;
     idle_socket.start_time = base::TimeTicks::Now();
 
     group.idle_sockets.push_back(idle_socket);
     IncrementIdleCount();
   } else {
     delete ptr;
   }
 
   // Process one pending request.
-  if (!group.pending_requests.empty()) {
-    Request r = group.pending_requests.front();
-    group.pending_requests.pop_front();
-    int rv = RequestSocket(r.handle, r.priority, NULL);
-    DCHECK(rv == OK);
-    r.callback->Run(rv);
-    return;
+  if (may_have_stalled_group_) {
+    // The highest priority pending request may belong to another group, so
+    // check all of them. Worst-case this takes |max_sockets_| iterations.
+    if (Group* top_group = FindTopStalledGroup())
+      StartPendingRequestForGroup(top_group);
+
+    // Check if there are any stalled groups left.
+    if (!FindTopStalledGroup())
+      may_have_stalled_group_ = false;
+
+  } else if (!group.pending_requests.empty()) {
+    // Otherwise just dequeue the first pending request of |group|.
+    StartPendingRequestForGroup(&group);
   }
 
   // Delete group if no longer needed.
-  if (group.active_socket_count == 0 && group.idle_sockets.empty()) {
-    DCHECK(group.pending_requests.empty());
+  if (group.empty())
     group_map_.erase(i);
+}
+
+void ClientSocketPool::StartPendingRequestForGroup(Group* group) {
+  Request r = group->pending_requests.front();
+  group->pending_requests.pop_front();
+  int rv = RequestSocket(r.handle, r.priority, NULL);
+  DCHECK(rv == OK);
+  r.callback->Run(rv);
+}
+
+// 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).
+ClientSocketPool::Group* ClientSocketPool::FindTopStalledGroup() {
+  Group* top_group = NULL;
+  for (GroupMap::iterator i = group_map_.begin(); i != group_map_.end(); ++i) {
+    Group* group = &i->second;
+    const RequestQueue& queue = group->pending_requests;
+    if (group->active_socket_count < max_sockets_per_group_ &&
+        !queue.empty() && (top_group == NULL || queue.front().priority >
+            top_group->pending_requests.front().priority)) {
+      top_group = group;
+    }
   }
+  return top_group;
 }
 
 }  // namespace net