We started redesigning GpuMemoryBuffer interface to handle multiple buffers [0].
[chromium-blink-merge.git] / net / socket / websocket_transport_client_socket_pool.cc
blobce4336800385f20349de982ec31d4312fe8270a1
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/log/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 HostResolver* host_resolver,
232 ClientSocketFactory* client_socket_factory,
233 NetLog* net_log)
234 : TransportClientSocketPool(max_sockets,
235 max_sockets_per_group,
236 host_resolver,
237 client_socket_factory,
238 net_log),
239 connect_job_delegate_(this),
240 pool_net_log_(net_log),
241 client_socket_factory_(client_socket_factory),
242 host_resolver_(host_resolver),
243 max_sockets_(max_sockets),
244 handed_out_socket_count_(0),
245 flushing_(false),
246 weak_factory_(this) {}
248 WebSocketTransportClientSocketPool::~WebSocketTransportClientSocketPool() {
249 // Clean up any pending connect jobs.
250 FlushWithError(ERR_ABORTED);
251 DCHECK(pending_connects_.empty());
252 DCHECK_EQ(0, handed_out_socket_count_);
253 DCHECK(stalled_request_queue_.empty());
254 DCHECK(stalled_request_map_.empty());
257 // static
258 void WebSocketTransportClientSocketPool::UnlockEndpoint(
259 ClientSocketHandle* handle) {
260 DCHECK(handle->is_initialized());
261 DCHECK(handle->socket());
262 IPEndPoint address;
263 if (handle->socket()->GetPeerAddress(&address) == OK)
264 WebSocketEndpointLockManager::GetInstance()->UnlockEndpoint(address);
267 int WebSocketTransportClientSocketPool::RequestSocket(
268 const std::string& group_name,
269 const void* params,
270 RequestPriority priority,
271 ClientSocketHandle* handle,
272 const CompletionCallback& callback,
273 const BoundNetLog& request_net_log) {
274 DCHECK(params);
275 const scoped_refptr<TransportSocketParams>& casted_params =
276 *static_cast<const scoped_refptr<TransportSocketParams>*>(params);
278 NetLogTcpClientSocketPoolRequestedSocket(request_net_log, &casted_params);
280 CHECK(!callback.is_null());
281 CHECK(handle);
283 request_net_log.BeginEvent(NetLog::TYPE_SOCKET_POOL);
285 if (ReachedMaxSocketsLimit() && !casted_params->ignore_limits()) {
286 request_net_log.AddEvent(NetLog::TYPE_SOCKET_POOL_STALLED_MAX_SOCKETS);
287 // TODO(ricea): Use emplace_back when C++11 becomes allowed.
288 StalledRequest request(
289 casted_params, priority, handle, callback, request_net_log);
290 stalled_request_queue_.push_back(request);
291 StalledRequestQueue::iterator iterator = stalled_request_queue_.end();
292 --iterator;
293 DCHECK_EQ(handle, iterator->handle);
294 // Because StalledRequestQueue is a std::list, its iterators are guaranteed
295 // to remain valid as long as the elements are not removed. As long as
296 // stalled_request_queue_ and stalled_request_map_ are updated in sync, it
297 // is safe to dereference an iterator in stalled_request_map_ to find the
298 // corresponding list element.
299 stalled_request_map_.insert(
300 StalledRequestMap::value_type(handle, iterator));
301 return ERR_IO_PENDING;
304 scoped_ptr<WebSocketTransportConnectJob> connect_job(
305 new WebSocketTransportConnectJob(group_name,
306 priority,
307 casted_params,
308 ConnectionTimeout(),
309 callback,
310 client_socket_factory_,
311 host_resolver_,
312 handle,
313 &connect_job_delegate_,
314 pool_net_log_,
315 request_net_log));
317 int rv = connect_job->Connect();
318 // Regardless of the outcome of |connect_job|, it will always be bound to
319 // |handle|, since this pool uses early-binding. So the binding is logged
320 // here, without waiting for the result.
321 request_net_log.AddEvent(
322 NetLog::TYPE_SOCKET_POOL_BOUND_TO_CONNECT_JOB,
323 connect_job->net_log().source().ToEventParametersCallback());
324 if (rv == OK) {
325 HandOutSocket(connect_job->PassSocket(),
326 connect_job->connect_timing(),
327 handle,
328 request_net_log);
329 request_net_log.EndEvent(NetLog::TYPE_SOCKET_POOL);
330 } else if (rv == ERR_IO_PENDING) {
331 // TODO(ricea): Implement backup job timer?
332 AddJob(handle, connect_job.Pass());
333 } else {
334 scoped_ptr<StreamSocket> error_socket;
335 connect_job->GetAdditionalErrorState(handle);
336 error_socket = connect_job->PassSocket();
337 if (error_socket) {
338 HandOutSocket(error_socket.Pass(),
339 connect_job->connect_timing(),
340 handle,
341 request_net_log);
345 if (rv != ERR_IO_PENDING) {
346 request_net_log.EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL, rv);
349 return rv;
352 void WebSocketTransportClientSocketPool::RequestSockets(
353 const std::string& group_name,
354 const void* params,
355 int num_sockets,
356 const BoundNetLog& net_log) {
357 NOTIMPLEMENTED();
360 void WebSocketTransportClientSocketPool::CancelRequest(
361 const std::string& group_name,
362 ClientSocketHandle* handle) {
363 DCHECK(!handle->is_initialized());
364 if (DeleteStalledRequest(handle))
365 return;
366 scoped_ptr<StreamSocket> socket = handle->PassSocket();
367 if (socket)
368 ReleaseSocket(handle->group_name(), socket.Pass(), handle->id());
369 if (!DeleteJob(handle))
370 pending_callbacks_.erase(handle);
371 if (!ReachedMaxSocketsLimit() && !stalled_request_queue_.empty())
372 ActivateStalledRequest();
375 void WebSocketTransportClientSocketPool::ReleaseSocket(
376 const std::string& group_name,
377 scoped_ptr<StreamSocket> socket,
378 int id) {
379 WebSocketEndpointLockManager::GetInstance()->UnlockSocket(socket.get());
380 CHECK_GT(handed_out_socket_count_, 0);
381 --handed_out_socket_count_;
382 if (!ReachedMaxSocketsLimit() && !stalled_request_queue_.empty())
383 ActivateStalledRequest();
386 void WebSocketTransportClientSocketPool::FlushWithError(int error) {
387 // Sockets which are in LOAD_STATE_CONNECTING are in danger of unlocking
388 // sockets waiting for the endpoint lock. If they connected synchronously,
389 // then OnConnectJobComplete(). The |flushing_| flag tells this object to
390 // ignore spurious calls to OnConnectJobComplete(). It is safe to ignore those
391 // calls because this method will delete the jobs and call their callbacks
392 // anyway.
393 flushing_ = true;
394 for (PendingConnectsMap::iterator it = pending_connects_.begin();
395 it != pending_connects_.end();
396 ++it) {
397 InvokeUserCallbackLater(
398 it->second->handle(), it->second->callback(), error);
399 delete it->second, it->second = NULL;
401 pending_connects_.clear();
402 for (StalledRequestQueue::iterator it = stalled_request_queue_.begin();
403 it != stalled_request_queue_.end();
404 ++it) {
405 InvokeUserCallbackLater(it->handle, it->callback, error);
407 stalled_request_map_.clear();
408 stalled_request_queue_.clear();
409 flushing_ = false;
412 void WebSocketTransportClientSocketPool::CloseIdleSockets() {
413 // We have no idle sockets.
416 int WebSocketTransportClientSocketPool::IdleSocketCount() const {
417 return 0;
420 int WebSocketTransportClientSocketPool::IdleSocketCountInGroup(
421 const std::string& group_name) const {
422 return 0;
425 LoadState WebSocketTransportClientSocketPool::GetLoadState(
426 const std::string& group_name,
427 const ClientSocketHandle* handle) const {
428 if (stalled_request_map_.find(handle) != stalled_request_map_.end())
429 return LOAD_STATE_WAITING_FOR_AVAILABLE_SOCKET;
430 if (pending_callbacks_.count(handle))
431 return LOAD_STATE_CONNECTING;
432 return LookupConnectJob(handle)->GetLoadState();
435 base::DictionaryValue* WebSocketTransportClientSocketPool::GetInfoAsValue(
436 const std::string& name,
437 const std::string& type,
438 bool include_nested_pools) const {
439 base::DictionaryValue* dict = new base::DictionaryValue();
440 dict->SetString("name", name);
441 dict->SetString("type", type);
442 dict->SetInteger("handed_out_socket_count", handed_out_socket_count_);
443 dict->SetInteger("connecting_socket_count", pending_connects_.size());
444 dict->SetInteger("idle_socket_count", 0);
445 dict->SetInteger("max_socket_count", max_sockets_);
446 dict->SetInteger("max_sockets_per_group", max_sockets_);
447 dict->SetInteger("pool_generation_number", 0);
448 return dict;
451 TimeDelta WebSocketTransportClientSocketPool::ConnectionTimeout() const {
452 return TimeDelta::FromSeconds(kTransportConnectJobTimeoutInSeconds);
455 bool WebSocketTransportClientSocketPool::IsStalled() const {
456 return !stalled_request_queue_.empty();
459 void WebSocketTransportClientSocketPool::OnConnectJobComplete(
460 int result,
461 WebSocketTransportConnectJob* job) {
462 DCHECK_NE(ERR_IO_PENDING, result);
464 scoped_ptr<StreamSocket> socket = job->PassSocket();
466 // See comment in FlushWithError.
467 if (flushing_) {
468 WebSocketEndpointLockManager::GetInstance()->UnlockSocket(socket.get());
469 return;
472 BoundNetLog request_net_log = job->request_net_log();
473 CompletionCallback callback = job->callback();
474 LoadTimingInfo::ConnectTiming connect_timing = job->connect_timing();
476 ClientSocketHandle* const handle = job->handle();
477 bool handed_out_socket = false;
479 if (result == OK) {
480 DCHECK(socket.get());
481 handed_out_socket = true;
482 HandOutSocket(socket.Pass(), connect_timing, handle, request_net_log);
483 request_net_log.EndEvent(NetLog::TYPE_SOCKET_POOL);
484 } else {
485 // If we got a socket, it must contain error information so pass that
486 // up so that the caller can retrieve it.
487 job->GetAdditionalErrorState(handle);
488 if (socket.get()) {
489 handed_out_socket = true;
490 HandOutSocket(socket.Pass(), connect_timing, handle, request_net_log);
492 request_net_log.EndEventWithNetErrorCode(NetLog::TYPE_SOCKET_POOL, result);
494 bool delete_succeeded = DeleteJob(handle);
495 DCHECK(delete_succeeded);
496 if (!handed_out_socket && !stalled_request_queue_.empty() &&
497 !ReachedMaxSocketsLimit())
498 ActivateStalledRequest();
499 InvokeUserCallbackLater(handle, callback, result);
502 void WebSocketTransportClientSocketPool::InvokeUserCallbackLater(
503 ClientSocketHandle* handle,
504 const CompletionCallback& callback,
505 int rv) {
506 DCHECK(!pending_callbacks_.count(handle));
507 pending_callbacks_.insert(handle);
508 base::MessageLoop::current()->PostTask(
509 FROM_HERE,
510 base::Bind(&WebSocketTransportClientSocketPool::InvokeUserCallback,
511 weak_factory_.GetWeakPtr(),
512 handle,
513 callback,
514 rv));
517 void WebSocketTransportClientSocketPool::InvokeUserCallback(
518 ClientSocketHandle* handle,
519 const CompletionCallback& callback,
520 int rv) {
521 if (pending_callbacks_.erase(handle))
522 callback.Run(rv);
525 bool WebSocketTransportClientSocketPool::ReachedMaxSocketsLimit() const {
526 return handed_out_socket_count_ >= max_sockets_ ||
527 base::checked_cast<int>(pending_connects_.size()) >=
528 max_sockets_ - handed_out_socket_count_;
531 void WebSocketTransportClientSocketPool::HandOutSocket(
532 scoped_ptr<StreamSocket> socket,
533 const LoadTimingInfo::ConnectTiming& connect_timing,
534 ClientSocketHandle* handle,
535 const BoundNetLog& net_log) {
536 DCHECK(socket);
537 handle->SetSocket(socket.Pass());
538 DCHECK_EQ(ClientSocketHandle::UNUSED, handle->reuse_type());
539 DCHECK_EQ(0, handle->idle_time().InMicroseconds());
540 handle->set_pool_id(0);
541 handle->set_connect_timing(connect_timing);
543 net_log.AddEvent(
544 NetLog::TYPE_SOCKET_POOL_BOUND_TO_SOCKET,
545 handle->socket()->NetLog().source().ToEventParametersCallback());
547 ++handed_out_socket_count_;
550 void WebSocketTransportClientSocketPool::AddJob(
551 ClientSocketHandle* handle,
552 scoped_ptr<WebSocketTransportConnectJob> connect_job) {
553 bool inserted =
554 pending_connects_.insert(PendingConnectsMap::value_type(
555 handle, connect_job.release())).second;
556 DCHECK(inserted);
559 bool WebSocketTransportClientSocketPool::DeleteJob(ClientSocketHandle* handle) {
560 PendingConnectsMap::iterator it = pending_connects_.find(handle);
561 if (it == pending_connects_.end())
562 return false;
563 // Deleting a ConnectJob which holds an endpoint lock can lead to a different
564 // ConnectJob proceeding to connect. If the connect proceeds synchronously
565 // (usually because of a failure) then it can trigger that job to be
566 // deleted. |it| remains valid because std::map guarantees that erase() does
567 // not invalid iterators to other entries.
568 delete it->second, it->second = NULL;
569 DCHECK(pending_connects_.find(handle) == it);
570 pending_connects_.erase(it);
571 return true;
574 const WebSocketTransportConnectJob*
575 WebSocketTransportClientSocketPool::LookupConnectJob(
576 const ClientSocketHandle* handle) const {
577 PendingConnectsMap::const_iterator it = pending_connects_.find(handle);
578 CHECK(it != pending_connects_.end());
579 return it->second;
582 void WebSocketTransportClientSocketPool::ActivateStalledRequest() {
583 DCHECK(!stalled_request_queue_.empty());
584 DCHECK(!ReachedMaxSocketsLimit());
585 // Usually we will only be able to activate one stalled request at a time,
586 // however if all the connects fail synchronously for some reason, we may be
587 // able to clear the whole queue at once.
588 while (!stalled_request_queue_.empty() && !ReachedMaxSocketsLimit()) {
589 StalledRequest request(stalled_request_queue_.front());
590 stalled_request_queue_.pop_front();
591 stalled_request_map_.erase(request.handle);
592 int rv = RequestSocket("ignored",
593 &request.params,
594 request.priority,
595 request.handle,
596 request.callback,
597 request.net_log);
598 // ActivateStalledRequest() never returns synchronously, so it is never
599 // called re-entrantly.
600 if (rv != ERR_IO_PENDING)
601 InvokeUserCallbackLater(request.handle, request.callback, rv);
605 bool WebSocketTransportClientSocketPool::DeleteStalledRequest(
606 ClientSocketHandle* handle) {
607 StalledRequestMap::iterator it = stalled_request_map_.find(handle);
608 if (it == stalled_request_map_.end())
609 return false;
610 stalled_request_queue_.erase(it->second);
611 stalled_request_map_.erase(it);
612 return true;
615 WebSocketTransportClientSocketPool::ConnectJobDelegate::ConnectJobDelegate(
616 WebSocketTransportClientSocketPool* owner)
617 : owner_(owner) {}
619 WebSocketTransportClientSocketPool::ConnectJobDelegate::~ConnectJobDelegate() {}
621 void
622 WebSocketTransportClientSocketPool::ConnectJobDelegate::OnConnectJobComplete(
623 int result,
624 ConnectJob* job) {
625 owner_->OnConnectJobComplete(result,
626 static_cast<WebSocketTransportConnectJob*>(job));
629 WebSocketTransportClientSocketPool::StalledRequest::StalledRequest(
630 const scoped_refptr<TransportSocketParams>& params,
631 RequestPriority priority,
632 ClientSocketHandle* handle,
633 const CompletionCallback& callback,
634 const BoundNetLog& net_log)
635 : params(params),
636 priority(priority),
637 handle(handle),
638 callback(callback),
639 net_log(net_log) {}
641 WebSocketTransportClientSocketPool::StalledRequest::~StalledRequest() {}
643 } // namespace net