Limit total number of sockets in the system....

net/socket/client_socket_pool_base.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/socket/client_socket_pool_base.h"
 
 #include "base/compiler_specific.h"
 #include "base/message_loop.h"
 #include "base/stl_util-inl.h"
 #include "base/time.h"
@@ skipping 29 matching lines @@
       delegate_(delegate),
       load_state_(LOAD_STATE_IDLE) {
   DCHECK(!group_name.empty());
   DCHECK(key_handle);
   DCHECK(delegate);
 }
 
 ConnectJob::~ConnectJob() {}
 
 ClientSocketPoolBase::ClientSocketPoolBase(
+    int max_sockets,
     int max_sockets_per_group,
     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),
-      connect_job_factory_(connect_job_factory) {}
+      may_have_stalled_group_(false),
+      connect_job_factory_(connect_job_factory) {
+  DCHECK_LE(0, max_sockets_per_group);
+  DCHECK_LE(max_sockets_per_group, max_sockets);
+}
 
 ClientSocketPoolBase::~ClientSocketPoolBase() {
   if (g_late_binding)
     CancelAllConnectJobs();
   // 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());
   DCHECK(connect_job_map_.empty());
@@ skipping 17 matching lines @@
     const HostResolver::RequestInfo& resolve_info,
     int priority,
     ClientSocketHandle* handle,
     CompletionCallback* callback) {
   DCHECK(!resolve_info.hostname().empty());
   DCHECK_GE(priority, 0);
   DCHECK(callback);
   Group& group = group_map_[group_name];
 
   // Can we make another active socket now?
-  if (!group.HasAvailableSocketSlot(max_sockets_per_group_)) {
+  if (ReachedMaxSocketsLimit() ||
+      !group.HasAvailableSocketSlot(max_sockets_per_group_)) {
+    if (ReachedMaxSocketsLimit()) {
+      // 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.
+      may_have_stalled_group_ = true;
+    }
     CHECK(callback);
     Request r(handle, callback, priority, resolve_info);
     InsertRequestIntoQueue(r, &group.pending_requests);
     return ERR_IO_PENDING;
   }
 
   while (!group.idle_sockets.empty()) {
     IdleSocket idle_socket = group.idle_sockets.back();
     group.idle_sockets.pop_back();
     DecrementIdleCount();
@@ skipping 10 matching lines @@
   CHECK(callback);
   Request r(handle, callback, priority, resolve_info);
   scoped_ptr<ConnectJob> connect_job(
       connect_job_factory_->NewConnectJob(group_name, r, this));
 
   int rv = connect_job->Connect();
   if (rv == OK) {
     HandOutSocket(connect_job->ReleaseSocket(), false /* not reused */,
                   handle, &group);
   } else if (rv == ERR_IO_PENDING) {
+    connecting_socket_count_++;
+
     ConnectJob* job = connect_job.release();
     if (g_late_binding) {
       CHECK(!ContainsKey(connect_job_map_, handle));
       InsertRequestIntoQueue(r, &group.pending_requests);
     } else {
       group.connecting_requests[handle] = r;
       CHECK(!ContainsKey(connect_job_map_, handle));
       connect_job_map_[handle] = job;
     }
     group.jobs.insert(job);
@@ skipping 151 matching lines @@
     timer_.Stop();
 }
 
 void ClientSocketPoolBase::DoReleaseSocket(const std::string& group_name,
                                            ClientSocket* socket) {
   GroupMap::iterator i = group_map_.find(group_name);
   CHECK(i != group_map_.end());
 
   Group& group = i->second;
 
+  CHECK(handed_out_socket_count_ > 0);
+  handed_out_socket_count_--;
+
   CHECK(group.active_socket_count > 0);
   group.active_socket_count--;
 
   const bool can_reuse = socket->IsConnectedAndIdle();
   if (can_reuse) {
     AddIdleSocket(socket, true /* used socket */, &group);
   } else {
     delete socket;
   }
 
   OnAvailableSocketSlot(group_name, &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).
+int ClientSocketPoolBase::FindTopStalledGroup(Group** group,
+                                              std::string* group_name) {
+  Group* top_group = NULL;
+  const std::string* top_group_name = NULL;
+  int stalled_group_count = 0;
+  for (GroupMap::iterator i = group_map_.begin();
+       i != group_map_.end(); ++i) {
+    Group& group = i->second;
+    const RequestQueue& queue = group.pending_requests;
+    if (queue.empty())
+      continue;
+    bool has_slot = group.HasAvailableSocketSlot(max_sockets_per_group_);
+    if (has_slot)
+      stalled_group_count++;
+    bool has_higher_priority = !top_group ||
+        group.TopPendingPriority() > top_group->TopPendingPriority();
+    if (has_slot && has_higher_priority) {
+      top_group = &group;
+      top_group_name = &i->first;
+    }
+  }
+  if (top_group) {
+    *group = top_group;
+    *group_name = *top_group_name;
+  }
+  return stalled_group_count;
+}
+
 void ClientSocketPoolBase::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());
   Group& group = group_it->second;
 
   const ClientSocketHandle* const key_handle = job->key_handle();
   scoped_ptr<ClientSocket> socket(job->ReleaseSocket());
 
@@ skipping 45 matching lines @@
     callback->Run(result);
   }
 }
 
 void ClientSocketPoolBase::EnableLateBindingOfSockets(bool enabled) {
   g_late_binding = enabled;
 }
 
 void ClientSocketPoolBase::RemoveConnectJob(
     const ClientSocketHandle* handle, ConnectJob *job, Group* group) {
+  CHECK(connecting_socket_count_ > 0);
+  connecting_socket_count_--;
+
   if (g_late_binding) {
     DCHECK(job);
     delete job;
   } else {
     ConnectJobMap::iterator it = connect_job_map_.find(handle);
     CHECK(it != connect_job_map_.end());
     job = it->second;
     delete job;
     connect_job_map_.erase(it);
     group->connecting_requests.erase(handle);
@@ skipping 10 matching lines @@
   GroupMap::iterator it = group_map_.find(group_name);
   if (it != group_map_.end()) {
     Group& group = it->second;
     if (group.HasAvailableSocketSlot(max_sockets_per_group_))
       OnAvailableSocketSlot(group_name, &group);
   }
 }
 
 void ClientSocketPoolBase::OnAvailableSocketSlot(const std::string& group_name,
                                                  Group* group) {
-  if (!group->pending_requests.empty()) {
+  if (may_have_stalled_group_) {
+    std::string top_group_name;
+    Group* top_group;
+    int stalled_group_count = FindTopStalledGroup(&top_group, &top_group_name);
+    if (stalled_group_count <= 1)
+      may_have_stalled_group_ = false;
+    if (stalled_group_count >= 1)
+      ProcessPendingRequest(top_group_name, top_group);
+  } else if (!group->pending_requests.empty()) {
     ProcessPendingRequest(group_name, group);
     // |group| may no longer be valid after this point.  Be careful not to
     // access it again.
   } else if (group->IsEmpty()) {
     // Delete |group| if no longer needed.  |group| will no longer be valid.
     group_map_.erase(group_name);
   }
 }
 
 void ClientSocketPoolBase::ProcessPendingRequest(const std::string& group_name,
@@ skipping 15 matching lines @@
 }
 
 void ClientSocketPoolBase::HandOutSocket(
     ClientSocket* socket,
     bool reused,
     ClientSocketHandle* handle,
     Group* group) {
   DCHECK(socket);
   handle->set_socket(socket);
   handle->set_is_reused(reused);
+
+  handed_out_socket_count_++;
   group->active_socket_count++;
 }
 
 void ClientSocketPoolBase::AddIdleSocket(
     ClientSocket* socket, bool used, Group* group) {
   DCHECK(socket);
   IdleSocket idle_socket;
   idle_socket.socket = socket;
   idle_socket.start_time = base::TimeTicks::Now();
   idle_socket.used = used;
@@ skipping 10 matching lines @@
     // Delete group if no longer needed.
     if (group.IsEmpty()) {
       CHECK(group.pending_requests.empty());
       group_map_.erase(i++);
     } else {
       ++i;
     }
   }
 }
 
+bool ClientSocketPoolBase::ReachedMaxSocketsLimit() const {
+  // Each connecting socket will eventually connect and be handed out.
+  int total = handed_out_socket_count_ + connecting_socket_count_;
+  DCHECK_LE(total, max_sockets_);
+  return total == max_sockets_;
+}
+
 }  // namespace net