Implement support for a cancelable SyncSocket.

base/sync_socket_win.cc

 // Copyright (c) 2011 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 <limits.h>
-#include <stdio.h>
-#include <windows.h>
-#include <sys/types.h>
+
 #include "base/logging.h"
+#include "base/win/scoped_handle.h"
+
+using base::win::ScopedHandle;
 
 
 namespace base {
 
 namespace {
 // This prefix used to be appended to pipe names for pipes
 // created in CreatePair.
 const wchar_t kPipePrefix[] = L"\\\\.\\pipe\\chrome.sync.";
 const size_t kPipePrefixSize = arraysize(kPipePrefix);
 const size_t kPathMax = 28;  // print length of process id + pair count.
 const size_t kPipePathMax = kPipePrefixSize + kPathMax + 1;
 
 // 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);
 
 const int kOutBufferSize = 4096;
 const int kInBufferSize = 4096;
 const int kDefaultTimeoutMilliSeconds = 1000;
 
 static const SyncSocket::Handle kInvalidHandle = INVALID_HANDLE_VALUE;
 
-}  // namespace
+bool CreatePairImpl(HANDLE* socket_a, HANDLE* socket_b, bool overlapped) {
+  DCHECK(socket_a != socket_b);
+  DCHECK(*socket_a == kInvalidHandle);
+  DCHECK(*socket_b == kInvalidHandle);
 
-bool SyncSocket::CreatePair(SyncSocket* pair[2]) {
-  Handle handles[2];
-  SyncSocket* tmp_sockets[2];
+  wchar_t name[kPipePathMax];
+  ScopedHandle handle_a;
+  DWORD flags = PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE;
+  if (overlapped)
+    flags |= FILE_FLAG_OVERLAPPED;
 
-  // Create the two SyncSocket objects first to avoid ugly cleanup issues.
-  tmp_sockets[0] = new SyncSocket(kInvalidHandle);
-  if (tmp_sockets[0] == NULL) {
-    return false;
-  }
-  tmp_sockets[1] = new SyncSocket(kInvalidHandle);
-  if (tmp_sockets[1] == NULL) {
-    delete tmp_sockets[0];
+  do {
+    unsigned int rnd_name;
+    if (rand_s(&rnd_name) != 0) {
+      NOTREACHED();
+      return false;
+    }
+
+    swprintf(name, kPipePathMax, L"%s%u.%lu",
+             kPipePrefix, GetCurrentProcessId(),
+             rnd_name);
+
+    handle_a.Set(CreateNamedPipeW(name,
+        flags, PIPE_TYPE_BYTE | PIPE_READMODE_BYTE, 1, kOutBufferSize,
+        kInBufferSize, kDefaultTimeoutMilliSeconds, NULL));
+
+    if (!handle_a.IsValid()) {
+      DWORD error = GetLastError();
+      if (error != ERROR_ACCESS_DENIED && error != ERROR_PIPE_BUSY)
+        return false;
+    }
+  } while (!handle_a.IsValid());
+
+  flags = SECURITY_SQOS_PRESENT | SECURITY_ANONYMOUS;
+  if (overlapped)
+    flags |= FILE_FLAG_OVERLAPPED;
+
+  ScopedHandle handle_b(CreateFileW(name,
+      GENERIC_READ | GENERIC_WRITE,
+      0,              // no sharing.
+      NULL,           // default security attributes.
+      OPEN_EXISTING,  // opens existing pipe.
+      flags,          // no impersonation.
+      NULL));         // no template file.
+  if (!handle_b.IsValid()) {
+    DPLOG(ERROR) << "CreateFileW failed";
     return false;
   }
 
-  wchar_t name[kPipePathMax];
-  do {
-    unsigned int rnd_name;
-    if (rand_s(&rnd_name) != 0)
-      return false;
-    swprintf(name, kPipePathMax, L"%s%u.%lu",
-             kPipePrefix, GetCurrentProcessId(),
-             rnd_name);
-    handles[0] = CreateNamedPipeW(
-        name,
-        PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE,
-        PIPE_TYPE_BYTE | PIPE_READMODE_BYTE,
-        1,
-        kOutBufferSize,
-        kInBufferSize,
-        kDefaultTimeoutMilliSeconds,
-        NULL);
-    if (handles[0] == INVALID_HANDLE_VALUE &&
-        GetLastError() != ERROR_ACCESS_DENIED &&
-        GetLastError() != ERROR_PIPE_BUSY) {
-      return false;
-    }
-  } while (handles[0] == INVALID_HANDLE_VALUE);
-  handles[1] = CreateFileW(name,
-                           GENERIC_READ | GENERIC_WRITE,
-                           0,              // no sharing.
-                           NULL,           // default security attributes.
-                           OPEN_EXISTING,  // opens existing pipe.
-                           SECURITY_SQOS_PRESENT | SECURITY_ANONYMOUS,
-                                           // no impersonation.
-                           NULL);          // no template file.
-  if (handles[1] == INVALID_HANDLE_VALUE) {
-    CloseHandle(handles[0]);
-    return false;
-  }
-  if (ConnectNamedPipe(handles[0], NULL) == FALSE) {
+  if (!ConnectNamedPipe(handle_a, NULL)) {
     DWORD error = GetLastError();
     if (error != ERROR_PIPE_CONNECTED) {
-      CloseHandle(handles[0]);
-      CloseHandle(handles[1]);
+      DPLOG(ERROR) << "ConnectNamedPipe failed";
       return false;
     }
   }
-  // Copy the handles out for successful return.
-  tmp_sockets[0]->handle_ = handles[0];
-  pair[0] = tmp_sockets[0];
-  tmp_sockets[1]->handle_ = handles[1];
-  pair[1] = tmp_sockets[1];
+
+  *socket_a = handle_a.Take();
+  *socket_b = handle_b.Take();
+
   return true;
 }
 
+// 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) {
+  // 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 len;
+    // The following statement is for 64 bit portability.
+    DWORD chunk = static_cast<DWORD>(
+        ((length - count) <= UINT_MAX) ? (length - count) : UINT_MAX);
+    // This is either the ReadFile or WriteFile call depending on whether
+    // we're receiving or sending data.
+    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) {
+          VLOG(1) << "Shutdown was signaled. Closing socket.";
+          socket->Close();
+          break;
+        } else {
+          GetOverlappedResult(file, &ol, &len, TRUE);
+        }
+      } else {
+        return (0 < count) ? count : 0;
+      }
+    }
+    count += len;
+  }
+  return count;
+}
+
+}  // namespace
+
+SyncSocket::SyncSocket() : handle_(kInvalidHandle) {}
+
+SyncSocket::~SyncSocket() {
+  Close();
+}
+
+// static
+bool SyncSocket::CreatePair(SyncSocket* socket_a, SyncSocket* socket_b) {
+  return CreatePairImpl(&socket_a->handle_, &socket_b->handle_, false);
+}
+
 bool SyncSocket::Close() {
-  if (handle_ == kInvalidHandle) {
+  if (handle_ == kInvalidHandle)
     return false;
-  }
+
   BOOL retval = CloseHandle(handle_);
   handle_ = kInvalidHandle;
   return retval ? true : false;
 }
 
 size_t SyncSocket::Send(const void* buffer, size_t length) {
   DCHECK_LE(length, kMaxMessageLength);
   size_t count = 0;
   while (count < length) {
     DWORD len;
     // The following statement is for 64 bit portability.
     DWORD chunk = static_cast<DWORD>(
-      ((length - count) <= UINT_MAX) ? (length - count) : UINT_MAX);
+        ((length - count) <= UINT_MAX) ? (length - count) : UINT_MAX);
     if (WriteFile(handle_, static_cast<const char*>(buffer) + count,
                   chunk, &len, NULL) == FALSE) {
       return (0 < count) ? count : 0;
     }
     count += len;
   }
   return count;
 }
 
 size_t SyncSocket::Receive(void* buffer, size_t length) {
   DCHECK_LE(length, kMaxMessageLength);
   size_t count = 0;
   while (count < length) {
     DWORD len;
     DWORD chunk = static_cast<DWORD>(
-      ((length - count) <= UINT_MAX) ? (length - count) : UINT_MAX);
+        ((length - count) <= UINT_MAX) ? (length - count) : UINT_MAX);
     if (ReadFile(handle_, static_cast<char*>(buffer) + count,
                  chunk, &len, NULL) == FALSE) {
       return (0 < count) ? count : 0;
     }
     count += len;
   }
   return count;
 }
 
 size_t SyncSocket::Peek() {
   DWORD available = 0;
   PeekNamedPipe(handle_, NULL, 0, NULL, &available, NULL);
   return available;
 }
 
+CancelableSyncSocket::CancelableSyncSocket()
+    : shutdown_event_(true, false), file_operation_(true, false) {
+}
+
+CancelableSyncSocket::CancelableSyncSocket(Handle handle)
+    : SyncSocket(handle), shutdown_event_(true, false),
+      file_operation_(true, false) {
+}
+
+bool CancelableSyncSocket::Shutdown() {
+  // This doesn't shut down the pipe immediately, but subsequent Receive or Send
+  // methods will fail straight away.
+  shutdown_event_.Signal();
+  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);
+}
+
+size_t CancelableSyncSocket::Receive(void* buffer, size_t length) {
+  return CancelableFileOperation(&ReadFile, handle_,
+      reinterpret_cast<char*>(buffer), length, &file_operation_,
+      &shutdown_event_, this);
+}
+
+// static
+bool CancelableSyncSocket::CreatePair(CancelableSyncSocket* socket_a,
+                                      CancelableSyncSocket* socket_b) {
+  return CreatePairImpl(&socket_a->handle_, &socket_b->handle_, true);
+}
+
+
 }  // namespace base