Add SyncSocket::ReceiveWithTimeout() and SyncSocket unit tests.

base/sync_socket_posix.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 "base/sync_socket.h"
 
 #include <errno.h>
+#include <fcntl.h>
 #include <limits.h>
-#include <fcntl.h>
 #include <stdio.h>
 #include <sys/ioctl.h>
 #include <sys/socket.h>
 #include <sys/types.h>
 
 #if defined(OS_SOLARIS)
 #include <sys/filio.h>
 #endif
 
 #include "base/file_util.h"
 #include "base/logging.h"
-
+#include "base/threading/thread_restrictions.h"
 
 namespace base {
 
 namespace {
 // To avoid users sending negative message lengths to Send/Receive
 // we clamp message lengths, which are size_t, to no more than INT_MAX.
 const size_t kMaxMessageLength = static_cast<size_t>(INT_MAX);
 
+// Writes |length| of |buffer| into |handle|.  Returns the number of bytes
+// written or zero on error.  |length| must be greater than 0.
+size_t SendHelper(SyncSocket::Handle handle,
+                  const void* buffer,
+                  size_t length) {
+  DCHECK_GT(length, 0u);
+  DCHECK_LE(length, kMaxMessageLength);
+  DCHECK_NE(handle, SyncSocket::kInvalidHandle);
+  const char* charbuffer = static_cast<const char*>(buffer);
+  const int len = file_util::WriteFileDescriptor(handle, charbuffer, length);
+  return len < 0 ? 0 : static_cast<size_t>(len);
+}
+
 }  // namespace
 
 const SyncSocket::Handle SyncSocket::kInvalidHandle = -1;
 
 SyncSocket::SyncSocket() : handle_(kInvalidHandle) {}
 
 SyncSocket::~SyncSocket() {
   Close();
 }
 
 // static
 bool SyncSocket::CreatePair(SyncSocket* socket_a, SyncSocket* socket_b) {
-  DCHECK(socket_a != socket_b);
-  DCHECK(socket_a->handle_ == kInvalidHandle);
-  DCHECK(socket_b->handle_ == kInvalidHandle);
+  DCHECK_NE(socket_a, socket_b);
+  DCHECK_EQ(socket_a->handle_, kInvalidHandle);
+  DCHECK_EQ(socket_b->handle_, kInvalidHandle);
 
 #if defined(OS_MACOSX)
   int nosigpipe = 1;
 #endif  // defined(OS_MACOSX)
 
   Handle handles[2] = { kInvalidHandle, kInvalidHandle };
   if (socketpair(AF_UNIX, SOCK_STREAM, 0, handles) != 0)
     goto cleanup;
 
 #if defined(OS_MACOSX)
@@ skipping 20 matching lines @@
   }
   if (handles[1] != kInvalidHandle) {
     if (HANDLE_EINTR(close(handles[1])) < 0)
       DPLOG(ERROR) << "close";
   }
 
   return false;
 }
 
 bool SyncSocket::Close() {
-  if (handle_ == kInvalidHandle) {
-    return false;
-  }
-  int retval = HANDLE_EINTR(close(handle_));
-  if (retval < 0)
-    DPLOG(ERROR) << "close";
+  if (handle_ == kInvalidHandle)
+    return true;
+  const int retval = HANDLE_EINTR(close(handle_));
+  DPLOG_IF(ERROR, retval < 0) << "close";
   handle_ = kInvalidHandle;
-  return (retval == 0);
+  return retval == 0;
 }
 
 size_t SyncSocket::Send(const void* buffer, size_t length) {
-  DCHECK_LE(length, kMaxMessageLength);
-  const char* charbuffer = static_cast<const char*>(buffer);
-  int len = file_util::WriteFileDescriptor(handle_, charbuffer, length);
-
-  return (len == -1) ? 0 : static_cast<size_t>(len);
+  ThreadRestrictions::AssertIOAllowed();
+  return SendHelper(handle_, buffer, length);
 }
 
 size_t SyncSocket::Receive(void* buffer, size_t length) {
+  ThreadRestrictions::AssertIOAllowed();
+  DCHECK_GT(length, 0u);
   DCHECK_LE(length, kMaxMessageLength);
+  DCHECK_NE(handle_, kInvalidHandle);
   char* charbuffer = static_cast<char*>(buffer);
   if (file_util::ReadFromFD(handle_, charbuffer, length))
     return length;
   return 0;
 }
 
+size_t SyncSocket::ReceiveWithTimeout(void* buffer,
+                                      size_t length,
+                                      TimeDelta timeout) {
+  ThreadRestrictions::AssertIOAllowed();
+  DCHECK_GT(length, 0u);
+  DCHECK_LE(length, kMaxMessageLength);
+  DCHECK_NE(handle_, kInvalidHandle);
+  DCHECK_LT(handle_, FD_SETSIZE);
+
+  // Only timeouts greater than zero and less than one second are allowed.
+  DCHECK_GT(timeout.InMicroseconds(), 0);
+  DCHECK_LT(timeout.InMicroseconds(),
+            base::TimeDelta::FromSeconds(1).InMicroseconds());
+
+  // Track the start time so we can reduce the timeout as data is read.
+  TimeTicks start_time = TimeTicks::Now();
+  const TimeTicks finish_time = start_time + timeout;
+
+  fd_set read_fds;
+  size_t bytes_read_total;
+  for (bytes_read_total = 0;
+       bytes_read_total < length && timeout.InMicroseconds() > 0;

[jar (doing other things), 2013/10/18 22:02:52]

nit: I don't know if this is time sensitive code (and I don't know how often it
has to cycle around and chew cycles)... but...

InMicroseconds() is a 64bit divide, which on Windows is done via a library call,
and is slow. (I recall being surprised when I single stepped through assembly
code a few years ago).  I don't know how this plays on Linux.

If you tracked timeout_ms as a separate variable (declare on line 139, update on
line 144), then you wouldn't have to do this conversion twice per iteration
(second time is on line 150).

Clarity seems very comparable.. code a tad shorter....  YMMV.

Your call.

[DaleCurtis, 2013/10/18 22:10:50]

I don't see where it maps to a divide?
https://code.google.com/p/chromium/codesearch#chromium/src/base/time/time.cc&...

[jar (doing other things), 2013/10/18 22:29:57]

<DOH!>  Nevermind.  I'm so used to looking at ToMilliseconds().  As you noted,
ToMicroseconds() uses the internal format, does no math (beyond the cast you
need to discard high-order bits... which is NBD).

Nevermind.  Sorry.

+       timeout = finish_time - base::TimeTicks::Now()) {
+    FD_ZERO(&read_fds);
+    FD_SET(handle_, &read_fds);
+
+    // Wait for data to become available.
+    struct timeval timeout_struct =
+        { 0, static_cast<suseconds_t>(timeout.InMicroseconds()) };
+    const int select_result =
+        select(handle_ + 1, &read_fds, NULL, NULL, &timeout_struct);
+    // Handle EINTR manually since we need to update the timeout value.
+    if (select_result == -1 && errno == EINTR)
+      continue;
+    if (select_result <= 0)
+      return bytes_read_total;
+
+    // select() only tells us that data is ready for reading, not how much.  We
+    // must Peek() for the amount ready for reading to avoid blocking.
+    DCHECK(FD_ISSET(handle_, &read_fds));
+    const size_t bytes_to_read = std::min(Peek(), length - bytes_read_total);
+
+    // There may be zero bytes to read if the socket at the other end closed.
+    if (!bytes_to_read)
+      return bytes_read_total;
+
+    const size_t bytes_received =
+        Receive(static_cast<char*>(buffer) + bytes_read_total, bytes_to_read);
+    bytes_read_total += bytes_received;
+    if (bytes_received != bytes_to_read)
+      return bytes_read_total;
+  }
+
+  return bytes_read_total;
+}
+
 size_t SyncSocket::Peek() {
-  int number_chars;
-  if (-1 == ioctl(handle_, FIONREAD, &number_chars)) {
+  DCHECK_NE(handle_, kInvalidHandle);
+  int number_chars = 0;
+  if (ioctl(handle_, FIONREAD, &number_chars) == -1) {
     // If there is an error in ioctl, signal that the channel would block.
     return 0;
   }
-  return (size_t) number_chars;
+  DCHECK_GE(number_chars, 0);
+  return number_chars;
 }
 
 CancelableSyncSocket::CancelableSyncSocket() {}
 CancelableSyncSocket::CancelableSyncSocket(Handle handle)
     : SyncSocket(handle) {
 }
 
 bool CancelableSyncSocket::Shutdown() {
-  return HANDLE_EINTR(shutdown(handle(), SHUT_RDWR)) >= 0;
+  DCHECK_NE(handle_, kInvalidHandle);
+  return HANDLE_EINTR(shutdown(handle_, SHUT_RDWR)) >= 0;
 }
 
 size_t CancelableSyncSocket::Send(const void* buffer, size_t length) {
-  long flags = 0;
-  flags = fcntl(handle_, F_GETFL, NULL);
+  DCHECK_GT(length, 0u);
+  DCHECK_LE(length, kMaxMessageLength);
+  DCHECK_NE(handle_, kInvalidHandle);
+
+  const long flags = fcntl(handle_, F_GETFL, NULL);
   if (flags != -1 && (flags & O_NONBLOCK) == 0) {
     // Set the socket to non-blocking mode for sending if its original mode
     // is blocking.
     fcntl(handle_, F_SETFL, flags | O_NONBLOCK);
   }
 
-  size_t len = SyncSocket::Send(buffer, length);
+  const size_t len = SendHelper(handle_, buffer, length);
 
   if (flags != -1 && (flags & O_NONBLOCK) == 0) {
     // Restore the original flags.
     fcntl(handle_, F_SETFL, flags);
   }
 
   return len;
 }
 
 // static
 bool CancelableSyncSocket::CreatePair(CancelableSyncSocket* socket_a,
                                       CancelableSyncSocket* socket_b) {
   return SyncSocket::CreatePair(socket_a, socket_b);
 }
 
 }  // namespace base