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;
 
 }  // namespace
 
@@ skipping 30 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),
+      has_pending_writes_(false),
+      ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)) {
 #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 15 matching lines @@
   sockaddr_in from;
   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) {

[michaeln, 2012/05/08 18:17:12]

Are there any consumers of this class that could be adversely affected by this
change in behavior? Not saying there is, just asking.

[Marshall, 2012/05/08 20:27:07]

From what I can tell TCPListenSocket is only used by HttpServer and unit tests
(tcp_listen_socket_unittest.cc. transport_client_socket_unittest.cc). HttpServer
is only used by devtools_http_handler_impl.cc.

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

It's also used by chrome_frame/test/test_server

-  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 (bytes == NULL && len == 0) {
+    // Executing the asynchronous send callback. Clear the pending flag.
+    DCHECK(has_pending_writes_);
+    has_pending_writes_ = false;
+
+    // Another send call may have already emptied the buffer.
+    if (send_data_.empty())
+      return;
+  }
+
+  if (send_data_.empty()) {
+    // Nothing in the buffer currently. Try sending the new data without adding
+    // to the buffer.
+    DCHECK(bytes);
+    DCHECK_GT(len, 0);
+    int sent = HANDLE_EINTR(send(socket_, bytes, len, 0));
+    if (sent == len)  // All data has been sent.
+      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
-        break;
+        // Don't try to re-send data after a socket error.
+        return;
       }
-      // Otherwise we would block, and now we have to wait for a retry.
-      // Fall through to PlatformThread::YieldCurrentThread()
+    }
+
+    if (sent > 0) {
+      // Add the remaining bytes to the send buffer.
+      send_data_.append(bytes + sent, len - sent);

[mmenke, 2012/05/08 15:51:57]

nit:  Since send_data_ is empty, suggest you just use .assign here, and below.

[Marshall, 2012/05/08 20:27:07]

Done.

     } 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;
+      // Add all bytes to the send buffer.
+      send_data_.append(bytes, len);
     }
-    base::PlatformThread::YieldCurrentThread();
+
+    if (!has_pending_writes_) {
+      // Fire a new asychronous continuation callback.
+      has_pending_writes_ = true;
+      MessageLoop::current()->PostTask(FROM_HERE,

[michaeln, 2012/05/08 18:17:12]

I would expect this to spin pretty hard when network backs up, retrying
repeatedly at a very high rate waiting on the comparatively slow network to make
progress. There probably need to be some delays in here alleviate that.

Ideally, this would not employ polling at all and instead listen for
notifications of previous writes completing so that new data may be written.
Given the narrowness of the use case, polling at a more reasonable rate maybe
with some backoff seems like an OK solution for now.

[Marshall, 2012/05/08 20:27:07]

I've implemented an exponential back-off using net::BackoffEntry. If a send call
blocks it will delay the next send call by a minimum of 100ms up to a maximum of
500ms if the calls continue to block. If the buffer is flushed completely the
delay will be reset to 0ms. If partial data is sent the back-off will be
reduced.

+          base::Bind(&TCPListenSocket::SendInternal, weak_factory_.GetWeakPtr(),
+                     static_cast<const char*>(NULL), 0));

[mmenke, 2012/05/08 15:51:57]

I think you should at least use a different function for the timer, to simplify
the if statements and class invariants.  As things are, the two uses of this
function make it pretty ugly.

[Marshall, 2012/05/08 20:27:07]

Done.

+    }
+    return;
+  }

[mmenke, 2012/05/08 15:51:57]

I wonder if we really need two copies of the send code.  Since we already do a
couple rather horrible things, performance wise (polling, the substr call), I
wonder if we wouldn't just be better off using a single copy of the send code,
and basically get rid of the block of code above, except the initial DCHECKs.

[Marshall, 2012/05/08 20:27:07]

Done.

+
+  if (bytes != NULL && len > 0) {
+    // Add the bytes to the end of the send data buffer.
+    send_data_.append(bytes, len);

[michaeln, 2012/05/08 18:17:12]

What if no progress is made for a very long time, how large should this buffer
be allowed to grow prior to giving up, and how long till we give up? And when
giving up, what does that look like? There is no error return value from
SendInternal and the family of Send methods.

Looks like this class needs care-and-feeding beyond the scope of just this CL.
For just this CL, I think you should make decisions about "how large" and "how
long" that work for your immediate use case and call it a day. And once the
class has "given up", dont try to send anymore after that for the life of the
connection (to start sending again would be bad because data would have been
lost and any data sent to the receiver beyond what was lost would look like
corruption).

[Marshall, 2012/05/08 20:27:07]

I've implemented a maximum buffer size of 5MB. In my testing the buffer tends to
reach a max size of between 1MB and 2MB, so 5MB seems like a safe margin.
Currently it will output LOG(ERROR) and refuse to send additional data if this
limit is reached. I guess we could take this a step further by closing the
socket connection and/or asserting in the code.

I agree that more work is needed on this class in the future.

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

In the medium to long term, we do intend to eventually get rid of this class in
favor of ServerSocket, so there is the question of just how much effort should
be spent on cleaning up this class.

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

"Tends to?"  How strong of an assurance do we have of that?  I really should
take a look at consumers of this class before signing off on anything.

[Marshall, 2012/05/08 21:04:57]

Not a very strong assurance at all, at least with DevTools. For example, it may
be sending a large HTML page at the same time that the renderer is blocked on a
sync message. In that case a significant portion of the HTML page may end up in
the buffer.

+  }
+
+  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();
+    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();
+      return;
+    }
+  }
+
+  if (sent > 0) {
+    // Remove the sent bytes from the buffer.
+    send_data_ = send_data_.substr(sent, len_left - sent);

[mmenke, 2012/05/08 17:08:46]

This is so ugly...you may want to use a scoped_array of DrainableIOBuffers or
something.

[Marshall, 2012/05/08 20:27:07]

The buffer can grow to > 1MB in size. Do we want to keep a dedicated multi-MB
buffer for TCPListenSocket? If so, a circular buffer implementation may be the
best option. What do you think?

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

I think that's reasonable.  The one problem with a ringbuffer is resizing one is
a little tricky, but really shouldn't be too bad at all.  Probably a better
solution than my suggestion, was just trying to keep things simple.

[Marshall, 2012/05/08 21:04:57]

Are there any currently existing ring buffer implementations that can be used
from net/base?

[mmenke, 2012/05/08 21:21:35]

I don't believe so.  Our main socket class (TcpClientSocket) uses a single
buffer each for reads and sends, passed in by whatever's using it.  It doesn't
support multiple sends at once.  Instead, it calls a function when a read/send
is completed, and it's ready for another request.

StreamSocket, which I'm not terribly familiar with, does basically what I
suggested - keeps a list of send buffers, and keeps on adding to it.  I'm not
quite sure what the class is used for.

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

Keeping a list of send buffers sounds like a good idea. We can then limit the
total pending data size while also avoiding the allocation/resize of a single
large buffer.

+  }
+
+  if (!has_pending_writes_) {

[mmenke, 2012/05/08 15:51:57]

May be a little simpler to use a OneShotTimer, rather than the weak_ptr_ /
has_pending_writes_ combination, though think it's fine as-is.

[Marshall, 2012/05/08 20:27:07]

Done.

+    // Fire a new asychronous continuation callback.
+    has_pending_writes_ = true;
+    MessageLoop::current()->PostTask(FROM_HERE,
+        base::Bind(&TCPListenSocket::SendInternal, weak_factory_.GetWeakPtr(),
+                   static_cast<const char*>(NULL), 0));
   }
 }
 
 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 142 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
 
 }  // namespace net