Include destination port for websocket throttling.

Before this change, a pending connection to 127.0.0.1:80 would block a
connection to 127.0.0.1:81 from starting. With this change, they are considered
to be distinct for throttling purposes.

Added a unit test to verify that pending connections to distinct ports do not
block each other.

Took the opportunity to simplify WebSocketThrottle a bit, using the IPEndPoint
as the map key directly instead of creating a string to be the key.

Modified the WebSocketJobSpdy[23]Tests to use port 80 both for the blocking
connection and the connection being throttled. Previously the blocking
connection used port 0, which meant that this change caused the throttling tests
to fail.

TEST=net_unittests, chrome

BUG=172202


Review URL: https://chromiumcodereview.appspot.com/12033072

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@181448 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 93 insertions, 57 deletions

diff --git a/net/websockets/websocket_job_spdy2_unittest.cc b/net/websockets/websocket_job_spdy2_unittest.cc
index 6af4df6..4154be4 100644
--- a/net/websockets/websocket_job_spdy2_unittest.cc
+++ b/net/websockets/websocket_job_spdy2_unittest.cc
@@ -406,9 +406,13 @@ class WebSocketJobSpdy2Test : public PlatformTest {
 
     websocket_->InitSocketStream(socket_.get());
     websocket_->set_context(context_.get());
+    // MockHostResolver resolves all hosts to 127.0.0.1; however, when we create
+    // a WebSocketJob purely to block another one in a throttling test, we don't
+    // perform a real connect. In that case, the following address is used
+    // instead.
     IPAddressNumber ip;
     ParseIPLiteralToNumber("127.0.0.1", &ip);
-    websocket_->addresses_ = AddressList::CreateFromIPAddress(ip, 0);
+    websocket_->addresses_ = AddressList::CreateFromIPAddress(ip, 80);
   }
   void SkipToConnecting() {
     websocket_->state_ = WebSocketJob::CONNECTING;
diff --git a/net/websockets/websocket_job_spdy3_unittest.cc b/net/websockets/websocket_job_spdy3_unittest.cc
index cfaf515..3777c5b 100644
--- a/net/websockets/websocket_job_spdy3_unittest.cc
+++ b/net/websockets/websocket_job_spdy3_unittest.cc
@@ -406,9 +406,13 @@ class WebSocketJobSpdy3Test : public PlatformTest {
 
     websocket_->InitSocketStream(socket_.get());
     websocket_->set_context(context_.get());
+    // MockHostResolver resolves all hosts to 127.0.0.1; however, when we create
+    // a WebSocketJob purely to block another one in a throttling test, we don't
+    // perform a real connect. In that case, the following address is used
+    // instead.
     IPAddressNumber ip;
     ParseIPLiteralToNumber("127.0.0.1", &ip);
-    websocket_->addresses_ = AddressList::CreateFromIPAddress(ip, 0);
+    websocket_->addresses_ = AddressList::CreateFromIPAddress(ip, 80);
   }
   void SkipToConnecting() {
     websocket_->state_ = WebSocketJob::CONNECTING;
diff --git a/net/websockets/websocket_throttle.cc b/net/websockets/websocket_throttle.cc
index 61b4086..3b3a6ce 100644
--- a/net/websockets/websocket_throttle.cc
+++ b/net/websockets/websocket_throttle.cc
@@ -4,10 +4,10 @@
 
 #include "net/websockets/websocket_throttle.h"
 
+#include <algorithm>
+#include <set>
 #include <string>
 
-#include "base/hash_tables.h"
-#include "base/memory/ref_counted.h"
 #include "base/memory/singleton.h"
 #include "base/message_loop.h"
 #include "base/string_number_conversions.h"
@@ -19,13 +19,6 @@
 
 namespace net {
 
-static std::string IPEndPointToHashkey(const IPEndPoint& endpoint) {
-  return base::StringPrintf("%d:%s",
-                            endpoint.GetFamily(),
-                            base::HexEncode(&endpoint.address()[0],
-                                            endpoint.address().size()).c_str());
-}
-
 WebSocketThrottle::WebSocketThrottle() {
 }
 
@@ -42,70 +35,56 @@ WebSocketThrottle* WebSocketThrottle::GetInstance() {
 void WebSocketThrottle::PutInQueue(WebSocketJob* job) {
   queue_.push_back(job);
   const AddressList& address_list = job->address_list();
-  base::hash_set<std::string> address_set;
+  std::set<IPEndPoint> address_set;
   for (AddressList::const_iterator addr_iter = address_list.begin();
        addr_iter != address_list.end();
        ++addr_iter) {
-    std::string addrkey = IPEndPointToHashkey(*addr_iter);
-
-    // If |addrkey| is already processed, don't do it again.
-    if (address_set.find(addrkey) != address_set.end())
+    const IPEndPoint& address = *addr_iter;
+    // If |address| is already processed, don't do it again.
+    if (!address_set.insert(address).second)
       continue;
-    address_set.insert(addrkey);
 
-    ConnectingAddressMap::iterator iter = addr_map_.find(addrkey);
+    ConnectingAddressMap::iterator iter = addr_map_.find(address);
     if (iter == addr_map_.end()) {
       ConnectingQueue* queue = new ConnectingQueue();
       queue->push_back(job);
-      addr_map_[addrkey] = queue;
+      addr_map_[address] = queue;
     } else {
       iter->second->push_back(job);
       job->SetWaiting();
-      DVLOG(1) << "Waiting on " << addrkey;
+      DVLOG(1) << "Waiting on " << address.ToString();
     }
   }
 }
 
 void WebSocketThrottle::RemoveFromQueue(WebSocketJob* job) {
-  bool in_queue = false;
-  for (ConnectingQueue::iterator iter = queue_.begin();
-       iter != queue_.end();
-       ++iter) {
-    if (*iter == job) {
-      queue_.erase(iter);
-      in_queue = true;
-      break;
-    }
-  }
-  if (!in_queue)
+  ConnectingQueue::iterator queue_iter =
+      std::find(queue_.begin(), queue_.end(), job);
+  if (queue_iter == queue_.end())
     return;
+  queue_.erase(queue_iter);
   const AddressList& address_list = job->address_list();
-  base::hash_set<std::string> address_set;
+  std::set<IPEndPoint> address_set;
   for (AddressList::const_iterator addr_iter = address_list.begin();
        addr_iter != address_list.end();
        ++addr_iter) {
-    std::string addrkey = IPEndPointToHashkey(*addr_iter);
-    // If |addrkey| is already processed, don't do it again.
-    if (address_set.find(addrkey) != address_set.end())
+    const IPEndPoint& address = *addr_iter;
+    // If |address| is already processed, don't do it again.
+    if (!address_set.insert(address).second)
       continue;
-    address_set.insert(addrkey);
 
-    ConnectingAddressMap::iterator iter = addr_map_.find(addrkey);
-    DCHECK(iter != addr_map_.end());
+    ConnectingAddressMap::iterator map_iter = addr_map_.find(address);
+    DCHECK(map_iter != addr_map_.end());
 
-    ConnectingQueue* queue = iter->second;
-    // Job may not be front of queue when job is closed early while waiting.
-    for (ConnectingQueue::iterator iter = queue->begin();
-         iter != queue->end();
-         ++iter) {
-      if (*iter == job) {
-        queue->erase(iter);
-        break;
-      }
-    }
+    ConnectingQueue* queue = map_iter->second;
+    // Job may not be front of queue if the socket is closed while waiting.
+    ConnectingQueue::iterator address_queue_iter =
+        std::find(queue->begin(), queue->end(), job);
+    if (address_queue_iter != queue->end())
+      queue->erase(address_queue_iter);
     if (queue->empty()) {
       delete queue;
-      addr_map_.erase(iter);
+      addr_map_.erase(map_iter);
     }
   }
 }
@@ -123,10 +102,10 @@ void WebSocketThrottle::WakeupSocketIfNecessary() {
     for (AddressList::const_iterator addr_iter = address_list.begin();
          addr_iter != address_list.end();
          ++addr_iter) {
-      std::string addrkey = IPEndPointToHashkey(*addr_iter);
-      ConnectingAddressMap::iterator iter = addr_map_.find(addrkey);
-      DCHECK(iter != addr_map_.end());
-      ConnectingQueue* queue = iter->second;
+      const IPEndPoint& address = *addr_iter;
+      ConnectingAddressMap::iterator map_iter = addr_map_.find(address);
+      DCHECK(map_iter != addr_map_.end());
+      ConnectingQueue* queue = map_iter->second;
       if (job != queue->front()) {
         should_wakeup = false;
         break;
diff --git a/net/websockets/websocket_throttle.h b/net/websockets/websocket_throttle.h
index 1c64658..713e8f8 100644
--- a/net/websockets/websocket_throttle.h
+++ b/net/websockets/websocket_throttle.h
@@ -6,9 +6,10 @@
 #define NET_WEBSOCKETS_WEBSOCKET_THROTTLE_H_
 
 #include <deque>
+#include <map>
 #include <string>
 
-#include "base/hash_tables.h"
+#include "net/base/ip_endpoint.h"
 #include "net/base/net_export.h"
 
 template <typename T> struct DefaultSingletonTraits;
@@ -48,10 +49,10 @@ class NET_EXPORT_PRIVATE WebSocketThrottle {
 
  private:
   typedef std::deque<WebSocketJob*> ConnectingQueue;
-  typedef base::hash_map<std::string, ConnectingQueue*> ConnectingAddressMap;
+  typedef std::map<IPEndPoint, ConnectingQueue*> ConnectingAddressMap;
 
   WebSocketThrottle();
-  virtual ~WebSocketThrottle();
+  ~WebSocketThrottle();
   friend struct DefaultSingletonTraits<WebSocketThrottle>;
 
   // Key: string of host's address.  Value: queue of sockets for the address.
@@ -59,6 +60,8 @@ class NET_EXPORT_PRIVATE WebSocketThrottle {
 
   // Queue of sockets for websockets in opening state.
   ConnectingQueue queue_;
+
+  DISALLOW_COPY_AND_ASSIGN(WebSocketThrottle);
 };
 
 }  // namespace net
diff --git a/net/websockets/websocket_throttle_unittest.cc b/net/websockets/websocket_throttle_unittest.cc
index 8778ffa..86b5861 100644
--- a/net/websockets/websocket_throttle_unittest.cc
+++ b/net/websockets/websocket_throttle_unittest.cc
@@ -33,7 +33,7 @@ namespace net {
 
 class WebSocketThrottleTest : public PlatformTest {
  protected:
-  IPEndPoint MakeAddr(int a1, int a2, int a3, int a4) {
+  static IPEndPoint MakeAddr(int a1, int a2, int a3, int a4) {
     IPAddressNumber ip;
     ip.push_back(a1);
     ip.push_back(a2);
@@ -306,4 +306,50 @@ TEST_F(WebSocketThrottleTest, NoThrottleForDuplicateAddress) {
   MessageLoopForIO::current()->RunUntilIdle();
 }
 
+// A connection should not be blocked by another connection to the same IP
+// with a different port.
+TEST_F(WebSocketThrottleTest, NoThrottleForDistinctPort) {
+  TestURLRequestContext context;
+  DummySocketStreamDelegate delegate;
+  IPAddressNumber localhost;
+  ParseIPLiteralToNumber("127.0.0.1", &localhost);
+  WebSocketJob::set_websocket_over_spdy_enabled(false);
+
+  // socket1: 127.0.0.1:80
+  scoped_refptr<WebSocketJob> w1(new WebSocketJob(&delegate));
+  scoped_refptr<SocketStream> s1(
+      new SocketStream(GURL("ws://localhost:80/"), w1.get()));
+  s1->set_context(&context);
+  w1->InitSocketStream(s1.get());
+  MockSocketStreamConnect(s1, AddressList::CreateFromIPAddress(localhost, 80));
+
+  DVLOG(1) << "connecting socket1";
+  TestCompletionCallback callback_s1;
+  // Trying to open connection to localhost:80 will start without waiting.
+  EXPECT_EQ(OK, w1->OnStartOpenConnection(s1, callback_s1.callback()));
+
+  // socket2: 127.0.0.1:81
+  scoped_refptr<WebSocketJob> w2(new WebSocketJob(&delegate));
+  scoped_refptr<SocketStream> s2(
+      new SocketStream(GURL("ws://localhost:81/"), w2.get()));
+  s2->set_context(&context);
+  w2->InitSocketStream(s2.get());
+  MockSocketStreamConnect(s2, AddressList::CreateFromIPAddress(localhost, 81));
+
+  DVLOG(1) << "connecting socket2";
+  TestCompletionCallback callback_s2;
+  // Trying to open connection to localhost:81 will start without waiting.
+  EXPECT_EQ(OK, w2->OnStartOpenConnection(s2, callback_s2.callback()));
+
+  DVLOG(1) << "closing socket1";
+  w1->OnClose(s1.get());
+  s1->DetachDelegate();
+
+  DVLOG(1) << "closing socket2";
+  w2->OnClose(s2.get());
+  s2->DetachDelegate();
+  DVLOG(1) << "Done";
+  MessageLoopForIO::current()->RunUntilIdle();
+}
+
 }