Refactor how ClientSocketPoolBaseHelper avoids re-entrancy.

net/socket/client_socket_pool_base.cc

 // Copyright (c) 2010 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/message_loop.h"
 #include "base/stats_counters.h"
@@ skipping 121 matching lines @@
     : 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),
       unused_idle_socket_timeout_(unused_idle_socket_timeout),
       used_idle_socket_timeout_(used_idle_socket_timeout),
       connect_job_factory_(connect_job_factory),
       backup_jobs_enabled_(false),
       ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)),
-      pool_generation_number_(0),
-      last_stalled_group_count_(0) {
+      pool_generation_number_(0) {
   DCHECK_LE(0, max_sockets_per_group);
   DCHECK_LE(max_sockets_per_group, max_sockets);
 
   NetworkChangeNotifier::AddObserver(this);
 }
 
 ClientSocketPoolBaseHelper::~ClientSocketPoolBaseHelper() {
   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();
   CHECK(group_map_.empty());
+  DCHECK(pending_callback_map_.empty());
   DCHECK_EQ(0, connecting_socket_count_);
 
   NetworkChangeNotifier::RemoveObserver(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
@@ skipping 16 matching lines @@
 
 int ClientSocketPoolBaseHelper::RequestSocket(
     const std::string& group_name,
     const Request* request) {
   request->net_log().BeginEvent(NetLog::TYPE_SOCKET_POOL, NULL);
   Group& group = group_map_[group_name];
 
   int rv = RequestSocketInternal(group_name, request);
   if (rv != ERR_IO_PENDING) {
     request->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL, NULL);
+    CHECK(!request->handle()->is_initialized());
     delete request;
   } else {
     InsertRequestIntoQueue(request, &group.pending_requests);
   }
   return rv;
 }
 
 int ClientSocketPoolBaseHelper::RequestSocketInternal(
     const std::string& group_name,
     const Request* request) {
@@ skipping 51 matching lines @@
 
     ConnectJob* job = connect_job.release();
     group.jobs.insert(job);
   } else {
     LogBoundConnectJobToRequest(connect_job->net_log().source(), request);
     connect_job->GetAdditionalErrorState(handle);
     ClientSocket* error_socket = connect_job->ReleaseSocket();
     if (error_socket) {
       HandOutSocket(error_socket, false /* not reused */, handle,
                     base::TimeDelta(), &group, request->net_log());
+    } else if (group.IsEmpty()) {
+      group_map_.erase(group_name);
     }
-    if (group.IsEmpty())
-      group_map_.erase(group_name);
   }
 
   return rv;
 }
 
-bool ClientSocketPoolBaseHelper::AssignIdleSocketToGroup(Group* group,
-    const Request* request) {
+bool ClientSocketPoolBaseHelper::AssignIdleSocketToGroup(
+    Group* group, const Request* request) {
   // Iterate through the list of idle sockets until we find one or exhaust
   // the list.
   while (!group->idle_sockets.empty()) {
     IdleSocket idle_socket = group->idle_sockets.back();
     group->idle_sockets.pop_back();
     DecrementIdleCount();
     if (idle_socket.socket->IsConnectedAndIdle()) {
       // We found one we can reuse!
       base::TimeDelta idle_time =
           base::TimeTicks::Now() - idle_socket.start_time;
@@ skipping 63 matching lines @@
   int rv = group.backup_job->Connect();
   connecting_socket_count_++;
   group.jobs.insert(group.backup_job);
   ConnectJob* job = group.backup_job;
   group.backup_job = NULL;
   if (rv != ERR_IO_PENDING)
     OnConnectJobComplete(rv, job);
 }
 
 void ClientSocketPoolBaseHelper::CancelRequest(
-    const std::string& group_name, const ClientSocketHandle* handle) {
-  // Running callbacks can cause the last outside reference to be released.
-  // Hold onto a reference.
-  scoped_refptr<ClientSocketPoolBaseHelper> ref_holder(this);
+    const std::string& group_name, ClientSocketHandle* handle) {
+  PendingCallbackMap::iterator callback_it = pending_callback_map_.find(handle);
+  if (callback_it != pending_callback_map_.end()) {
+    int result = callback_it->second.result;
+    pending_callback_map_.erase(callback_it);
+    ClientSocket* socket = handle->release_socket();
+    if (socket) {
+      if (result != OK)
+        socket->Disconnect();
+      ReleaseSocket(handle->group_name(), socket, handle->id());
+    }
+    return;
+  }
 
   CHECK(ContainsKey(group_map_, group_name));
 
   Group& group = group_map_[group_name];
 
   // Search pending_requests for matching handle.
   RequestQueue::iterator it = group.pending_requests.begin();
   for (; it != group.pending_requests.end(); ++it) {
     if ((*it)->handle() == handle) {
       const Request* req = RemoveRequestFromQueue(it, &group.pending_requests);
@@ skipping 19 matching lines @@
     const std::string& group_name) const {
   GroupMap::const_iterator i = group_map_.find(group_name);
   CHECK(i != group_map_.end());
 
   return i->second.idle_sockets.size();
 }
 
 LoadState ClientSocketPoolBaseHelper::GetLoadState(
     const std::string& group_name,
     const ClientSocketHandle* handle) const {
+  if (ContainsKey(pending_callback_map_, handle))
+    return LOAD_STATE_CONNECTING;
+
   if (!ContainsKey(group_map_, group_name)) {
     NOTREACHED() << "ClientSocketPool does not contain group: " << group_name
                  << " for handle: " << handle;
     return LOAD_STATE_IDLE;
   }
 
   // Can't use operator[] since it is non-const.
   const Group& group = group_map_.find(group_name)->second;
 
   // Search pending_requests for matching handle.
@@ skipping 68 matching lines @@
 }
 
 void ClientSocketPoolBaseHelper::DecrementIdleCount() {
   if (--idle_socket_count_ == 0)
     timer_.Stop();
 }
 
 void ClientSocketPoolBaseHelper::ReleaseSocket(const std::string& group_name,
                                                ClientSocket* socket,
                                                int id) {
-  // Running callbacks can cause the last outside reference to be released.
-  // Hold onto a reference.
-  scoped_refptr<ClientSocketPoolBaseHelper> ref_holder(this);
-
   GroupMap::iterator i = group_map_.find(group_name);
   CHECK(i != group_map_.end());
 
   Group& group = i->second;
 
   CHECK_GT(handed_out_socket_count_, 0);
   handed_out_socket_count_--;
 
   CHECK_GT(group.active_socket_count, 0);
   group.active_socket_count--;
 
   const bool can_reuse = socket->IsConnectedAndIdle() &&
       id == pool_generation_number_;
   if (can_reuse) {
     // Add it to the idle list.
     AddIdleSocket(socket, true /* used socket */, &group);
-    OnAvailableSocketSlot(group_name, MayHaveStalledGroups());
+    OnAvailableSocketSlot(group_name, &group);
   } else {
     delete socket;
   }
-  // Check to see if there are stalled groups that can resume now.
+
   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;
-  last_stalled_group_count_ = FindTopStalledGroup(&top_group, &top_group_name);
-  if (!last_stalled_group_count_)
+  if (!FindTopStalledGroup(&top_group, &top_group_name))
     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;
     }
   }
 
   // Note:  we don't loop on waking stalled groups.  If the stalled group is at
   //        its limit, may be left with other stalled groups that could be
-  //        waken.  This isn't optimal, but there is no starvation, so to avoid
+  //        woken.  This isn't optimal, but there is no starvation, so to avoid
   //        the looping we leave it at this.
-  OnAvailableSocketSlot(top_group_name, false);
-}
-
-bool ClientSocketPoolBaseHelper::MayHaveStalledGroups() {
-  return last_stalled_group_count_ > 0 || ReachedMaxSocketsLimit();
+  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).
-int ClientSocketPoolBaseHelper::FindTopStalledGroup(Group** group,
-                                                    std::string* group_name) {
+bool ClientSocketPoolBaseHelper::FindTopStalledGroup(Group** group,
+                                                     std::string* group_name) {
   Group* top_group = NULL;
   const std::string* top_group_name = NULL;
-  int stalled_group_count = 0;
+  bool has_stalled_group = false;
   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_unused_slot =
-        group.HasAvailableSocketSlot(max_sockets_per_group_) &&
-        group.pending_requests.size() > group.jobs.size();
-    if (has_unused_slot) {
-      stalled_group_count++;
+    if (group.IsStalled(max_sockets_per_group_)) {
+      has_stalled_group = true;
       bool has_higher_priority = !top_group ||
           group.TopPendingPriority() < top_group->TopPendingPriority();
       if (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;
+  return has_stalled_group;
 }
 
 void ClientSocketPoolBaseHelper::OnConnectJobComplete(
     int result, ConnectJob* job) {
-  // Running callbacks can cause the last outside reference to be released.
-  // Hold onto a reference.
-  scoped_refptr<ClientSocketPoolBaseHelper> ref_holder(this);
-
   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;
 
   scoped_ptr<ClientSocket> socket(job->ReleaseSocket());
 
   BoundNetLog job_log = job->net_log();
 
   if (result == OK) {
     DCHECK(socket.get());
     RemoveConnectJob(job, &group);
     if (!group.pending_requests.empty()) {
       scoped_ptr<const Request> r(RemoveRequestFromQueue(
           group.pending_requests.begin(), &group.pending_requests));
       LogBoundConnectJobToRequest(job_log.source(), r.get());
       HandOutSocket(
           socket.release(), false /* unused socket */, r->handle(),
           base::TimeDelta(), &group, r->net_log());
       r->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL, NULL);
-      r->callback()->Run(result);
+      InvokeUserCallbackLater(r->handle(), r->callback(), result);
     } else {
       AddIdleSocket(socket.release(), false /* unused socket */, &group);
-      OnAvailableSocketSlot(group_name, MayHaveStalledGroups());
+      OnAvailableSocketSlot(group_name, &group);
+      CheckForStalledSocketGroups();
     }
   } else {
     // If we got a socket, it must contain error information so pass that
     // up so that the caller can retrieve it.
     bool handed_out_socket = false;
     if (!group.pending_requests.empty()) {
       scoped_ptr<const Request> r(RemoveRequestFromQueue(
           group.pending_requests.begin(), &group.pending_requests));
       LogBoundConnectJobToRequest(job_log.source(), r.get());
       job->GetAdditionalErrorState(r->handle());
       RemoveConnectJob(job, &group);
       if (socket.get()) {
         handed_out_socket = true;
         HandOutSocket(socket.release(), false /* unused socket */, r->handle(),
                       base::TimeDelta(), &group, r->net_log());
       }
       r->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL,
                             new NetLogIntegerParameter("net_error", result));
-      if (socket.get()) {
-        handed_out_socket = true;
-        HandOutSocket(
-            socket.release(), false /* unused socket */, r->handle(),
-            base::TimeDelta(), &group, r->net_log());
-      }
-      r->callback()->Run(result);
+      InvokeUserCallbackLater(r->handle(), r->callback(), result);
     } else {
       RemoveConnectJob(job, &group);
     }
-    if (!handed_out_socket)
-      OnAvailableSocketSlot(group_name, MayHaveStalledGroups());
+    if (!handed_out_socket) {
+      OnAvailableSocketSlot(group_name, &group);
+      CheckForStalledSocketGroups();
+    }
   }
 }
 
 void ClientSocketPoolBaseHelper::OnIPAddressChanged() {
   Flush();
 }
 
 void ClientSocketPoolBaseHelper::Flush() {
   pool_generation_number_++;
   CancelAllConnectJobs();
@@ skipping 12 matching lines @@
   // If we've got no more jobs for this group, then we no longer need a
   // backup job either.
   if (group->jobs.empty())
     group->CleanupBackupJob();
 
   DCHECK(job);
   delete job;
 }
 
 void ClientSocketPoolBaseHelper::OnAvailableSocketSlot(
-    const std::string& group_name, bool was_at_socket_limit) {
-  // Go back to the message loop before processing the request wakeup
-  // so that we don't get recursive and lengthy stacks.
-  MessageLoop::current()->PostTask(FROM_HERE,
-      NewRunnableMethod(
-          this,
-          &ClientSocketPoolBaseHelper::ProcessPendingRequest,
-          group_name,
-          was_at_socket_limit));
+    const std::string& group_name, Group* group) {
+  if (!group->pending_requests.empty())
+    ProcessPendingRequest(group_name, group);
+
+  if (group->IsEmpty())
+    group_map_.erase(group_name);
 }
 
 void ClientSocketPoolBaseHelper::ProcessPendingRequest(
-    const std::string& group_name, bool was_at_socket_limit) {
-  GroupMap::iterator it = group_map_.find(group_name);
-  if (it != group_map_.end()) {
-    Group& group = it->second;
-    if (!group.pending_requests.empty()) {
-      int rv = RequestSocketInternal(group_name,
-                                     *group.pending_requests.begin());
-      if (rv != ERR_IO_PENDING) {
-        scoped_ptr<const Request> request(RemoveRequestFromQueue(
-              group.pending_requests.begin(), &group.pending_requests));
+    const std::string& group_name, Group* group) {
+  int rv = RequestSocketInternal(group_name,
+                                 *group->pending_requests.begin());
+  if (rv != ERR_IO_PENDING) {
+    scoped_ptr<const Request> request(RemoveRequestFromQueue(
+          group->pending_requests.begin(), &group->pending_requests));
 
-        scoped_refptr<NetLog::EventParameters> params;
-        if (rv != OK)
-          params = new NetLogIntegerParameter("net_error", rv);
-        request->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL, params);
-        request->callback()->Run(rv);
-      }
-
-      // |group| may no longer be valid after this point.  Be careful not to
-      // access it again.
-      if (group.IsEmpty()) {
-        // Delete |group| if no longer needed.  |group| will no longer be valid.
-        group_map_.erase(group_name);
-      }
-    }
+    scoped_refptr<NetLog::EventParameters> params;
+    if (rv != OK)
+      params = new NetLogIntegerParameter("net_error", rv);
+    request->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL, params);
+    InvokeUserCallbackLater(
+        request->handle(), request->callback(), rv);
   }
-
-  if (was_at_socket_limit)
-    CheckForStalledSocketGroups();
 }
 
 void ClientSocketPoolBaseHelper::HandOutSocket(
     ClientSocket* socket,
     bool reused,
     ClientSocketHandle* handle,
     base::TimeDelta idle_time,
     Group* group,
     const BoundNetLog& net_log) {
   DCHECK(socket);
@@ skipping 74 matching lines @@
       if (group.IsEmpty())
         group_map_.erase(i);
 
       return;
     }
   }
 
   LOG(DFATAL) << "No idle socket found to close!.";
 }
 
+void ClientSocketPoolBaseHelper::InvokeUserCallbackLater(
+    ClientSocketHandle* handle, CompletionCallback* callback, int rv) {
+  CHECK(!ContainsKey(pending_callback_map_, handle));
+  pending_callback_map_[handle] = CallbackResultPair(callback, rv);
+  MessageLoop::current()->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(
+          this,
+          &ClientSocketPoolBaseHelper::InvokeUserCallback,
+          handle));
+}
+
+void ClientSocketPoolBaseHelper::InvokeUserCallback(
+    ClientSocketHandle* handle) {
+  PendingCallbackMap::iterator it = pending_callback_map_.find(handle);
+
+  // Exit if the request has already been cancelled.
+  if (it == pending_callback_map_.end())
+    return;
+
+  CHECK(!handle->is_initialized());
+  CompletionCallback* callback = it->second.callback;
+  int result = it->second.result;
+  pending_callback_map_.erase(it);
+  callback->Run(result);
+}
+
 }  // namespace internal
 
 }  // namespace net