Switch MessagePumpForIO to use completion ports on Windows.

Cleanup the separation between MessagePumpForUI and 
MessagePumpForIO, and convert the latter to use Completion
Ports instead of MsgWaitForMultipleobjects to sleep
when idle.

Remove all traces of Windows messages from MessagePumpForIO,
remove the transitional API of completion port notifications
and remove WatchObject API.

Modify all callers of RegisterIOHandler so that they are no
longer using RegisterIOContext, and also handle properly
the new semantics of completion ports (notifications even when
the IO completes immediately).

Add a new interface to allow proper cleanup of disk cache (to
replace code that was waiting for pending APCs from the destructor).

Add a way for the message pump to perform cleanup of abandoned IO.


BUG=B/1344358, 3497, 3630
TESt=unit tests
R=darin


Review URL: http://codereview.chromium.org/8156

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@5021 0039d316-1c4b-4281-b951-d872f2087c98

1 files changed, 87 insertions, 99 deletions

diff --git a/base/message_loop_unittest.cc b/base/message_loop_unittest.cc
index 83be2b1..d959659 100644
--- a/base/message_loop_unittest.cc
+++ b/base/message_loop_unittest.cc
@@ -1056,68 +1056,6 @@ void RunTest_NonNestableInNestedLoop(MessageLoop::Type message_loop_type) {
 
 #if defined(OS_WIN)
 
-class AutoresetWatcher : public MessageLoopForIO::Watcher {
- public:
-  explicit AutoresetWatcher(HANDLE signal) : signal_(signal) {
-  }
-  virtual void OnObjectSignaled(HANDLE object);
- private:
-  HANDLE signal_;
-};
-
-void AutoresetWatcher::OnObjectSignaled(HANDLE object) {
-  MessageLoopForIO::current()->WatchObject(object, NULL);
-  ASSERT_TRUE(SetEvent(signal_));
-}
-
-class AutoresetTask : public Task {
- public:
-  AutoresetTask(HANDLE object, MessageLoopForIO::Watcher* watcher)
-    : object_(object), watcher_(watcher) {}
-  virtual void Run() {
-    MessageLoopForIO::current()->WatchObject(object_, watcher_);
-  }
-
- private:
-  HANDLE object_;
-  MessageLoopForIO::Watcher* watcher_;
-};
-
-void RunTest_AutoresetEvents(MessageLoop::Type message_loop_type) {
-  MessageLoop loop(message_loop_type);
-
-  SECURITY_ATTRIBUTES attributes;
-  attributes.nLength = sizeof(attributes);
-  attributes.bInheritHandle = false;
-  attributes.lpSecurityDescriptor = NULL;
-
-  // Init an autoreset and a manual reset events.
-  HANDLE autoreset = CreateEvent(&attributes, FALSE, FALSE, NULL);
-  HANDLE callback_called = CreateEvent(&attributes, TRUE, FALSE, NULL);
-  ASSERT_TRUE(NULL != autoreset);
-  ASSERT_TRUE(NULL != callback_called);
-
-  Thread thread("Autoreset test");
-  Thread::Options options;
-  options.message_loop_type = message_loop_type;
-  ASSERT_TRUE(thread.StartWithOptions(options));
-
-  MessageLoop* thread_loop = thread.message_loop();
-  ASSERT_TRUE(NULL != thread_loop);
-
-  AutoresetWatcher watcher(callback_called);
-  AutoresetTask* task = new AutoresetTask(autoreset, &watcher);
-  thread_loop->PostTask(FROM_HERE, task);
-  Sleep(100);  // Make sure the thread runs and sleeps for lack of work.
-
-  ASSERT_TRUE(SetEvent(autoreset));
-
-  DWORD result = WaitForSingleObject(callback_called, 1000);
-  EXPECT_EQ(WAIT_OBJECT_0, result);
-
-  thread.Stop();
-}
-
 class DispatcherImpl : public MessageLoopForUI::Dispatcher {
  public:
   DispatcherImpl() : dispatch_count_(0) {}
@@ -1150,68 +1088,68 @@ void RunTest_Dispatcher(MessageLoop::Type message_loop_type) {
 
 class TestIOHandler : public MessageLoopForIO::IOHandler {
  public:
-  TestIOHandler(const wchar_t* name, HANDLE signal);
+  TestIOHandler(const wchar_t* name, HANDLE signal, bool wait);
 
-  virtual void OnIOCompleted(OVERLAPPED* context, DWORD bytes_transfered,
-                             DWORD error);
+  virtual void OnIOCompleted(MessageLoopForIO::IOContext* context,
+                             DWORD bytes_transfered, DWORD error);
 
-  HANDLE file() { return file_.Get(); }
-  void* buffer() { return buffer_; }
-  OVERLAPPED* context() { return &context_; }
+  void Init();
+  void WaitForIO();
+  OVERLAPPED* context() { return &context_.overlapped; }
   DWORD size() { return sizeof(buffer_); }
 
  private:
   char buffer_[48];
-  OVERLAPPED context_;
+  MessageLoopForIO::IOContext context_;
   HANDLE signal_;
   ScopedHandle file_;
-  ScopedHandle event_;
+  bool wait_;
 };
 
-TestIOHandler::TestIOHandler(const wchar_t* name, HANDLE signal)
-    : signal_(signal) {
+TestIOHandler::TestIOHandler(const wchar_t* name, HANDLE signal, bool wait)
+    : signal_(signal), wait_(wait) {
   memset(buffer_, 0, sizeof(buffer_));
   memset(&context_, 0, sizeof(context_));
-
-  event_.Set(CreateEvent(NULL, TRUE, FALSE, NULL));
-  EXPECT_TRUE(event_.IsValid());
-  context_.hEvent =  event_.Get();
+  context_.handler = this;
 
   file_.Set(CreateFile(name, GENERIC_READ, 0, NULL, OPEN_EXISTING,
                        FILE_FLAG_OVERLAPPED, NULL));
   EXPECT_TRUE(file_.IsValid());
 }
 
-void TestIOHandler::OnIOCompleted(OVERLAPPED* context, DWORD bytes_transfered,
-                                  DWORD error) {
+void TestIOHandler::Init() {
+  MessageLoopForIO::current()->RegisterIOHandler(file_, this);
+
+  DWORD read;
+  EXPECT_FALSE(ReadFile(file_, buffer_, size(), &read, context()));
+  EXPECT_EQ(ERROR_IO_PENDING, GetLastError());
+  if (wait_)
+    WaitForIO();
+}
+
+void TestIOHandler::OnIOCompleted(MessageLoopForIO::IOContext* context,
+                                  DWORD bytes_transfered, DWORD error) {
   ASSERT_TRUE(context == &context_);
-  MessageLoopForIO::current()->RegisterIOContext(context, NULL);
   ASSERT_TRUE(SetEvent(signal_));
 }
 
+void TestIOHandler::WaitForIO() {
+  EXPECT_TRUE(MessageLoopForIO::current()->WaitForIOCompletion(300, this));
+  EXPECT_TRUE(MessageLoopForIO::current()->WaitForIOCompletion(400, this));
+}
+
 class IOHandlerTask : public Task {
  public:
   explicit IOHandlerTask(TestIOHandler* handler) : handler_(handler) {}
-  virtual void Run();
+  virtual void Run() {
+    handler_->Init();
+  }
 
  private:
   TestIOHandler* handler_;
 };
 
-void IOHandlerTask::Run() {
-  MessageLoopForIO::current()->RegisterIOHandler(handler_->file(), handler_);
-  MessageLoopForIO::current()->RegisterIOContext(handler_->context(), handler_);
-
-  DWORD read;
-  EXPECT_FALSE(ReadFile(handler_->file(), handler_->buffer(), handler_->size(),
-               &read, handler_->context()));
-  EXPECT_EQ(ERROR_IO_PENDING, GetLastError());
-}
-
 void RunTest_IOHandler() {
-  // This test requires an IO loop.
-  MessageLoop loop(MessageLoop::TYPE_IO);
-
   ScopedHandle callback_called(CreateEvent(NULL, TRUE, FALSE, NULL));
   ASSERT_TRUE(callback_called.IsValid());
 
@@ -1228,7 +1166,7 @@ void RunTest_IOHandler() {
   MessageLoop* thread_loop = thread.message_loop();
   ASSERT_TRUE(NULL != thread_loop);
 
-  TestIOHandler handler(kPipeName, callback_called);
+  TestIOHandler handler(kPipeName, callback_called, false);
   IOHandlerTask* task = new IOHandlerTask(&handler);
   thread_loop->PostTask(FROM_HERE, task);
   Sleep(100);  // Make sure the thread runs and sleeps for lack of work.
@@ -1243,6 +1181,57 @@ void RunTest_IOHandler() {
   thread.Stop();
 }
 
+void RunTest_WaitForIO() {
+  ScopedHandle callback1_called(CreateEvent(NULL, TRUE, FALSE, NULL));
+  ScopedHandle callback2_called(CreateEvent(NULL, TRUE, FALSE, NULL));
+  ASSERT_TRUE(callback1_called.IsValid());
+  ASSERT_TRUE(callback2_called.IsValid());
+
+  const wchar_t* kPipeName1 = L"\\\\.\\pipe\\iohandler_pipe1";
+  const wchar_t* kPipeName2 = L"\\\\.\\pipe\\iohandler_pipe2";
+  ScopedHandle server1(CreateNamedPipe(kPipeName1, PIPE_ACCESS_OUTBOUND, 0, 1,
+                                       0, 0, 0, NULL));
+  ScopedHandle server2(CreateNamedPipe(kPipeName2, PIPE_ACCESS_OUTBOUND, 0, 1,
+                                       0, 0, 0, NULL));
+  ASSERT_TRUE(server1.IsValid());
+  ASSERT_TRUE(server2.IsValid());
+
+  Thread thread("IOHandler test");
+  Thread::Options options;
+  options.message_loop_type = MessageLoop::TYPE_IO;
+  ASSERT_TRUE(thread.StartWithOptions(options));
+
+  MessageLoop* thread_loop = thread.message_loop();
+  ASSERT_TRUE(NULL != thread_loop);
+
+  TestIOHandler handler1(kPipeName1, callback1_called, false);
+  TestIOHandler handler2(kPipeName2, callback2_called, true);
+  IOHandlerTask* task1 = new IOHandlerTask(&handler1);
+  IOHandlerTask* task2 = new IOHandlerTask(&handler2);
+  thread_loop->PostTask(FROM_HERE, task1);
+  Sleep(100);  // Make sure the thread runs and sleeps for lack of work.
+  thread_loop->PostTask(FROM_HERE, task2);
+  Sleep(100);
+
+  // At this time handler1 is waiting to be called, and the thread is waiting
+  // on the Init method of handler2, filtering only handler2 callbacks.
+
+  const char buffer[] = "Hello there!";
+  DWORD written;
+  EXPECT_TRUE(WriteFile(server1, buffer, sizeof(buffer), &written, NULL));
+  Sleep(200);
+  EXPECT_EQ(WAIT_TIMEOUT, WaitForSingleObject(callback1_called, 0)) <<
+      "handler1 has not been called";
+
+  EXPECT_TRUE(WriteFile(server2, buffer, sizeof(buffer), &written, NULL));
+
+  HANDLE objects[2] = { callback1_called.Get(), callback2_called.Get() };
+  DWORD result = WaitForMultipleObjects(2, objects, TRUE, 1000);
+  EXPECT_EQ(WAIT_OBJECT_0, result);
+
+  thread.Stop();
+}
+
 #endif  // defined(OS_WIN)
 
 }  // namespace
@@ -1379,11 +1368,6 @@ TEST(MessageLoopTest, NonNestableInNestedLoop) {
 }
 
 #if defined(OS_WIN)
-TEST(MessageLoopTest, AutoresetEvents) {
-  // This test requires an IO loop
-  RunTest_AutoresetEvents(MessageLoop::TYPE_IO);
-}
-
 TEST(MessageLoopTest, Dispatcher) {
   // This test requires a UI loop
   RunTest_Dispatcher(MessageLoop::TYPE_UI);
@@ -1392,4 +1376,8 @@ TEST(MessageLoopTest, Dispatcher) {
 TEST(MessageLoopTest, IOHandler) {
   RunTest_IOHandler();
 }
+
+TEST(MessageLoopTest, WaitForIO) {
+  RunTest_WaitForIO();
+}
 #endif  // defined(OS_WIN)