A helper class for calling RegisterWaitForSingleObject with the objective of finding out when objects get signaled.

The API is pretty simple:
-  consumer can associate a Task with a HANDLE via AddWatch method
-  when the HANDLE is signaled, we run the Task on the thread that called AddWatch
-  the consumer can call CancelWatch to abort the watch
-  a watch is one-shot:  after the object is signaled, the consumer has to call AddWatch again to wait a second time
-  if the ObjectWatcher instance is destroyed, it cancels all associated watches.

Implementation details:
-  Uses RegisterWaitForSingleObject to run a function on the wait thread.  This avoids extra thread marshaling to a worker thread.
-  Uses a Task to get back onto the origin thread.  This is possible due to the UnregisterWaitEx(INVALID_HANDLE_VALUE) call.
-  Once on the origin thread, runs the consumer's Task directly.

This is the first part of changes I want to make to MessageLoop::WatchObject.  For now, I have not made any changes to existing code.

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

5 files changed, 411 insertions, 0 deletions

diff --git a/base/build/base.vcproj b/base/build/base.vcproj
index e4d47856..68dbce3 100644
--- a/base/build/base.vcproj
+++ b/base/build/base.vcproj
@@ -390,6 +390,14 @@
 			>
 		</File>
 		<File
+			RelativePath="..\object_watcher.cc"
+			>
+		</File>
+		<File
+			RelativePath="..\object_watcher.h"
+			>
+		</File>
+		<File
 			RelativePath="..\observer_list.h"
 			>
 		</File>
diff --git a/base/build/base_unittests.vcproj b/base/build/base_unittests.vcproj
index 69d3948..ae77e67 100644
--- a/base/build/base_unittests.vcproj
+++ b/base/build/base_unittests.vcproj
@@ -236,6 +236,10 @@
 				>
 			</File>
 			<File
+				RelativePath="..\object_watcher_unittest.cc"
+				>
+			</File>
+			<File
 				RelativePath="..\path_service_unittest.cc"
 				>
 			</File>
diff --git a/base/object_watcher.cc b/base/object_watcher.cc
new file mode 100644
index 0000000..95c0b64
--- /dev/null
+++ b/base/object_watcher.cc
@@ -0,0 +1,148 @@
+// Copyright 2008, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//    * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//    * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//    * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "base/object_watcher.h"
+
+#include "base/message_loop.h"
+#include "base/logging.h"
+
+namespace base {
+
+//-----------------------------------------------------------------------------
+
+struct ObjectWatcher::Watch : public Task {
+  ObjectWatcher* watcher;    // The associated ObjectWatcher instance
+  HANDLE object;             // The object being watched
+  HANDLE wait_object;        // Returned by RegisterWaitForSingleObject
+  MessageLoop* origin_loop;  // Used to get back to the origin thread
+  scoped_ptr<Task> task;     // Task to notify when signaled
+  bool did_signal;           // DoneWaiting was called
+
+  virtual void Run() {
+    // The watcher may have already been torn down, in which case we need to
+    // just get out of dodge.
+    if (!watcher)
+      return;
+    
+    // Put this on the stack since CancelWatch deletes task.  It is a good
+    // to call CancelWatch before running the task because we want to allow
+    // the consumer to call AddWatch again inside Run.
+    Task* task_to_run = task.release();
+    
+    watcher->CancelWatch(object);
+
+    task_to_run->Run();
+    delete task_to_run;
+  }
+};
+
+//-----------------------------------------------------------------------------
+
+ObjectWatcher::ObjectWatcher() {
+}
+
+ObjectWatcher::~ObjectWatcher() {
+  // Cancel any watches that may still exist.
+  while (!watches_.empty())
+    CancelWatch(watches_.begin()->first);
+}
+
+bool ObjectWatcher::AddWatch(const tracked_objects::Location& from_here,
+                             HANDLE object, Task* task) {
+  task->SetBirthPlace(from_here);
+
+  linked_ptr<Watch>& watch = watches_[object];
+  CHECK(!watch.get()) << "Already watched!";
+
+  watch.reset(new Watch());
+  watch->watcher = this;
+  watch->object = object;
+  watch->origin_loop = MessageLoop::current();
+  watch->task.reset(task);
+  watch->did_signal = false;
+
+  // Since our job is to just notice when an object is signaled and report the
+  // result back to this thread, we can just run on a Windows wait thread.
+  DWORD wait_flags = WT_EXECUTEINWAITTHREAD | WT_EXECUTEONLYONCE;
+
+  if (!RegisterWaitForSingleObject(&watch->wait_object, object, DoneWaiting,
+                                   watch.get(), INFINITE, wait_flags)) {
+    NOTREACHED() << "RegisterWaitForSingleObject failed: " << GetLastError();
+    watches_.erase(object);
+    return false;
+  }
+
+  return true;
+}
+
+bool ObjectWatcher::CancelWatch(HANDLE object) {
+  WatchMap::iterator i = watches_.find(object);
+  if (i == watches_.end())
+    return false;
+
+  Watch* watch = i->second.get();
+  
+  // If DoneWaiting is in progress, we wait for it to finish.  We know whether
+  // DoneWaiting happened or not by inspecting the did_signal flag.
+  if (!UnregisterWaitEx(watch->wait_object, INVALID_HANDLE_VALUE)) {
+    NOTREACHED() << "UnregisterWaitEx failed: " << GetLastError();
+    return false;
+  }
+
+  // If DoneWaiting was called, then the watch would have been posted as a
+  // task, and will therefore be deleted by the MessageLoop.  Otherwise, we
+  // need to take care to delete it here.
+  if (watch->did_signal)
+    i->second.release();
+
+  // If the watch has been posted, then we need to make sure it knows not to do
+  // anything once it is run.
+  watch->watcher = NULL;
+
+  // Delete the task object now so that everything, from the perspective of the
+  // consumer, is cleaned up once we return from CancelWatch.
+  watch->task.reset();
+
+  watches_.erase(i);
+  return true;
+}
+
+// static
+void CALLBACK ObjectWatcher::DoneWaiting(void* param, BOOLEAN timed_out) {
+  DCHECK(!timed_out);
+
+  Watch* watch = static_cast<Watch*>(param);
+
+  // Record that we ran this function.
+  watch->did_signal = true;
+
+  watch->origin_loop->PostTask(FROM_HERE, watch);
+}
+
+}  // namespace base
diff --git a/base/object_watcher.h b/base/object_watcher.h
new file mode 100644
index 0000000..58431b1
--- /dev/null
+++ b/base/object_watcher.h
@@ -0,0 +1,84 @@
+// Copyright 2008, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//    * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//    * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//    * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#ifndef BASE_OBJECT_WATCHER_H_
+#define BASE_OBJECT_WATCHER_H_
+
+#include <windows.h>
+
+#include <map>
+
+#include "base/linked_ptr.h"
+#include "base/tracked.h"
+
+class Task;
+
+namespace base {
+
+// A class that enables support for asynchronously waiting for Windows objects
+// to become signaled.  Supports waiting on more than 64 objects.
+class ObjectWatcher {
+ public:
+  ObjectWatcher();
+  ~ObjectWatcher();
+
+  // When the object is signaled, the given task is run on the thread where
+  // Watch is called.  The ObjectWatcher assumes ownership of the task and
+  // will ensure that it gets deleted eventually.
+  //
+  // NOTE: It is an error to call this method on an object that is already
+  // being watched by this ObjectWatcher.
+  //
+  // Returns true if the watch was added.  Otherwise, false is returned.
+  //
+  bool AddWatch(const tracked_objects::Location& from_here, HANDLE object,
+                Task* task);
+
+  // Stops watching the given object.  Does nothing if the watch has already
+  // completed.  If the watch is still active, then it is canceled, and the
+  // associated task is deleted.
+  //
+  // Returns true if the watch was canceled.  Otherwise, false is returned.
+  //
+  bool CancelWatch(HANDLE object);
+
+ private:
+  // Called on a background thread when done waiting.
+  static void CALLBACK DoneWaiting(void* param, BOOLEAN timed_out);
+
+  // Passed as the param argument to the above methods.
+  struct Watch;
+
+  typedef std::map<HANDLE, linked_ptr<Watch>> WatchMap;
+  WatchMap watches_;
+};
+
+}  // namespace base
+
+#endif  // BASE_OBJECT_WATCHER_H_
diff --git a/base/object_watcher_unittest.cc b/base/object_watcher_unittest.cc
new file mode 100644
index 0000000..16e6ae5
--- /dev/null
+++ b/base/object_watcher_unittest.cc
@@ -0,0 +1,167 @@
+// Copyright 2008, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//    * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//    * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//    * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "base/message_loop.h"
+#include "base/object_watcher.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace {
+typedef testing::Test ObjectWatcherTest;
+
+class DecrementCountTask : public Task {
+ public:
+  DecrementCountTask(int* counter) : counter_(counter) {
+  }
+  virtual void Run() {
+    if (--(*counter_) == 0)
+      MessageLoop::current()->Quit();
+  }
+ private:
+  int* counter_;
+};
+
+}
+
+TEST(ObjectWatcherTest, BasicSignal) {
+  base::ObjectWatcher watcher;
+
+  // A manual-reset event that is not yet signaled.
+  HANDLE event = CreateEvent(NULL, TRUE, FALSE, NULL);
+
+  bool ok = watcher.AddWatch(FROM_HERE, event, new MessageLoop::QuitTask());
+  EXPECT_TRUE(ok);
+  
+  SetEvent(event);
+
+  MessageLoop::current()->Run();
+
+  CloseHandle(event);
+}
+
+TEST(ObjectWatcherTest, BasicCancel) {
+  base::ObjectWatcher watcher;
+
+  // A manual-reset event that is not yet signaled.
+  HANDLE event = CreateEvent(NULL, TRUE, FALSE, NULL);
+
+  bool ok = watcher.AddWatch(FROM_HERE, event, new MessageLoop::QuitTask());
+  EXPECT_TRUE(ok);
+  
+  watcher.CancelWatch(event);
+
+  CloseHandle(event);
+}
+
+TEST(ObjectWatcherTest, ManySignal) {
+  base::ObjectWatcher watcher;
+
+  const int kNumObjects = 2000;
+
+  int counter = kNumObjects;
+  HANDLE events[kNumObjects];
+  
+  for (int i = 0; i < kNumObjects; ++i) {
+    // A manual-reset event that is not yet signaled.
+    events[i] = CreateEvent(NULL, TRUE, FALSE, NULL);
+
+    bool ok = watcher.AddWatch(
+        FROM_HERE, events[i], new DecrementCountTask(&counter));
+    EXPECT_TRUE(ok);
+  }
+   
+  for (int i = 0; i < kNumObjects; ++i) 
+    SetEvent(events[i]);
+
+  MessageLoop::current()->Run();
+
+  for (int i = 0; i < kNumObjects; ++i)
+    CloseHandle(events[i]);
+}
+
+TEST(ObjectWatcherTest, ManyCancel) {
+  base::ObjectWatcher watcher;
+
+  const int kNumObjects = 2000;
+
+  int counter = kNumObjects;
+  HANDLE events[kNumObjects];
+  
+  for (int i = 0; i < kNumObjects; ++i) {
+    // A manual-reset event that is not yet signaled.
+    events[i] = CreateEvent(NULL, TRUE, FALSE, NULL);
+
+    bool ok = watcher.AddWatch(
+        FROM_HERE, events[i], new DecrementCountTask(&counter));
+    EXPECT_TRUE(ok);
+  }
+   
+  for (int i = 0; i < kNumObjects; ++i) 
+    watcher.CancelWatch(events[i]);
+
+  for (int i = 0; i < kNumObjects; ++i)
+    CloseHandle(events[i]);
+}
+
+TEST(ObjectWatcherTest, CancelAfterSet) {
+  base::ObjectWatcher watcher;
+
+  const int kNumObjects = 50;
+
+  int counter = kNumObjects;
+  HANDLE events[kNumObjects];
+  
+  for (int i = 0; i < kNumObjects; ++i) {
+    // A manual-reset event that is not yet signaled.
+    events[i] = CreateEvent(NULL, TRUE, FALSE, NULL);
+
+    bool ok = watcher.AddWatch(
+        FROM_HERE, events[i], new DecrementCountTask(&counter));
+    EXPECT_TRUE(ok);
+  }
+   
+  for (int i = 0; i < kNumObjects; ++i) {
+    SetEvent(events[i]);
+    
+    // Let the background thread do its business
+    SleepEx(10, TRUE);
+    
+    // Occasionally pump some tasks.  Sometimes we cancel a watch after its
+    // task has run and other times we do nothing.
+    if (i % 3 == 0)
+      MessageLoop::current()->RunAllPending();
+
+    if (watcher.CancelWatch(events[i]))
+      --counter;
+  }
+  
+  EXPECT_EQ(0, counter);
+
+  for (int i = 0; i < kNumObjects; ++i)
+    CloseHandle(events[i]);
+}