Reland 10000004 - Make the CancellableSyncSocket non-blocking on Send, and blocking on Receive

base/sync_socket_win.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 "base/logging.h"
 #include "base/win/scoped_handle.h"
 
 namespace base {
@@ skipping 96 matching lines @@
 
 // Template function that supports calling ReadFile or WriteFile in an
 // overlapped fashion and waits for IO completion.  The function also waits
 // on an event that can be used to cancel the operation.  If the operation
 // is cancelled, the function returns and closes the relevant socket object.
 template <typename BufferType, typename Function>
 size_t CancelableFileOperation(Function operation, HANDLE file,
                                BufferType* buffer, size_t length,
                                base::WaitableEvent* io_event,
                                base::WaitableEvent* cancel_event,
-                               CancelableSyncSocket* socket) {
+                               CancelableSyncSocket* socket,
+                               DWORD timeout_in_ms) {
   // The buffer must be byte size or the length check won't make much sense.
   COMPILE_ASSERT(sizeof(buffer[0]) == sizeof(char), incorrect_buffer_type);
   DCHECK_LE(length, kMaxMessageLength);
 
   OVERLAPPED ol = {0};
   ol.hEvent = io_event->handle();
   size_t count = 0;
   while (count < length) {
     DWORD chunk = GetNextChunkSize(count, length);
     // This is either the ReadFile or WriteFile call depending on whether
     // we're receiving or sending data.
     DWORD len = 0;
     BOOL ok = operation(file, static_cast<BufferType*>(buffer) + count, chunk,
                         &len, &ol);
     if (!ok) {
       if (::GetLastError() == ERROR_IO_PENDING) {
-        base::WaitableEvent* events[] = { io_event, cancel_event };
-        size_t signaled = WaitableEvent::WaitMany(events, arraysize(events));
-        if (signaled == 1) {
+        HANDLE events[] = { io_event->handle(), cancel_event->handle() };
+        int wait_result = WaitForMultipleObjects(
+            arraysize(events), events, FALSE, timeout_in_ms);
+        if (wait_result == (WAIT_OBJECT_0 + 0)) {
+          GetOverlappedResult(file, &ol, &len, TRUE);
+        } else if (wait_result == (WAIT_OBJECT_0 + 1)) {
           VLOG(1) << "Shutdown was signaled. Closing socket.";
           CancelIo(file);
           socket->Close();
           count = 0;
           break;
         } else {
-          GetOverlappedResult(file, &ol, &len, TRUE);
+          // Timeout happened.
+          DCHECK_EQ(WAIT_TIMEOUT, wait_result);
+          if (!CancelIo(file)){
+            DLOG(WARNING) << "CancelIo() failed";
+          }
+          break;
         }
       } else {
-        return (0 < count) ? count : 0;
+        break;
       }
     }
+
     count += len;
+
+    // Quit the operation if we can't write/read anymore.
+    if (len != chunk)
+      break;
   }
-  return count;
+
+  return (count > 0) ? count : 0;
 }
 
 }  // namespace
 
 const SyncSocket::Handle SyncSocket::kInvalidHandle = INVALID_HANDLE_VALUE;
 
 SyncSocket::SyncSocket() : handle_(kInvalidHandle) {}
 
 SyncSocket::~SyncSocket() {
   Close();
@@ skipping 65 matching lines @@
   return true;
 }
 
 bool CancelableSyncSocket::Close() {
   bool ret = SyncSocket::Close();
   shutdown_event_.Reset();
   return ret;
 }
 
 size_t CancelableSyncSocket::Send(const void* buffer, size_t length) {
-  return CancelableFileOperation(&WriteFile, handle_,
-      reinterpret_cast<const char*>(buffer), length, &file_operation_,
-      &shutdown_event_, this);
+  static const DWORD kWaitTimeOutInMs = 500;
+  return CancelableFileOperation(
+      &WriteFile, handle_, reinterpret_cast<const char*>(buffer),
+      length, &file_operation_, &shutdown_event_, this, kWaitTimeOutInMs);
 }
 
 size_t CancelableSyncSocket::Receive(void* buffer, size_t length) {
   return CancelableFileOperation(&ReadFile, handle_,
       reinterpret_cast<char*>(buffer), length, &file_operation_,
-      &shutdown_event_, this);
+      &shutdown_event_, this, INFINITE);
 }
 
 // static
 bool CancelableSyncSocket::CreatePair(CancelableSyncSocket* socket_a,
                                       CancelableSyncSocket* socket_b) {
   return CreatePairImpl(&socket_a->handle_, &socket_b->handle_, true);
 }
 
 
 }  // namespace base