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

[mmenke, 2012/05/09 19:20:08]

nit:  1024 * 1024 * 5

[Marshall, 2012/05/10 16:28:05]

Done.

+
+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
 
 #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_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
   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_error_)
+    return;
+
+  if (send_pending_size_ + len > kMaxSendBufSize) {
+    // Too much of a backup, stop trying to add more data.
+    LOG(ERROR) << "send failed: buffer overrun";
+    send_buffers_.clear();
+    send_pending_size_ = 0;
+    send_error_ = true;
+    return;
   }
+
+  // Add a new buffer to the send buffer list.
+  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_backoff_.ShouldRejectRequest())

[michaeln, 2012/05/09 20:55:34]

For clarity, consider replacing this test with !send_timer_.IsRunning()... i
think thats the more proximate and obvious condition that matters here.

If it's possible for ShouldReject to return true when the timer is not
running... that would be a problem since the data added above would never get
sent. Its not real obvious that ShouldReject (that backoff class is a bit of a
black box) won't dont that under some circumstances.

[Marshall, 2012/05/10 16:28:05]

Done.

+    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 list.
+  if (send_buffers_.empty())
+    return;
+
+  int total_sent = 0;
+
+  // Send data until all buffers have been sent or a call would block.
+  do {

[mmenke, 2012/05/09 19:20:08]

I think this may be a little simpler if you used a
"while(!send_buffers_.empty())", and removed the if statement above.

Alternatively, you could use a while(true), and do the exit check where the
continue is now.

[Marshall, 2012/05/10 16:28:05]

Done.

+    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 == len_left) {
+      // All data has been sent. Remove the buffer from the list and move on
+      // to the next buffer, if any.
+      send_buffers_.erase(send_buffers_.begin());
+      total_sent += sent;
+      continue;
+    }
+
+    if (sent == kSocketError) {

[michaeln, 2012/05/09 20:55:34]

Is it possible for send() to return any other error code? If it is this could
infinite loop. Maybe we don't need to defend against that in production code
(not sure), probably worth at least a DCHECK that sent > 0 || sent ==
kSocketError.

[Marshall, 2012/05/10 16:28:05]

Done.

[Marshall, 2012/05/10 19:01:09]

I changed this to an "else { NOTREACHED(); break; }" so that if it does happen
we don't get stuck in an infinite loop.

+#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 if (sent > 0) {

[michaeln, 2012/05/09 20:55:34]

nit: might be nice to have the sent everything and sent something cases adjacent
to one another, as coded the error handling case is wedged between those two
success cases

[Marshall, 2012/05/10 16:28:05]

Done.

+      // The buffer was partially sent.
+      buffer->DidConsume(sent);
+      total_sent += sent;
+    }
+  } while (!send_buffers_.empty());
+
+  if (total_sent > 0) {
+    // Data has been sent.

[michaeln, 2012/05/09 20:55:34]

nit: some of these comments are stating the obvious, consider removing those
that don't really add value

[Marshall, 2012/05/10 16:28:05]

I've removed some. Let me know if you think more should be removed.

+    send_pending_size_ -= total_sent;
+    DCHECK_GE(send_pending_size_, 0);
+
+    // Clear the back-off delay.
+    send_backoff_.Reset();
+  } else {
+    // No data has been sent. Increase the back-off delay.
+    send_backoff_.InformOfRequest(false);
+  }
+
+  if (!send_buffers_.empty() && !send_timer_.IsRunning()) {
+    // Schedule a timer to continue sending data asynchronously.
+    send_timer_.Start(FROM_HERE, send_backoff_.GetTimeUntilRelease(),
+                      this, &TCPListenSocket::SendData);
+  }
+}
+
 }  // namespace net