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);
 
 }  // 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;

[jar (doing other things), 2013/10/10 02:01:28]

Why has this changed to false?

[DaleCurtis, 2013/10/10 17:37:34]

CancelableSyncSocket::Shutdown() will result in the handle being closed and set
to kInvalidHandle.  As such I thought it made sense for Close() to not fail if
called after Shutdown().

+  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) {
+  ThreadRestrictions::AssertIOAllowed();
+  DCHECK_GT(length, 0u);
   DCHECK_LE(length, kMaxMessageLength);
+  DCHECK_NE(handle_, kInvalidHandle);
   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);
+  const int len = file_util::WriteFileDescriptor(handle_, charbuffer, length);
+  return len < 0 ? 0 : static_cast<size_t>(len);
 }
 
 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);

[DaleCurtis, 2013/10/11 22:48:24]

I made this a DCHECK(), but I'm wondering if it should be a CHECK or handled
error instead.

+
+  // Only timeouts greater than zero and less than one second are allowed.
+  DCHECK_GT(timeout.InMicroseconds(), 0);
+  DCHECK(timeout.InMicroseconds() < Time::kMicrosecondsPerSecond);

[jar (doing other things), 2013/10/10 02:01:28]

nit: Much easier to read is something like:

timeout < TimeDelta::FromSeconds(1)

The constants were really meant for internal use, and you should let that class
handle all such arithmetic whenever possible.

I don't know if it would work, but best would be:
DCHECK_LT(timeout, TimeDelta::FromSeconds(1));

[DaleCurtis, 2013/10/11 22:48:24]

I like the second one, but it requires adding an ostream<< operator to TimeDelta
a la:

#include <ostream>
...
inline std::ostream& operator<<(std::ostream& os, const TimeDelta& delta) {
  os << delta.InMicroseconds() << "μs";
  return os;
}

This was raised back in '09 and saw some disagreement apparently.  Time to
reconsider?

https://groups.google.com/forum/#!topic/chromium-dev/swpKeYfcCOc

+
+  // Track the start time so we can reduce the timeout as data is read.
+  TimeTicks start_time = TimeTicks::Now();
+
+  fd_set read_fds;
+  size_t bytes_read_total = 0;
+  char* charbuffer = static_cast<char*>(buffer);
+  do {
+    FD_ZERO(&read_fds);
+    FD_SET(handle_, &read_fds);
+
+    // Wait for data to become available.
+    struct timeval timeout_struct = { 0, 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(charbuffer + bytes_read_total, bytes_to_read);

[DaleCurtis, 2013/10/11 22:48:24]

I just realized that Receive might hit and handle an EINTR internally.  Since we
should be guaranteed not to block, it seems like this is okay in terms of the
timeout though.  What do you think?

+    bytes_read_total += bytes_received;
+    if (bytes_received != bytes_to_read)
+      return bytes_read_total;
+  } while (bytes_read_total < length &&
+           (timeout -= (TimeTicks::Now() - start_time)) > TimeDelta());

[jar (doing other things), 2013/10/10 02:01:28]

This looks wrong.  

It appears that you're decrementing |timeout| repeatedly.  If you cycle through
this loop a few times, you can easily exhaust |timeout|, without spending a full
|timeout| period.

I think the correct test to continue is:

TimeTicks::Now() - start_time < timeout

[DaleCurtis, 2013/10/10 17:37:34]

Can you elaborate?  We need to update the timeout value we give to select() so
that the time spent both in select() and the loop doesn't exceed timeout.  Are
you proposing a different update mechanism than repeated subtraction of the
delta?

[jar (doing other things), 2013/10/11 02:36:12]

The following is an explanation of how your code appears to go wrong.

Example: You are asked to get all the data you can for the next 500ms (i.e.,
|timeout| == 500ms).  

After only waiting for 50ms, select() returns and indicates that data is ready. 
You read the data.  
You realize that 50ms (Now() - start_time) has passed since being called, and
you reduce |timeout| to 450ms.  
Things are looking good as you loop back to line 133.
You enter select with a delay of 450ms to wait, which is dead on.
Select returns with more data ready after only 10ms.  
You are now 60ms into your attempted wait of "no more than 500ms."
You read the data... and now the bug surfaces...
Line 162 calculates that you are indeed 60ms into your wait, and further reduces
|timeout| by 60ms <yikes!?!>.  
The value of timeout is now a mere 390ms, as it was reduced by 50ms the first
time through line 162, and then by 60ms this time.  
Sadly, we have only waited a total of 60ms since we were called.

We *should* next call select() with a full 440ms remaining, but we instead call
it with a mere 390ms. :-(

Another few passes like this (no more than 10?) and we'll diminish |timeout| to
zero (and return), even though we won't have spent much time in this function
:-(.

To say it another way: You shouldn't diminish |timeout| by the duration you have
run since start_time each pass through the loop.  You should reset |timeout| to
the remnant duration.

Perhaps you were assuming that you reset |start_time| every pass in the loop. 
Please note that it is never updated (that I could find).

As a matter of style, pushing a "-=" assignment into the already complex
conditional of a while loop adds a lot of readability issues, and may be a part
of why this code caused confusion (and made it harder to see the error).  To get
rid of this, you should probably (in this case) avoid using "continue" on line
143, and instead use an indented block to hold the code from lines 144 to 160. 

My original comment focused on the error in the while loop condition shown
above.  The following is the skeletal code that you could use to have a correct
value of |timeout| to feed to the timeout_struct on line 138 (and still make a
minimal number of calls to Now()).

TimeTicks finish_time = TimeTicks::Now() + timeout; // new line 129
...
do {  // Current line 133.
  if (timeout < TimeDelta())
    break;

  ... etc. Lines 134 onwards

  }  // Close new block around lines 144-160.
  if (bytes_read_total >= length)
    break;
  timeout = finish_time - TimeTicks::Now();
} while (1);

[DaleCurtis, 2013/10/11 22:48:24]

Ack, I shouldn't have missed that. :( Thanks for your patient and detailed
explanation.  I started writing up your suggestion, but didn't like the extra
conditionals, indentation, and while(1) it introduced.

I've switched it over to a for() loop which allows the EINTR and exit
conditionals to be simplified, while using a construct more appropriate for
containing assignments.

+  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() {
+  DCHECK_NE(handle_, kInvalidHandle);
   return HANDLE_EINTR(shutdown(handle(), SHUT_RDWR)) >= 0;
 }
 
 size_t CancelableSyncSocket::Send(const void* buffer, size_t length) {
+  DCHECK_GT(length, 0u);
+  DCHECK_LE(length, kMaxMessageLength);
+  DCHECK_NE(handle_, kInvalidHandle);
+
   long flags = 0;
   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);
 
   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