This source file includes following definitions.
- idle_
- PassSocket
- SetSocket
- NotifyDelegateOfCompletion
- ResetTimer
- LogConnectStart
- LogConnectCompletion
- OnTimeout
- net_log_
- weak_factory_
- result
- IsStalled
- AddLowerLayeredPool
- AddHigherLayeredPool
- RemoveHigherLayeredPool
- RequestSocket
- RequestSockets
- RequestSocketInternal
- AssignIdleSocketToRequest
- LogBoundConnectJobToRequest
- CancelRequest
- HasGroup
- CloseIdleSockets
- IdleSocketCountInGroup
- GetLoadState
- GetInfoAsValue
- IsUsable
- ShouldCleanup
- CleanupIdleSockets
- GetOrCreateGroup
- RemoveGroup
- RemoveGroup
- connect_backup_jobs_enabled
- set_connect_backup_jobs_enabled
- EnableConnectBackupJobs
- IncrementIdleCount
- DecrementIdleCount
- cleanup_timer_enabled
- set_cleanup_timer_enabled
- StartIdleSocketTimer
- ReleaseSocket
- CheckForStalledSocketGroups
- FindTopStalledGroup
- OnConnectJobComplete
- OnIPAddressChanged
- FlushWithError
- RemoveConnectJob
- OnAvailableSocketSlot
- ProcessPendingRequest
- HandOutSocket
- AddIdleSocket
- CancelAllConnectJobs
- CancelAllRequestsWithError
- ReachedMaxSocketsLimit
- CloseOneIdleSocket
- CloseOneIdleSocketExceptInGroup
- CloseOneIdleConnectionInHigherLayeredPool
- InvokeUserCallbackLater
- InvokeUserCallback
- TryToCloseSocketsInLayeredPools
- active_socket_count_
- StartBackupJobTimer
- BackupJobTimerIsRunning
- TryToUseUnassignedConnectJob
- AddJob
- RemoveJob
- OnBackupJobTimerFired
- SanityCheck
- RemoveAllJobs
- GetNextPendingRequest
- HasConnectJobForHandle
- InsertPendingRequest
- PopNextPendingRequest
- FindAndRemovePendingRequest
- RemovePendingRequest
#include "net/socket/client_socket_pool_base.h"
#include "base/compiler_specific.h"
#include "base/format_macros.h"
#include "base/logging.h"
#include "base/message_loop/message_loop.h"
#include "base/metrics/stats_counters.h"
#include "base/stl_util.h"
#include "base/strings/string_util.h"
#include "base/time/time.h"
#include "base/values.h"
#include "net/base/net_errors.h"
#include "net/base/net_log.h"
using base::TimeDelta;
namespace net {
namespace {
bool g_cleanup_timer_enabled = true;
const int kCleanupInterval = 10;
bool g_connect_backup_jobs_enabled = true;
}
ConnectJob::ConnectJob(const std::string& group_name,
base::TimeDelta timeout_duration,
RequestPriority priority,
Delegate* delegate,
const BoundNetLog& net_log)
: group_name_(group_name),
timeout_duration_(timeout_duration),
priority_(priority),
delegate_(delegate),
net_log_(net_log),
idle_(true) {
DCHECK(!group_name.empty());
DCHECK(delegate);
net_log.BeginEvent(NetLog::TYPE_SOCKET_POOL_CONNECT_JOB,
NetLog::StringCallback("group_name", &group_name_));
}
ConnectJob::~ConnectJob() {
net_log().EndEvent(NetLog::TYPE_SOCKET_POOL_CONNECT_JOB);
}
scoped_ptr<StreamSocket> ConnectJob::PassSocket() {
return socket_.Pass();
}
int ConnectJob::Connect() {
if (timeout_duration_ != base::TimeDelta())
timer_.Start(FROM_HERE, timeout_duration_, this, &ConnectJob::OnTimeout);
idle_ = false;
LogConnectStart();
int rv = ConnectInternal();
if (rv != ERR_IO_PENDING) {
LogConnectCompletion(rv);
delegate_ = NULL;
}
return rv;
}
void ConnectJob::SetSocket(scoped_ptr<StreamSocket> socket) {
if (socket) {
net_log().AddEvent(NetLog::TYPE_CONNECT_JOB_SET_SOCKET,
socket->NetLog().source().ToEventParametersCallback());
}
socket_ = socket.Pass();
}
void ConnectJob::NotifyDelegateOfCompletion(int rv) {
Delegate* delegate = delegate_;
delegate_ = NULL;
LogConnectCompletion(rv);
delegate->OnConnectJobComplete(rv, this);
}
void ConnectJob::ResetTimer(base::TimeDelta remaining_time) {
timer_.Stop();
timer_.Start(FROM_HERE, remaining_time, this, &ConnectJob::OnTimeout);
}
void ConnectJob::LogConnectStart() {
connect_timing_.connect_start = base::TimeTicks::Now();
net_log().BeginEvent(NetLog::TYPE_SOCKET_POOL_CONNECT_JOB_CONNECT);
}
void ConnectJob::LogConnectCompletion(int net_error) {
connect_timing_.connect_end = base::TimeTicks::Now();
net_log().EndEventWithNetErrorCode(
NetLog::TYPE_SOCKET_POOL_CONNECT_JOB_CONNECT, net_error);
}
void ConnectJob::OnTimeout() {
SetSocket(scoped_ptr<StreamSocket>());
net_log_.AddEvent(NetLog::TYPE_SOCKET_POOL_CONNECT_JOB_TIMED_OUT);
NotifyDelegateOfCompletion(ERR_TIMED_OUT);
}
namespace internal {
ClientSocketPoolBaseHelper::Request::Request(
ClientSocketHandle* handle,
const CompletionCallback& callback,
RequestPriority priority,
bool ignore_limits,
Flags flags,
const BoundNetLog& net_log)
: handle_(handle),
callback_(callback),
priority_(priority),
ignore_limits_(ignore_limits),
flags_(flags),
net_log_(net_log) {
if (ignore_limits_)
DCHECK_EQ(priority_, MAXIMUM_PRIORITY);
}
ClientSocketPoolBaseHelper::Request::~Request() {}
ClientSocketPoolBaseHelper::ClientSocketPoolBaseHelper(
HigherLayeredPool* pool,
int max_sockets,
int max_sockets_per_group,
base::TimeDelta unused_idle_socket_timeout,
base::TimeDelta used_idle_socket_timeout,
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),
use_cleanup_timer_(g_cleanup_timer_enabled),
unused_idle_socket_timeout_(unused_idle_socket_timeout),
used_idle_socket_timeout_(used_idle_socket_timeout),
connect_job_factory_(connect_job_factory),
connect_backup_jobs_enabled_(false),
pool_generation_number_(0),
pool_(pool),
weak_factory_(this) {
DCHECK_LE(0, max_sockets_per_group);
DCHECK_LE(max_sockets_per_group, max_sockets);
NetworkChangeNotifier::AddIPAddressObserver(this);
}
ClientSocketPoolBaseHelper::~ClientSocketPoolBaseHelper() {
FlushWithError(ERR_ABORTED);
DCHECK(group_map_.empty());
DCHECK(pending_callback_map_.empty());
DCHECK_EQ(0, connecting_socket_count_);
CHECK(higher_pools_.empty());
NetworkChangeNotifier::RemoveIPAddressObserver(this);
for (std::set<LowerLayeredPool*>::iterator it = lower_pools_.begin();
it != lower_pools_.end();
++it) {
(*it)->RemoveHigherLayeredPool(pool_);
}
}
ClientSocketPoolBaseHelper::CallbackResultPair::CallbackResultPair()
: result(OK) {
}
ClientSocketPoolBaseHelper::CallbackResultPair::CallbackResultPair(
const CompletionCallback& callback_in, int result_in)
: callback(callback_in),
result(result_in) {
}
ClientSocketPoolBaseHelper::CallbackResultPair::~CallbackResultPair() {}
bool ClientSocketPoolBaseHelper::IsStalled() const {
for (std::set<LowerLayeredPool*>::const_iterator it = lower_pools_.begin();
it != lower_pools_.end();
++it) {
if ((*it)->IsStalled())
return true;
}
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->IsStalledOnPoolMaxSockets(max_sockets_per_group_))
return true;
}
return false;
}
void ClientSocketPoolBaseHelper::AddLowerLayeredPool(
LowerLayeredPool* lower_pool) {
DCHECK(pool_);
CHECK(!ContainsKey(lower_pools_, lower_pool));
lower_pools_.insert(lower_pool);
lower_pool->AddHigherLayeredPool(pool_);
}
void ClientSocketPoolBaseHelper::AddHigherLayeredPool(
HigherLayeredPool* higher_pool) {
CHECK(higher_pool);
CHECK(!ContainsKey(higher_pools_, higher_pool));
higher_pools_.insert(higher_pool);
}
void ClientSocketPoolBaseHelper::RemoveHigherLayeredPool(
HigherLayeredPool* higher_pool) {
CHECK(higher_pool);
CHECK(ContainsKey(higher_pools_, higher_pool));
higher_pools_.erase(higher_pool);
}
int ClientSocketPoolBaseHelper::RequestSocket(
const std::string& group_name,
scoped_ptr<const Request> request) {
CHECK(!request->callback().is_null());
CHECK(request->handle());
if (!use_cleanup_timer_)
CleanupIdleSockets(false);
request->net_log().BeginEvent(NetLog::TYPE_SOCKET_POOL);
Group* group = GetOrCreateGroup(group_name);
int rv = RequestSocketInternal(group_name, *request);
if (rv != ERR_IO_PENDING) {
request->net_log().EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL, rv);
CHECK(!request->handle()->is_initialized());
request.reset();
} else {
group->InsertPendingRequest(request.Pass());
if (group->IsStalledOnPoolMaxSockets(max_sockets_per_group_)) {
base::MessageLoop::current()->PostTask(
FROM_HERE,
base::Bind(
&ClientSocketPoolBaseHelper::TryToCloseSocketsInLayeredPools,
weak_factory_.GetWeakPtr()));
}
}
return rv;
}
void ClientSocketPoolBaseHelper::RequestSockets(
const std::string& group_name,
const Request& request,
int num_sockets) {
DCHECK(request.callback().is_null());
DCHECK(!request.handle());
if (!use_cleanup_timer_)
CleanupIdleSockets(false);
if (num_sockets > max_sockets_per_group_) {
num_sockets = max_sockets_per_group_;
}
request.net_log().BeginEvent(
NetLog::TYPE_SOCKET_POOL_CONNECTING_N_SOCKETS,
NetLog::IntegerCallback("num_sockets", num_sockets));
Group* group = GetOrCreateGroup(group_name);
bool deleted_group = false;
int rv = OK;
for (int num_iterations_left = num_sockets;
group->NumActiveSocketSlots() < num_sockets &&
num_iterations_left > 0 ; num_iterations_left--) {
rv = RequestSocketInternal(group_name, request);
if (rv < 0 && rv != ERR_IO_PENDING) {
if (!ContainsKey(group_map_, group_name))
deleted_group = true;
break;
}
if (!ContainsKey(group_map_, group_name)) {
NOTREACHED();
deleted_group = true;
break;
}
}
if (!deleted_group && group->IsEmpty())
RemoveGroup(group_name);
if (rv == ERR_IO_PENDING)
rv = OK;
request.net_log().EndEventWithNetErrorCode(
NetLog::TYPE_SOCKET_POOL_CONNECTING_N_SOCKETS, rv);
}
int ClientSocketPoolBaseHelper::RequestSocketInternal(
const std::string& group_name,
const Request& request) {
ClientSocketHandle* const handle = request.handle();
const bool preconnecting = !handle;
Group* group = GetOrCreateGroup(group_name);
if (!(request.flags() & NO_IDLE_SOCKETS)) {
if (AssignIdleSocketToRequest(request, group))
return OK;
}
if (!preconnecting && group->TryToUseUnassignedConnectJob())
return ERR_IO_PENDING;
if (!group->HasAvailableSocketSlot(max_sockets_per_group_) &&
!request.ignore_limits()) {
request.net_log().AddEvent(
NetLog::TYPE_SOCKET_POOL_STALLED_MAX_SOCKETS_PER_GROUP);
return ERR_IO_PENDING;
}
if (ReachedMaxSocketsLimit() && !request.ignore_limits()) {
if (idle_socket_count() > 0) {
bool closed = CloseOneIdleSocketExceptInGroup(group);
if (preconnecting && !closed)
return ERR_PRECONNECT_MAX_SOCKET_LIMIT;
} else {
request.net_log().AddEvent(NetLog::TYPE_SOCKET_POOL_STALLED_MAX_SOCKETS);
return ERR_IO_PENDING;
}
}
scoped_ptr<ConnectJob> connect_job(
connect_job_factory_->NewConnectJob(group_name, request, this));
int rv = connect_job->Connect();
if (rv == OK) {
LogBoundConnectJobToRequest(connect_job->net_log().source(), request);
if (!preconnecting) {
HandOutSocket(connect_job->PassSocket(), ClientSocketHandle::UNUSED,
connect_job->connect_timing(), handle, base::TimeDelta(),
group, request.net_log());
} else {
AddIdleSocket(connect_job->PassSocket(), group);
}
} else if (rv == ERR_IO_PENDING) {
if (connect_backup_jobs_enabled_ && group->IsEmpty()) {
group->StartBackupJobTimer(group_name, this);
}
connecting_socket_count_++;
group->AddJob(connect_job.Pass(), preconnecting);
} else {
LogBoundConnectJobToRequest(connect_job->net_log().source(), request);
scoped_ptr<StreamSocket> error_socket;
if (!preconnecting) {
DCHECK(handle);
connect_job->GetAdditionalErrorState(handle);
error_socket = connect_job->PassSocket();
}
if (error_socket) {
HandOutSocket(error_socket.Pass(), ClientSocketHandle::UNUSED,
connect_job->connect_timing(), handle, base::TimeDelta(),
group, request.net_log());
} else if (group->IsEmpty()) {
RemoveGroup(group_name);
}
}
return rv;
}
bool ClientSocketPoolBaseHelper::AssignIdleSocketToRequest(
const Request& request, Group* group) {
std::list<IdleSocket>* idle_sockets = group->mutable_idle_sockets();
std::list<IdleSocket>::iterator idle_socket_it = idle_sockets->end();
for (std::list<IdleSocket>::iterator it = idle_sockets->begin();
it != idle_sockets->end();) {
if (!it->IsUsable()) {
DecrementIdleCount();
delete it->socket;
it = idle_sockets->erase(it);
continue;
}
if (it->socket->WasEverUsed()) {
idle_socket_it = it;
}
++it;
}
if (idle_socket_it == idle_sockets->end() && !idle_sockets->empty())
idle_socket_it = idle_sockets->begin();
if (idle_socket_it != idle_sockets->end()) {
DecrementIdleCount();
base::TimeDelta idle_time =
base::TimeTicks::Now() - idle_socket_it->start_time;
IdleSocket idle_socket = *idle_socket_it;
idle_sockets->erase(idle_socket_it);
ClientSocketHandle::SocketReuseType reuse_type =
idle_socket.socket->WasEverUsed() ?
ClientSocketHandle::REUSED_IDLE :
ClientSocketHandle::UNUSED_IDLE;
HandOutSocket(
scoped_ptr<StreamSocket>(idle_socket.socket),
reuse_type,
LoadTimingInfo::ConnectTiming(),
request.handle(),
idle_time,
group,
request.net_log());
return true;
}
return false;
}
void ClientSocketPoolBaseHelper::LogBoundConnectJobToRequest(
const NetLog::Source& connect_job_source, const Request& request) {
request.net_log().AddEvent(NetLog::TYPE_SOCKET_POOL_BOUND_TO_CONNECT_JOB,
connect_job_source.ToEventParametersCallback());
}
void ClientSocketPoolBaseHelper::CancelRequest(
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);
scoped_ptr<StreamSocket> socket = handle->PassSocket();
if (socket) {
if (result != OK)
socket->Disconnect();
ReleaseSocket(handle->group_name(), socket.Pass(), handle->id());
}
return;
}
CHECK(ContainsKey(group_map_, group_name));
Group* group = GetOrCreateGroup(group_name);
scoped_ptr<const Request> request =
group->FindAndRemovePendingRequest(handle);
if (request) {
request->net_log().AddEvent(NetLog::TYPE_CANCELLED);
request->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL);
if (group->jobs().size() > group->pending_request_count() &&
ReachedMaxSocketsLimit()) {
RemoveConnectJob(*group->jobs().begin(), group);
CheckForStalledSocketGroups();
}
}
}
bool ClientSocketPoolBaseHelper::HasGroup(const std::string& group_name) const {
return ContainsKey(group_map_, group_name);
}
void ClientSocketPoolBaseHelper::CloseIdleSockets() {
CleanupIdleSockets(true);
DCHECK_EQ(0, idle_socket_count_);
}
int ClientSocketPoolBaseHelper::IdleSocketCountInGroup(
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;
}
const Group& group = *group_map_.find(group_name)->second;
if (group.HasConnectJobForHandle(handle)) {
LoadState max_state = LOAD_STATE_IDLE;
for (ConnectJobSet::const_iterator job_it = group.jobs().begin();
job_it != group.jobs().end(); ++job_it) {
max_state = std::max(max_state, (*job_it)->GetLoadState());
}
return max_state;
}
if (group.IsStalledOnPoolMaxSockets(max_sockets_per_group_))
return LOAD_STATE_WAITING_FOR_STALLED_SOCKET_POOL;
return LOAD_STATE_WAITING_FOR_AVAILABLE_SOCKET;
}
base::DictionaryValue* ClientSocketPoolBaseHelper::GetInfoAsValue(
const std::string& name, const std::string& type) const {
base::DictionaryValue* dict = new base::DictionaryValue();
dict->SetString("name", name);
dict->SetString("type", type);
dict->SetInteger("handed_out_socket_count", handed_out_socket_count_);
dict->SetInteger("connecting_socket_count", connecting_socket_count_);
dict->SetInteger("idle_socket_count", idle_socket_count_);
dict->SetInteger("max_socket_count", max_sockets_);
dict->SetInteger("max_sockets_per_group", max_sockets_per_group_);
dict->SetInteger("pool_generation_number", pool_generation_number_);
if (group_map_.empty())
return dict;
base::DictionaryValue* all_groups_dict = new base::DictionaryValue();
for (GroupMap::const_iterator it = group_map_.begin();
it != group_map_.end(); it++) {
const Group* group = it->second;
base::DictionaryValue* group_dict = new base::DictionaryValue();
group_dict->SetInteger("pending_request_count",
group->pending_request_count());
if (group->has_pending_requests()) {
group_dict->SetString(
"top_pending_priority",
RequestPriorityToString(group->TopPendingPriority()));
}
group_dict->SetInteger("active_socket_count", group->active_socket_count());
base::ListValue* idle_socket_list = new base::ListValue();
std::list<IdleSocket>::const_iterator idle_socket;
for (idle_socket = group->idle_sockets().begin();
idle_socket != group->idle_sockets().end();
idle_socket++) {
int source_id = idle_socket->socket->NetLog().source().id;
idle_socket_list->Append(new base::FundamentalValue(source_id));
}
group_dict->Set("idle_sockets", idle_socket_list);
base::ListValue* connect_jobs_list = new base::ListValue();
std::set<ConnectJob*>::const_iterator job = group->jobs().begin();
for (job = group->jobs().begin(); job != group->jobs().end(); job++) {
int source_id = (*job)->net_log().source().id;
connect_jobs_list->Append(new base::FundamentalValue(source_id));
}
group_dict->Set("connect_jobs", connect_jobs_list);
group_dict->SetBoolean("is_stalled",
group->IsStalledOnPoolMaxSockets(
max_sockets_per_group_));
group_dict->SetBoolean("backup_job_timer_is_running",
group->BackupJobTimerIsRunning());
all_groups_dict->SetWithoutPathExpansion(it->first, group_dict);
}
dict->Set("groups", all_groups_dict);
return dict;
}
bool ClientSocketPoolBaseHelper::IdleSocket::IsUsable() const {
if (socket->WasEverUsed())
return socket->IsConnectedAndIdle();
return socket->IsConnected();
}
bool ClientSocketPoolBaseHelper::IdleSocket::ShouldCleanup(
base::TimeTicks now,
base::TimeDelta timeout) const {
bool timed_out = (now - start_time) >= timeout;
if (timed_out)
return true;
return !IsUsable();
}
void ClientSocketPoolBaseHelper::CleanupIdleSockets(bool force) {
if (idle_socket_count_ == 0)
return;
base::TimeTicks now = base::TimeTicks::Now();
GroupMap::iterator i = group_map_.begin();
while (i != group_map_.end()) {
Group* group = i->second;
std::list<IdleSocket>::iterator j = group->mutable_idle_sockets()->begin();
while (j != group->idle_sockets().end()) {
base::TimeDelta timeout =
j->socket->WasEverUsed() ?
used_idle_socket_timeout_ : unused_idle_socket_timeout_;
if (force || j->ShouldCleanup(now, timeout)) {
delete j->socket;
j = group->mutable_idle_sockets()->erase(j);
DecrementIdleCount();
} else {
++j;
}
}
if (group->IsEmpty()) {
RemoveGroup(i++);
} else {
++i;
}
}
}
ClientSocketPoolBaseHelper::Group* ClientSocketPoolBaseHelper::GetOrCreateGroup(
const std::string& group_name) {
GroupMap::iterator it = group_map_.find(group_name);
if (it != group_map_.end())
return it->second;
Group* group = new Group;
group_map_[group_name] = group;
return group;
}
void ClientSocketPoolBaseHelper::RemoveGroup(const std::string& group_name) {
GroupMap::iterator it = group_map_.find(group_name);
CHECK(it != group_map_.end());
RemoveGroup(it);
}
void ClientSocketPoolBaseHelper::RemoveGroup(GroupMap::iterator it) {
delete it->second;
group_map_.erase(it);
}
bool ClientSocketPoolBaseHelper::connect_backup_jobs_enabled() {
return g_connect_backup_jobs_enabled;
}
bool ClientSocketPoolBaseHelper::set_connect_backup_jobs_enabled(bool enabled) {
bool old_value = g_connect_backup_jobs_enabled;
g_connect_backup_jobs_enabled = enabled;
return old_value;
}
void ClientSocketPoolBaseHelper::EnableConnectBackupJobs() {
connect_backup_jobs_enabled_ = g_connect_backup_jobs_enabled;
}
void ClientSocketPoolBaseHelper::IncrementIdleCount() {
if (++idle_socket_count_ == 1 && use_cleanup_timer_)
StartIdleSocketTimer();
}
void ClientSocketPoolBaseHelper::DecrementIdleCount() {
if (--idle_socket_count_ == 0)
timer_.Stop();
}
bool ClientSocketPoolBaseHelper::cleanup_timer_enabled() {
return g_cleanup_timer_enabled;
}
bool ClientSocketPoolBaseHelper::set_cleanup_timer_enabled(bool enabled) {
bool old_value = g_cleanup_timer_enabled;
g_cleanup_timer_enabled = enabled;
return old_value;
}
void ClientSocketPoolBaseHelper::StartIdleSocketTimer() {
timer_.Start(FROM_HERE, TimeDelta::FromSeconds(kCleanupInterval), this,
&ClientSocketPoolBaseHelper::OnCleanupTimerFired);
}
void ClientSocketPoolBaseHelper::ReleaseSocket(const std::string& group_name,
scoped_ptr<StreamSocket> socket,
int id) {
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->DecrementActiveSocketCount();
const bool can_reuse = socket->IsConnectedAndIdle() &&
id == pool_generation_number_;
if (can_reuse) {
AddIdleSocket(socket.Pass(), group);
OnAvailableSocketSlot(group_name, group);
} else {
socket.reset();
}
CheckForStalledSocketGroups();
}
void ClientSocketPoolBaseHelper::CheckForStalledSocketGroups() {
std::string top_group_name;
Group* top_group = NULL;
if (!FindTopStalledGroup(&top_group, &top_group_name)) {
for (std::set<LowerLayeredPool*>::iterator it = lower_pools_.begin();
it != lower_pools_.end();
++it) {
if ((*it)->IsStalled()) {
CloseOneIdleSocket();
break;
}
}
return;
}
if (ReachedMaxSocketsLimit()) {
if (idle_socket_count() > 0) {
CloseOneIdleSocket();
} else {
return;
}
}
OnAvailableSocketSlot(top_group_name, top_group);
}
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::const_iterator i = group_map_.begin();
i != group_map_.end(); ++i) {
Group* curr_group = i->second;
if (!curr_group->has_pending_requests())
continue;
if (curr_group->IsStalledOnPoolMaxSockets(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());
Group* group = group_it->second;
scoped_ptr<StreamSocket> socket = job->PassSocket();
BoundNetLog job_log = job->net_log();
LoadTimingInfo::ConnectTiming connect_timing = job->connect_timing();
if (result == OK) {
DCHECK(socket.get());
RemoveConnectJob(job, group);
scoped_ptr<const Request> request = group->PopNextPendingRequest();
if (request) {
LogBoundConnectJobToRequest(job_log.source(), *request);
HandOutSocket(
socket.Pass(), ClientSocketHandle::UNUSED, connect_timing,
request->handle(), base::TimeDelta(), group, request->net_log());
request->net_log().EndEvent(NetLog::TYPE_SOCKET_POOL);
InvokeUserCallbackLater(request->handle(), request->callback(), result);
} else {
AddIdleSocket(socket.Pass(), group);
OnAvailableSocketSlot(group_name, group);
CheckForStalledSocketGroups();
}
} else {
bool handed_out_socket = false;
scoped_ptr<const Request> request = group->PopNextPendingRequest();
if (request) {
LogBoundConnectJobToRequest(job_log.source(), *request);
job->GetAdditionalErrorState(request->handle());
RemoveConnectJob(job, group);
if (socket.get()) {
handed_out_socket = true;
HandOutSocket(socket.Pass(), ClientSocketHandle::UNUSED,
connect_timing, request->handle(), base::TimeDelta(),
group, request->net_log());
}
request->net_log().EndEventWithNetErrorCode(
NetLog::TYPE_SOCKET_POOL, result);
InvokeUserCallbackLater(request->handle(), request->callback(), result);
} else {
RemoveConnectJob(job, group);
}
if (!handed_out_socket) {
OnAvailableSocketSlot(group_name, group);
CheckForStalledSocketGroups();
}
}
}
void ClientSocketPoolBaseHelper::OnIPAddressChanged() {
FlushWithError(ERR_NETWORK_CHANGED);
}
void ClientSocketPoolBaseHelper::FlushWithError(int error) {
pool_generation_number_++;
CancelAllConnectJobs();
CloseIdleSockets();
CancelAllRequestsWithError(error);
}
void ClientSocketPoolBaseHelper::RemoveConnectJob(ConnectJob* job,
Group* group) {
CHECK_GT(connecting_socket_count_, 0);
connecting_socket_count_--;
DCHECK(group);
group->RemoveJob(job);
}
void ClientSocketPoolBaseHelper::OnAvailableSocketSlot(
const std::string& group_name, Group* group) {
DCHECK(ContainsKey(group_map_, group_name));
if (group->IsEmpty()) {
RemoveGroup(group_name);
} else if (group->has_pending_requests()) {
ProcessPendingRequest(group_name, group);
}
}
void ClientSocketPoolBaseHelper::ProcessPendingRequest(
const std::string& group_name, Group* group) {
const Request* next_request = group->GetNextPendingRequest();
DCHECK(next_request);
int rv = RequestSocketInternal(group_name, *next_request);
if (rv != ERR_IO_PENDING) {
scoped_ptr<const Request> request = group->PopNextPendingRequest();
DCHECK(request);
if (group->IsEmpty())
RemoveGroup(group_name);
request->net_log().EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL, rv);
InvokeUserCallbackLater(request->handle(), request->callback(), rv);
}
}
void ClientSocketPoolBaseHelper::HandOutSocket(
scoped_ptr<StreamSocket> socket,
ClientSocketHandle::SocketReuseType reuse_type,
const LoadTimingInfo::ConnectTiming& connect_timing,
ClientSocketHandle* handle,
base::TimeDelta idle_time,
Group* group,
const BoundNetLog& net_log) {
DCHECK(socket);
handle->SetSocket(socket.Pass());
handle->set_reuse_type(reuse_type);
handle->set_idle_time(idle_time);
handle->set_pool_id(pool_generation_number_);
handle->set_connect_timing(connect_timing);
if (handle->is_reused()) {
net_log.AddEvent(
NetLog::TYPE_SOCKET_POOL_REUSED_AN_EXISTING_SOCKET,
NetLog::IntegerCallback(
"idle_ms", static_cast<int>(idle_time.InMilliseconds())));
}
net_log.AddEvent(
NetLog::TYPE_SOCKET_POOL_BOUND_TO_SOCKET,
handle->socket()->NetLog().source().ToEventParametersCallback());
handed_out_socket_count_++;
group->IncrementActiveSocketCount();
}
void ClientSocketPoolBaseHelper::AddIdleSocket(
scoped_ptr<StreamSocket> socket,
Group* group) {
DCHECK(socket);
IdleSocket idle_socket;
idle_socket.socket = socket.release();
idle_socket.start_time = base::TimeTicks::Now();
group->mutable_idle_sockets()->push_back(idle_socket);
IncrementIdleCount();
}
void ClientSocketPoolBaseHelper::CancelAllConnectJobs() {
for (GroupMap::iterator i = group_map_.begin(); i != group_map_.end();) {
Group* group = i->second;
connecting_socket_count_ -= group->jobs().size();
group->RemoveAllJobs();
if (group->IsEmpty()) {
RemoveGroup(i++);
} else {
++i;
}
}
DCHECK_EQ(0, connecting_socket_count_);
}
void ClientSocketPoolBaseHelper::CancelAllRequestsWithError(int error) {
for (GroupMap::iterator i = group_map_.begin(); i != group_map_.end();) {
Group* group = i->second;
while (true) {
scoped_ptr<const Request> request = group->PopNextPendingRequest();
if (!request)
break;
InvokeUserCallbackLater(request->handle(), request->callback(), error);
}
if (group->IsEmpty()) {
RemoveGroup(i++);
} else {
++i;
}
}
}
bool ClientSocketPoolBaseHelper::ReachedMaxSocketsLimit() const {
int total = handed_out_socket_count_ + connecting_socket_count_ +
idle_socket_count();
if (total < max_sockets_)
return false;
return true;
}
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;
}
}
return false;
}
bool ClientSocketPoolBaseHelper::CloseOneIdleConnectionInHigherLayeredPool() {
for (std::set<HigherLayeredPool*>::const_iterator it = higher_pools_.begin();
it != higher_pools_.end(); ++it) {
if ((*it)->CloseOneIdleConnection())
return true;
}
return false;
}
void ClientSocketPoolBaseHelper::InvokeUserCallbackLater(
ClientSocketHandle* handle, const CompletionCallback& callback, int rv) {
CHECK(!ContainsKey(pending_callback_map_, handle));
pending_callback_map_[handle] = CallbackResultPair(callback, rv);
base::MessageLoop::current()->PostTask(
FROM_HERE,
base::Bind(&ClientSocketPoolBaseHelper::InvokeUserCallback,
weak_factory_.GetWeakPtr(), handle));
}
void ClientSocketPoolBaseHelper::InvokeUserCallback(
ClientSocketHandle* handle) {
PendingCallbackMap::iterator it = pending_callback_map_.find(handle);
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);
}
void ClientSocketPoolBaseHelper::TryToCloseSocketsInLayeredPools() {
while (IsStalled()) {
if (!CloseOneIdleConnectionInHigherLayeredPool())
return;
}
}
ClientSocketPoolBaseHelper::Group::Group()
: unassigned_job_count_(0),
pending_requests_(NUM_PRIORITIES),
active_socket_count_(0) {}
ClientSocketPoolBaseHelper::Group::~Group() {
DCHECK_EQ(0u, unassigned_job_count_);
}
void ClientSocketPoolBaseHelper::Group::StartBackupJobTimer(
const std::string& group_name,
ClientSocketPoolBaseHelper* pool) {
if (BackupJobTimerIsRunning())
return;
backup_job_timer_.Start(
FROM_HERE, pool->ConnectRetryInterval(),
base::Bind(&Group::OnBackupJobTimerFired, base::Unretained(this),
group_name, pool));
}
bool ClientSocketPoolBaseHelper::Group::BackupJobTimerIsRunning() const {
return backup_job_timer_.IsRunning();
}
bool ClientSocketPoolBaseHelper::Group::TryToUseUnassignedConnectJob() {
SanityCheck();
if (unassigned_job_count_ == 0)
return false;
--unassigned_job_count_;
return true;
}
void ClientSocketPoolBaseHelper::Group::AddJob(scoped_ptr<ConnectJob> job,
bool is_preconnect) {
SanityCheck();
if (is_preconnect)
++unassigned_job_count_;
jobs_.insert(job.release());
}
void ClientSocketPoolBaseHelper::Group::RemoveJob(ConnectJob* job) {
scoped_ptr<ConnectJob> owned_job(job);
SanityCheck();
std::set<ConnectJob*>::iterator it = jobs_.find(job);
if (it != jobs_.end()) {
jobs_.erase(it);
} else {
NOTREACHED();
}
size_t job_count = jobs_.size();
if (job_count < unassigned_job_count_)
unassigned_job_count_ = job_count;
if (jobs_.empty())
backup_job_timer_.Stop();
}
void ClientSocketPoolBaseHelper::Group::OnBackupJobTimerFired(
std::string group_name,
ClientSocketPoolBaseHelper* pool) {
if (jobs_.empty()) {
NOTREACHED();
return;
}
if (pool->ReachedMaxSocketsLimit() ||
!HasAvailableSocketSlot(pool->max_sockets_per_group_) ||
(*jobs_.begin())->GetLoadState() == LOAD_STATE_RESOLVING_HOST) {
StartBackupJobTimer(group_name, pool);
return;
}
if (pending_requests_.empty())
return;
scoped_ptr<ConnectJob> backup_job =
pool->connect_job_factory_->NewConnectJob(
group_name, *pending_requests_.FirstMax().value(), pool);
backup_job->net_log().AddEvent(NetLog::TYPE_BACKUP_CONNECT_JOB_CREATED);
SIMPLE_STATS_COUNTER("socket.backup_created");
int rv = backup_job->Connect();
pool->connecting_socket_count_++;
ConnectJob* raw_backup_job = backup_job.get();
AddJob(backup_job.Pass(), false);
if (rv != ERR_IO_PENDING)
pool->OnConnectJobComplete(rv, raw_backup_job);
}
void ClientSocketPoolBaseHelper::Group::SanityCheck() {
DCHECK_LE(unassigned_job_count_, jobs_.size());
}
void ClientSocketPoolBaseHelper::Group::RemoveAllJobs() {
SanityCheck();
STLDeleteElements(&jobs_);
unassigned_job_count_ = 0;
backup_job_timer_.Stop();
}
const ClientSocketPoolBaseHelper::Request*
ClientSocketPoolBaseHelper::Group::GetNextPendingRequest() const {
return
pending_requests_.empty() ? NULL : pending_requests_.FirstMax().value();
}
bool ClientSocketPoolBaseHelper::Group::HasConnectJobForHandle(
const ClientSocketHandle* handle) const {
size_t i = 0;
for (RequestQueue::Pointer pointer = pending_requests_.FirstMax();
!pointer.is_null() && i < jobs_.size();
pointer = pending_requests_.GetNextTowardsLastMin(pointer), ++i) {
if (pointer.value()->handle() == handle)
return true;
}
return false;
}
void ClientSocketPoolBaseHelper::Group::InsertPendingRequest(
scoped_ptr<const Request> request) {
RequestPriority priority = request->priority();
if (request->ignore_limits()) {
DCHECK_EQ(priority, MAXIMUM_PRIORITY);
pending_requests_.InsertAtFront(request.release(), priority);
} else {
pending_requests_.Insert(request.release(), priority);
}
}
scoped_ptr<const ClientSocketPoolBaseHelper::Request>
ClientSocketPoolBaseHelper::Group::PopNextPendingRequest() {
if (pending_requests_.empty())
return scoped_ptr<const ClientSocketPoolBaseHelper::Request>();
return RemovePendingRequest(pending_requests_.FirstMax());
}
scoped_ptr<const ClientSocketPoolBaseHelper::Request>
ClientSocketPoolBaseHelper::Group::FindAndRemovePendingRequest(
ClientSocketHandle* handle) {
for (RequestQueue::Pointer pointer = pending_requests_.FirstMax();
!pointer.is_null();
pointer = pending_requests_.GetNextTowardsLastMin(pointer)) {
if (pointer.value()->handle() == handle) {
scoped_ptr<const Request> request = RemovePendingRequest(pointer);
return request.Pass();
}
}
return scoped_ptr<const ClientSocketPoolBaseHelper::Request>();
}
scoped_ptr<const ClientSocketPoolBaseHelper::Request>
ClientSocketPoolBaseHelper::Group::RemovePendingRequest(
const RequestQueue::Pointer& pointer) {
scoped_ptr<const Request> request(pointer.value());
pending_requests_.Erase(pointer);
if (pending_requests_.empty())
backup_job_timer_.Stop();
return request.Pass();
}
}
}