src/base/object_watcher_shell.cc - cobalt - Git at Google

 /*
  * Copyright 2012 Google Inc. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 #include "base/object_watcher_shell.h"

 #include <stack>

 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/synchronization/lock.h"
 #include "base/synchronization/waitable_event.h"

 namespace base {

 struct Watch {
   ObjectWatcher* watcher;             // associated ObjectWatcher instance
   int object;                         // the file descriptor being watched
   int watch_handle;                   // used for uniquely identifying watches
   MessageLoop* origin_loop;           // the origin thread's message loop
   ObjectWatcher::Delegate* delegate;  // delegate to notify when signaled
   bool did_signal;                    // set when DoneWaiting is called
   MessagePumpShell::Mode mode;        // callback on read, write or both?

   void Run() {
     // The watcher may have already been torn down, in which case we need to
     // just get out of dodge.
     if (!watcher)
       return;

     DCHECK(did_signal);
     watcher->StopWatching();

     delegate->OnObjectSignaled(object);
   }
 };

 // To emulate the behavior of the former Task system:
 static void WatchTask(Watch* watch) {
   watch->Run();
   delete watch;
 }

 ObjectWatchMultiplexer* ObjectWatchMultiplexer::s_instance = NULL;

 ObjectWatchMultiplexer::ObjectWatchMultiplexer()
     : base::SimpleThread(
         "ObjectWatchMultiplexer Thread",
         base::SimpleThread::Options(
             // TODO: Pass these into ObjectWatchMultiplexer.
             0 /* kObjectWatcherThreadStackSize */,
             kThreadPriority_Default /* kObjectWatcherThreadPriority */,
             kNoThreadAffinity /* kNetworkIOThreadAffinity */)),
         non_empty_list_event_(true, false),
         max_watch_handle_(0),
         should_recompose_pollfd_array_(true),
         exit_(false) {
   DCHECK(!s_instance);
   s_instance = this;
   Start();
 }

 ObjectWatchMultiplexer::~ObjectWatchMultiplexer() {
   DCHECK(s_instance);
   Join();
   s_instance = NULL;
 }

 // blocking call to exit the internal thread
 void ObjectWatchMultiplexer::Join() {
   {
     // RecomposePollFDArray may reset the event, but can only be called
     // while this lock is held.  To break the race between these two, grab
     // the lock before setting exit_ and signalling the event, and make sure
     // that exit_ is checked before resetting the event.
     base::AutoLock lock(add_watch_list_lock_);
     exit_ = true;
     // wake up the thread in case it's waiting on the empty list event
     non_empty_list_event_.Signal();
   }
   // in any event it should timeout eventually and see the exit flag..
   base::SimpleThread::Join();
 }

 void ObjectWatchMultiplexer::AddWatch(Watch* watch) {
   DCHECK(watch);
   DCHECK_EQ(watch->watch_handle, 0);
   // grab the mutex to modify the watch map et al
   base::AutoLock lock(add_watch_list_lock_);
   should_recompose_pollfd_array_ = true;
   watch->watch_handle = ++max_watch_handle_;
   new_watch_map_.insert(WatchMap::value_type(watch->object, watch));
   non_empty_list_event_.Signal();
 }

 void ObjectWatchMultiplexer::RemoveWatch(Watch* watch) {
   DCHECK(watch);
   base::AutoLock lock(remove_watch_list_lock_);
   WatchMap::iterator start = watch_map_.lower_bound(watch->object);
   WatchMap::iterator end = watch_map_.upper_bound(watch->object);
   WatchMap::iterator it;
   // search within those elements having the same file descriptor
   for (it = start; it != end; it++)
     if (it->second->watch_handle == watch->watch_handle)
       break;
   if (it != end) {
     watch_map_.erase(it);
     should_recompose_pollfd_array_ = true;
   } else {
     base::AutoLock lock(add_watch_list_lock_);
     // not found, search new_watch_map_
     WatchMap::iterator start = new_watch_map_.lower_bound(watch->object);
     WatchMap::iterator end = new_watch_map_.upper_bound(watch->object);
     for (it = start; it != end; it++)
       if (it->second->watch_handle == watch->watch_handle)
         break;
     if (it != end) {
       new_watch_map_.erase(it);
       should_recompose_pollfd_array_ = true;
     }
   }
 }

 void ObjectWatchMultiplexer::Run() {
   // we keep a stack of pointers to watches needing a callback and
   // subsequent removal from the map
   std::stack<WatchMap::iterator> watch_removal_stack;
   // separate stack of callbacks to process after releasing the
   // WatchMap mutex
   std::stack<Watch*> watch_callback_stack;

   while (!exit_) {
     // We need both locks here. Order is important
     remove_watch_list_lock_.Acquire();
     add_watch_list_lock_.Acquire();
     if (should_recompose_pollfd_array_)
       RecomposePollFDArray();
     add_watch_list_lock_.Release();
     remove_watch_list_lock_.Release();

     if (!pollfd_array_.empty()) {
       // timeout on a 100ms to wake up and see if we need
       // to modify the array of fds or exit..
       int event_count =
           poll(&pollfd_array_.front(), pollfd_array_.size(), kPollTimeout);
       // negative value returned means error...
       if (event_count < 0) {
         // Sleep for kPollTimeout, otherwise this thread could busy-wait until
         // poll returns a non-error value
         base::PlatformThread::Sleep(
             base::TimeDelta::FromMilliseconds(kPollTimeout));
       }

       // we need the mutex to access the map. AddWatch() is on a different
       // map object so we dont need the add_watch_list_lock_ here.
       remove_watch_list_lock_.Acquire();
       DCHECK_LE(event_count, static_cast<int>(pollfd_array_.size()));
       for (int j = 0; j < pollfd_array_.size() && event_count > 0; ++j) {
         if (pollfd_array_[j].revents != 0) {
           event_count--;
           WatchMap::iterator start =
               watch_map_.lower_bound(pollfd_array_[j].fd);
           WatchMap::iterator stop =
               watch_map_.upper_bound(pollfd_array_[j].fd);
           for (WatchMap::iterator it = start; it != stop; it++) {
             // oh STL, your capacity for obfuscation never ceases to amaze.
             // this if/else block tries to match interest for each fd event
             // to the associated watchers.  If a watcher was watching for
             // reads and a read happened, then we add it to the stack of
             // callbacks to process.  Else if it was watching for writes and
             // a write happened then we do the same.
             // note that signaled watches are going to become the property
             // of their respectively owned MessageLoops and will be deleted
             // once executed, and so will be removed from the hash_map
             if ((pollfd_array_[j].revents & POLLIN) &&
                 (it->second->mode & MessagePumpShell::WATCH_READ)) {
               watch_removal_stack.push(it);
               watch_callback_stack.push(it->second);
             } else if ((pollfd_array_[j].revents & POLLOUT) &&
                         (it->second->mode & MessagePumpShell::WATCH_WRITE)) {
               watch_removal_stack.push(it);
               watch_callback_stack.push(it->second);
             }
           }
         }
       }  // for(int j = 0; j < pollfd_array_.size() && event_count > 0; ++j)

       // process all callbacks.  this must unfortunately be done inside
       // of the mutex to prevent the contention on watch objects being
       // deleted due to a call to ObjectWatcher::StopWatching that could
       // occur while the watch is on this callback stack
       while (!watch_callback_stack.empty()) {
         Watch* watch = watch_callback_stack.top();
         watch_callback_stack.pop();
         // signal this watch to run in the original thread context
         // of the calling thread message loop.  The watch will be
         // posted as a task to the MessageLoop which will take
         // ownership of the object and delete it
         watch->did_signal = true;
         watch->origin_loop->PostTask(FROM_HERE, base::Bind(WatchTask, watch));
       }
       // if we're going to signal something we should recompose
       // the array as we are going to be removing watches
       if (!watch_removal_stack.empty())
         should_recompose_pollfd_array_ = true;
       // now remove the callback elements from the current hash_map
       while (!watch_removal_stack.empty()) {
         watch_map_.erase(watch_removal_stack.top());
         watch_removal_stack.pop();
       }

       remove_watch_list_lock_.Release();
     } else {  // if (!pollfd_array_.empty())
       // sleep until flagged that something has been added to the list
       DCHECK(watch_map_.empty());
       // NOTE: You can't DCHECK(new_watch_map_.empty()) because we let go
       // of the lock already and new stuff could have been added before we
       // get here.  If that happens, the event will have already been
       // signaled and we won't have to wait.
       non_empty_list_event_.Wait();
     }  // if (!pollfd_array_.empty())
   }    // while (!exit_)
 }

 void ObjectWatchMultiplexer::RecomposePollFDArray() {
   remove_watch_list_lock_.AssertAcquired();
   add_watch_list_lock_.AssertAcquired();
   // process new requests and move them into working map
   WatchMap::iterator it;
   for (it = new_watch_map_.begin(); it != new_watch_map_.end(); it++) {
     watch_map_.insert(*it);
   }
   new_watch_map_.clear();
   // count the number of unique file descriptors in the map
   int count = 0;
   int last_fd = -1;
   for (it = watch_map_.begin(); it != watch_map_.end(); it++) {
     if (it->first != last_fd) {
       last_fd = it->first;
       count++;
     }
   }
   pollfd_array_.resize(count);

   // init pollfd_array_
   if (!pollfd_array_.empty()) {
     int j = -1;
     last_fd = -1;
     for (it = watch_map_.begin(); it != watch_map_.end(); it++) {
       if (it->first != last_fd) {
         // we keep j pointing at this current element for ORing in the watch
         // flags
         j++;
         last_fd = it->first;
         pollfd_array_[j].fd = last_fd;
         // assume we care about both, we will filter based on mode on callback
         pollfd_array_[j].events = 0;
         if (it->second->mode & MessagePumpShell::WATCH_READ)
           pollfd_array_[j].events |= POLLIN;
         if (it->second->mode & MessagePumpShell::WATCH_WRITE)
           pollfd_array_[j].events |= POLLOUT;
         pollfd_array_[j].revents = 0;
       } else {  // if (it->first != last_fd)
         // OR in the watch flags to the already initialized pollfd struct
         if (it->second->mode & MessagePumpShell::WATCH_READ)
           pollfd_array_[j].events |= POLLIN;
         if (it->second->mode & MessagePumpShell::WATCH_WRITE)
           pollfd_array_[j].events |= POLLOUT;
       }     // if (it->first != last_fd)
     }       // for (it = watch_map_.begin(); it != watch_map_.end(); it++)
   } else {  // if (!pollfd_array_.empty())
     // reset the empty event if we've completely emptied the watch list
     DCHECK(watch_map_.empty());
     DCHECK(new_watch_map_.empty());
     // Join may have signalled the event because we are exitting.  If so,
     // don't reset the event.  That would lead to an infinite wait on
     // shutdown.
     if (!exit_)
       non_empty_list_event_.Reset();
   }  // if (!pollfd_array_.empty())
   // reset flag
   should_recompose_pollfd_array_ = false;
 }

 ObjectWatcher::ObjectWatcher() : watch_(NULL) {
 }

 ObjectWatcher::~ObjectWatcher() {
   StopWatching();
 }

 bool ObjectWatcher::StartWatching(int object,
                                   MessagePumpShell::Mode mode,
                                   Delegate* delegate) {
   DCHECK(ObjectWatchMultiplexer::GetInstance() != NULL);
   if (watch_) {
     NOTREACHED() << "Already watching an object";
     return false;
   }

   watch_ = new Watch;
   watch_->watcher = this;
   watch_->object = object;
   watch_->mode = mode;
   watch_->origin_loop = MessageLoop::current();
   watch_->delegate = delegate;
   watch_->did_signal = false;
   watch_->watch_handle = 0;

   ObjectWatchMultiplexer::GetInstance()->AddWatch(watch_);

   // We need to know if the current message loop is going away so we can
   // prevent the wait thread from trying to access a dead message loop.
   MessageLoop::current()->AddDestructionObserver(this);
   return true;
 }

 bool ObjectWatcher::StopWatching() {
   if (!watch_)
     return false;

   DCHECK(ObjectWatchMultiplexer::GetInstance() != NULL);

   // make sure stop call happens on same thread as start call
   DCHECK(watch_->origin_loop == MessageLoop::current());

   // this will block until this watch has been removed from the mux
   ObjectWatchMultiplexer::GetInstance()->RemoveWatch(watch_);
   // let the watch know that the watcher has died, in case the watch is
   // still sitting in a message loop.  See Watch::Run() to see that it
   // will bug out in that event.
   watch_->watcher = NULL;

   // if the watch was signaled, then it is now property of the MessageLoop
   // and will be deleted there.  It has also been safely removed from the
   // internal list of watches in the ObjectWatchMux.  Therefore if it hasn't
   // been signaled we delete it here to avoid leaking the watch
   if (!watch_->did_signal)
     delete watch_;

   watch_ = NULL;

   MessageLoop::current()->RemoveDestructionObserver(this);
   return true;
 }

 void ObjectWatcher::WillDestroyCurrentMessageLoop() {
   // Need to shutdown the watch so that we don't try to access the MessageLoop
   // after this point.
   StopWatching();
 }

 }  // namespace base
	/*
	* Copyright 2012 Google Inc. All Rights Reserved.
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	#include "base/object_watcher_shell.h"

	#include <stack>

	#include "base/bind.h"
	#include "base/logging.h"
	#include "base/synchronization/lock.h"
	#include "base/synchronization/waitable_event.h"

	namespace base {

	struct Watch {
	ObjectWatcher* watcher; // associated ObjectWatcher instance
	int object; // the file descriptor being watched
	int watch_handle; // used for uniquely identifying watches
	MessageLoop* origin_loop; // the origin thread's message loop
	ObjectWatcher::Delegate* delegate; // delegate to notify when signaled
	bool did_signal; // set when DoneWaiting is called
	MessagePumpShell::Mode mode; // callback on read, write or both?

	void Run() {
	// The watcher may have already been torn down, in which case we need to
	// just get out of dodge.
	if (!watcher)
	return;

	DCHECK(did_signal);
	watcher->StopWatching();

	delegate->OnObjectSignaled(object);
	}
	};

	// To emulate the behavior of the former Task system:
	static void WatchTask(Watch* watch) {
	watch->Run();
	delete watch;
	}

	ObjectWatchMultiplexer* ObjectWatchMultiplexer::s_instance = NULL;

	ObjectWatchMultiplexer::ObjectWatchMultiplexer()
	: base::SimpleThread(
	"ObjectWatchMultiplexer Thread",
	base::SimpleThread::Options(
	// TODO: Pass these into ObjectWatchMultiplexer.
	0 /* kObjectWatcherThreadStackSize */,
	kThreadPriority_Default /* kObjectWatcherThreadPriority */,
	kNoThreadAffinity /* kNetworkIOThreadAffinity */)),
	non_empty_list_event_(true, false),
	max_watch_handle_(0),
	should_recompose_pollfd_array_(true),
	exit_(false) {
	DCHECK(!s_instance);
	s_instance = this;
	Start();
	}

	ObjectWatchMultiplexer::~ObjectWatchMultiplexer() {
	DCHECK(s_instance);
	Join();
	s_instance = NULL;
	}

	// blocking call to exit the internal thread
	void ObjectWatchMultiplexer::Join() {
	{
	// RecomposePollFDArray may reset the event, but can only be called
	// while this lock is held. To break the race between these two, grab
	// the lock before setting exit_ and signalling the event, and make sure
	// that exit_ is checked before resetting the event.
	base::AutoLock lock(add_watch_list_lock_);
	exit_ = true;
	// wake up the thread in case it's waiting on the empty list event
	non_empty_list_event_.Signal();
	}
	// in any event it should timeout eventually and see the exit flag..
	base::SimpleThread::Join();
	}

	void ObjectWatchMultiplexer::AddWatch(Watch* watch) {
	DCHECK(watch);
	DCHECK_EQ(watch->watch_handle, 0);
	// grab the mutex to modify the watch map et al
	base::AutoLock lock(add_watch_list_lock_);
	should_recompose_pollfd_array_ = true;
	watch->watch_handle = ++max_watch_handle_;
	new_watch_map_.insert(WatchMap::value_type(watch->object, watch));
	non_empty_list_event_.Signal();
	}

	void ObjectWatchMultiplexer::RemoveWatch(Watch* watch) {
	DCHECK(watch);
	base::AutoLock lock(remove_watch_list_lock_);
	WatchMap::iterator start = watch_map_.lower_bound(watch->object);
	WatchMap::iterator end = watch_map_.upper_bound(watch->object);
	WatchMap::iterator it;
	// search within those elements having the same file descriptor
	for (it = start; it != end; it++)
	if (it->second->watch_handle == watch->watch_handle)
	break;
	if (it != end) {
	watch_map_.erase(it);
	should_recompose_pollfd_array_ = true;
	} else {
	base::AutoLock lock(add_watch_list_lock_);
	// not found, search new_watch_map_
	WatchMap::iterator start = new_watch_map_.lower_bound(watch->object);
	WatchMap::iterator end = new_watch_map_.upper_bound(watch->object);
	for (it = start; it != end; it++)
	if (it->second->watch_handle == watch->watch_handle)
	break;
	if (it != end) {
	new_watch_map_.erase(it);
	should_recompose_pollfd_array_ = true;
	}
	}
	}

	void ObjectWatchMultiplexer::Run() {
	// we keep a stack of pointers to watches needing a callback and
	// subsequent removal from the map
	std::stack<WatchMap::iterator> watch_removal_stack;
	// separate stack of callbacks to process after releasing the
	// WatchMap mutex
	std::stack<Watch*> watch_callback_stack;

	while (!exit_) {
	// We need both locks here. Order is important
	remove_watch_list_lock_.Acquire();
	add_watch_list_lock_.Acquire();
	if (should_recompose_pollfd_array_)
	RecomposePollFDArray();
	add_watch_list_lock_.Release();
	remove_watch_list_lock_.Release();

	if (!pollfd_array_.empty()) {
	// timeout on a 100ms to wake up and see if we need
	// to modify the array of fds or exit..
	int event_count =
	poll(&pollfd_array_.front(), pollfd_array_.size(), kPollTimeout);
	// negative value returned means error...
	if (event_count < 0) {
	// Sleep for kPollTimeout, otherwise this thread could busy-wait until
	// poll returns a non-error value
	base::PlatformThread::Sleep(
	base::TimeDelta::FromMilliseconds(kPollTimeout));
	}

	// we need the mutex to access the map. AddWatch() is on a different
	// map object so we dont need the add_watch_list_lock_ here.
	remove_watch_list_lock_.Acquire();
	DCHECK_LE(event_count, static_cast<int>(pollfd_array_.size()));
	for (int j = 0; j < pollfd_array_.size() && event_count > 0; ++j) {
	if (pollfd_array_[j].revents != 0) {
	event_count--;
	WatchMap::iterator start =
	watch_map_.lower_bound(pollfd_array_[j].fd);
	WatchMap::iterator stop =
	watch_map_.upper_bound(pollfd_array_[j].fd);
	for (WatchMap::iterator it = start; it != stop; it++) {
	// oh STL, your capacity for obfuscation never ceases to amaze.
	// this if/else block tries to match interest for each fd event
	// to the associated watchers. If a watcher was watching for
	// reads and a read happened, then we add it to the stack of
	// callbacks to process. Else if it was watching for writes and
	// a write happened then we do the same.
	// note that signaled watches are going to become the property
	// of their respectively owned MessageLoops and will be deleted
	// once executed, and so will be removed from the hash_map
	if ((pollfd_array_[j].revents & POLLIN) &&
	(it->second->mode & MessagePumpShell::WATCH_READ)) {
	watch_removal_stack.push(it);
	watch_callback_stack.push(it->second);
	} else if ((pollfd_array_[j].revents & POLLOUT) &&
	(it->second->mode & MessagePumpShell::WATCH_WRITE)) {
	watch_removal_stack.push(it);
	watch_callback_stack.push(it->second);
	}
	}
	}
	} // for(int j = 0; j < pollfd_array_.size() && event_count > 0; ++j)

	// process all callbacks. this must unfortunately be done inside
	// of the mutex to prevent the contention on watch objects being
	// deleted due to a call to ObjectWatcher::StopWatching that could
	// occur while the watch is on this callback stack
	while (!watch_callback_stack.empty()) {
	Watch* watch = watch_callback_stack.top();
	watch_callback_stack.pop();
	// signal this watch to run in the original thread context
	// of the calling thread message loop. The watch will be
	// posted as a task to the MessageLoop which will take
	// ownership of the object and delete it
	watch->did_signal = true;
	watch->origin_loop->PostTask(FROM_HERE, base::Bind(WatchTask, watch));
	}
	// if we're going to signal something we should recompose
	// the array as we are going to be removing watches
	if (!watch_removal_stack.empty())
	should_recompose_pollfd_array_ = true;
	// now remove the callback elements from the current hash_map
	while (!watch_removal_stack.empty()) {
	watch_map_.erase(watch_removal_stack.top());
	watch_removal_stack.pop();
	}

	remove_watch_list_lock_.Release();
	} else { // if (!pollfd_array_.empty())
	// sleep until flagged that something has been added to the list
	DCHECK(watch_map_.empty());
	// NOTE: You can't DCHECK(new_watch_map_.empty()) because we let go
	// of the lock already and new stuff could have been added before we
	// get here. If that happens, the event will have already been
	// signaled and we won't have to wait.
	non_empty_list_event_.Wait();
	} // if (!pollfd_array_.empty())
	} // while (!exit_)
	}

	void ObjectWatchMultiplexer::RecomposePollFDArray() {
	remove_watch_list_lock_.AssertAcquired();
	add_watch_list_lock_.AssertAcquired();
	// process new requests and move them into working map
	WatchMap::iterator it;
	for (it = new_watch_map_.begin(); it != new_watch_map_.end(); it++) {
	watch_map_.insert(*it);
	}
	new_watch_map_.clear();
	// count the number of unique file descriptors in the map
	int count = 0;
	int last_fd = -1;
	for (it = watch_map_.begin(); it != watch_map_.end(); it++) {
	if (it->first != last_fd) {
	last_fd = it->first;
	count++;
	}
	}
	pollfd_array_.resize(count);

	// init pollfd_array_
	if (!pollfd_array_.empty()) {
	int j = -1;
	last_fd = -1;
	for (it = watch_map_.begin(); it != watch_map_.end(); it++) {
	if (it->first != last_fd) {
	// we keep j pointing at this current element for ORing in the watch
	// flags
	j++;
	last_fd = it->first;
	pollfd_array_[j].fd = last_fd;
	// assume we care about both, we will filter based on mode on callback
	pollfd_array_[j].events = 0;
	if (it->second->mode & MessagePumpShell::WATCH_READ)
	pollfd_array_[j].events \|= POLLIN;
	if (it->second->mode & MessagePumpShell::WATCH_WRITE)
	pollfd_array_[j].events \|= POLLOUT;
	pollfd_array_[j].revents = 0;
	} else { // if (it->first != last_fd)
	// OR in the watch flags to the already initialized pollfd struct
	if (it->second->mode & MessagePumpShell::WATCH_READ)
	pollfd_array_[j].events \|= POLLIN;
	if (it->second->mode & MessagePumpShell::WATCH_WRITE)
	pollfd_array_[j].events \|= POLLOUT;
	} // if (it->first != last_fd)
	} // for (it = watch_map_.begin(); it != watch_map_.end(); it++)
	} else { // if (!pollfd_array_.empty())
	// reset the empty event if we've completely emptied the watch list
	DCHECK(watch_map_.empty());
	DCHECK(new_watch_map_.empty());
	// Join may have signalled the event because we are exitting. If so,
	// don't reset the event. That would lead to an infinite wait on
	// shutdown.
	if (!exit_)
	non_empty_list_event_.Reset();
	} // if (!pollfd_array_.empty())
	// reset flag
	should_recompose_pollfd_array_ = false;
	}

	ObjectWatcher::ObjectWatcher() : watch_(NULL) {
	}

	ObjectWatcher::~ObjectWatcher() {
	StopWatching();
	}

	bool ObjectWatcher::StartWatching(int object,
	MessagePumpShell::Mode mode,
	Delegate* delegate) {
	DCHECK(ObjectWatchMultiplexer::GetInstance() != NULL);
	if (watch_) {
	NOTREACHED() << "Already watching an object";
	return false;
	}

	watch_ = new Watch;
	watch_->watcher = this;
	watch_->object = object;
	watch_->mode = mode;
	watch_->origin_loop = MessageLoop::current();
	watch_->delegate = delegate;
	watch_->did_signal = false;
	watch_->watch_handle = 0;

	ObjectWatchMultiplexer::GetInstance()->AddWatch(watch_);

	// We need to know if the current message loop is going away so we can
	// prevent the wait thread from trying to access a dead message loop.
	MessageLoop::current()->AddDestructionObserver(this);
	return true;
	}

	bool ObjectWatcher::StopWatching() {
	if (!watch_)
	return false;

	DCHECK(ObjectWatchMultiplexer::GetInstance() != NULL);

	// make sure stop call happens on same thread as start call
	DCHECK(watch_->origin_loop == MessageLoop::current());

	// this will block until this watch has been removed from the mux
	ObjectWatchMultiplexer::GetInstance()->RemoveWatch(watch_);
	// let the watch know that the watcher has died, in case the watch is
	// still sitting in a message loop. See Watch::Run() to see that it
	// will bug out in that event.
	watch_->watcher = NULL;

	// if the watch was signaled, then it is now property of the MessageLoop
	// and will be deleted there. It has also been safely removed from the
	// internal list of watches in the ObjectWatchMux. Therefore if it hasn't
	// been signaled we delete it here to avoid leaking the watch
	if (!watch_->did_signal)
	delete watch_;

	watch_ = NULL;

	MessageLoop::current()->RemoveDestructionObserver(this);
	return true;
	}

	void ObjectWatcher::WillDestroyCurrentMessageLoop() {
	// Need to shutdown the watch so that we don't try to access the MessageLoop
	// after this point.
	StopWatching();
	}

	} // namespace base