net/websockets/websocket_throttle.cc

   1 // Copyright (c) 2012 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/websockets/websocket_throttle.h"
   6
   7 #include <string>
   8
   9 #include "base/hash_tables.h"
  10 #include "base/memory/ref_counted.h"
  11 #include "base/memory/singleton.h"
  12 #include "base/message_loop.h"
  13 #include "base/string_number_conversions.h"
  14 #include "base/string_util.h"
  15 #include "base/stringprintf.h"
  16 #include "net/base/io_buffer.h"
  17 #include "net/socket_stream/socket_stream.h"
  18 #include "net/websockets/websocket_job.h"
  19
  20 namespace net {
  21
  22 static std::string IPEndPointToHashkey(const IPEndPoint& endpoint) {
  23   return base::StringPrintf("%d:%s",
  24                             endpoint.GetFamily(),
  25                             base::HexEncode(&endpoint.address()[0],
  26                                             endpoint.address().size()).c_str());
  27 }
  28
  29 WebSocketThrottle::WebSocketThrottle() {
  30 }
  31
  32 WebSocketThrottle::~WebSocketThrottle() {
  33   DCHECK(queue_.empty());
  34   DCHECK(addr_map_.empty());
  35 }
  36
  37 // static
  38 WebSocketThrottle* WebSocketThrottle::GetInstance() {
  39   return Singleton<WebSocketThrottle>::get();
  40 }
  41
  42 void WebSocketThrottle::PutInQueue(WebSocketJob* job) {
  43   queue_.push_back(job);
  44   const AddressList& address_list = job->address_list();
  45   base::hash_set<std::string> address_set;
  46   for (AddressList::const_iterator addr_iter = address_list.begin();
  47        addr_iter != address_list.end();
  48        ++addr_iter) {
  49     std::string addrkey = IPEndPointToHashkey(*addr_iter);
  50
  51     // If |addrkey| is already processed, don't do it again.
  52     if (address_set.find(addrkey) != address_set.end())
  53       continue;
  54     address_set.insert(addrkey);
  55
  56     ConnectingAddressMap::iterator iter = addr_map_.find(addrkey);
  57     if (iter == addr_map_.end()) {
  58       ConnectingQueue* queue = new ConnectingQueue();
  59       queue->push_back(job);
  60       addr_map_[addrkey] = queue;
  61     } else {
  62       iter->second->push_back(job);
  63       job->SetWaiting();
  64       DVLOG(1) << "Waiting on " << addrkey;
  65     }
  66   }
  67 }
  68
  69 void WebSocketThrottle::RemoveFromQueue(WebSocketJob* job) {
  70   bool in_queue = false;
  71   for (ConnectingQueue::iterator iter = queue_.begin();
  72        iter != queue_.end();
  73        ++iter) {
  74     if (*iter == job) {
  75       queue_.erase(iter);
  76       in_queue = true;
  77       break;
  78     }
  79   }
  80   if (!in_queue)
  81     return;
  82   const AddressList& address_list = job->address_list();
  83   base::hash_set<std::string> address_set;
  84   for (AddressList::const_iterator addr_iter = address_list.begin();
  85        addr_iter != address_list.end();
  86        ++addr_iter) {
  87     std::string addrkey = IPEndPointToHashkey(*addr_iter);
  88     // If |addrkey| is already processed, don't do it again.
  89     if (address_set.find(addrkey) != address_set.end())
  90       continue;
  91     address_set.insert(addrkey);
  92
  93     ConnectingAddressMap::iterator iter = addr_map_.find(addrkey);
  94     DCHECK(iter != addr_map_.end());
  95
  96     ConnectingQueue* queue = iter->second;
  97     // Job may not be front of queue when job is closed early while waiting.
  98     for (ConnectingQueue::iterator iter = queue->begin();
  99          iter != queue->end();
 100          ++iter) {
 101       if (*iter == job) {
 102         queue->erase(iter);
 103         break;
 104       }
 105     }
 106     if (queue->empty()) {
 107       delete queue;
 108       addr_map_.erase(iter);
 109     }
 110   }
 111 }
 112
 113 void WebSocketThrottle::WakeupSocketIfNecessary() {
 114   for (ConnectingQueue::iterator iter = queue_.begin();
 115        iter != queue_.end();
 116        ++iter) {
 117     WebSocketJob* job = *iter;
 118     if (!job->IsWaiting())
 119       continue;
 120
 121     bool should_wakeup = true;
 122     const AddressList& address_list = job->address_list();
 123     for (AddressList::const_iterator addr_iter = address_list.begin();
 124          addr_iter != address_list.end();
 125          ++addr_iter) {
 126       std::string addrkey = IPEndPointToHashkey(*addr_iter);
 127       ConnectingAddressMap::iterator iter = addr_map_.find(addrkey);
 128       DCHECK(iter != addr_map_.end());
 129       ConnectingQueue* queue = iter->second;
 130       if (job != queue->front()) {
 131         should_wakeup = false;
 132         break;
 133       }
 134     }
 135     if (should_wakeup)
 136       job->Wakeup();
 137   }
 138 }
 139
 140 }  // namespace net