Updating trunk VERSION from 2139.0 to 2140.0
[chromium-blink-merge.git] / net / socket / websocket_transport_client_socket_pool.cc
blob15ec028cb18a665f7518a339e73bb7ab33ed7f18
1 // Copyright 2014 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.
5 #include "net/socket/websocket_transport_client_socket_pool.h"
7 #include <algorithm>
9 #include "base/compiler_specific.h"
10 #include "base/logging.h"
11 #include "base/numerics/safe_conversions.h"
12 #include "base/strings/string_util.h"
13 #include "base/time/time.h"
14 #include "base/values.h"
15 #include "net/base/net_errors.h"
16 #include "net/base/net_log.h"
17 #include "net/socket/client_socket_handle.h"
18 #include "net/socket/client_socket_pool_base.h"
19 #include "net/socket/websocket_endpoint_lock_manager.h"
20 #include "net/socket/websocket_transport_connect_sub_job.h"
22 namespace net {
24 namespace {
26 using base::TimeDelta;
28 // TODO(ricea): For now, we implement a global timeout for compatability with
29 // TransportConnectJob. Since WebSocketTransportConnectJob controls the address
30 // selection process more tightly, it could do something smarter here.
31 const int kTransportConnectJobTimeoutInSeconds = 240; // 4 minutes.
33 } // namespace
35 WebSocketTransportConnectJob::WebSocketTransportConnectJob(
36 const std::string& group_name,
37 RequestPriority priority,
38 const scoped_refptr<TransportSocketParams>& params,
39 TimeDelta timeout_duration,
40 const CompletionCallback& callback,
41 ClientSocketFactory* client_socket_factory,
42 HostResolver* host_resolver,
43 ClientSocketHandle* handle,
44 Delegate* delegate,
45 NetLog* pool_net_log,
46 const BoundNetLog& request_net_log)
47 : ConnectJob(group_name,
48 timeout_duration,
49 priority,
50 delegate,
51 BoundNetLog::Make(pool_net_log, NetLog::SOURCE_CONNECT_JOB)),
52 helper_(params, client_socket_factory, host_resolver, &connect_timing_),
53 race_result_(TransportConnectJobHelper::CONNECTION_LATENCY_UNKNOWN),
54 handle_(handle),
55 callback_(callback),
56 request_net_log_(request_net_log),
57 had_ipv4_(false),
58 had_ipv6_(false) {
59 helper_.SetOnIOComplete(this);
62 WebSocketTransportConnectJob::~WebSocketTransportConnectJob() {}
64 LoadState WebSocketTransportConnectJob::GetLoadState() const {
65 LoadState load_state = LOAD_STATE_RESOLVING_HOST;
66 if (ipv6_job_)
67 load_state = ipv6_job_->GetLoadState();
68 // This method should return LOAD_STATE_CONNECTING in preference to
69 // LOAD_STATE_WAITING_FOR_AVAILABLE_SOCKET when possible because "waiting for
70 // available socket" implies that nothing is happening.
71 if (ipv4_job_ && load_state != LOAD_STATE_CONNECTING)
72 load_state = ipv4_job_->GetLoadState();
73 return load_state;
76 int WebSocketTransportConnectJob::DoResolveHost() {
77 return helper_.DoResolveHost(priority(), net_log());
80 int WebSocketTransportConnectJob::DoResolveHostComplete(int result) {
81 return helper_.DoResolveHostComplete(result, net_log());
84 int WebSocketTransportConnectJob::DoTransportConnect() {
85 AddressList ipv4_addresses;
86 AddressList ipv6_addresses;
87 int result = ERR_UNEXPECTED;
88 helper_.set_next_state(
89 TransportConnectJobHelper::STATE_TRANSPORT_CONNECT_COMPLETE);
91 for (AddressList::const_iterator it = helper_.addresses().begin();
92 it != helper_.addresses().end();
93 ++it) {
94 switch (it->GetFamily()) {
95 case ADDRESS_FAMILY_IPV4:
96 ipv4_addresses.push_back(*it);
97 break;
99 case ADDRESS_FAMILY_IPV6:
100 ipv6_addresses.push_back(*it);
101 break;
103 default:
104 DVLOG(1) << "Unexpected ADDRESS_FAMILY: " << it->GetFamily();
105 break;
109 if (!ipv4_addresses.empty()) {
110 had_ipv4_ = true;
111 ipv4_job_.reset(new WebSocketTransportConnectSubJob(
112 ipv4_addresses, this, SUB_JOB_IPV4));
115 if (!ipv6_addresses.empty()) {
116 had_ipv6_ = true;
117 ipv6_job_.reset(new WebSocketTransportConnectSubJob(
118 ipv6_addresses, this, SUB_JOB_IPV6));
119 result = ipv6_job_->Start();
120 switch (result) {
121 case OK:
122 SetSocket(ipv6_job_->PassSocket());
123 race_result_ =
124 had_ipv4_
125 ? TransportConnectJobHelper::CONNECTION_LATENCY_IPV6_RACEABLE
126 : TransportConnectJobHelper::CONNECTION_LATENCY_IPV6_SOLO;
127 return result;
129 case ERR_IO_PENDING:
130 if (ipv4_job_) {
131 // This use of base::Unretained is safe because |fallback_timer_| is
132 // owned by this object.
133 fallback_timer_.Start(
134 FROM_HERE,
135 TimeDelta::FromMilliseconds(
136 TransportConnectJobHelper::kIPv6FallbackTimerInMs),
137 base::Bind(&WebSocketTransportConnectJob::StartIPv4JobAsync,
138 base::Unretained(this)));
140 return result;
142 default:
143 ipv6_job_.reset();
147 DCHECK(!ipv6_job_);
148 if (ipv4_job_) {
149 result = ipv4_job_->Start();
150 if (result == OK) {
151 SetSocket(ipv4_job_->PassSocket());
152 race_result_ =
153 had_ipv6_
154 ? TransportConnectJobHelper::CONNECTION_LATENCY_IPV4_WINS_RACE
155 : TransportConnectJobHelper::CONNECTION_LATENCY_IPV4_NO_RACE;
159 return result;
162 int WebSocketTransportConnectJob::DoTransportConnectComplete(int result) {
163 if (result == OK)
164 helper_.HistogramDuration(race_result_);
165 return result;
168 void WebSocketTransportConnectJob::OnSubJobComplete(
169 int result,
170 WebSocketTransportConnectSubJob* job) {
171 if (result == OK) {
172 switch (job->type()) {
173 case SUB_JOB_IPV4:
174 race_result_ =
175 had_ipv6_
176 ? TransportConnectJobHelper::CONNECTION_LATENCY_IPV4_WINS_RACE
177 : TransportConnectJobHelper::CONNECTION_LATENCY_IPV4_NO_RACE;
178 break;
180 case SUB_JOB_IPV6:
181 race_result_ =
182 had_ipv4_
183 ? TransportConnectJobHelper::CONNECTION_LATENCY_IPV6_RACEABLE
184 : TransportConnectJobHelper::CONNECTION_LATENCY_IPV6_SOLO;
185 break;
187 SetSocket(job->PassSocket());
189 // Make sure all connections are cancelled even if this object fails to be
190 // deleted.
191 ipv4_job_.reset();
192 ipv6_job_.reset();
193 } else {
194 switch (job->type()) {
195 case SUB_JOB_IPV4:
196 ipv4_job_.reset();
197 break;
199 case SUB_JOB_IPV6:
200 ipv6_job_.reset();
201 if (ipv4_job_ && !ipv4_job_->started()) {
202 fallback_timer_.Stop();
203 result = ipv4_job_->Start();
204 if (result != ERR_IO_PENDING) {
205 OnSubJobComplete(result, ipv4_job_.get());
206 return;
209 break;
211 if (ipv4_job_ || ipv6_job_)
212 return;
214 helper_.OnIOComplete(this, result);
217 void WebSocketTransportConnectJob::StartIPv4JobAsync() {
218 DCHECK(ipv4_job_);
219 int result = ipv4_job_->Start();
220 if (result != ERR_IO_PENDING)
221 OnSubJobComplete(result, ipv4_job_.get());
224 int WebSocketTransportConnectJob::ConnectInternal() {
225 return helper_.DoConnectInternal(this);
228 WebSocketTransportClientSocketPool::WebSocketTransportClientSocketPool(
229 int max_sockets,
230 int max_sockets_per_group,
231 ClientSocketPoolHistograms* histograms,
232 HostResolver* host_resolver,
233 ClientSocketFactory* client_socket_factory,
234 NetLog* net_log)
235 : TransportClientSocketPool(max_sockets,
236 max_sockets_per_group,
237 histograms,
238 host_resolver,
239 client_socket_factory,
240 net_log),
241 connect_job_delegate_(this),
242 histograms_(histograms),
243 pool_net_log_(net_log),
244 client_socket_factory_(client_socket_factory),
245 host_resolver_(host_resolver),
246 max_sockets_(max_sockets),
247 handed_out_socket_count_(0),
248 flushing_(false),
249 weak_factory_(this) {}
251 WebSocketTransportClientSocketPool::~WebSocketTransportClientSocketPool() {
252 // Clean up any pending connect jobs.
253 FlushWithError(ERR_ABORTED);
254 DCHECK(pending_connects_.empty());
255 DCHECK_EQ(0, handed_out_socket_count_);
256 DCHECK(stalled_request_queue_.empty());
257 DCHECK(stalled_request_map_.empty());
260 // static
261 void WebSocketTransportClientSocketPool::UnlockEndpoint(
262 ClientSocketHandle* handle) {
263 DCHECK(handle->is_initialized());
264 DCHECK(handle->socket());
265 IPEndPoint address;
266 if (handle->socket()->GetPeerAddress(&address) == OK)
267 WebSocketEndpointLockManager::GetInstance()->UnlockEndpoint(address);
270 int WebSocketTransportClientSocketPool::RequestSocket(
271 const std::string& group_name,
272 const void* params,
273 RequestPriority priority,
274 ClientSocketHandle* handle,
275 const CompletionCallback& callback,
276 const BoundNetLog& request_net_log) {
277 DCHECK(params);
278 const scoped_refptr<TransportSocketParams>& casted_params =
279 *static_cast<const scoped_refptr<TransportSocketParams>*>(params);
281 NetLogTcpClientSocketPoolRequestedSocket(request_net_log, &casted_params);
283 CHECK(!callback.is_null());
284 CHECK(handle);
286 request_net_log.BeginEvent(NetLog::TYPE_SOCKET_POOL);
288 if (ReachedMaxSocketsLimit() && !casted_params->ignore_limits()) {
289 request_net_log.AddEvent(NetLog::TYPE_SOCKET_POOL_STALLED_MAX_SOCKETS);
290 // TODO(ricea): Use emplace_back when C++11 becomes allowed.
291 StalledRequest request(
292 casted_params, priority, handle, callback, request_net_log);
293 stalled_request_queue_.push_back(request);
294 StalledRequestQueue::iterator iterator = stalled_request_queue_.end();
295 --iterator;
296 DCHECK_EQ(handle, iterator->handle);
297 // Because StalledRequestQueue is a std::list, its iterators are guaranteed
298 // to remain valid as long as the elements are not removed. As long as
299 // stalled_request_queue_ and stalled_request_map_ are updated in sync, it
300 // is safe to dereference an iterator in stalled_request_map_ to find the
301 // corresponding list element.
302 stalled_request_map_.insert(
303 StalledRequestMap::value_type(handle, iterator));
304 return ERR_IO_PENDING;
307 scoped_ptr<WebSocketTransportConnectJob> connect_job(
308 new WebSocketTransportConnectJob(group_name,
309 priority,
310 casted_params,
311 ConnectionTimeout(),
312 callback,
313 client_socket_factory_,
314 host_resolver_,
315 handle,
316 &connect_job_delegate_,
317 pool_net_log_,
318 request_net_log));
320 int rv = connect_job->Connect();
321 // Regardless of the outcome of |connect_job|, it will always be bound to
322 // |handle|, since this pool uses early-binding. So the binding is logged
323 // here, without waiting for the result.
324 request_net_log.AddEvent(
325 NetLog::TYPE_SOCKET_POOL_BOUND_TO_CONNECT_JOB,
326 connect_job->net_log().source().ToEventParametersCallback());
327 if (rv == OK) {
328 HandOutSocket(connect_job->PassSocket(),
329 connect_job->connect_timing(),
330 handle,
331 request_net_log);
332 request_net_log.EndEvent(NetLog::TYPE_SOCKET_POOL);
333 } else if (rv == ERR_IO_PENDING) {
334 // TODO(ricea): Implement backup job timer?
335 AddJob(handle, connect_job.Pass());
336 } else {
337 scoped_ptr<StreamSocket> error_socket;
338 connect_job->GetAdditionalErrorState(handle);
339 error_socket = connect_job->PassSocket();
340 if (error_socket) {
341 HandOutSocket(error_socket.Pass(),
342 connect_job->connect_timing(),
343 handle,
344 request_net_log);
348 if (rv != ERR_IO_PENDING) {
349 request_net_log.EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL, rv);
352 return rv;
355 void WebSocketTransportClientSocketPool::RequestSockets(
356 const std::string& group_name,
357 const void* params,
358 int num_sockets,
359 const BoundNetLog& net_log) {
360 NOTIMPLEMENTED();
363 void WebSocketTransportClientSocketPool::CancelRequest(
364 const std::string& group_name,
365 ClientSocketHandle* handle) {
366 DCHECK(!handle->is_initialized());
367 if (DeleteStalledRequest(handle))
368 return;
369 scoped_ptr<StreamSocket> socket = handle->PassSocket();
370 if (socket)
371 ReleaseSocket(handle->group_name(), socket.Pass(), handle->id());
372 if (!DeleteJob(handle))
373 pending_callbacks_.erase(handle);
374 if (!ReachedMaxSocketsLimit() && !stalled_request_queue_.empty())
375 ActivateStalledRequest();
378 void WebSocketTransportClientSocketPool::ReleaseSocket(
379 const std::string& group_name,
380 scoped_ptr<StreamSocket> socket,
381 int id) {
382 WebSocketEndpointLockManager::GetInstance()->UnlockSocket(socket.get());
383 CHECK_GT(handed_out_socket_count_, 0);
384 --handed_out_socket_count_;
385 if (!ReachedMaxSocketsLimit() && !stalled_request_queue_.empty())
386 ActivateStalledRequest();
389 void WebSocketTransportClientSocketPool::FlushWithError(int error) {
390 // Sockets which are in LOAD_STATE_CONNECTING are in danger of unlocking
391 // sockets waiting for the endpoint lock. If they connected synchronously,
392 // then OnConnectJobComplete(). The |flushing_| flag tells this object to
393 // ignore spurious calls to OnConnectJobComplete(). It is safe to ignore those
394 // calls because this method will delete the jobs and call their callbacks
395 // anyway.
396 flushing_ = true;
397 for (PendingConnectsMap::iterator it = pending_connects_.begin();
398 it != pending_connects_.end();
399 ++it) {
400 InvokeUserCallbackLater(
401 it->second->handle(), it->second->callback(), error);
402 delete it->second, it->second = NULL;
404 pending_connects_.clear();
405 for (StalledRequestQueue::iterator it = stalled_request_queue_.begin();
406 it != stalled_request_queue_.end();
407 ++it) {
408 InvokeUserCallbackLater(it->handle, it->callback, error);
410 stalled_request_map_.clear();
411 stalled_request_queue_.clear();
412 flushing_ = false;
415 void WebSocketTransportClientSocketPool::CloseIdleSockets() {
416 // We have no idle sockets.
419 int WebSocketTransportClientSocketPool::IdleSocketCount() const {
420 return 0;
423 int WebSocketTransportClientSocketPool::IdleSocketCountInGroup(
424 const std::string& group_name) const {
425 return 0;
428 LoadState WebSocketTransportClientSocketPool::GetLoadState(
429 const std::string& group_name,
430 const ClientSocketHandle* handle) const {
431 if (stalled_request_map_.find(handle) != stalled_request_map_.end())
432 return LOAD_STATE_WAITING_FOR_AVAILABLE_SOCKET;
433 if (pending_callbacks_.count(handle))
434 return LOAD_STATE_CONNECTING;
435 return LookupConnectJob(handle)->GetLoadState();
438 base::DictionaryValue* WebSocketTransportClientSocketPool::GetInfoAsValue(
439 const std::string& name,
440 const std::string& type,
441 bool include_nested_pools) const {
442 base::DictionaryValue* dict = new base::DictionaryValue();
443 dict->SetString("name", name);
444 dict->SetString("type", type);
445 dict->SetInteger("handed_out_socket_count", handed_out_socket_count_);
446 dict->SetInteger("connecting_socket_count", pending_connects_.size());
447 dict->SetInteger("idle_socket_count", 0);
448 dict->SetInteger("max_socket_count", max_sockets_);
449 dict->SetInteger("max_sockets_per_group", max_sockets_);
450 dict->SetInteger("pool_generation_number", 0);
451 return dict;
454 TimeDelta WebSocketTransportClientSocketPool::ConnectionTimeout() const {
455 return TimeDelta::FromSeconds(kTransportConnectJobTimeoutInSeconds);
458 ClientSocketPoolHistograms* WebSocketTransportClientSocketPool::histograms()
459 const {
460 return histograms_;
463 bool WebSocketTransportClientSocketPool::IsStalled() const {
464 return !stalled_request_queue_.empty();
467 void WebSocketTransportClientSocketPool::OnConnectJobComplete(
468 int result,
469 WebSocketTransportConnectJob* job) {
470 DCHECK_NE(ERR_IO_PENDING, result);
472 scoped_ptr<StreamSocket> socket = job->PassSocket();
474 // See comment in FlushWithError.
475 if (flushing_) {
476 WebSocketEndpointLockManager::GetInstance()->UnlockSocket(socket.get());
477 return;
480 BoundNetLog request_net_log = job->request_net_log();
481 CompletionCallback callback = job->callback();
482 LoadTimingInfo::ConnectTiming connect_timing = job->connect_timing();
484 ClientSocketHandle* const handle = job->handle();
485 bool handed_out_socket = false;
487 if (result == OK) {
488 DCHECK(socket.get());
489 handed_out_socket = true;
490 HandOutSocket(socket.Pass(), connect_timing, handle, request_net_log);
491 request_net_log.EndEvent(NetLog::TYPE_SOCKET_POOL);
492 } else {
493 // If we got a socket, it must contain error information so pass that
494 // up so that the caller can retrieve it.
495 job->GetAdditionalErrorState(handle);
496 if (socket.get()) {
497 handed_out_socket = true;
498 HandOutSocket(socket.Pass(), connect_timing, handle, request_net_log);
500 request_net_log.EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL, result);
502 bool delete_succeeded = DeleteJob(handle);
503 DCHECK(delete_succeeded);
504 if (!handed_out_socket && !stalled_request_queue_.empty() &&
505 !ReachedMaxSocketsLimit())
506 ActivateStalledRequest();
507 InvokeUserCallbackLater(handle, callback, result);
510 void WebSocketTransportClientSocketPool::InvokeUserCallbackLater(
511 ClientSocketHandle* handle,
512 const CompletionCallback& callback,
513 int rv) {
514 DCHECK(!pending_callbacks_.count(handle));
515 pending_callbacks_.insert(handle);
516 base::MessageLoop::current()->PostTask(
517 FROM_HERE,
518 base::Bind(&WebSocketTransportClientSocketPool::InvokeUserCallback,
519 weak_factory_.GetWeakPtr(),
520 handle,
521 callback,
522 rv));
525 void WebSocketTransportClientSocketPool::InvokeUserCallback(
526 ClientSocketHandle* handle,
527 const CompletionCallback& callback,
528 int rv) {
529 if (pending_callbacks_.erase(handle))
530 callback.Run(rv);
533 bool WebSocketTransportClientSocketPool::ReachedMaxSocketsLimit() const {
534 return handed_out_socket_count_ >= max_sockets_ ||
535 base::checked_cast<int>(pending_connects_.size()) >=
536 max_sockets_ - handed_out_socket_count_;
539 void WebSocketTransportClientSocketPool::HandOutSocket(
540 scoped_ptr<StreamSocket> socket,
541 const LoadTimingInfo::ConnectTiming& connect_timing,
542 ClientSocketHandle* handle,
543 const BoundNetLog& net_log) {
544 DCHECK(socket);
545 handle->SetSocket(socket.Pass());
546 DCHECK_EQ(ClientSocketHandle::UNUSED, handle->reuse_type());
547 DCHECK_EQ(0, handle->idle_time().InMicroseconds());
548 handle->set_pool_id(0);
549 handle->set_connect_timing(connect_timing);
551 net_log.AddEvent(
552 NetLog::TYPE_SOCKET_POOL_BOUND_TO_SOCKET,
553 handle->socket()->NetLog().source().ToEventParametersCallback());
555 ++handed_out_socket_count_;
558 void WebSocketTransportClientSocketPool::AddJob(
559 ClientSocketHandle* handle,
560 scoped_ptr<WebSocketTransportConnectJob> connect_job) {
561 bool inserted =
562 pending_connects_.insert(PendingConnectsMap::value_type(
563 handle, connect_job.release())).second;
564 DCHECK(inserted);
567 bool WebSocketTransportClientSocketPool::DeleteJob(ClientSocketHandle* handle) {
568 PendingConnectsMap::iterator it = pending_connects_.find(handle);
569 if (it == pending_connects_.end())
570 return false;
571 // Deleting a ConnectJob which holds an endpoint lock can lead to a different
572 // ConnectJob proceeding to connect. If the connect proceeds synchronously
573 // (usually because of a failure) then it can trigger that job to be
574 // deleted. |it| remains valid because std::map guarantees that erase() does
575 // not invalid iterators to other entries.
576 delete it->second, it->second = NULL;
577 DCHECK(pending_connects_.find(handle) == it);
578 pending_connects_.erase(it);
579 return true;
582 const WebSocketTransportConnectJob*
583 WebSocketTransportClientSocketPool::LookupConnectJob(
584 const ClientSocketHandle* handle) const {
585 PendingConnectsMap::const_iterator it = pending_connects_.find(handle);
586 CHECK(it != pending_connects_.end());
587 return it->second;
590 void WebSocketTransportClientSocketPool::ActivateStalledRequest() {
591 DCHECK(!stalled_request_queue_.empty());
592 DCHECK(!ReachedMaxSocketsLimit());
593 // Usually we will only be able to activate one stalled request at a time,
594 // however if all the connects fail synchronously for some reason, we may be
595 // able to clear the whole queue at once.
596 while (!stalled_request_queue_.empty() && !ReachedMaxSocketsLimit()) {
597 StalledRequest request(stalled_request_queue_.front());
598 stalled_request_queue_.pop_front();
599 stalled_request_map_.erase(request.handle);
600 int rv = RequestSocket("ignored",
601 &request.params,
602 request.priority,
603 request.handle,
604 request.callback,
605 request.net_log);
606 // ActivateStalledRequest() never returns synchronously, so it is never
607 // called re-entrantly.
608 if (rv != ERR_IO_PENDING)
609 InvokeUserCallbackLater(request.handle, request.callback, rv);
613 bool WebSocketTransportClientSocketPool::DeleteStalledRequest(
614 ClientSocketHandle* handle) {
615 StalledRequestMap::iterator it = stalled_request_map_.find(handle);
616 if (it == stalled_request_map_.end())
617 return false;
618 stalled_request_queue_.erase(it->second);
619 stalled_request_map_.erase(it);
620 return true;
623 WebSocketTransportClientSocketPool::ConnectJobDelegate::ConnectJobDelegate(
624 WebSocketTransportClientSocketPool* owner)
625 : owner_(owner) {}
627 WebSocketTransportClientSocketPool::ConnectJobDelegate::~ConnectJobDelegate() {}
629 void
630 WebSocketTransportClientSocketPool::ConnectJobDelegate::OnConnectJobComplete(
631 int result,
632 ConnectJob* job) {
633 owner_->OnConnectJobComplete(result,
634 static_cast<WebSocketTransportConnectJob*>(job));
637 WebSocketTransportClientSocketPool::StalledRequest::StalledRequest(
638 const scoped_refptr<TransportSocketParams>& params,
639 RequestPriority priority,
640 ClientSocketHandle* handle,
641 const CompletionCallback& callback,
642 const BoundNetLog& net_log)
643 : params(params),
644 priority(priority),
645 handle(handle),
646 callback(callback),
647 net_log(net_log) {}
649 WebSocketTransportClientSocketPool::StalledRequest::~StalledRequest() {}
651 } // namespace net