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

net/base/tcp_listen_socket.cc

 // Copyright (c) 2012 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 "build/build_config.h"
 
 #if defined(OS_WIN)
 // winsock2.h must be included first in order to ensure it is included before
 // windows.h.
 #include <winsock2.h>
 #elif defined(OS_POSIX)
 #include <errno.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #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"
 #include "net/base/net_util.h"
 #include "net/base/tcp_listen_socket.h"
 
 #if defined(OS_WIN)
 typedef int socklen_t;
+#include "net/base/winsock_init.h"
 #endif  // defined(OS_WIN)
 
 namespace net {
 
 namespace {
 
 const int kReadBufSize = 4096;
+const int kMaxSendBufSize = 1024*1024*5;  // 5MB
+
+const net::BackoffEntry::Policy kSendDataBackoffPolicy = {
+  // Number of initial errors (in sequence) to ignore before applying
+  // exponential back-off rules.
+  0,
+
+  // Initial delay for exponential back-off in ms.
+  100,

[mmenke, 2012/05/08 21:23:38]

If we're sending megabytes locally, 100 milliseconds might be too long.

[Marshall, 2012/05/08 21:44:51]

Reduced to 25ms.

+
+  // 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.
+  500,
+
+  // 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
 
 #if defined(OS_WIN)
 const SOCKET TCPListenSocket::kInvalidSocket = INVALID_SOCKET;
 const int TCPListenSocket::kSocketError = SOCKET_ERROR;
 #elif defined(OS_POSIX)
 const SOCKET TCPListenSocket::kInvalidSocket = -1;
 const int TCPListenSocket::kSocketError = -1;
 #endif
@@ skipping 23 matching lines @@
     has_pending_reads_ = false;
     Read();
   }
 }
 
 TCPListenSocket::TCPListenSocket(SOCKET s,
                                  ListenSocket::ListenSocketDelegate *del)
     : ListenSocket(del),
       socket_(s),
       reads_paused_(false),
-      has_pending_reads_(false) {
+      has_pending_reads_(false),
+      send_data_error_(false),
+      send_data_backoff_(&kSendDataBackoffPolicy) {
 #if defined(OS_WIN)
   socket_event_ = WSACreateEvent();
   // TODO(ibrar): error handling in case of socket_event_ == WSA_INVALID_EVENT
   WatchSocket(NOT_WAITING);
 #elif defined(OS_POSIX)
   wait_state_ = NOT_WAITING;
 #endif
 }
 
 TCPListenSocket::~TCPListenSocket() {
 #if defined(OS_WIN)
   if (socket_event_) {
     WSACloseEvent(socket_event_);
     socket_event_ = WSA_INVALID_EVENT;
   }
 #endif
   CloseSocket(socket_);
 }
 
 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)
     // Allow rapid reuse.
     static const int kOn = 1;
     setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &kOn, sizeof(kOn));
 #endif
     sockaddr_in addr;
     memset(&addr, 0, sizeof(addr));
     addr.sin_family = AF_INET;
@@ skipping 16 matching lines @@
   socklen_t from_len = sizeof(from);
   SOCKET conn =
       HANDLE_EINTR(accept(s, reinterpret_cast<sockaddr*>(&from), &from_len));
   if (conn != kInvalidSocket) {
     SetNonBlocking(conn);
   }
   return conn;
 }
 
 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_data_error_)
+    return;
+
+  // Add all bytes to the send buffer.
+  if (send_data_.empty()) {
+    send_data_.assign(bytes, len);
+  } else if (send_data_.size() > kMaxSendBufSize) {

[mmenke, 2012/05/08 20:51:43]

Should this be send_data_.size() + len?  We'd still be allowing the first send
to exceed the limit.

[Marshall, 2012/05/08 21:44:51]

Done.

+    // Too much of a backup, stop trying to add more data.
+    LOG(ERROR) << "send failed: buffer overrun";
+    send_data_error_ = true;

[mmenke, 2012/05/08 20:51:43]

send_data_.clear()?

[Marshall, 2012/05/08 21:44:51]

Done.

+    return;
+  } else {
+    send_data_.append(bytes, len);
   }
+
+  if (!send_data_backoff_.ShouldRejectRequest())
+    SendData();
 }
 
 void TCPListenSocket::Listen() {
   int backlog = 10;  // TODO(erikkay): maybe don't allow any backlog?
   listen(socket_, backlog);
   // TODO(erikkay): error handling
 #if defined(OS_POSIX)
   WatchSocket(WAITING_ACCEPT);
 #endif
 }
@@ skipping 140 matching lines @@
 }
 
 void TCPListenSocket::OnFileCanWriteWithoutBlocking(int fd) {
   // MessagePumpLibevent callback, we don't listen for write events
   // so we shouldn't ever reach here.
   NOTREACHED();
 }
 
 #endif
 
+void TCPListenSocket::SendData() {
+  // Another send call may have already emptied the buffer.
+  if (send_data_.empty())
+    return;
+
+  int len_left = static_cast<int>(send_data_.length());
+  int sent = HANDLE_EINTR(send(socket_, send_data_.c_str(), len_left, 0));
+  if (sent == len_left) {
+    // All data has been sent.
+    send_data_.clear();
+
+    // Successfully sending all data clears the backoff state.
+    send_data_backoff_.Reset();
+    return;
+  }
+
+  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_data_.clear();
+      send_data_error_ = true;
+      return;
+    }
+
+    // Blocking is considered a failure.
+    send_data_backoff_.InformOfRequest(false);
+  }
+
+  if (sent > 0) {
+    // Remove the sent bytes from the buffer.
+    send_data_ = send_data_.substr(sent, len_left - sent);
+
+    // Sending any data is considered a success.
+    send_data_backoff_.InformOfRequest(true);
+  }
+
+  if (!send_timer_.IsRunning()) {
+    // Schedule a timer to continue sending data asynchronously.
+    send_timer_.Start(FROM_HERE, send_data_backoff_.GetTimeUntilRelease(),
+                      this, &TCPListenSocket::SendData);
+  }
+}
+
 }  // namespace net