1// Copyright (c) 2011 The Chromium Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5#include "net/socket/client_socket_pool_base.h"
6
7#include "base/compiler_specific.h"
8#include "base/format_macros.h"
9#include "base/message_loop.h"
10#include "base/metrics/stats_counters.h"
11#include "base/stl_util-inl.h"
12#include "base/string_util.h"
13#include "base/time.h"
14#include "base/values.h"
15#include "net/base/net_log.h"
16#include "net/base/net_errors.h"
17#include "net/socket/client_socket_handle.h"
18
19using base::TimeDelta;
20
21namespace {
22
23// Indicate whether we should enable idle socket cleanup timer. When timer is
24// disabled, sockets are closed next time a socket request is made.
25#ifdef ANDROID
26bool g_cleanup_timer_enabled = false;
27#else
28bool g_cleanup_timer_enabled = true;
29#endif
30
31// The timeout value, in seconds, used to clean up idle sockets that can't be
32// reused.
33//
34// Note: It's important to close idle sockets that have received data as soon
35// as possible because the received data may cause BSOD on Windows XP under
36// some conditions.  See http://crbug.com/4606.
37const int kCleanupInterval = 10;  // DO NOT INCREASE THIS TIMEOUT.
38
39// Indicate whether or not we should establish a new transport layer connection
40// after a certain timeout has passed without receiving an ACK.
41bool g_connect_backup_jobs_enabled = true;
42
43}  // namespace
44
45namespace net {
46
47ConnectJob::ConnectJob(const std::string& group_name,
48                       base::TimeDelta timeout_duration,
49                       Delegate* delegate,
50                       const BoundNetLog& net_log)
51    : group_name_(group_name),
52      timeout_duration_(timeout_duration),
53      delegate_(delegate),
54      net_log_(net_log),
55      idle_(true),
56      preconnect_state_(NOT_PRECONNECT) {
57  DCHECK(!group_name.empty());
58  DCHECK(delegate);
59  net_log.BeginEvent(NetLog::TYPE_SOCKET_POOL_CONNECT_JOB, NULL);
60}
61
62ConnectJob::~ConnectJob() {
63  net_log().EndEvent(NetLog::TYPE_SOCKET_POOL_CONNECT_JOB, NULL);
64}
65
66void ConnectJob::Initialize(bool is_preconnect) {
67  if (is_preconnect)
68    preconnect_state_ = UNUSED_PRECONNECT;
69  else
70    preconnect_state_ = NOT_PRECONNECT;
71}
72
73int ConnectJob::Connect() {
74  if (timeout_duration_ != base::TimeDelta())
75    timer_.Start(timeout_duration_, this, &ConnectJob::OnTimeout);
76
77  idle_ = false;
78
79  LogConnectStart();
80
81  int rv = ConnectInternal();
82
83  if (rv != ERR_IO_PENDING) {
84    LogConnectCompletion(rv);
85    delegate_ = NULL;
86  }
87
88  return rv;
89}
90
91void ConnectJob::UseForNormalRequest() {
92  DCHECK_EQ(UNUSED_PRECONNECT, preconnect_state_);
93  preconnect_state_ = USED_PRECONNECT;
94}
95
96void ConnectJob::set_socket(ClientSocket* socket) {
97  if (socket) {
98    net_log().AddEvent(NetLog::TYPE_CONNECT_JOB_SET_SOCKET, make_scoped_refptr(
99          new NetLogSourceParameter("source_dependency",
100                                    socket->NetLog().source())));
101  }
102  socket_.reset(socket);
103}
104
105void ConnectJob::NotifyDelegateOfCompletion(int rv) {
106  // The delegate will delete |this|.
107  Delegate *delegate = delegate_;
108  delegate_ = NULL;
109
110  LogConnectCompletion(rv);
111  delegate->OnConnectJobComplete(rv, this);
112}
113
114void ConnectJob::ResetTimer(base::TimeDelta remaining_time) {
115  timer_.Stop();
116  timer_.Start(remaining_time, this, &ConnectJob::OnTimeout);
117}
118
119void ConnectJob::LogConnectStart() {
120  net_log().BeginEvent(NetLog::TYPE_SOCKET_POOL_CONNECT_JOB_CONNECT,
121      make_scoped_refptr(new NetLogStringParameter("group_name", group_name_)));
122}
123
124void ConnectJob::LogConnectCompletion(int net_error) {
125  net_log().EndEventWithNetErrorCode(
126      NetLog::TYPE_SOCKET_POOL_CONNECT_JOB_CONNECT, net_error);
127}
128
129void ConnectJob::OnTimeout() {
130  // Make sure the socket is NULL before calling into |delegate|.
131  set_socket(NULL);
132
133  net_log_.AddEvent(NetLog::TYPE_SOCKET_POOL_CONNECT_JOB_TIMED_OUT, NULL);
134
135  NotifyDelegateOfCompletion(ERR_TIMED_OUT);
136}
137
138namespace internal {
139
140ClientSocketPoolBaseHelper::Request::Request(
141    ClientSocketHandle* handle,
142    CompletionCallback* callback,
143    RequestPriority priority,
144    bool ignore_limits,
145    Flags flags,
146    const BoundNetLog& net_log
147#ifdef ANDROID
148    , bool valid_uid, uid_t calling_uid
149#endif
150    )
151    : handle_(handle),
152      callback_(callback),
153      priority_(priority),
154      ignore_limits_(ignore_limits),
155      flags_(flags),
156      net_log_(net_log)
157#ifdef ANDROID
158      , valid_uid_(valid_uid), calling_uid_(calling_uid)
159#endif
160      {}
161
162ClientSocketPoolBaseHelper::Request::~Request() {}
163
164ClientSocketPoolBaseHelper::ClientSocketPoolBaseHelper(
165    int max_sockets,
166    int max_sockets_per_group,
167    base::TimeDelta unused_idle_socket_timeout,
168    base::TimeDelta used_idle_socket_timeout,
169    ConnectJobFactory* connect_job_factory)
170    : idle_socket_count_(0),
171      connecting_socket_count_(0),
172      handed_out_socket_count_(0),
173      max_sockets_(max_sockets),
174      max_sockets_per_group_(max_sockets_per_group),
175      use_cleanup_timer_(g_cleanup_timer_enabled),
176      unused_idle_socket_timeout_(unused_idle_socket_timeout),
177      used_idle_socket_timeout_(used_idle_socket_timeout),
178      connect_job_factory_(connect_job_factory),
179      connect_backup_jobs_enabled_(false),
180      pool_generation_number_(0),
181      method_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
182  DCHECK_LE(0, max_sockets_per_group);
183  DCHECK_LE(max_sockets_per_group, max_sockets);
184
185  NetworkChangeNotifier::AddIPAddressObserver(this);
186}
187
188ClientSocketPoolBaseHelper::~ClientSocketPoolBaseHelper() {
189  // Clean up any idle sockets and pending connect jobs.  Assert that we have no
190  // remaining active sockets or pending requests.  They should have all been
191  // cleaned up prior to |this| being destroyed.
192  Flush();
193  DCHECK(group_map_.empty());
194  DCHECK(pending_callback_map_.empty());
195  DCHECK_EQ(0, connecting_socket_count_);
196
197  NetworkChangeNotifier::RemoveIPAddressObserver(this);
198}
199
200// InsertRequestIntoQueue inserts the request into the queue based on
201// priority.  Highest priorities are closest to the front.  Older requests are
202// prioritized over requests of equal priority.
203//
204// static
205void ClientSocketPoolBaseHelper::InsertRequestIntoQueue(
206    const Request* r, RequestQueue* pending_requests) {
207  RequestQueue::iterator it = pending_requests->begin();
208  while (it != pending_requests->end() && r->priority() >= (*it)->priority())
209    ++it;
210  pending_requests->insert(it, r);
211}
212
213// static
214const ClientSocketPoolBaseHelper::Request*
215ClientSocketPoolBaseHelper::RemoveRequestFromQueue(
216    const RequestQueue::iterator& it, Group* group) {
217  const Request* req = *it;
218  group->mutable_pending_requests()->erase(it);
219  // If there are no more requests, we kill the backup timer.
220  if (group->pending_requests().empty())
221    group->CleanupBackupJob();
222  return req;
223}
224
225int ClientSocketPoolBaseHelper::RequestSocket(
226    const std::string& group_name,
227    const Request* request) {
228  CHECK(request->callback());
229  CHECK(request->handle());
230
231  // Cleanup any timed-out idle sockets if no timer is used.
232  if (!use_cleanup_timer_)
233    CleanupIdleSockets(false);
234
235  request->net_log().BeginEvent(NetLog::TYPE_SOCKET_POOL, NULL);
236  Group* group = GetOrCreateGroup(group_name);
237
238  int rv = RequestSocketInternal(group_name, request);
239  if (rv != ERR_IO_PENDING) {
240    request->net_log().EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL, rv);
241    CHECK(!request->handle()->is_initialized());
242    delete request;
243  } else {
244    InsertRequestIntoQueue(request, group->mutable_pending_requests());
245  }
246  return rv;
247}
248
249void ClientSocketPoolBaseHelper::RequestSockets(
250    const std::string& group_name,
251    const Request& request,
252    int num_sockets) {
253  DCHECK(!request.callback());
254  DCHECK(!request.handle());
255
256  // Cleanup any timed out idle sockets if no timer is used.
257  if (!use_cleanup_timer_)
258    CleanupIdleSockets(false);
259
260  if (num_sockets > max_sockets_per_group_) {
261    num_sockets = max_sockets_per_group_;
262  }
263
264  request.net_log().BeginEvent(
265      NetLog::TYPE_SOCKET_POOL_CONNECTING_N_SOCKETS,
266      make_scoped_refptr(new NetLogIntegerParameter(
267          "num_sockets", num_sockets)));
268
269  Group* group = GetOrCreateGroup(group_name);
270
271  // RequestSocketsInternal() may delete the group.
272  bool deleted_group = false;
273
274  int rv = OK;
275  for (int num_iterations_left = num_sockets;
276       group->NumActiveSocketSlots() < num_sockets &&
277       num_iterations_left > 0 ; num_iterations_left--) {
278    rv = RequestSocketInternal(group_name, &request);
279    if (rv < 0 && rv != ERR_IO_PENDING) {
280      // We're encountering a synchronous error.  Give up.
281      if (!ContainsKey(group_map_, group_name))
282        deleted_group = true;
283      break;
284    }
285    if (!ContainsKey(group_map_, group_name)) {
286      // Unexpected.  The group should only be getting deleted on synchronous
287      // error.
288      NOTREACHED();
289      deleted_group = true;
290      break;
291    }
292  }
293
294  if (!deleted_group && group->IsEmpty())
295    RemoveGroup(group_name);
296
297  if (rv == ERR_IO_PENDING)
298    rv = OK;
299  request.net_log().EndEventWithNetErrorCode(
300      NetLog::TYPE_SOCKET_POOL_CONNECTING_N_SOCKETS, rv);
301}
302
303int ClientSocketPoolBaseHelper::RequestSocketInternal(
304    const std::string& group_name,
305    const Request* request) {
306  DCHECK_GE(request->priority(), 0);
307  ClientSocketHandle* const handle = request->handle();
308  const bool preconnecting = !handle;
309  Group* group = GetOrCreateGroup(group_name);
310
311  if (!(request->flags() & NO_IDLE_SOCKETS)) {
312    // Try to reuse a socket.
313    if (AssignIdleSocketToGroup(request, group))
314      return OK;
315  }
316
317  if (!preconnecting && group->TryToUsePreconnectConnectJob())
318    return ERR_IO_PENDING;
319
320  // Can we make another active socket now?
321  if (!group->HasAvailableSocketSlot(max_sockets_per_group_) &&
322      !request->ignore_limits()) {
323    request->net_log().AddEvent(
324        NetLog::TYPE_SOCKET_POOL_STALLED_MAX_SOCKETS_PER_GROUP, NULL);
325    return ERR_IO_PENDING;
326  }
327
328  if (ReachedMaxSocketsLimit() && !request->ignore_limits()) {
329    if (idle_socket_count() > 0) {
330      bool closed = CloseOneIdleSocketExceptInGroup(group);
331      if (preconnecting && !closed)
332        return ERR_PRECONNECT_MAX_SOCKET_LIMIT;
333    } else {
334      // We could check if we really have a stalled group here, but it requires
335      // a scan of all groups, so just flip a flag here, and do the check later.
336      request->net_log().AddEvent(
337          NetLog::TYPE_SOCKET_POOL_STALLED_MAX_SOCKETS, NULL);
338      return ERR_IO_PENDING;
339    }
340  }
341
342  // We couldn't find a socket to reuse, so allocate and connect a new one.
343  scoped_ptr<ConnectJob> connect_job(
344      connect_job_factory_->NewConnectJob(group_name, *request, this));
345
346  connect_job->Initialize(preconnecting);
347  int rv = connect_job->Connect();
348  if (rv == OK) {
349    LogBoundConnectJobToRequest(connect_job->net_log().source(), request);
350    if (!preconnecting) {
351      HandOutSocket(connect_job->ReleaseSocket(), false /* not reused */,
352                    handle, base::TimeDelta(), group, request->net_log());
353    } else {
354      AddIdleSocket(connect_job->ReleaseSocket(), group);
355    }
356  } else if (rv == ERR_IO_PENDING) {
357    // If we don't have any sockets in this group, set a timer for potentially
358    // creating a new one.  If the SYN is lost, this backup socket may complete
359    // before the slow socket, improving end user latency.
360    if (connect_backup_jobs_enabled_ &&
361        group->IsEmpty() && !group->HasBackupJob() &&
362        handle) {
363      group->StartBackupSocketTimer(group_name, this);
364    }
365
366    connecting_socket_count_++;
367
368    group->AddJob(connect_job.release());
369  } else {
370    LogBoundConnectJobToRequest(connect_job->net_log().source(), request);
371    ClientSocket* error_socket = NULL;
372    if (!preconnecting) {
373      DCHECK(handle);
374      connect_job->GetAdditionalErrorState(handle);
375      error_socket = connect_job->ReleaseSocket();
376    }
377    if (error_socket) {
378      HandOutSocket(error_socket, false /* not reused */, handle,
379                    base::TimeDelta(), group, request->net_log());
380    } else if (group->IsEmpty()) {
381      RemoveGroup(group_name);
382    }
383  }
384
385  return rv;
386}
387
388bool ClientSocketPoolBaseHelper::AssignIdleSocketToGroup(
389    const Request* request, Group* group) {
390  std::list<IdleSocket>* idle_sockets = group->mutable_idle_sockets();
391  std::list<IdleSocket>::iterator idle_socket_it = idle_sockets->end();
392
393  // Iterate through the idle sockets forwards (oldest to newest)
394  //   * Delete any disconnected ones.
395  //   * If we find a used idle socket, assign to |idle_socket|.  At the end,
396  //   the |idle_socket_it| will be set to the newest used idle socket.
397  for (std::list<IdleSocket>::iterator it = idle_sockets->begin();
398       it != idle_sockets->end();) {
399    if (!it->socket->IsConnectedAndIdle()) {
400      DecrementIdleCount();
401      delete it->socket;
402      it = idle_sockets->erase(it);
403      continue;
404    }
405
406    if (it->socket->WasEverUsed()) {
407      // We found one we can reuse!
408      idle_socket_it = it;
409    }
410
411    ++it;
412  }
413
414  // If we haven't found an idle socket, that means there are no used idle
415  // sockets.  Pick the oldest (first) idle socket (FIFO).
416
417  if (idle_socket_it == idle_sockets->end() && !idle_sockets->empty())
418    idle_socket_it = idle_sockets->begin();
419
420  if (idle_socket_it != idle_sockets->end()) {
421    DecrementIdleCount();
422    base::TimeDelta idle_time =
423        base::TimeTicks::Now() - idle_socket_it->start_time;
424    IdleSocket idle_socket = *idle_socket_it;
425    idle_sockets->erase(idle_socket_it);
426    HandOutSocket(
427        idle_socket.socket,
428        idle_socket.socket->WasEverUsed(),
429        request->handle(),
430        idle_time,
431        group,
432        request->net_log());
433    return true;
434  }
435
436  return false;
437}
438
439// static
440void ClientSocketPoolBaseHelper::LogBoundConnectJobToRequest(
441    const NetLog::Source& connect_job_source, const Request* request) {
442  request->net_log().AddEvent(
443      NetLog::TYPE_SOCKET_POOL_BOUND_TO_CONNECT_JOB,
444      make_scoped_refptr(new NetLogSourceParameter(
445          "source_dependency", connect_job_source)));
446}
447
448void ClientSocketPoolBaseHelper::CancelRequest(
449    const std::string& group_name, ClientSocketHandle* handle) {
450  PendingCallbackMap::iterator callback_it = pending_callback_map_.find(handle);
451  if (callback_it != pending_callback_map_.end()) {
452    int result = callback_it->second.result;
453    pending_callback_map_.erase(callback_it);
454    ClientSocket* socket = handle->release_socket();
455    if (socket) {
456      if (result != OK)
457        socket->Disconnect();
458      ReleaseSocket(handle->group_name(), socket, handle->id());
459    }
460    return;
461  }
462
463  CHECK(ContainsKey(group_map_, group_name));
464
465  Group* group = GetOrCreateGroup(group_name);
466
467  // Search pending_requests for matching handle.
468  RequestQueue::iterator it = group->mutable_pending_requests()->begin();
469  for (; it != group->pending_requests().end(); ++it) {
470    if ((*it)->handle() == handle) {
471      scoped_ptr<const Request> req(RemoveRequestFromQueue(it, group));
472      req->net_log().AddEvent(NetLog::TYPE_CANCELLED, NULL);
473      req->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL, NULL);
474
475      // We let the job run, unless we're at the socket limit.
476      if (group->jobs().size() && ReachedMaxSocketsLimit()) {
477        RemoveConnectJob(*group->jobs().begin(), group);
478        CheckForStalledSocketGroups();
479      }
480      break;
481    }
482  }
483}
484
485bool ClientSocketPoolBaseHelper::HasGroup(const std::string& group_name) const {
486  return ContainsKey(group_map_, group_name);
487}
488
489void ClientSocketPoolBaseHelper::CloseIdleSockets() {
490  CleanupIdleSockets(true);
491  DCHECK_EQ(0, idle_socket_count_);
492}
493
494int ClientSocketPoolBaseHelper::IdleSocketCountInGroup(
495    const std::string& group_name) const {
496  GroupMap::const_iterator i = group_map_.find(group_name);
497  CHECK(i != group_map_.end());
498
499  return i->second->idle_sockets().size();
500}
501
502LoadState ClientSocketPoolBaseHelper::GetLoadState(
503    const std::string& group_name,
504    const ClientSocketHandle* handle) const {
505  if (ContainsKey(pending_callback_map_, handle))
506    return LOAD_STATE_CONNECTING;
507
508  if (!ContainsKey(group_map_, group_name)) {
509    NOTREACHED() << "ClientSocketPool does not contain group: " << group_name
510                 << " for handle: " << handle;
511    return LOAD_STATE_IDLE;
512  }
513
514  // Can't use operator[] since it is non-const.
515  const Group& group = *group_map_.find(group_name)->second;
516
517  // Search pending_requests for matching handle.
518  RequestQueue::const_iterator it = group.pending_requests().begin();
519  for (size_t i = 0; it != group.pending_requests().end(); ++it, ++i) {
520    if ((*it)->handle() == handle) {
521      if (i < group.jobs().size()) {
522        LoadState max_state = LOAD_STATE_IDLE;
523        for (ConnectJobSet::const_iterator job_it = group.jobs().begin();
524             job_it != group.jobs().end(); ++job_it) {
525          max_state = std::max(max_state, (*job_it)->GetLoadState());
526        }
527        return max_state;
528      } else {
529        // TODO(wtc): Add a state for being on the wait list.
530        // See http://www.crbug.com/5077.
531        return LOAD_STATE_IDLE;
532      }
533    }
534  }
535
536  NOTREACHED();
537  return LOAD_STATE_IDLE;
538}
539
540DictionaryValue* ClientSocketPoolBaseHelper::GetInfoAsValue(
541    const std::string& name, const std::string& type) const {
542  DictionaryValue* dict = new DictionaryValue();
543  dict->SetString("name", name);
544  dict->SetString("type", type);
545  dict->SetInteger("handed_out_socket_count", handed_out_socket_count_);
546  dict->SetInteger("connecting_socket_count", connecting_socket_count_);
547  dict->SetInteger("idle_socket_count", idle_socket_count_);
548  dict->SetInteger("max_socket_count", max_sockets_);
549  dict->SetInteger("max_sockets_per_group", max_sockets_per_group_);
550  dict->SetInteger("pool_generation_number", pool_generation_number_);
551
552  if (group_map_.empty())
553    return dict;
554
555  DictionaryValue* all_groups_dict = new DictionaryValue();
556  for (GroupMap::const_iterator it = group_map_.begin();
557       it != group_map_.end(); it++) {
558    const Group* group = it->second;
559    DictionaryValue* group_dict = new DictionaryValue();
560
561    group_dict->SetInteger("pending_request_count",
562                           group->pending_requests().size());
563    if (!group->pending_requests().empty()) {
564      group_dict->SetInteger("top_pending_priority",
565                             group->TopPendingPriority());
566    }
567
568    group_dict->SetInteger("active_socket_count", group->active_socket_count());
569
570    ListValue* idle_socket_list = new ListValue();
571    std::list<IdleSocket>::const_iterator idle_socket;
572    for (idle_socket = group->idle_sockets().begin();
573         idle_socket != group->idle_sockets().end();
574         idle_socket++) {
575      int source_id = idle_socket->socket->NetLog().source().id;
576      idle_socket_list->Append(Value::CreateIntegerValue(source_id));
577    }
578    group_dict->Set("idle_sockets", idle_socket_list);
579
580    ListValue* connect_jobs_list = new ListValue();
581    std::set<ConnectJob*>::const_iterator job = group->jobs().begin();
582    for (job = group->jobs().begin(); job != group->jobs().end(); job++) {
583      int source_id = (*job)->net_log().source().id;
584      connect_jobs_list->Append(Value::CreateIntegerValue(source_id));
585    }
586    group_dict->Set("connect_jobs", connect_jobs_list);
587
588    group_dict->SetBoolean("is_stalled",
589                           group->IsStalled(max_sockets_per_group_));
590    group_dict->SetBoolean("has_backup_job", group->HasBackupJob());
591
592    all_groups_dict->SetWithoutPathExpansion(it->first, group_dict);
593  }
594  dict->Set("groups", all_groups_dict);
595  return dict;
596}
597
598bool ClientSocketPoolBaseHelper::IdleSocket::ShouldCleanup(
599    base::TimeTicks now,
600    base::TimeDelta timeout) const {
601  bool timed_out = (now - start_time) >= timeout;
602  if (timed_out)
603    return true;
604  if (socket->WasEverUsed())
605    return !socket->IsConnectedAndIdle();
606  return !socket->IsConnected();
607}
608
609void ClientSocketPoolBaseHelper::CleanupIdleSockets(bool force) {
610  if (idle_socket_count_ == 0)
611    return;
612
613  // Current time value. Retrieving it once at the function start rather than
614  // inside the inner loop, since it shouldn't change by any meaningful amount.
615  base::TimeTicks now = base::TimeTicks::Now();
616
617  GroupMap::iterator i = group_map_.begin();
618  while (i != group_map_.end()) {
619    Group* group = i->second;
620
621    std::list<IdleSocket>::iterator j = group->mutable_idle_sockets()->begin();
622    while (j != group->idle_sockets().end()) {
623      base::TimeDelta timeout =
624          j->socket->WasEverUsed() ?
625          used_idle_socket_timeout_ : unused_idle_socket_timeout_;
626      if (force || j->ShouldCleanup(now, timeout)) {
627        delete j->socket;
628        j = group->mutable_idle_sockets()->erase(j);
629        DecrementIdleCount();
630      } else {
631        ++j;
632      }
633    }
634
635    // Delete group if no longer needed.
636    if (group->IsEmpty()) {
637      RemoveGroup(i++);
638    } else {
639      ++i;
640    }
641  }
642}
643
644ClientSocketPoolBaseHelper::Group* ClientSocketPoolBaseHelper::GetOrCreateGroup(
645    const std::string& group_name) {
646  GroupMap::iterator it = group_map_.find(group_name);
647  if (it != group_map_.end())
648    return it->second;
649  Group* group = new Group;
650  group_map_[group_name] = group;
651  return group;
652}
653
654void ClientSocketPoolBaseHelper::RemoveGroup(const std::string& group_name) {
655  GroupMap::iterator it = group_map_.find(group_name);
656  CHECK(it != group_map_.end());
657
658  RemoveGroup(it);
659}
660
661void ClientSocketPoolBaseHelper::RemoveGroup(GroupMap::iterator it) {
662  delete it->second;
663  group_map_.erase(it);
664}
665
666// static
667bool ClientSocketPoolBaseHelper::connect_backup_jobs_enabled() {
668  return g_connect_backup_jobs_enabled;
669}
670
671// static
672bool ClientSocketPoolBaseHelper::set_connect_backup_jobs_enabled(bool enabled) {
673  bool old_value = g_connect_backup_jobs_enabled;
674  g_connect_backup_jobs_enabled = enabled;
675  return old_value;
676}
677
678void ClientSocketPoolBaseHelper::EnableConnectBackupJobs() {
679  connect_backup_jobs_enabled_ = g_connect_backup_jobs_enabled;
680}
681
682void ClientSocketPoolBaseHelper::IncrementIdleCount() {
683  if (++idle_socket_count_ == 1)
684    StartIdleSocketTimer();
685}
686
687void ClientSocketPoolBaseHelper::DecrementIdleCount() {
688  if (--idle_socket_count_ == 0)
689    timer_.Stop();
690}
691
692// static
693bool ClientSocketPoolBaseHelper::cleanup_timer_enabled() {
694  return g_cleanup_timer_enabled;
695}
696
697// static
698bool ClientSocketPoolBaseHelper::set_cleanup_timer_enabled(bool enabled) {
699  bool old_value = g_cleanup_timer_enabled;
700  g_cleanup_timer_enabled = enabled;
701  return old_value;
702}
703
704void ClientSocketPoolBaseHelper::StartIdleSocketTimer() {
705  timer_.Start(TimeDelta::FromSeconds(kCleanupInterval), this,
706               &ClientSocketPoolBaseHelper::OnCleanupTimerFired);
707}
708
709void ClientSocketPoolBaseHelper::ReleaseSocket(const std::string& group_name,
710                                               ClientSocket* socket,
711                                               int id) {
712  GroupMap::iterator i = group_map_.find(group_name);
713  CHECK(i != group_map_.end());
714
715  Group* group = i->second;
716
717  CHECK_GT(handed_out_socket_count_, 0);
718  handed_out_socket_count_--;
719
720  CHECK_GT(group->active_socket_count(), 0);
721  group->DecrementActiveSocketCount();
722
723  const bool can_reuse = socket->IsConnectedAndIdle() &&
724      id == pool_generation_number_;
725  if (can_reuse) {
726    // Add it to the idle list.
727    AddIdleSocket(socket, group);
728    OnAvailableSocketSlot(group_name, group);
729  } else {
730    delete socket;
731  }
732
733  CheckForStalledSocketGroups();
734}
735
736void ClientSocketPoolBaseHelper::CheckForStalledSocketGroups() {
737  // If we have idle sockets, see if we can give one to the top-stalled group.
738  std::string top_group_name;
739  Group* top_group = NULL;
740  if (!FindTopStalledGroup(&top_group, &top_group_name))
741    return;
742
743  if (ReachedMaxSocketsLimit()) {
744    if (idle_socket_count() > 0) {
745      CloseOneIdleSocket();
746    } else {
747      // We can't activate more sockets since we're already at our global
748      // limit.
749      return;
750    }
751  }
752
753  // Note:  we don't loop on waking stalled groups.  If the stalled group is at
754  //        its limit, may be left with other stalled groups that could be
755  //        woken.  This isn't optimal, but there is no starvation, so to avoid
756  //        the looping we leave it at this.
757  OnAvailableSocketSlot(top_group_name, top_group);
758}
759
760// Search for the highest priority pending request, amongst the groups that
761// are not at the |max_sockets_per_group_| limit. Note: for requests with
762// the same priority, the winner is based on group hash ordering (and not
763// insertion order).
764bool ClientSocketPoolBaseHelper::FindTopStalledGroup(Group** group,
765                                                     std::string* group_name) {
766  Group* top_group = NULL;
767  const std::string* top_group_name = NULL;
768  bool has_stalled_group = false;
769  for (GroupMap::iterator i = group_map_.begin();
770       i != group_map_.end(); ++i) {
771    Group* curr_group = i->second;
772    const RequestQueue& queue = curr_group->pending_requests();
773    if (queue.empty())
774      continue;
775    if (curr_group->IsStalled(max_sockets_per_group_)) {
776      has_stalled_group = true;
777      bool has_higher_priority = !top_group ||
778          curr_group->TopPendingPriority() < top_group->TopPendingPriority();
779      if (has_higher_priority) {
780        top_group = curr_group;
781        top_group_name = &i->first;
782      }
783    }
784  }
785
786  if (top_group) {
787    *group = top_group;
788    *group_name = *top_group_name;
789  }
790  return has_stalled_group;
791}
792
793void ClientSocketPoolBaseHelper::OnConnectJobComplete(
794    int result, ConnectJob* job) {
795  DCHECK_NE(ERR_IO_PENDING, result);
796  const std::string group_name = job->group_name();
797  GroupMap::iterator group_it = group_map_.find(group_name);
798  CHECK(group_it != group_map_.end());
799  Group* group = group_it->second;
800
801  scoped_ptr<ClientSocket> socket(job->ReleaseSocket());
802
803  BoundNetLog job_log = job->net_log();
804
805  if (result == OK) {
806    DCHECK(socket.get());
807    RemoveConnectJob(job, group);
808    if (!group->pending_requests().empty()) {
809      scoped_ptr<const Request> r(RemoveRequestFromQueue(
810          group->mutable_pending_requests()->begin(), group));
811      LogBoundConnectJobToRequest(job_log.source(), r.get());
812      HandOutSocket(
813          socket.release(), false /* unused socket */, r->handle(),
814          base::TimeDelta(), group, r->net_log());
815      r->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL, NULL);
816      InvokeUserCallbackLater(r->handle(), r->callback(), result);
817    } else {
818      AddIdleSocket(socket.release(), group);
819      OnAvailableSocketSlot(group_name, group);
820      CheckForStalledSocketGroups();
821    }
822  } else {
823    // If we got a socket, it must contain error information so pass that
824    // up so that the caller can retrieve it.
825    bool handed_out_socket = false;
826    if (!group->pending_requests().empty()) {
827      scoped_ptr<const Request> r(RemoveRequestFromQueue(
828          group->mutable_pending_requests()->begin(), group));
829      LogBoundConnectJobToRequest(job_log.source(), r.get());
830      job->GetAdditionalErrorState(r->handle());
831      RemoveConnectJob(job, group);
832      if (socket.get()) {
833        handed_out_socket = true;
834        HandOutSocket(socket.release(), false /* unused socket */, r->handle(),
835                      base::TimeDelta(), group, r->net_log());
836      }
837      r->net_log().EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL,
838                                            result);
839      InvokeUserCallbackLater(r->handle(), r->callback(), result);
840    } else {
841      RemoveConnectJob(job, group);
842    }
843    if (!handed_out_socket) {
844      OnAvailableSocketSlot(group_name, group);
845      CheckForStalledSocketGroups();
846    }
847  }
848}
849
850void ClientSocketPoolBaseHelper::OnIPAddressChanged() {
851  Flush();
852}
853
854void ClientSocketPoolBaseHelper::Flush() {
855  pool_generation_number_++;
856  CancelAllConnectJobs();
857  CloseIdleSockets();
858  AbortAllRequests();
859}
860
861void ClientSocketPoolBaseHelper::RemoveConnectJob(ConnectJob* job,
862                                                  Group* group) {
863  CHECK_GT(connecting_socket_count_, 0);
864  connecting_socket_count_--;
865
866  DCHECK(group);
867  DCHECK(ContainsKey(group->jobs(), job));
868  group->RemoveJob(job);
869
870  // If we've got no more jobs for this group, then we no longer need a
871  // backup job either.
872  if (group->jobs().empty())
873    group->CleanupBackupJob();
874
875  DCHECK(job);
876  delete job;
877}
878
879void ClientSocketPoolBaseHelper::OnAvailableSocketSlot(
880    const std::string& group_name, Group* group) {
881  DCHECK(ContainsKey(group_map_, group_name));
882  if (group->IsEmpty())
883    RemoveGroup(group_name);
884  else if (!group->pending_requests().empty())
885    ProcessPendingRequest(group_name, group);
886}
887
888void ClientSocketPoolBaseHelper::ProcessPendingRequest(
889    const std::string& group_name, Group* group) {
890  int rv = RequestSocketInternal(group_name,
891                                 *group->pending_requests().begin());
892  if (rv != ERR_IO_PENDING) {
893    scoped_ptr<const Request> request(RemoveRequestFromQueue(
894          group->mutable_pending_requests()->begin(), group));
895    if (group->IsEmpty())
896      RemoveGroup(group_name);
897
898    request->net_log().EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL, rv);
899    InvokeUserCallbackLater(request->handle(), request->callback(), rv);
900  }
901}
902
903void ClientSocketPoolBaseHelper::HandOutSocket(
904    ClientSocket* socket,
905    bool reused,
906    ClientSocketHandle* handle,
907    base::TimeDelta idle_time,
908    Group* group,
909    const BoundNetLog& net_log) {
910  DCHECK(socket);
911  handle->set_socket(socket);
912  handle->set_is_reused(reused);
913  handle->set_idle_time(idle_time);
914  handle->set_pool_id(pool_generation_number_);
915
916  if (reused) {
917    net_log.AddEvent(
918        NetLog::TYPE_SOCKET_POOL_REUSED_AN_EXISTING_SOCKET,
919        make_scoped_refptr(new NetLogIntegerParameter(
920            "idle_ms", static_cast<int>(idle_time.InMilliseconds()))));
921  }
922
923  net_log.AddEvent(NetLog::TYPE_SOCKET_POOL_BOUND_TO_SOCKET,
924                   make_scoped_refptr(new NetLogSourceParameter(
925                       "source_dependency", socket->NetLog().source())));
926
927  handed_out_socket_count_++;
928  group->IncrementActiveSocketCount();
929}
930
931void ClientSocketPoolBaseHelper::AddIdleSocket(
932    ClientSocket* socket, Group* group) {
933  DCHECK(socket);
934  IdleSocket idle_socket;
935  idle_socket.socket = socket;
936  idle_socket.start_time = base::TimeTicks::Now();
937
938  group->mutable_idle_sockets()->push_back(idle_socket);
939  IncrementIdleCount();
940}
941
942void ClientSocketPoolBaseHelper::CancelAllConnectJobs() {
943  for (GroupMap::iterator i = group_map_.begin(); i != group_map_.end();) {
944    Group* group = i->second;
945    connecting_socket_count_ -= group->jobs().size();
946    group->RemoveAllJobs();
947
948    // Delete group if no longer needed.
949    if (group->IsEmpty()) {
950      // RemoveGroup() will call .erase() which will invalidate the iterator,
951      // but i will already have been incremented to a valid iterator before
952      // RemoveGroup() is called.
953      RemoveGroup(i++);
954    } else {
955      ++i;
956    }
957  }
958  DCHECK_EQ(0, connecting_socket_count_);
959}
960
961void ClientSocketPoolBaseHelper::AbortAllRequests() {
962  for (GroupMap::iterator i = group_map_.begin(); i != group_map_.end();) {
963    Group* group = i->second;
964
965    RequestQueue pending_requests;
966    pending_requests.swap(*group->mutable_pending_requests());
967    for (RequestQueue::iterator it2 = pending_requests.begin();
968         it2 != pending_requests.end(); ++it2) {
969      scoped_ptr<const Request> request(*it2);
970      InvokeUserCallbackLater(
971          request->handle(), request->callback(), ERR_ABORTED);
972    }
973
974    // Delete group if no longer needed.
975    if (group->IsEmpty()) {
976      // RemoveGroup() will call .erase() which will invalidate the iterator,
977      // but i will already have been incremented to a valid iterator before
978      // RemoveGroup() is called.
979      RemoveGroup(i++);
980    } else {
981      ++i;
982    }
983  }
984}
985
986bool ClientSocketPoolBaseHelper::ReachedMaxSocketsLimit() const {
987  // Each connecting socket will eventually connect and be handed out.
988  int total = handed_out_socket_count_ + connecting_socket_count_ +
989      idle_socket_count();
990  // There can be more sockets than the limit since some requests can ignore
991  // the limit
992  if (total < max_sockets_)
993    return false;
994  return true;
995}
996
997void ClientSocketPoolBaseHelper::CloseOneIdleSocket() {
998  CloseOneIdleSocketExceptInGroup(NULL);
999}
1000
1001bool ClientSocketPoolBaseHelper::CloseOneIdleSocketExceptInGroup(
1002    const Group* exception_group) {
1003  CHECK_GT(idle_socket_count(), 0);
1004
1005  for (GroupMap::iterator i = group_map_.begin(); i != group_map_.end(); ++i) {
1006    Group* group = i->second;
1007    if (exception_group == group)
1008      continue;
1009    std::list<IdleSocket>* idle_sockets = group->mutable_idle_sockets();
1010
1011    if (!idle_sockets->empty()) {
1012      delete idle_sockets->front().socket;
1013      idle_sockets->pop_front();
1014      DecrementIdleCount();
1015      if (group->IsEmpty())
1016        RemoveGroup(i);
1017
1018      return true;
1019    }
1020  }
1021
1022  if (!exception_group)
1023    LOG(DFATAL) << "No idle socket found to close!.";
1024
1025  return false;
1026}
1027
1028void ClientSocketPoolBaseHelper::InvokeUserCallbackLater(
1029    ClientSocketHandle* handle, CompletionCallback* callback, int rv) {
1030  CHECK(!ContainsKey(pending_callback_map_, handle));
1031  pending_callback_map_[handle] = CallbackResultPair(callback, rv);
1032  MessageLoop::current()->PostTask(
1033      FROM_HERE,
1034      method_factory_.NewRunnableMethod(
1035          &ClientSocketPoolBaseHelper::InvokeUserCallback,
1036          handle));
1037}
1038
1039void ClientSocketPoolBaseHelper::InvokeUserCallback(
1040    ClientSocketHandle* handle) {
1041  PendingCallbackMap::iterator it = pending_callback_map_.find(handle);
1042
1043  // Exit if the request has already been cancelled.
1044  if (it == pending_callback_map_.end())
1045    return;
1046
1047  CHECK(!handle->is_initialized());
1048  CompletionCallback* callback = it->second.callback;
1049  int result = it->second.result;
1050  pending_callback_map_.erase(it);
1051  callback->Run(result);
1052}
1053
1054ClientSocketPoolBaseHelper::Group::Group()
1055    : active_socket_count_(0),
1056      ALLOW_THIS_IN_INITIALIZER_LIST(method_factory_(this)) {}
1057
1058ClientSocketPoolBaseHelper::Group::~Group() {
1059  CleanupBackupJob();
1060}
1061
1062void ClientSocketPoolBaseHelper::Group::StartBackupSocketTimer(
1063    const std::string& group_name,
1064    ClientSocketPoolBaseHelper* pool) {
1065  // Only allow one timer pending to create a backup socket.
1066  if (!method_factory_.empty())
1067    return;
1068
1069  MessageLoop::current()->PostDelayedTask(
1070      FROM_HERE,
1071      method_factory_.NewRunnableMethod(
1072          &Group::OnBackupSocketTimerFired, group_name, pool),
1073      pool->ConnectRetryIntervalMs());
1074}
1075
1076bool ClientSocketPoolBaseHelper::Group::TryToUsePreconnectConnectJob() {
1077  for (std::set<ConnectJob*>::iterator it = jobs_.begin();
1078       it != jobs_.end(); ++it) {
1079    ConnectJob* job = *it;
1080    if (job->is_unused_preconnect()) {
1081      job->UseForNormalRequest();
1082      return true;
1083    }
1084  }
1085  return false;
1086}
1087
1088void ClientSocketPoolBaseHelper::Group::OnBackupSocketTimerFired(
1089    std::string group_name,
1090    ClientSocketPoolBaseHelper* pool) {
1091  // If there are no more jobs pending, there is no work to do.
1092  // If we've done our cleanups correctly, this should not happen.
1093  if (jobs_.empty()) {
1094    NOTREACHED();
1095    return;
1096  }
1097
1098  // If our backup job is waiting on DNS, or if we can't create any sockets
1099  // right now due to limits, just reset the timer.
1100  if (pool->ReachedMaxSocketsLimit() ||
1101      !HasAvailableSocketSlot(pool->max_sockets_per_group_) ||
1102      (*jobs_.begin())->GetLoadState() == LOAD_STATE_RESOLVING_HOST) {
1103    StartBackupSocketTimer(group_name, pool);
1104    return;
1105  }
1106
1107  if (pending_requests_.empty()) {
1108    LOG(DFATAL) << "No pending request for backup job.";
1109    return;
1110  }
1111
1112  ConnectJob* backup_job = pool->connect_job_factory_->NewConnectJob(
1113      group_name, **pending_requests_.begin(), pool);
1114  backup_job->net_log().AddEvent(NetLog::TYPE_SOCKET_BACKUP_CREATED, NULL);
1115  SIMPLE_STATS_COUNTER("socket.backup_created");
1116  int rv = backup_job->Connect();
1117  pool->connecting_socket_count_++;
1118  AddJob(backup_job);
1119  if (rv != ERR_IO_PENDING)
1120    pool->OnConnectJobComplete(rv, backup_job);
1121}
1122
1123void ClientSocketPoolBaseHelper::Group::RemoveAllJobs() {
1124  // Delete active jobs.
1125  STLDeleteElements(&jobs_);
1126
1127  // Cancel pending backup job.
1128  method_factory_.RevokeAll();
1129}
1130
1131}  // namespace internal
1132
1133}  // namespace net
1134