Last set of fixes to make the src/mojo/edk pass the page cycler.

mojo/edk/system/raw_channel_win.cc

 // Copyright 2013 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 "mojo/edk/system/raw_channel.h"
 
 #include <windows.h>
 
 #include "base/bind.h"
 #include "base/lazy_instance.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop/message_loop.h"
 #include "base/process/process.h"
 #include "base/synchronization/lock.h"
-#include "base/win/object_watcher.h"
+#include "base/win/scoped_handle.h"
 #include "base/win/windows_version.h"
 #include "mojo/edk/embedder/platform_handle.h"
 #include "mojo/edk/system/transport_data.h"
 #include "mojo/public/cpp/system/macros.h"
 
 #define STATUS_CANCELLED 0xC0000120
 #define STATUS_PIPE_BROKEN 0xC000014B
 
-// We can't use IO completion ports if we send a message pipe. The reason is
-// that the only way to stop an existing IOCP is by closing the pipe handle.
-// See https://msdn.microsoft.com/en-us/library/windows/hardware/ff545834(v=vs.85).aspx
-bool g_use_iocp = false;
-
-// Manual reset per
-// Doc for overlapped I/O says use manual per
-//   https://msdn.microsoft.com/en-us/library/windows/desktop/ms684342(v=vs.85).aspx
-// However using an auto-reset event makes the perf test 5x faster and also
-// works since we don't wait on the event elsewhere or call GetOverlappedResult
-// before it fires.
-bool g_use_autoreset_event = true;
-
 namespace mojo {
 namespace edk {
 
 namespace {
 
 struct MOJO_ALIGNAS(8) SerializedHandle {
   DWORD handle_pid;
   HANDLE handle;
 };
 
 class VistaOrHigherFunctions {
  public:
   VistaOrHigherFunctions()
       : is_vista_or_higher_(
           base::win::GetVersion() >= base::win::VERSION_VISTA),
-        set_file_completion_notification_modes_(nullptr),
         cancel_io_ex_(nullptr),
         get_file_information_by_handle_ex_(nullptr) {
     if (!is_vista_or_higher_)
       return;
 
     HMODULE module = GetModuleHandleW(L"kernel32.dll");
-    set_file_completion_notification_modes_ =
-        reinterpret_cast<SetFileCompletionNotificationModesFunc>(
-            GetProcAddress(module, "SetFileCompletionNotificationModes"));
-    DCHECK(set_file_completion_notification_modes_);
-
     cancel_io_ex_ =
         reinterpret_cast<CancelIoExFunc>(GetProcAddress(module, "CancelIoEx"));
     DCHECK(cancel_io_ex_);
 
     get_file_information_by_handle_ex_ =
         reinterpret_cast<GetFileInformationByHandleExFunc>(
             GetProcAddress(module, "GetFileInformationByHandleEx"));
     DCHECK(get_file_information_by_handle_ex_);
   }
 
   bool is_vista_or_higher() const { return is_vista_or_higher_; }
 
-  BOOL SetFileCompletionNotificationModes(HANDLE handle, UCHAR flags) {
-    return set_file_completion_notification_modes_(handle, flags);
-  }
-
   BOOL CancelIoEx(HANDLE handle, LPOVERLAPPED overlapped) {
     return cancel_io_ex_(handle, overlapped);
   }
 
   BOOL GetFileInformationByHandleEx(HANDLE handle,
                                     FILE_INFO_BY_HANDLE_CLASS file_info_class,
                                     LPVOID file_info,
                                     DWORD buffer_size) {
     return get_file_information_by_handle_ex_(
         handle, file_info_class, file_info, buffer_size);
   }
 
  private:
-  using SetFileCompletionNotificationModesFunc = BOOL(WINAPI*)(HANDLE, UCHAR);
   using CancelIoExFunc = BOOL(WINAPI*)(HANDLE, LPOVERLAPPED);
   using GetFileInformationByHandleExFunc = BOOL(WINAPI*)(
       HANDLE, FILE_INFO_BY_HANDLE_CLASS, LPVOID, DWORD);
 
   bool is_vista_or_higher_;
-  SetFileCompletionNotificationModesFunc
-      set_file_completion_notification_modes_;
   CancelIoExFunc cancel_io_ex_;
   GetFileInformationByHandleExFunc get_file_information_by_handle_ex_;
 };
 
 base::LazyInstance<VistaOrHigherFunctions> g_vista_or_higher_functions =
     LAZY_INSTANCE_INITIALIZER;
 
 class RawChannelWin final : public RawChannel {
  public:
   RawChannelWin(ScopedPlatformHandle handle)
-      : handle_(handle.Pass()),
-        io_handler_(nullptr),
-        skip_completion_port_on_success_(
-            g_use_iocp &&
-            g_vista_or_higher_functions.Get().is_vista_or_higher()) {
+      : handle_(handle.Pass()), io_handler_(nullptr) {
     DCHECK(handle_.is_valid());
   }
   ~RawChannelWin() override {
     DCHECK(!io_handler_);
   }
 
  private:
-  // RawChannelIOHandler receives OS notifications for I/O completion. It must
-  // be created on the I/O thread.
+  // RawChannelIOHandler receives OS notifications for I/O completion. Currently
+  // this object is only used on the IO thread, other than ReleaseHandle. But
+  // there's nothing preventing using this on other threads.
   //
   // It manages its own destruction. Destruction happens on the I/O thread when
   // all the following conditions are satisfied:
   //   - |DetachFromOwnerNoLock()| has been called;
   //   - there is no pending read;
   //   - there is no pending write.
-  class RawChannelIOHandler : public base::MessageLoopForIO::IOHandler,
-    public base::win::ObjectWatcher::Delegate {
+  class RawChannelIOHandler {
    public:
     RawChannelIOHandler(RawChannelWin* owner,
                         ScopedPlatformHandle handle)
         : handle_(handle.Pass()),
           owner_(owner),
           suppress_self_destruct_(false),
           pending_read_(false),
           pending_write_(false),
-          platform_handles_written_(0) {
+          platform_handles_written_(0),
+          read_event_(CreateEvent(NULL, FALSE, FALSE, NULL)),
+          write_event_(CreateEvent(NULL, FALSE, FALSE, NULL)),
+          read_wait_object_(NULL),
+          write_wait_object_(NULL),
+          read_event_signalled_(false),
+          write_event_signalled_(false),
+          message_loop_for_io_(base::MessageLoop::current()->task_runner()),
+          weak_ptr_factory_(this) {
       memset(&read_context_.overlapped, 0, sizeof(read_context_.overlapped));
       memset(&write_context_.overlapped, 0, sizeof(write_context_.overlapped));
-      if (g_use_iocp) {
-        owner_->message_loop_for_io()->RegisterIOHandler(
-            handle_.get().handle, this);
-        read_context_.handler = this;
-        write_context_.handler = this;
-      } else {
-        read_event = CreateEvent(
-            NULL, g_use_autoreset_event ? FALSE : TRUE, FALSE, NULL);
-        write_event = CreateEvent(
-            NULL, g_use_autoreset_event ? FALSE : TRUE, FALSE, NULL);
-        read_context_.overlapped.hEvent = read_event;
-        write_context_.overlapped.hEvent = write_event;
+      read_context_.overlapped.hEvent = read_event_.Get();
+      write_context_.overlapped.hEvent = write_event_.Get();
 
-
-        if (g_use_autoreset_event) {
-          read_watcher_.StartWatchingMultipleTimes(read_event, this);
-          write_watcher_.StartWatchingMultipleTimes(write_event, this);
-        }
-      }
+      this_weakptr_ = weak_ptr_factory_.GetWeakPtr();
+      RegisterWaitForSingleObject(&read_wait_object_, read_event_.Get(),
+          ReadCompleted, this, INFINITE, WT_EXECUTEINWAITTHREAD);
+      RegisterWaitForSingleObject(&write_wait_object_, write_event_.Get(),
+          WriteCompleted, this, INFINITE, WT_EXECUTEINWAITTHREAD);
     }
 
-    ~RawChannelIOHandler() override {
+    ~RawChannelIOHandler() {
+      if (read_wait_object_)
+        UnregisterWaitEx(read_wait_object_, INVALID_HANDLE_VALUE);
+
+      if (write_wait_object_)
+        UnregisterWaitEx(write_wait_object_, INVALID_HANDLE_VALUE);
       DCHECK(ShouldSelfDestruct());
     }
 
     HANDLE handle() const { return handle_.get().handle; }
 
     // The following methods are only called by the owner on the I/O thread.
     bool pending_read() const {
       DCHECK(owner_);
       DCHECK_EQ(base::MessageLoop::current(), owner_->message_loop_for_io());
       return pending_read_;
     }
 
     base::MessageLoopForIO::IOContext* read_context() {
       DCHECK(owner_);
       DCHECK_EQ(base::MessageLoop::current(), owner_->message_loop_for_io());
       return &read_context_;
     }
 
-    // Instructs the object to wait for an |OnIOCompleted()| notification.
+    // Instructs the object to wait for an OnObjectSignaled notification.
     void OnPendingReadStarted() {
       DCHECK(owner_);
       DCHECK_EQ(base::MessageLoop::current(), owner_->message_loop_for_io());
       DCHECK(!pending_read_);
       pending_read_ = true;
+      read_event_signalled_ = false;
     }
 
     // The following methods are only called by the owner under
     // |owner_->write_lock()|.
     bool pending_write_no_lock() const {
       DCHECK(owner_);
       owner_->write_lock().AssertAcquired();
       return pending_write_;
     }
 
     base::MessageLoopForIO::IOContext* write_context_no_lock() {
       DCHECK(owner_);
       owner_->write_lock().AssertAcquired();
       return &write_context_;
     }
-    // Instructs the object to wait for an |OnIOCompleted()| notification.
+
+    // Instructs the object to wait for an OnObjectSignaled notification.
     void OnPendingWriteStartedNoLock(size_t platform_handles_written) {
       DCHECK(owner_);
       owner_->write_lock().AssertAcquired();
       DCHECK(!pending_write_);
       pending_write_ = true;
+      write_event_signalled_ = false;
       platform_handles_written_ = platform_handles_written;
     }
 
-    // |base::MessageLoopForIO::IOHandler| implementation:
-    // Must be called on the I/O thread. It could be called before or after
-    // detached from the owner.
-    void OnIOCompleted(base::MessageLoopForIO::IOContext* context,
-                       DWORD bytes_transferred,
-                       DWORD error) override {
-      DCHECK(!owner_ ||
-             base::MessageLoop::current() == owner_->message_loop_for_io());
-
-      // Suppress self-destruction inside |OnReadCompleted()|, etc. (in case
-      // they result in a call to |Shutdown()|).
-      bool old_suppress_self_destruct = suppress_self_destruct_;
-      suppress_self_destruct_ = true;
-
-      if (context == &read_context_)
-        OnReadCompleted(bytes_transferred, error);
-      else if (context == &write_context_)
-        OnWriteCompleted(bytes_transferred, error);
-      else
-        NOTREACHED();
-
-      // Maybe allow self-destruction again.
-      suppress_self_destruct_ = old_suppress_self_destruct;
-
-      if (ShouldSelfDestruct())
-        delete this;
-    }
-
-    // Must be called on the I/O thread under |owner_->write_lock()|.
+    // Must be called on the I/O thread under read and write locks.
     // After this call, the owner must not make any further calls on this
     // object, and therefore the object is used on the I/O thread exclusively
     // (if it stays alive).
     void DetachFromOwnerNoLock(scoped_ptr<ReadBuffer> read_buffer,
                                scoped_ptr<WriteBuffer> write_buffer) {
       DCHECK(owner_);
       DCHECK_EQ(base::MessageLoop::current(), owner_->message_loop_for_io());
-      //owner_->write_lock().AssertAcquired();
+      owner_->read_lock().AssertAcquired();
+      owner_->write_lock().AssertAcquired();
 
       // If read/write is pending, we have to retain the corresponding buffer.
       if (pending_read_)
         preserved_read_buffer_after_detach_ = read_buffer.Pass();
       if (pending_write_)
         preserved_write_buffer_after_detach_ = write_buffer.Pass();
 
       owner_ = nullptr;
       if (ShouldSelfDestruct())
         delete this;
     }
 
     ScopedPlatformHandle ReleaseHandle(
         std::vector<char>* serialized_read_buffer,
         std::vector<char>* serialized_write_buffer) {
       // Cancel pending IO calls.
       if (g_vista_or_higher_functions.Get().is_vista_or_higher()) {
         g_vista_or_higher_functions.Get().CancelIoEx(handle(), nullptr);
       } else {
         CHECK(false) << "TODO(jam): handle XP";
       }
 
       size_t additional_bytes_read = 0;
       if (pending_read_) {
+        bool wait = false;
+        UnregisterWaitEx(read_wait_object_, INVALID_HANDLE_VALUE);
+        read_wait_object_ = NULL;
+        if (!read_event_signalled_)
+          wait = true;
         DWORD bytes_read_dword = 0;
 
-        DWORD old_bytes = read_context_.overlapped.InternalHigh;
-
         // Since we cancelled pending IO calls above, we need to know if the
         // read did succeed (i.e. it completed and there's a pending task posted
         // to alert us) or if it was cancelled. This important because if the
         // read completed, we don't want to serialize those bytes again.
-        //TODO(jam): for XP, can return TRUE here to wait. also below.
+        // TODO(jam): for XP, can return TRUE here to wait. also below.
         BOOL rv = GetOverlappedResult(
-            handle(), &read_context_.overlapped, &bytes_read_dword, FALSE);
-        DCHECK_EQ(old_bytes, bytes_read_dword);
-        if (rv && read_context_.overlapped.Internal != STATUS_CANCELLED) {
-          additional_bytes_read =
-              static_cast<size_t>(read_context_.overlapped.InternalHigh);
-        }
+            handle(), &read_context_.overlapped, &bytes_read_dword,
+            wait ? TRUE : FALSE);
+        if (rv && read_context_.overlapped.Internal != STATUS_CANCELLED)
+          additional_bytes_read = bytes_read_dword;
         pending_read_ = false;
       }
 
       RawChannel::WriteBuffer* write_buffer = owner_->write_buffer_no_lock();
 
       size_t additional_bytes_written = 0;
       size_t additional_platform_handles_written = 0;
       if (pending_write_) {
+        bool wait = false;
+        UnregisterWaitEx(write_wait_object_, INVALID_HANDLE_VALUE);
+        write_wait_object_ = NULL;
+        if (!write_event_signalled_)
+          wait = true;
+
         DWORD bytes_written_dword = 0;
-        DWORD old_bytes = write_context_.overlapped.InternalHigh;
 
         // See comment above.
         BOOL rv = GetOverlappedResult(
-            handle(), &write_context_.overlapped, &bytes_written_dword, FALSE);
+            handle(), &write_context_.overlapped, &bytes_written_dword,
+            wait ? TRUE : FALSE);
 
-        DCHECK_EQ(old_bytes, bytes_written_dword);
         if (rv && write_context_.overlapped.Internal != STATUS_CANCELLED) {
           CHECK(!write_buffer->IsEmpty());
 
           additional_bytes_written = static_cast<size_t>(bytes_written_dword);
           additional_platform_handles_written = platform_handles_written_;
           platform_handles_written_ = 0;
         }
         pending_write_ = false;
       }
 
       owner_->SerializeReadBuffer(
           additional_bytes_read, serialized_read_buffer);
       owner_->SerializeWriteBuffer(
-          serialized_write_buffer, additional_bytes_written,
-          additional_platform_handles_written);
+          additional_bytes_written, additional_platform_handles_written,
+          serialized_write_buffer);
+
+      // There's a PostTask inside RawChannel because an error over the channel
+      // occurred. We need to propagate this, otherwise the object using this
+      // channel will never get a peer-closed signal.
+      if (owner_->pending_error()) {
+        handle_.reset();

[yzshen1, 2015/10/14 21:37:00]

Just to make sure I understand: does this part need to be changed in later CLs
according to our discussion about "write error doesn't stop reads"?

[jam, 2015/10/14 22:35:35]

yep

+        serialized_read_buffer->clear();
+        serialized_write_buffer->clear();
+        return ScopedPlatformHandle();
+      }
 
       return ScopedPlatformHandle(handle_.release());
     }
 
-    void OnObjectSignaled(HANDLE object) override {
-      // Since this is called on the IO thread, no locks needed for owner_.
-      bool handle_is_valid = false;
-      if (owner_)
-        owner_->read_lock().Acquire();
-      handle_is_valid = handle_.is_valid();
-      if (owner_)
-        owner_->read_lock().Release();
-      if (!handle_is_valid) {
-        if (object == read_event)
-          pending_read_ = false;
-        else
-          pending_write_ = false;
-        if (ShouldSelfDestruct())
-          delete this;
-        return;
-      }
-
-      if (object == read_event) {
-        OnIOCompleted(&read_context_, read_context_.overlapped.InternalHigh,
-                      read_context_.overlapped.Internal);
-
-      } else {
-        CHECK(object == write_event);
-        OnIOCompleted(&write_context_, write_context_.overlapped.InternalHigh,
-                      write_context_.overlapped.Internal);
-      }
-    }
-    HANDLE read_event, write_event;
-    base::win::ObjectWatcher read_watcher_, write_watcher_;
-
    private:
     // Returns true if |owner_| has been reset and there is not pending read or
     // write.
     // Must be called on the I/O thread.
     bool ShouldSelfDestruct() const {
       if (owner_ || suppress_self_destruct_)
         return false;
 
       // Note: Detached, hence no lock needed for |pending_write_|.
       return !pending_read_ && !pending_write_;
     }
 
     // Must be called on the I/O thread. It may be called before or after
     // detaching from the owner.
     void OnReadCompleted(DWORD bytes_read, DWORD error) {
       DCHECK(!owner_ ||
              base::MessageLoop::current() == owner_->message_loop_for_io());
       DCHECK(suppress_self_destruct_);
+      if (!owner_) {
+        pending_read_ = false;
+        return;
+      }
 
-      if (g_use_autoreset_event && !pending_read_)
+      // Must acquire the read lock before we update pending_read_, since
+      // otherwise there is a race condition in ReleaseHandle if this method
+      // sets it to false but ReleaseHandle acquired read lock. It would then
+      // think there's no pending read and miss the read bytes.
+      base::AutoLock locker(owner_->read_lock());
+
+      // This can happen if ReleaseHandle was called and it set pending_read to
+      // false. We don't want to call owner_->OnReadCompletedNoLock since the
+      // read_buffer has been freed.
+      if (!pending_read_)
         return;
 
       CHECK(pending_read_);
       pending_read_ = false;
-      if (!owner_)
-        return;
 
       // Note: |OnReadCompleted()| may detach us from |owner_|.
-      if (error == ERROR_SUCCESS ||
-          (g_use_autoreset_event && error == ERROR_NO_MORE_ITEMS)) {
+      if (error == ERROR_SUCCESS || error == ERROR_NO_MORE_ITEMS) {
         DCHECK_GT(bytes_read, 0u);
-        owner_->OnReadCompleted(IO_SUCCEEDED, bytes_read);
-      } else if (error == ERROR_BROKEN_PIPE  ||
-                 (g_use_autoreset_event && error == STATUS_PIPE_BROKEN)) {
+        owner_->OnReadCompletedNoLock(IO_SUCCEEDED, bytes_read);
+      } else if (error == ERROR_BROKEN_PIPE  || error == STATUS_PIPE_BROKEN) {

[yzshen1, 2015/10/14 21:37:00]

I vaguely remember that you mentioned that at the end we only need one error
code here, right?

         DCHECK_EQ(bytes_read, 0u);
-        owner_->OnReadCompleted(IO_FAILED_SHUTDOWN, 0);
+        owner_->OnReadCompletedNoLock(IO_FAILED_SHUTDOWN, 0);
       } else {
         DCHECK_EQ(bytes_read, 0u);
         LOG(WARNING) << "ReadFile: " << logging::SystemErrorCodeToString(error);
-        owner_->OnReadCompleted(IO_FAILED_UNKNOWN, 0);
+        owner_->OnReadCompletedNoLock(IO_FAILED_UNKNOWN, 0);
       }
     }
 
     // Must be called on the I/O thread. It may be called before or after
     // detaching from the owner.
     void OnWriteCompleted(DWORD bytes_written, DWORD error) {
       DCHECK(!owner_ ||
              base::MessageLoop::current() == owner_->message_loop_for_io());
       DCHECK(suppress_self_destruct_);
 
       if (!owner_) {
         // No lock needed.
         CHECK(pending_write_);
         pending_write_ = false;
         return;
       }
 
-      {
-        base::AutoLock locker(owner_->write_lock());
-        if (g_use_autoreset_event && !pending_write_)
-          return;
+      base::AutoLock locker(owner_->write_lock());
+      // This can happen if ReleaseHandle was called and it set pending_write to
+      // false. We don't want to call owner_->OnWriteCompletedNoLock since the
+      // write_buffer has been freed.
+      if (!pending_write_)
+        return;
 
-        CHECK(pending_write_);
-        pending_write_ = false;
-      }
+      CHECK(pending_write_);
+      pending_write_ = false;
 
       // Note: |OnWriteCompleted()| may detach us from |owner_|.
-      if (error == ERROR_SUCCESS ||
-         (g_use_autoreset_event && error == ERROR_NO_MORE_ITEMS)) {
+      if (error == ERROR_SUCCESS || error == ERROR_NO_MORE_ITEMS) {
         // Reset |platform_handles_written_| before calling |OnWriteCompleted()|
         // since that function may call back to this class and set it again.
         size_t local_platform_handles_written = platform_handles_written_;
         platform_handles_written_ = 0;
-        owner_->OnWriteCompleted(IO_SUCCEEDED, local_platform_handles_written,
-                                 bytes_written);
-      } else if (error == ERROR_BROKEN_PIPE  ||
-                 (g_use_autoreset_event && error == STATUS_PIPE_BROKEN)) {
-        owner_->OnWriteCompleted(IO_FAILED_SHUTDOWN, 0, 0);
+        owner_->OnWriteCompletedNoLock(
+            IO_SUCCEEDED, local_platform_handles_written, bytes_written);
+      } else if (error == ERROR_BROKEN_PIPE  || error == STATUS_PIPE_BROKEN) {
+        owner_->OnWriteCompletedNoLock(IO_FAILED_SHUTDOWN, 0, 0);
       } else {
         LOG(WARNING) << "WriteFile: "
                      << logging::SystemErrorCodeToString(error);
-        owner_->OnWriteCompleted(IO_FAILED_UNKNOWN, 0, 0);
+        owner_->OnWriteCompletedNoLock(IO_FAILED_UNKNOWN, 0, 0);
       }
     }
 
+    void OnObjectSignaled(HANDLE object) {
+      DCHECK(!owner_ ||
+             base::MessageLoop::current() == owner_->message_loop_for_io());
+      // Since this is called on the IO thread, no locks needed for owner_.
+      bool handle_is_valid = false;
+      if (owner_)
+        owner_->read_lock().Acquire();
+      handle_is_valid = handle_.is_valid();
+      if (owner_)
+        owner_->read_lock().Release();
+      if (!handle_is_valid) {
+        if (object == read_event_.Get())
+          pending_read_ = false;
+        else
+          pending_write_ = false;
+        if (ShouldSelfDestruct())
+          delete this;
+        return;
+      }
+
+      // Suppress self-destruction inside |OnReadCompleted()|, etc. (in case
+      // they result in a call to |Shutdown()|).
+      bool old_suppress_self_destruct = suppress_self_destruct_;
+      suppress_self_destruct_ = true;
+      if (object == read_event_.Get()) {
+        OnReadCompleted(read_context_.overlapped.InternalHigh,
+                        read_context_.overlapped.Internal);
+      } else {
+        CHECK(object == write_event_.Get());
+        OnWriteCompleted(write_context_.overlapped.InternalHigh,
+                         write_context_.overlapped.Internal);
+      }
+
+      // Maybe allow self-destruction again.
+      suppress_self_destruct_ = old_suppress_self_destruct;
+
+      if (ShouldSelfDestruct())
+        delete this;
+    }
+
+    static void CALLBACK ReadCompleted(void* param, BOOLEAN timed_out) {
+      DCHECK(!timed_out);
+      // The destructor blocks on any callbacks that are in flight, so we know
+      // that that is always a pointer to a valid RawChannelIOHandler.
+      RawChannelIOHandler* that = static_cast<RawChannelIOHandler*>(param);
+      that->read_event_signalled_ = true;

[yzshen1, 2015/10/14 21:37:00]

Is it safe to set the value outside of lock? ReleaseHandle could check the value
on any thread, right?

[jam, 2015/10/14 22:35:35]

ReleaseHandle only checks it after calling UnregisterWaitEx, which guarantees
that if this callback would have run, it already has completed.

+      that->message_loop_for_io_->PostTask(
+          FROM_HERE,
+          base::Bind(&RawChannelIOHandler::OnObjectSignaled,
+                     that->this_weakptr_, that->read_event_.Get()));
+    }
+
+    static void CALLBACK WriteCompleted(void* param, BOOLEAN timed_out) {
+      DCHECK(!timed_out);
+      // The destructor blocks on any callbacks that are in flight, so we know
+      // that that is always a pointer to a valid RawChannelIOHandler.
+      RawChannelIOHandler* that = static_cast<RawChannelIOHandler*>(param);
+      that->write_event_signalled_ = true;

[yzshen1, 2015/10/14 21:37:00]

ditto.

+      that->message_loop_for_io_->PostTask(
+          FROM_HERE,
+          base::Bind(&RawChannelIOHandler::OnObjectSignaled,
+                     that->this_weakptr_, that->write_event_.Get()));
+    }
+
     ScopedPlatformHandle handle_;
 
     // |owner_| is reset on the I/O thread under |owner_->write_lock()|.
     // Therefore, it may be used on any thread under lock; or on the I/O thread
     // without locking.
     RawChannelWin* owner_;
 
     // The following members must be used on the I/O thread.
     scoped_ptr<ReadBuffer> preserved_read_buffer_after_detach_;
     scoped_ptr<WriteBuffer> preserved_write_buffer_after_detach_;
     bool suppress_self_destruct_;
 
     bool pending_read_;
     base::MessageLoopForIO::IOContext read_context_;
 
     // The following members must be used under |owner_->write_lock()| while the
     // object is still attached to the owner, and only on the I/O thread
     // afterwards.
     bool pending_write_;
     size_t platform_handles_written_;
     base::MessageLoopForIO::IOContext write_context_;
 
+    base::win::ScopedHandle read_event_;
+    base::win::ScopedHandle write_event_;
+
+    HANDLE read_wait_object_;
+    HANDLE write_wait_object_;
+
+    // Since we use auto-reset event, these variables let ReleaseHandle know if
+    // UnregisterWaitEx ended up running a callback or not.
+    bool read_event_signalled_;
+    bool write_event_signalled_;
+
+    // These are used by the callbacks for the wait event watchers.
+    scoped_refptr<base::SingleThreadTaskRunner> message_loop_for_io_;
+    base::WeakPtr<RawChannelIOHandler> this_weakptr_;
+    base::WeakPtrFactory<RawChannelIOHandler> weak_ptr_factory_;
+
     MOJO_DISALLOW_COPY_AND_ASSIGN(RawChannelIOHandler);
   };
 
   ScopedPlatformHandle ReleaseHandleNoLock(
       std::vector<char>* serialized_read_buffer,
       std::vector<char>* serialized_write_buffer) override {
     if (handle_.is_valid()) {
       // SetInitialBuffer could have been called on main thread before OnInit
       // is called on Io thread. and in meantime releasehandle called.
       SerializeReadBuffer(0u, serialized_read_buffer);
 
       // We could have been given messages to write before OnInit.
-      SerializeWriteBuffer(serialized_write_buffer, 0u, 0u);
+      SerializeWriteBuffer(0u, 0u, serialized_write_buffer);
 
       return ScopedPlatformHandle(PlatformHandle(handle_.release().handle));
     }
 
     return io_handler_->ReleaseHandle(serialized_read_buffer,
                                       serialized_write_buffer);
   }
+
   PlatformHandle HandleForDebuggingNoLock() override {
     if (handle_.is_valid())
       return handle_.get();
 
     if (!io_handler_)
       return PlatformHandle();
 
     return PlatformHandle(io_handler_->handle());
   }
 
@@ skipping 12 matching lines @@
     if (!result) {
       DWORD error = GetLastError();
       if (error == ERROR_BROKEN_PIPE)
         return IO_FAILED_SHUTDOWN;
       if (error != ERROR_IO_PENDING) {
         LOG(WARNING) << "ReadFile: " << logging::SystemErrorCodeToString(error);
         return IO_FAILED_UNKNOWN;
       }
     }
 
-    if (result && skip_completion_port_on_success_) {
-      DWORD bytes_read_dword = 0;
-      BOOL get_size_result = GetOverlappedResult(
-          io_handler_->handle(), &io_handler_->read_context()->overlapped,
-          &bytes_read_dword, FALSE);
-      DPCHECK(get_size_result);
-      *bytes_read = bytes_read_dword;
-      return IO_SUCCEEDED;
-    }
-
-    if (!g_use_autoreset_event) {
-      if (!g_use_iocp) {
-        io_handler_->read_watcher_.StartWatchingOnce(
-            io_handler_->read_event, io_handler_);
-      }
-    }
     // If the read is pending or the read has succeeded but we don't skip
     // completion port on success, instruct |io_handler_| to wait for the
     // completion packet.
     //
     // TODO(yzshen): It seems there isn't document saying that all error cases
     // (other than ERROR_IO_PENDING) are guaranteed to *not* queue a completion
-    // packet. If we do get one for errors,
-    // |RawChannelIOHandler::OnIOCompleted()| will crash so we will learn about
-    // it.
-
+    // packet. If we do get one for errors, OnObjectSignaled()| will crash so we
+    // will learn about it.
     io_handler_->OnPendingReadStarted();
     return IO_PENDING;
   }
 
   IOResult ScheduleRead() override {
     if (!io_handler_)
       return IO_PENDING;  // OnInit could have earlied out.
 
     DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
     DCHECK(io_handler_);
     DCHECK(!io_handler_->pending_read());
 
     size_t bytes_read = 0;
-    IOResult io_result = Read(&bytes_read);
-    if (io_result == IO_SUCCEEDED) {
-      DCHECK(skip_completion_port_on_success_);
+    return Read(&bytes_read);
+  }
 
-      // We have finished reading successfully. Queue a notification manually.
-      io_handler_->OnPendingReadStarted();
-      // |io_handler_| won't go away before the task is run, so it is safe to
-      // use |base::Unretained()|.
-      message_loop_for_io()->PostTask(
-          FROM_HERE, base::Bind(&RawChannelIOHandler::OnIOCompleted,
-                                base::Unretained(io_handler_),
-                                base::Unretained(io_handler_->read_context()),
-                                static_cast<DWORD>(bytes_read), ERROR_SUCCESS));
-      return IO_PENDING;
-    }
-
-    return io_result;
-  }
   ScopedPlatformHandleVectorPtr GetReadPlatformHandles(
       size_t num_platform_handles,
       const void* platform_handle_table) override {
     // TODO(jam): this code will have to be updated once it's used in a sandbox
     // and the receiving process doesn't have duplicate permission for the
     // receiver. Once there's a broker and we have a connection to it (possibly
     // through ConnectionManager), then we can make a sync IPC to it here to get
     // a token for this handle, and it will duplicate the handle to is process.
     // Then we pass the token to the receiver, which will then make a sync call
     // to the broker to get a duplicated handle. This will also allow us to
@@ skipping 73 matching lines @@
       DWORD error = GetLastError();
       if (error == ERROR_BROKEN_PIPE)
         return IO_FAILED_SHUTDOWN;
       if (error != ERROR_IO_PENDING) {
         LOG(WARNING) << "WriteFile: "
                      << logging::SystemErrorCodeToString(error);
         return IO_FAILED_UNKNOWN;
       }
     }
 
-    if (result && skip_completion_port_on_success_) {
-      *platform_handles_written = num_platform_handles;
-      *bytes_written = bytes_written_dword;
-      return IO_SUCCEEDED;
-    }
-
-    if (!g_use_autoreset_event) {
-      if (!g_use_iocp) {
-        io_handler_->write_watcher_.StartWatchingOnce(
-            io_handler_->write_event, io_handler_);
-      }
-    }
     // If the write is pending or the write has succeeded but we don't skip
     // completion port on success, instruct |io_handler_| to wait for the
     // completion packet.
     //
     // TODO(yzshen): it seems there isn't document saying that all error cases
     // (other than ERROR_IO_PENDING) are guaranteed to *not* queue a completion
-    // packet. If we do get one for errors,
-    // |RawChannelIOHandler::OnIOCompleted()| will crash so we will learn about
-    // it.
+    // packet. If we do get one for errors, OnObjectSignaled will crash so we
+    // will learn about it.
 
     io_handler_->OnPendingWriteStartedNoLock(num_platform_handles);
     return IO_PENDING;
   }
 
   IOResult ScheduleWriteNoLock() override {
     write_lock().AssertAcquired();
 
     DCHECK(io_handler_);
     DCHECK(!io_handler_->pending_write_no_lock());
 
     size_t platform_handles_written = 0;
     size_t bytes_written = 0;
-    IOResult io_result = WriteNoLock(&platform_handles_written, &bytes_written);
-    if (io_result == IO_SUCCEEDED) {
-      DCHECK(skip_completion_port_on_success_);
-
-      // We have finished writing successfully. Queue a notification manually.
-      io_handler_->OnPendingWriteStartedNoLock(platform_handles_written);
-      // |io_handler_| won't go away before that task is run, so it is safe to
-      // use |base::Unretained()|.
-      message_loop_for_io()->PostTask(
-          FROM_HERE,
-          base::Bind(&RawChannelIOHandler::OnIOCompleted,
-                     base::Unretained(io_handler_),
-                     base::Unretained(io_handler_->write_context_no_lock()),
-                     static_cast<DWORD>(bytes_written), ERROR_SUCCESS));
-      return IO_PENDING;
-    }
-
-    return io_result;
+    return WriteNoLock(&platform_handles_written, &bytes_written);
   }
 
   void OnInit() override {
     DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 
     if (!handle_.is_valid()) {
       LOG(ERROR) << "Note: RawChannelWin " << this
                  << " early exiting in OnInit because no handle";
       return;
     }
 
     DCHECK(handle_.is_valid());
-    if (skip_completion_port_on_success_) {
-      // I don't know how this can fail (unless |handle_| is bad, in which case
-      // it's a bug in our code).
-      CHECK(g_vista_or_higher_functions.Get().
-          SetFileCompletionNotificationModes(
-              handle_.get().handle, FILE_SKIP_COMPLETION_PORT_ON_SUCCESS));
-    }
-
     DCHECK(!io_handler_);
     io_handler_ = new RawChannelIOHandler(this, handle_.Pass());
   }
 
   void OnShutdownNoLock(scoped_ptr<ReadBuffer> read_buffer,
                         scoped_ptr<WriteBuffer> write_buffer) override {
     // happens on shutdown if didn't call init when doing createduplicate
     if (message_loop_for_io()) {
       DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
     }
@@ skipping 22 matching lines @@
 
     io_handler_->DetachFromOwnerNoLock(read_buffer.Pass(), write_buffer.Pass());
     io_handler_ = nullptr;
   }
 
   // Passed to |io_handler_| during initialization.
   ScopedPlatformHandle handle_;
 
   RawChannelIOHandler* io_handler_;
 
-  const bool skip_completion_port_on_success_;
-
   MOJO_DISALLOW_COPY_AND_ASSIGN(RawChannelWin);
 };
 
 
 }  // namespace
 
 // -----------------------------------------------------------------------------
 
 RawChannel* RawChannel::Create(ScopedPlatformHandle handle) {
   return new RawChannelWin(handle.Pass());
@@ skipping 22 matching lines @@
     return filepath1 == filepath2;
   } else {
     // TODO: XP: see http://stackoverflow.com/questions/65170/how-to-get-name-associated-with-open-handle/5286888#5286888
     CHECK(false) << "TODO(jam): handle XP";
     return false;
   }
 }
 
 }  // namespace edk
 }  // namespace mojo