Notify NQE of TCP RTT values

net/socket/tcp_socket_unittest.cc

 // Copyright 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/socket/tcp_socket.h"
 
+#include <stddef.h>
 #include <string.h>
 
 #include <string>
 #include <vector>
 
 #include "base/memory/ref_counted.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/test/simple_test_tick_clock.h"
+#include "base/time/time.h"
 #include "net/base/address_list.h"
 #include "net/base/io_buffer.h"
 #include "net/base/ip_endpoint.h"
 #include "net/base/ip_endpoint.h"
 #include "net/base/net_errors.h"
 #include "net/base/sockaddr_storage.h"
+#include "net/base/socket_performance_watcher.h"
 #include "net/base/test_completion_callback.h"
 #include "net/socket/tcp_client_socket.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "testing/platform_test.h"
 
 namespace net {
 
 namespace {
+
+class TestSocketPerformanceWatcher : public SocketPerformanceWatcher {
+ public:
+  explicit TestSocketPerformanceWatcher(bool should_notify_updated_rtt)
+      : should_notify_updated_rtt_(should_notify_updated_rtt),
+        connection_changed_count_(0u),
+        rtt_notification_count_(0u) {}
+  ~TestSocketPerformanceWatcher() override {}
+
+  bool ShouldNotifyUpdatedRTT() const override {
+    return should_notify_updated_rtt_;
+  }
+
+  void OnUpdatedRTTAvailable(const base::TimeDelta& rtt) override {
+    rtt_notification_count_++;
+  }
+
+  void OnConnectionChanged() override { connection_changed_count_++; }
+
+  size_t rtt_notification_count() const { return rtt_notification_count_; }
+
+  size_t connection_changed_count() const { return connection_changed_count_; }
+
+ private:
+  const bool should_notify_updated_rtt_;
+  size_t connection_changed_count_;
+  size_t rtt_notification_count_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestSocketPerformanceWatcher);
+};
+
 const int kListenBacklog = 5;
 
 class TCPSocketTest : public PlatformTest {
  protected:
-  TCPSocketTest() : socket_(NULL, NetLog::Source()) {
-  }
+  TCPSocketTest() : socket_(NULL, NULL, NetLog::Source()) {}
 
   void SetUpListenIPv4() {
     ASSERT_EQ(OK, socket_.Open(ADDRESS_FAMILY_IPV4));
     ASSERT_EQ(OK, socket_.Bind(IPEndPoint(IPAddress::IPv4Localhost(), 0)));
     ASSERT_EQ(OK, socket_.Listen(kListenBacklog));
     ASSERT_EQ(OK, socket_.GetLocalAddress(&local_address_));
   }
 
   void SetUpListenIPv6(bool* success) {
     *success = false;
@@ skipping 11 matching lines @@
 
   void TestAcceptAsync() {
     TestCompletionCallback accept_callback;
     scoped_ptr<TCPSocket> accepted_socket;
     IPEndPoint accepted_address;
     ASSERT_EQ(ERR_IO_PENDING,
               socket_.Accept(&accepted_socket, &accepted_address,
                              accept_callback.callback()));
 
     TestCompletionCallback connect_callback;
-    TCPClientSocket connecting_socket(local_address_list(),
-                                      NULL, NetLog::Source());
+    TCPClientSocket connecting_socket(local_address_list(), NULL, NULL,
+                                      NetLog::Source());
     connecting_socket.Connect(connect_callback.callback());
 
     EXPECT_EQ(OK, connect_callback.WaitForResult());
     EXPECT_EQ(OK, accept_callback.WaitForResult());
 
     EXPECT_TRUE(accepted_socket.get());
 
     // Both sockets should be on the loopback network interface.
     EXPECT_EQ(accepted_address.address(), local_address_.address());
   }
 
+#if defined(TCP_INFO) || defined(OS_LINUX)
+  // Tests that notifications to Socket Performance Watcher (SPW) are delivered
+  // correctly. |advance_ticks| is the duration by which the clock is advanced
+  // before a message is read. |should_notify_updated_rtt| is true if the SPW
+  // is interested in receiving RTT notifications. |num_messages| is the number
+  // of messages that are written/read by the sockets.
+  // |expect_connection_changed_count| is the expected number of connection
+  // change notifications received by the SPW. |expect_rtt_notification_count|
+  // is the expected number of RTT notifications received by the SPW.
+  // This test works by writing |num_messages| to the socket. A different
+  // socket (with a SPW attached to it) reads the messages.
+  void TestSPWNotifications(const base::TimeDelta& advance_ticks,
+                            bool should_notify_updated_rtt,
+                            size_t num_messages,
+                            size_t expect_connection_changed_count,
+                            size_t expect_rtt_notification_count) {
+    ASSERT_NO_FATAL_FAILURE(SetUpListenIPv4());
+
+    scoped_ptr<base::SimpleTestTickClock> tick_clock(
+        new base::SimpleTestTickClock());
+    base::SimpleTestTickClock* tick_clock_ptr = tick_clock.get();
+    tick_clock_ptr->SetNowTicks(base::TimeTicks::Now());
+
+    TestCompletionCallback connect_callback;
+
+    scoped_ptr<TestSocketPerformanceWatcher> watcher(
+        new TestSocketPerformanceWatcher(should_notify_updated_rtt));
+    TestSocketPerformanceWatcher* watcher_ptr = watcher.get();
+
+    TCPSocket connecting_socket(std::move(watcher), NULL, NetLog::Source());
+    connecting_socket.SetTickClockForTesting(std::move(tick_clock));
+
+    int result = connecting_socket.Open(ADDRESS_FAMILY_IPV4);
+    ASSERT_EQ(OK, result);
+    connecting_socket.Connect(local_address_, connect_callback.callback());

[Ryan Sleevi, 2016/04/11 22:06:56]

You never seem to check the result here? Seems like it should be around line 154

[tbansal1, 2016/04/12 16:14:33]

Not sure if I understand the comment.

Result is checked below (Connect never succeeds synchronously) on Line 160:
EXPECT_EQ(OK, connect_callback.WaitForResult());

[Ryan Sleevi, 2016/04/12 21:08:31]

Ah, I missed that.

[tbansal1, 2016/04/12 22:39:19]

Acknowledged.

+
+    TestCompletionCallback accept_callback;
+    scoped_ptr<TCPSocket> accepted_socket;
+    IPEndPoint accepted_address;
+    result = socket_.Accept(&accepted_socket, &accepted_address,
+                            accept_callback.callback());
+    ASSERT_EQ(OK, accept_callback.GetResult(result));
+
+    ASSERT_TRUE(accepted_socket.get());
+
+    // Both sockets should be on the loopback network interface.
+    EXPECT_EQ(accepted_address.address(), local_address_.address());
+
+    EXPECT_EQ(OK, connect_callback.WaitForResult());

[Ryan Sleevi, 2016/04/12 21:08:31]

Should it be an ASSERT?

[tbansal1, 2016/04/12 22:39:19]

Done.

+
+    for (size_t i = 0; i < num_messages; ++i) {
+      tick_clock_ptr->Advance(advance_ticks);
+
+      // Use a 1 byte message so that it is written and read in one attempt.

[Ryan Sleevi, 2016/04/11 22:06:57]

There's no guarantee that this statement will apply.

[tbansal1, 2016/04/12 16:14:34]

I do not understand this comment. 

Is it possible for a 1 byte message to be broken down to smaller messages by
TCP? Also, since the buffer is empty, the message should be written. Right?

[Ryan Sleevi, 2016/04/12 21:08:31]

It's possible that you'll get 0 bytes written with an asynchronous completion.
No, that's not guaranteed here. That is, don't depend/assume on kernel buffering
semantics (we've had other tests that failed here)

[tbansal1, 2016/04/12 22:39:19]

Acknowledged. I have rewritten the comment.

+      const std::string message("t");
+      std::vector<char> buffer(message.size());

[Ryan Sleevi, 2016/04/11 22:06:57]

This just seems wasteful; why not

std::vector<char> buffer('t', 1); ?

[tbansal1, 2016/04/12 16:14:34]

I simplified the test significantly.

+      ASSERT_EQ(1u, message.size());

[Ryan Sleevi, 2016/04/11 22:06:57]

Seems... unnecessary?

[tbansal1, 2016/04/12 16:14:34]

Done.

+
+      size_t bytes_written = 0;
+      scoped_refptr<IOBufferWithSize> write_buffer(
+          new IOBufferWithSize(message.size() - bytes_written));
+      memmove(write_buffer->data(), message.data() + bytes_written,
+              message.size() - bytes_written);
+
+      TestCompletionCallback write_callback;
+      int write_result = accepted_socket->Write(
+          write_buffer.get(), write_buffer->size(), write_callback.callback());
+      write_result = write_callback.GetResult(write_result);

[Ryan Sleevi, 2016/04/11 22:06:57]

Combine
write_callback.GetResult(accepted_socket->Write(...));

Also, not guaranteed it will synchronously complete. You should order it as

- Start async write operation
- Start async read operation
- Ensure both are completed and have results.

[tbansal1, 2016/04/12 16:14:34]

This test is similar to "ReadWrite" below. Also, would not GetResult() block
until the data has been written to the socket?

Since the socket write buffer is empty, what else could be the reason for this
to not succeed?

[Ryan Sleevi, 2016/04/12 21:08:31]

You can't assume that the underlying implementation will do this. That's what is
"not guaranteed"

For example, Windows implements local sockets (127.0.0.1) as a series of named
pipes, and the first write may not complete until there's also a read operation
enqueued. While I realize this is "Linux Only", it's about writing to the
interface. There's no guarantee that the kernel buffer won't be, say, 0 bytes -
so you need to handle it as an asynchronous operation, and not just assume
synchronicity.

[tbansal1, 2016/04/12 22:39:19]

Thanks. I did not know that some operating systems may special case local
sockets and may not write to socket until a read operation is enqueued. That
sounds broken because on real networks there is no way to know if a read
operation is enqueued or not on the remote endpoint.

+      ASSERT_TRUE(write_result >= 0);

[Ryan Sleevi, 2016/04/11 22:06:57]

ASSERT_GE

[tbansal1, 2016/04/12 16:14:34]

Done.

+      bytes_written += write_result;
+      ASSERT_TRUE(bytes_written <= message.size());
+      ASSERT_EQ(bytes_written, message.size());
+
+      size_t bytes_read = 0;
+      scoped_refptr<IOBufferWithSize> read_buffer(
+          new IOBufferWithSize(message.size() - bytes_read));

[Ryan Sleevi, 2016/04/11 22:06:57]

bytes_read is always 0 (see line 185)

[tbansal1, 2016/04/12 16:14:34]

Done.

+      TestCompletionCallback read_callback;
+      int read_result = connecting_socket.Read(
+          read_buffer.get(), read_buffer->size(), read_callback.callback());
+      read_result = read_callback.GetResult(read_result);
+      ASSERT_TRUE(read_result >= 0);

[Ryan Sleevi, 2016/04/11 22:06:57]

GE

[tbansal1, 2016/04/12 16:14:34]

Done.

+      ASSERT_TRUE(bytes_read + read_result <= message.size());

[Ryan Sleevi, 2016/04/11 22:06:56]

LE

buffer.size()

[tbansal1, 2016/04/12 16:14:33]

Done.

+      memmove(&buffer[bytes_read], read_buffer->data(), read_result);

[Ryan Sleevi, 2016/04/11 22:06:57]

Seems unnecessary

[tbansal1, 2016/04/12 16:14:34]

Done.

+      bytes_read += read_result;

[Ryan Sleevi, 2016/04/11 22:06:57]

Seems unnecessary given the above?

[tbansal1, 2016/04/12 16:14:33]

Done.

+      ASSERT_EQ(bytes_read, message.size());
+
+      std::string received_message(buffer.begin(), buffer.end());

[Ryan Sleevi, 2016/04/11 22:06:57]

You copy back into a string, which is also unnecessary, given that vector has an
operator== defined.

[tbansal1, 2016/04/12 16:14:33]

Done.

+      ASSERT_EQ(message, received_message);
+    }
+    EXPECT_EQ(expect_connection_changed_count,
+              watcher_ptr->connection_changed_count());
+    EXPECT_EQ(expect_rtt_notification_count,
+              watcher_ptr->rtt_notification_count());
+  }
+#endif  // defined(TCP_INFO) || defined(OS_LINUX)
+
   AddressList local_address_list() const {
     return AddressList(local_address_);
   }
 
   TCPSocket socket_;
   IPEndPoint local_address_;
 };
 
 // Test listening and accepting with a socket bound to an IPv4 address.
 TEST_F(TCPSocketTest, Accept) {
   ASSERT_NO_FATAL_FAILURE(SetUpListenIPv4());
 
   TestCompletionCallback connect_callback;
   // TODO(yzshen): Switch to use TCPSocket when it supports client socket
   // operations.
-  TCPClientSocket connecting_socket(local_address_list(),
-                                    NULL, NetLog::Source());
+  TCPClientSocket connecting_socket(local_address_list(), NULL, NULL,
+                                    NetLog::Source());
   connecting_socket.Connect(connect_callback.callback());
 
   TestCompletionCallback accept_callback;
   scoped_ptr<TCPSocket> accepted_socket;
   IPEndPoint accepted_address;
   int result = socket_.Accept(&accepted_socket, &accepted_address,
                               accept_callback.callback());
   if (result == ERR_IO_PENDING)
     result = accept_callback.WaitForResult();
   ASSERT_EQ(OK, result);
@@ skipping 37 matching lines @@
 
   TestCompletionCallback accept_callback;
   scoped_ptr<TCPSocket> accepted_socket;
   IPEndPoint accepted_address;
 
   ASSERT_EQ(ERR_IO_PENDING,
             socket_.Accept(&accepted_socket, &accepted_address,
                            accept_callback.callback()));
 
   TestCompletionCallback connect_callback;
-  TCPClientSocket connecting_socket(local_address_list(),
-                                    NULL, NetLog::Source());
+  TCPClientSocket connecting_socket(local_address_list(), NULL, NULL,
+                                    NetLog::Source());
   connecting_socket.Connect(connect_callback.callback());
 
   TestCompletionCallback connect_callback2;
-  TCPClientSocket connecting_socket2(local_address_list(),
-                                     NULL, NetLog::Source());
+  TCPClientSocket connecting_socket2(local_address_list(), NULL, NULL,
+                                     NetLog::Source());
   connecting_socket2.Connect(connect_callback2.callback());
 
   EXPECT_EQ(OK, accept_callback.WaitForResult());
 
   TestCompletionCallback accept_callback2;
   scoped_ptr<TCPSocket> accepted_socket2;
   IPEndPoint accepted_address2;
 
   int result = socket_.Accept(&accepted_socket2, &accepted_address2,
                               accept_callback2.callback());
@@ skipping 13 matching lines @@
 }
 
 // Test listening and accepting with a socket bound to an IPv6 address.
 TEST_F(TCPSocketTest, AcceptIPv6) {
   bool initialized = false;
   ASSERT_NO_FATAL_FAILURE(SetUpListenIPv6(&initialized));
   if (!initialized)
     return;
 
   TestCompletionCallback connect_callback;
-  TCPClientSocket connecting_socket(local_address_list(),
-                                    NULL, NetLog::Source());
+  TCPClientSocket connecting_socket(local_address_list(), NULL, NULL,
+                                    NetLog::Source());
   connecting_socket.Connect(connect_callback.callback());
 
   TestCompletionCallback accept_callback;
   scoped_ptr<TCPSocket> accepted_socket;
   IPEndPoint accepted_address;
   int result = socket_.Accept(&accepted_socket, &accepted_address,
                               accept_callback.callback());
   if (result == ERR_IO_PENDING)
     result = accept_callback.WaitForResult();
   ASSERT_EQ(OK, result);
 
   EXPECT_TRUE(accepted_socket.get());
 
   // Both sockets should be on the loopback network interface.
   EXPECT_EQ(accepted_address.address(), local_address_.address());
 
   EXPECT_EQ(OK, connect_callback.WaitForResult());
 }
 
 TEST_F(TCPSocketTest, ReadWrite) {
   ASSERT_NO_FATAL_FAILURE(SetUpListenIPv4());
 
   TestCompletionCallback connect_callback;
-  TCPSocket connecting_socket(NULL, NetLog::Source());
+  TCPSocket connecting_socket(NULL, NULL, NetLog::Source());
   int result = connecting_socket.Open(ADDRESS_FAMILY_IPV4);
   ASSERT_EQ(OK, result);
   connecting_socket.Connect(local_address_, connect_callback.callback());
 
   TestCompletionCallback accept_callback;
   scoped_ptr<TCPSocket> accepted_socket;
   IPEndPoint accepted_address;
   result = socket_.Accept(&accepted_socket, &accepted_address,
                           accept_callback.callback());
   ASSERT_EQ(OK, accept_callback.GetResult(result));
@@ skipping 35 matching lines @@
     ASSERT_TRUE(read_result >= 0);
     ASSERT_TRUE(bytes_read + read_result <= message.size());
     memmove(&buffer[bytes_read], read_buffer->data(), read_result);
     bytes_read += read_result;
   }
 
   std::string received_message(buffer.begin(), buffer.end());
   ASSERT_EQ(message, received_message);
 }
 
+// These tests require kernel support for tcp_info struct, and so they are
+// enabled only on certain platforms.
+#if defined(TCP_INFO) || defined(OS_LINUX)
+// If SocketPerformanceWatcher::ShouldNotifyUpdatedRTT always returns false,
+// then the wtatcher should not receive any notifications.
+TEST_F(TCPSocketTest, SPWNotInterested) {
+  TestSPWNotifications(base::TimeDelta::FromSeconds(0), false, 2u, 0u, 0u);
+}
+
+// One notification should be received when the socket connects. No additional
+// notifications should be received when the message is read because the clock
+// is not advanced.
+TEST_F(TCPSocketTest, SPWNoAdvance) {
+  TestSPWNotifications(base::TimeDelta::FromSeconds(0), true, 2u, 0u, 1u);
+}
+
+// One notification should be received when the socket connects. One
+// additional notification should be received for each message read since this
+// test advances clock by 2 seconds (which is longer than the minimum interval
+// between consecutive notifications) before every read.
+TEST_F(TCPSocketTest, SPWAdvance) {
+  TestSPWNotifications(base::TimeDelta::FromSeconds(2), true, 2u, 0u, 3u);
+}
+#endif  // defined(TCP_INFO) || defined(OS_LINUX)
+
 }  // namespace
 }  // namespace net