Change TCPListenSocket::SendInternal to use a non-blocking implementation.

HttpServer uses TCPListenSocket to create a server that runs on the browser process IO thread. The IO thread should not be blocked as this can introduce jank and deadlocks.


BUG=125191
TEST=TCPListenSocketTest.*, DevTools works (see bug report)

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

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

4 files changed, 172 insertions, 31 deletions

diff --git a/net/base/tcp_listen_socket.cc b/net/base/tcp_listen_socket.cc
index 6463822..8fb4734 100644
--- a/net/base/tcp_listen_socket.cc
+++ b/net/base/tcp_listen_socket.cc
@@ -17,6 +17,7 @@
 #include "net/base/net_errors.h"
 #endif
 
+#include "base/bind.h"
 #include "base/eintr_wrapper.h"
 #include "base/sys_byteorder.h"
 #include "base/threading/platform_thread.h"
@@ -25,6 +26,7 @@
 
 #if defined(OS_WIN)
 typedef int socklen_t;
+#include "net/base/winsock_init.h"
 #endif  // defined(OS_WIN)
 
 namespace net {
@@ -32,6 +34,33 @@ namespace net {
 namespace {
 
 const int kReadBufSize = 4096;
+const int kMaxSendBufSize = 1024 * 1024 * 5;  // 5MB
+
+const net::BackoffEntry::Policy kSendBackoffPolicy = {
+  // Number of initial errors (in sequence) to ignore before applying
+  // exponential back-off rules.
+  0,
+
+  // Initial delay for exponential back-off in ms.
+  25,
+
+  // Factor by which the waiting time will be multiplied.
+  2,
+
+  // Fuzzing percentage. ex: 10% will spread requests randomly
+  // between 90%-100% of the calculated time.
+  0,
+
+  // Maximum amount of time we are willing to delay our request in ms.
+  100,
+
+  // Time to keep an entry from being discarded even when it
+  // has no significant state, -1 to never discard.
+  -1,
+
+  // Don't use initial delay unless the last request was an error.
+  false,
+};
 
 }  // namespace
 
@@ -75,7 +104,10 @@ TCPListenSocket::TCPListenSocket(SOCKET s,
     : ListenSocket(del),
       socket_(s),
       reads_paused_(false),
-      has_pending_reads_(false) {
+      has_pending_reads_(false),
+      send_pending_size_(0),
+      send_error_(false),
+      send_backoff_(&kSendBackoffPolicy) {
 #if defined(OS_WIN)
   socket_event_ = WSACreateEvent();
   // TODO(ibrar): error handling in case of socket_event_ == WSA_INVALID_EVENT
@@ -96,6 +128,10 @@ TCPListenSocket::~TCPListenSocket() {
 }
 
 SOCKET TCPListenSocket::CreateAndBind(const std::string& ip, int port) {
+#if defined(OS_WIN)
+  EnsureWinsockInit();
+#endif
+
   SOCKET s = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
   if (s != kInvalidSocket) {
 #if defined(OS_POSIX)
@@ -132,33 +168,27 @@ SOCKET TCPListenSocket::Accept(SOCKET s) {
 }
 
 void TCPListenSocket::SendInternal(const char* bytes, int len) {
-  char* send_buf = const_cast<char *>(bytes);
-  int len_left = len;
-  while (true) {
-    int sent = HANDLE_EINTR(send(socket_, send_buf, len_left, 0));
-    if (sent == len_left) {  // A shortcut to avoid extraneous checks.
-      break;
-    }
-    if (sent == kSocketError) {
-#if defined(OS_WIN)
-      if (WSAGetLastError() != WSAEWOULDBLOCK) {
-        LOG(ERROR) << "send failed: WSAGetLastError()==" << WSAGetLastError();
-#elif defined(OS_POSIX)
-      if (errno != EWOULDBLOCK && errno != EAGAIN) {
-        LOG(ERROR) << "send failed: errno==" << errno;
-#endif
-        break;
-      }
-      // Otherwise we would block, and now we have to wait for a retry.
-      // Fall through to PlatformThread::YieldCurrentThread()
-    } else {
-      // sent != len_left according to the shortcut above.
-      // Shift the buffer start and send the remainder after a short while.
-      send_buf += sent;
-      len_left -= sent;
-    }
-    base::PlatformThread::YieldCurrentThread();
+  DCHECK(bytes);
+  DCHECK_GT(len, 0);
+
+  if (send_error_)
+    return;
+
+  if (send_pending_size_ + len > kMaxSendBufSize) {
+    LOG(ERROR) << "send failed: buffer overrun";
+    send_buffers_.clear();
+    send_pending_size_ = 0;
+    send_error_ = true;
+    return;
   }
+
+  scoped_refptr<IOBuffer> buffer(new IOBuffer(len));
+  memcpy(buffer->data(), bytes, len);
+  send_buffers_.push_back(new DrainableIOBuffer(buffer, len));
+  send_pending_size_ += len;
+
+  if (!send_timer_.IsRunning())
+    SendData();
 }
 
 void TCPListenSocket::Listen() {
@@ -319,4 +349,63 @@ void TCPListenSocket::OnFileCanWriteWithoutBlocking(int fd) {
 
 #endif
 
+void TCPListenSocket::SendData() {
+  DCHECK(!send_buffers_.empty());
+
+  int total_sent = 0;
+
+  // Send data until all buffers have been sent or a call would block.
+  while (!send_buffers_.empty()) {
+    scoped_refptr<DrainableIOBuffer> buffer = send_buffers_.front();
+
+    int len_left = buffer->BytesRemaining();
+    int sent = HANDLE_EINTR(send(socket_, buffer->data(), len_left, 0));
+    if (sent > 0) {
+      if (sent == len_left)
+        send_buffers_.pop_front();
+      else
+        buffer->DidConsume(sent);
+
+      total_sent += sent;
+    } else if (sent == kSocketError) {
+#if defined(OS_WIN)
+      if (WSAGetLastError() != WSAEWOULDBLOCK) {
+        LOG(ERROR) << "send failed: WSAGetLastError()==" << WSAGetLastError();
+#elif defined(OS_POSIX)
+      if (errno != EWOULDBLOCK && errno != EAGAIN) {
+        LOG(ERROR) << "send failed: errno==" << errno;
+#endif
+        // Don't try to re-send data after a socket error.
+        send_buffers_.clear();
+        send_pending_size_ = 0;
+        send_error_ = true;
+        return;
+      }
+
+      // The call would block. Don't send any more data at this time.
+      break;
+    } else {
+      NOTREACHED();
+      break;
+    }
+  }
+
+  if (total_sent > 0) {
+    send_pending_size_ -= total_sent;
+    DCHECK_GE(send_pending_size_, 0);
+
+    // Clear the back-off delay.
+    send_backoff_.Reset();
+  } else {
+    // Increase the back-off delay.
+    send_backoff_.InformOfRequest(false);
+  }
+
+  if (!send_buffers_.empty()) {
+    DCHECK(!send_timer_.IsRunning());
+    send_timer_.Start(FROM_HERE, send_backoff_.GetTimeUntilRelease(),
+                      this, &TCPListenSocket::SendData);
+  }
+}
+
 }  // namespace net
diff --git a/net/base/tcp_listen_socket.h b/net/base/tcp_listen_socket.h
index ef3f3e6..32aac27 100644
--- a/net/base/tcp_listen_socket.h
+++ b/net/base/tcp_listen_socket.h
@@ -12,6 +12,8 @@
 #define NET_BASE_TCP_LISTEN_SOCKET_H_
 #pragma once
 
+#include <list>
+
 #include "build/build_config.h"
 
 #if defined(OS_WIN)
@@ -26,6 +28,10 @@
 
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
+#include "base/memory/ref_counted.h"
+#include "base/timer.h"
+#include "net/base/backoff_entry.h"
+#include "net/base/io_buffer.h"
 #include "net/base/listen_socket.h"
 #include "net/base/net_export.h"
 
@@ -101,8 +107,17 @@ class NET_EXPORT TCPListenSocket : public ListenSocket,
   SOCKET socket_;
 
  private:
+  void SendData();
+
   bool reads_paused_;
   bool has_pending_reads_;
+  std::list<scoped_refptr<DrainableIOBuffer> > send_buffers_;
+  int send_pending_size_;
+  bool send_error_;
+  net::BackoffEntry send_backoff_;
+
+  // Used to continue sending data asynchronously.
+  base::OneShotTimer<TCPListenSocket> send_timer_;
 
   DISALLOW_COPY_AND_ASSIGN(TCPListenSocket);
 };
diff --git a/net/base/tcp_listen_socket_unittest.cc b/net/base/tcp_listen_socket_unittest.cc
index 30cde3cb1..423691b 100644
--- a/net/base/tcp_listen_socket_unittest.cc
+++ b/net/base/tcp_listen_socket_unittest.cc
@@ -119,10 +119,9 @@ void TCPListenSocketTester::Shutdown() {
 
 void TCPListenSocketTester::Listen() {
   server_ = DoListen();
-  if (server_) {
-    server_->AddRef();
-    ReportAction(TCPListenSocketTestAction(ACTION_LISTEN));
-  }
+  ASSERT_TRUE(server_);
+  server_->AddRef();
+  ReportAction(TCPListenSocketTestAction(ACTION_LISTEN));
 }
 
 void TCPListenSocketTester::SendFromTester() {
@@ -179,6 +178,38 @@ void TCPListenSocketTester::TestServerSend() {
   ASSERT_STREQ(buf, kHelloWorld);
 }
 
+void TCPListenSocketTester::TestServerSendMultiple() {
+  // Send enough data to exceed the socket receive window. 20kb is probably a
+  // safe bet.
+  int send_count = (1024*20) / (sizeof(kHelloWorld)-1);
+  int i;
+
+  // Send multiple writes. Since no reading is occuring the data should be
+  // buffered in TCPListenSocket.
+  for (i = 0; i < send_count; ++i) {
+    loop_->PostTask(FROM_HERE, base::Bind(
+        &TCPListenSocketTester::SendFromTester, this));
+    NextAction();
+    ASSERT_EQ(ACTION_SEND, last_action_.type());
+  }
+
+  // Make multiple reads. All of the data should eventually be returned.
+  char buf[sizeof(kHelloWorld)];
+  const int buf_len = sizeof(kHelloWorld);
+  for (i = 0; i < send_count; ++i) {
+    unsigned recv_len = 0;
+    while (recv_len < buf_len-1) {
+      int r = HANDLE_EINTR(recv(test_socket_, buf, buf_len-1, 0));
+      ASSERT_GE(r, 0);
+      recv_len += static_cast<unsigned>(r);
+      if (!r)
+        break;
+    }
+    buf[recv_len] = 0;
+    ASSERT_STREQ(buf, kHelloWorld);
+  }
+}
+
 bool TCPListenSocketTester::Send(SOCKET sock, const std::string& str) {
   int len = static_cast<int>(str.length());
   int send_len = HANDLE_EINTR(send(sock, str.data(), len, 0));
@@ -248,4 +279,8 @@ TEST_F(TCPListenSocketTest, ServerSend) {
   tester_->TestServerSend();
 }
 
+TEST_F(TCPListenSocketTest, ServerSendMultiple) {
+  tester_->TestServerSendMultiple();
+}
+
 }  // namespace net
diff --git a/net/base/tcp_listen_socket_unittest.h b/net/base/tcp_listen_socket_unittest.h
index 4e24b30..a8c2b89 100644
--- a/net/base/tcp_listen_socket_unittest.h
+++ b/net/base/tcp_listen_socket_unittest.h
@@ -91,6 +91,8 @@ class TCPListenSocketTester :
   void TestClientSendLong();
   // verify a send/read from server to client
   void TestServerSend();
+  // verify multiple sends and reads from server to client.
+  void TestServerSendMultiple();
 
   virtual bool Send(SOCKET sock, const std::string& str);