kqueue implementation of FilePathWatcher on Mac.

content/common/file_path_watcher/file_path_watcher_mac.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 "content/common/file_path_watcher/file_path_watcher.h"
 
-#include <CoreServices/CoreServices.h>
-#include <set>
-
-#include "base/file_path.h"
+#include <fcntl.h>
+#include <sys/event.h>
+#include <sys/param.h>
+
+#include <vector>
+
 #include "base/file_util.h"
-#include "base/logging.h"
-#include "base/mac/scoped_cftyperef.h"
 #include "base/message_loop.h"
-#include "base/singleton.h"
-#include "base/time.h"
-
-// Note to future well meaning engineers. Unless kqueue semantics have changed
-// considerably, do NOT try to reimplement this class using kqueue. The main
-// problem is that this class requires the ability to watch a directory
-// and notice changes to any files within it. A kqueue on a directory can watch
-// for creation and deletion of files, but not for modifications to files within
-// the directory. To do this with the current kqueue semantics would require
-// kqueueing every file in the directory, and file descriptors are a limited
-// resource. If you have a good idea on how to get around this, the source for a
-// reasonable implementation of this class using kqueues is attached here:
-// http://code.google.com/p/chromium/issues/detail?id=54822#c13
+#include "base/message_loop_proxy.h"
 
 namespace {
 
-// The latency parameter passed to FSEventsStreamCreate().
-const CFAbsoluteTime kEventLatencySeconds = 0.3;
-
-// Mac-specific file watcher implementation based on the FSEvents API.
+// Mac-specific file watcher implementation based on kqueue.
+// Originally it was based on FSEvents so that the semantics were equivalent
+// on Linux, OSX and Windows where it was able to detect:
+// - file creation/deletion/modification in a watched directory
+// - file creation/deletion/modification for a watched file
+// - modifications to the paths to a watched object that would affect the
+//   object such as renaming/attibute changes etc.
+// The FSEvents version did all of the above except handling attribute changes
+// to path components. Unfortunately FSEvents appears to have an issue where the
+// current implementation (Mac OS X 10.6.7) sometimes drops events and doesn't
+// send notifications. See
+// http://code.google.com/p/chromium/issues/detail?id=54822#c31 for source that
+// will reproduce the problem. The kqueue implementation will handle all of the
+// items in the list above except for detecting modifications to files in a
+// watched directory. It will detect the creation and deletion of files, just
+// not the modification of files. It does however detect the attribute changes
+// that the FSEvents impl would miss.

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

Maybe also mention that FSEvents needs a runloop backed MessageLoop, so it adds
additional complexity.

[dmac, 2011/03/30 17:18:47]

Done.

 class FilePathWatcherImpl : public FilePathWatcher::PlatformDelegate,
+                            public MessageLoopForIO::Watcher,
                             public MessageLoop::DestructionObserver {
  public:
-  FilePathWatcherImpl();
-
-  // Called from the FSEvents callback whenever there is a change to the paths
-  void OnFilePathChanged();
-
-  // (Re-)Initialize the event stream to start reporting events from
-  // |start_event|.
-  void UpdateEventStream(FSEventStreamEventId start_event);
+  FilePathWatcherImpl() : kqueue_(-1) {}
+  virtual ~FilePathWatcherImpl() {}
+
+  // MessageLoopForIO::Watcher overrides.
+  virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE;
+  virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE;
+
+  // MessageLoop::DestructionObserver overrides.
+  virtual void WillDestroyCurrentMessageLoop() OVERRIDE;
 
   // FilePathWatcher::PlatformDelegate overrides.
   virtual bool Watch(const FilePath& path,
-                     FilePathWatcher::Delegate* delegate,
-                     base::MessageLoopProxy* loop) OVERRIDE;
+                     FilePathWatcher::Delegate* delegate) OVERRIDE;
   virtual void Cancel() OVERRIDE;
 
-  // Deletion of the FilePathWatcher will call Cancel() to dispose of this
-  // object in the right thread. This also observes destruction of the required
-  // cleanup thread, in case it quits before Cancel() is called.
-  virtual void WillDestroyCurrentMessageLoop() OVERRIDE;
-
-  scoped_refptr<base::MessageLoopProxy> run_loop_message_loop() {
-    return run_loop_message_loop_;
-  }
-
  private:
-  virtual ~FilePathWatcherImpl() {}
-
-  // Destroy the event stream.
-  void DestroyEventStream();
-
-  // Start observing the destruction of the |run_loop_message_loop_| thread,
-  // and watching the FSEventStream.
-  void StartObserverAndEventStream(FSEventStreamEventId start_event);
-
-  // Cleans up and stops observing the |run_loop_message_loop_| thread.
-  void CancelOnMessageLoopThread() OVERRIDE;
-
-  // Delegate to notify upon changes.
+  class EventData {
+   public:
+    EventData(const FilePath& path, const FilePath::StringType& subdir)
+        : path_(path), subdir_(subdir) { }
+    FilePath path_;  // Full path to this item.
+    FilePath::StringType subdir_;  // Path to any sub item.
+  };
+  typedef std::vector<struct kevent> EventVector;

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

Ah, does this mean we consume one file descriptor per path component? Is this
acceptable?

[dmac, 2011/03/30 17:18:47]

That's kind of a requirement for the kqueue implementation, unless you have a
solution to it?

+
+  virtual void CancelOnMessageLoopThread() OVERRIDE;

[Mark Mentovai, 2011/03/29 21:27:25]

This deserves a comment. Does it mean that it can be called fro any thread and
asks for work to be done on MessageLoopThread, or that it must be called from
MessageLoopThread?

[dmac, 2011/03/30 17:18:47]

Done.

+
+  // Fills |events| with one kevent per component in |path|.
+  // Returns the number of valid events created where a valid event is
+  // defined as one that has a ident (file descriptor) field != -1.
+  static int EventsForPath(FilePath path, EventVector *events);
+
+  // Release a kevent generated by EventsForPath.
+  static void ReleaseEvent(struct kevent& event);

[Mark Mentovai, 2011/03/29 21:27:25]

It’s unorthodox in our style to pass a non-const reference. Can this be a
pointer?

[dmac, 2011/03/30 17:18:47]

I'm only doing it this way because I use foreach below to close things and that
passes struct kevent& in. If you feel strongly about it, I can get rid of the
for each and pass them in by pointer.

+
+  // Returns a file descriptor that will not block the system from deleting
+  // the file it references.
+  static int FileDescriptorForPath(const FilePath& path);
+
+  // Closes |*fd| and sets |*fd| to -1.
+  static void CloseFileDescriptor(int *fd);

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

use int* instead of int *

[dmac, 2011/03/30 17:18:47]

Done.

+
+  EventVector events_;
+  scoped_refptr<base::MessageLoopProxy> io_message_loop_;
+  MessageLoopForIO::FileDescriptorWatcher kqueue_watcher_;
   scoped_refptr<FilePathWatcher::Delegate> delegate_;
-
-  // Target path to watch (passed to delegate).
   FilePath target_;
-
-  // Keep track of the last modified time of the file.  We use nulltime
-  // to represent the file not existing.
-  base::Time last_modified_;
-
-  // The time at which we processed the first notification with the
-  // |last_modified_| time stamp.
-  base::Time first_notification_;
-
-  // Backend stream we receive event callbacks from (strong reference).
-  FSEventStreamRef fsevent_stream_;
-
-  // Run loop for FSEventStream to run on.
-  scoped_refptr<base::MessageLoopProxy> run_loop_message_loop_;
+  int kqueue_;
 
   DISALLOW_COPY_AND_ASSIGN(FilePathWatcherImpl);
 };
 
-// The callback passed to FSEventStreamCreate().
-void FSEventsCallback(ConstFSEventStreamRef stream,
-                      void* event_watcher, size_t num_events,
-                      void* event_paths, const FSEventStreamEventFlags flags[],
-                      const FSEventStreamEventId event_ids[]) {
-  FilePathWatcherImpl* watcher =
-      reinterpret_cast<FilePathWatcherImpl*>(event_watcher);
-  DCHECK(watcher->run_loop_message_loop()->BelongsToCurrentThread());
-
-  bool root_changed = false;
-  FSEventStreamEventId root_change_at = FSEventStreamGetLatestEventId(stream);
-  for (size_t i = 0; i < num_events; i++) {
-    if (flags[i] & kFSEventStreamEventFlagRootChanged)
-      root_changed = true;
-    if (event_ids[i])
-      root_change_at = std::min(root_change_at, event_ids[i]);
-  }
-
-  // Reinitialize the event stream if we find changes to the root. This is
-  // necessary since FSEvents doesn't report any events for the subtree after
-  // the directory to be watched gets created.
-  if (root_changed) {
-    // Resetting the event stream from within the callback fails (FSEvents spews
-    // bad file descriptor errors), so post a task to do the reset.
-    watcher->run_loop_message_loop()->PostTask(FROM_HERE,
-        NewRunnableMethod(watcher, &FilePathWatcherImpl::UpdateEventStream,
-                          root_change_at));
-  }
-
-  watcher->OnFilePathChanged();
-}
-
-// FilePathWatcherImpl implementation:
-
-FilePathWatcherImpl::FilePathWatcherImpl()
-    : fsevent_stream_(NULL) {
-}
-
-void FilePathWatcherImpl::OnFilePathChanged() {
-  // Switch to the CFRunLoop based thread if necessary, so we can tear down
-  // the event stream.
-  if (!message_loop()->BelongsToCurrentThread()) {
-    message_loop()->PostTask(
-        FROM_HERE,
-        NewRunnableMethod(this, &FilePathWatcherImpl::OnFilePathChanged));
+void FilePathWatcherImpl::ReleaseEvent(struct kevent& event) {
+  EventData* entry = reinterpret_cast<EventData*>(event.udata);
+  delete entry;
+  event.udata = NULL;
+}
+
+int FilePathWatcherImpl::EventsForPath(FilePath path, EventVector *events) {

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

Asterisk by the type.

[dmac, 2011/03/30 17:18:47]

Done.

+  DCHECK(MessageLoopForIO::current());
+  // Make sure that we are working with a clean slate.
+  DCHECK(events->empty());
+
+  std::vector<FilePath::StringType> components;
+  path.GetComponents(&components);
+
+  if (components.size() < 1) {
+    return -1;
+  }
+
+  int last_existing_entry = 0;
+  FilePath built_path;
+  bool path_still_exists = true;
+  for(std::vector<FilePath::StringType>::iterator i = components.begin();
+      i != components.end(); ++i) {
+    if (i == components.begin()) {
+      built_path = FilePath(*i);
+    } else {
+      built_path = built_path.Append(*i);
+    }
+    int fd = -1;
+    if (path_still_exists) {
+      fd = FileDescriptorForPath(built_path);

[Mark Mentovai, 2011/03/29 21:27:25]

Do you have to worry about closing these ever? You only seem to close them in
some cases.

[dmac, 2011/03/30 17:18:47]

Good catch. ReleaseEvent now handles closing them.

+      if (fd == -1) {
+        path_still_exists = false;
+      } else {
+        ++last_existing_entry;
+      }
+    }
+    FilePath::StringType subdir = (i != (components.end() - 1)) ? *(i + 1) : "";
+    EventData* data = new EventData(built_path, subdir);
+    struct kevent event;
+    EV_SET(&event, fd, EVFILT_VNODE, (EV_ADD | EV_CLEAR | EV_RECEIPT),
+           (NOTE_DELETE | NOTE_WRITE | NOTE_ATTRIB
+            | NOTE_RENAME | NOTE_REVOKE | NOTE_EXTEND), 0, data);

[Mark Mentovai, 2011/03/29 21:27:25]

Put the operator at the end of the preceding line instead of the beginning of
this one. This is a Chromium-only requirement, not part of standard Google
style.

http://www.chromium.org/developers/coding-style#Code_Formatting

[dmac, 2011/03/30 17:18:47]

Done.

+    events->push_back(event);
+  }
+  return last_existing_entry;
+}
+
+int FilePathWatcherImpl::FileDescriptorForPath(const FilePath& path) {
+  return HANDLE_EINTR(open(path.value().c_str(), O_EVTONLY | O_NONBLOCK));
+}
+
+void FilePathWatcherImpl::CloseFileDescriptor(int *fd) {
+  if (*fd == -1) {
     return;
   }

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

No need for curlies for single-line conditionals (also at other places), but if
you like them, keep them :)

[dmac, 2011/03/30 17:18:47]

I like them. With all the python programmers we have around who think that
indentation is going to save them, I prefer to be explicit about my blocks.

 
-  DCHECK(message_loop()->BelongsToCurrentThread());
-  DCHECK(!target_.empty());
-
-  base::PlatformFileInfo file_info;
-  bool file_exists = file_util::GetFileInfo(target_, &file_info);
-  if (file_exists && (last_modified_.is_null() ||
-      last_modified_ != file_info.last_modified)) {
-    last_modified_ = file_info.last_modified;
-    first_notification_ = base::Time::Now();
+  if (HANDLE_EINTR(close(*fd)) != 0) {
+    PLOG(ERROR) << "close";
+  }
+  *fd = -1;
+}
+
+void FilePathWatcherImpl::OnFileCanReadWithoutBlocking(int fd) {

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

This is a very long function that handles separate aspects, i.e. detecting
events and updating watches. Maybe break it up a bit?

[dmac, 2011/03/30 17:18:47]

I tried breaking it up a bit, and it didn't really break cleanly. Hopefully the
added comments will help.

+  DCHECK(MessageLoopForIO::current());
+  CHECK_EQ(fd, kqueue_);
+  std::vector<struct kevent> updates(events_.size());

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

Maybe a scoped_array is more appropriate?

[Mark Mentovai, 2011/03/30 13:59:08]

Mattias Nissler wrote:
No, std::vector is. :)

[Mattias Nissler (ping if slow), 2011/03/30 14:07:52]

Can you give reasons? He doesn't use any vector-specific goodness, so what's bad
about a simple heap-allocated array (which is what vector alos does behind the
scenes)?

[dmac, 2011/03/30 17:18:47]

This discussion was taken to email ;-) Left as a vector for now.

+  struct timespec timeout = {0, 0};
+  int count = kevent(kqueue_, NULL, 0, &updates[0], updates.size(), &timeout);

[Mark Mentovai, 2011/03/29 21:27:25]

HANDLE_EINTR.

[dmac, 2011/03/30 17:18:47]

Done.

+  if (count < 0) {
+    PLOG(ERROR) << "kevent";
+    delegate_->OnError();
+    Cancel();
+    return;
+  }
+  bool update_watches = false;
+  bool send_notification = false;
+
+  for (int i = 0; i < count; ++i) {
+    EventVector::iterator event(events_.begin());
+    for ( ; event != events_.end(); ++event) {
+      if (event->ident == updates[i].ident)
+        break;
+    }
+    CHECK(event != events_.end());

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

trusting the kernel to only pass valid fds? We've been bitten quite a few times
on windows because of this, so you may want to switch this over to
check-NOTREACHED()-continue.

[dmac, 2011/03/30 17:18:47]

I've never seen this on the Mac (not to say it can't happen). I'd feel
uncomfortable attempting to continue in that case, so I'm going to chicken out,
and error on it. I can't think of a way to unittest this case though.

+    EventData* event_data = reinterpret_cast<EventData*>(event->udata);
+    bool target_file_affected = event_data->subdir_.empty();
+    if (updates[i].fflags & NOTE_DELETE || updates[i].fflags & NOTE_REVOKE) {
+      // This is no longer accessible, so close the descriptor for it.
+      CloseFileDescriptor(reinterpret_cast<int*>(&event->ident));
+      update_watches = true;
+    } else if (updates[i].fflags & NOTE_RENAME) {
+      target_file_affected = true;
+      for (; event != events_.end(); ++event) {
+        // Close all nodes from the event down.
+        CloseFileDescriptor(reinterpret_cast<int*>(&event->ident));
+      }
+      update_watches = true;
+    } else if (updates[i].fflags & NOTE_WRITE) {
+      if (!target_file_affected) {
+        // In a directory this means a file has been added or deleted.
+        // Deletions are taken care of above so check for addition.
+        EventVector::iterator next_event = event + 1;
+        EventData* next_event_data =
+          reinterpret_cast<EventData*>(next_event->udata);

[Mark Mentovai, 2011/03/29 21:27:25]

Continuation lines are indented by 4.

[dmac, 2011/03/30 17:18:47]

Done.

+        if (next_event->ident == static_cast<uintptr_t>(-1)) {
+          next_event->ident = FileDescriptorForPath(next_event_data->path_);
+          if (next_event->ident != static_cast<uintptr_t>(-1)) {
+            update_watches = true;
+            if (next_event_data->subdir_.empty()) {
+              target_file_affected = true;
+            }
+          }
+        }
+      }
+    }
+    send_notification |= target_file_affected;
+  }
+
+  // Iterate over events adding kevents for items that exist to the kqueue.
+  // Then check to see if new components in the path have been created.
+  // Repeat until no new components in the path are detected.
+  // This is to get around races in directory creation in a watched path.
+  while (update_watches) {
+    size_t valid;
+    for (valid = 0; valid < events_.size(); ++valid) {
+      if (events_[valid].ident == static_cast<uintptr_t>(-1)) {
+        break;
+      }
+    }
+    count = kevent(kqueue_, &events_[0], valid, NULL, 0, NULL);

[Mark Mentovai, 2011/03/29 21:27:25]

HANDLE_EINTR (mostly just paranoia in this case).

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

Don't we need EV_ERROR checking for the responses?

[dmac, 2011/03/30 17:18:47]

Done.

[dmac, 2011/03/30 17:18:47]

Done.

+    if (count < 0) {
+      PLOG(ERROR) << "kevent";
+      delegate_->OnError();
+      Cancel();
+      return;
+    }
+    update_watches = false;
+    for(; valid < events_.size(); ++valid) {

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

space after for

[dmac, 2011/03/30 17:18:47]

Done.

+      EventData* event_data =
+          reinterpret_cast<EventData*>(events_[valid].udata);
+      events_[valid].ident = FileDescriptorForPath(event_data->path_);
+      if (events_[valid].ident != static_cast<uintptr_t>(-1)) {
+        update_watches = true;
+        if (event_data->subdir_.empty()) {
+          send_notification = true;
+        }
+      } else {
+        break;
+      }
+    }
+  }
+
+  if (send_notification) {
     delegate_->OnFilePathChanged(target_);
-  } else if (file_exists && !first_notification_.is_null()) {
-    // The target's last modification time is equal to what's on record. This
-    // means that either an unrelated event occurred, or the target changed
-    // again (file modification times only have a resolution of 1s). Comparing
-    // file modification times against the wall clock is not reliable to find
-    // out whether the change is recent, since this code might just run too
-    // late. Moreover, there's no guarantee that file modification time and wall
-    // clock times come from the same source.
-    //
-    // Instead, the time at which the first notification carrying the current
-    // |last_notified_| time stamp is recorded. Later notifications that find
-    // the same file modification time only need to be forwarded until wall
-    // clock has advanced one second from the initial notification. After that
-    // interval, client code is guaranteed to having seen the current revision
-    // of the file.
-    if (base::Time::Now() - first_notification_ >
-        base::TimeDelta::FromSeconds(1)) {
-      // Stop further notifications for this |last_modification_| time stamp.
-      first_notification_ = base::Time();
-    }
-    delegate_->OnFilePathChanged(target_);
-  } else if (!file_exists && !last_modified_.is_null()) {
-    last_modified_ = base::Time();
-    delegate_->OnFilePathChanged(target_);
-  }
+  }
+}
+
+void FilePathWatcherImpl::OnFileCanWriteWithoutBlocking(int fd) {
+  NOTREACHED();
+}
+
+void FilePathWatcherImpl::WillDestroyCurrentMessageLoop() {
+  CancelOnMessageLoopThread();
 }
 
 bool FilePathWatcherImpl::Watch(const FilePath& path,
-                                FilePathWatcher::Delegate* delegate,
-                                base::MessageLoopProxy* loop) {
-  DCHECK(target_.value().empty());
+                                FilePathWatcher::Delegate* delegate) {
   DCHECK(MessageLoopForIO::current());
-
-  set_message_loop(base::MessageLoopProxy::CreateForCurrentThread());
-  run_loop_message_loop_ = loop;
+  DCHECK(target_.value().empty());  // Can only watch one path.
+  DCHECK(delegate);
+  DCHECK_EQ(kqueue_, -1);
+
+  delegate_ = delegate;
   target_ = path;
-  delegate_ = delegate;
-
-  FSEventStreamEventId start_event = FSEventsGetCurrentEventId();
-
-  base::PlatformFileInfo file_info;
-  if (file_util::GetFileInfo(target_, &file_info)) {
-    last_modified_ = file_info.last_modified;
-    first_notification_ = base::Time::Now();
-  }
-
-  run_loop_message_loop()->PostTask(FROM_HERE,
-      NewRunnableMethod(this, &FilePathWatcherImpl::StartObserverAndEventStream,
-                        start_event));
-
-  return true;
-}
-
-void FilePathWatcherImpl::StartObserverAndEventStream(
-    FSEventStreamEventId start_event) {
-  DCHECK(run_loop_message_loop()->BelongsToCurrentThread());
+
   MessageLoop::current()->AddDestructionObserver(this);
-  UpdateEventStream(start_event);
+  io_message_loop_ = base::MessageLoopProxy::CreateForCurrentThread();
+
+  kqueue_ = kqueue();
+  if (kqueue_ == -1) {
+    PLOG(ERROR) << "kqueue";
+    return false;
+  }
+
+  int last_entry = EventsForPath(target_, &events_);
+  std::vector<struct kevent> responses(last_entry);

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

a scoped_array might be more appropriate here?

[dmac, 2011/03/30 17:18:47]

comment about vector vs scoped_array above.

+
+  if (HANDLE_EINTR(kevent(kqueue_, &events_[0], last_entry, &responses[0],
+                          last_entry, NULL)) == -1) {
+    PLOG(ERROR) << "kevent";
+    return false;

[Mattias Nissler (ping if slow), 2011/03/30 11:30:52]

Do we need to close the directory FDs when we hit an error here and below?

[dmac, 2011/03/30 17:18:47]

They would be closed off in the dtor, but I can see that closing them off
earlier may make sense here.

+  }
+
+  bool error_occurred = false;
+  for(int i = 0; i < last_entry; ++i) {
+    if (responses[i].flags & EV_ERROR && responses[i].data) {
+      LOG(ERROR) << "Error: " << responses[i].data;
+      error_occurred = true;
+    }
+  }
+  if (error_occurred) {
+    return false;
+  }
+
+  return MessageLoopForIO::current()->WatchFileDescriptor(
+      kqueue_, true, MessageLoopForIO::WATCH_READ, &kqueue_watcher_, this);
 }
 
 void FilePathWatcherImpl::Cancel() {
-  if (!run_loop_message_loop().get()) {
-    // Watch was never called, so exit.
+  base::MessageLoopProxy* proxy = io_message_loop_.get();
+  if (!proxy) {
     set_cancelled();
     return;
   }
-
-  // Switch to the CFRunLoop based thread if necessary, so we can tear down
-  // the event stream.
-  if (!run_loop_message_loop()->BelongsToCurrentThread()) {
-    run_loop_message_loop()->PostTask(FROM_HERE,
-                                      new FilePathWatcher::CancelTask(this));
-  } else {
-    CancelOnMessageLoopThread();
-  }
+  if (!proxy->BelongsToCurrentThread()) {
+    proxy->PostTask(FROM_HERE,
+                    NewRunnableMethod(this, &FilePathWatcherImpl::Cancel));
+    return;
+  }
+  CancelOnMessageLoopThread();
 }
 
 void FilePathWatcherImpl::CancelOnMessageLoopThread() {
-  set_cancelled();
-  if (fsevent_stream_) {
-    DestroyEventStream();
+  if (!is_cancelled()) {
+    set_cancelled();
+    kqueue_watcher_.StopWatchingFileDescriptor();
+    CloseFileDescriptor(&kqueue_);
+    std::for_each(events_.begin(), events_.end(), ReleaseEvent);
+    events_.clear();
+    io_message_loop_.release();
     MessageLoop::current()->RemoveDestructionObserver(this);
     delegate_ = NULL;
   }
 }
 
-void FilePathWatcherImpl::WillDestroyCurrentMessageLoop() {
-  CancelOnMessageLoopThread();
-}
-
-void FilePathWatcherImpl::UpdateEventStream(FSEventStreamEventId start_event) {
-  DCHECK(run_loop_message_loop()->BelongsToCurrentThread());
-  DCHECK(MessageLoopForUI::current());
-
-  // It can happen that the watcher gets canceled while tasks that call this
-  // function are still in flight, so abort if this situation is detected.
-  if (is_cancelled())
-    return;
-
-  if (fsevent_stream_)
-    DestroyEventStream();
-
-  base::mac::ScopedCFTypeRef<CFStringRef> cf_path(CFStringCreateWithCString(
-      NULL, target_.value().c_str(), kCFStringEncodingMacHFS));
-  base::mac::ScopedCFTypeRef<CFStringRef> cf_dir_path(CFStringCreateWithCString(
-      NULL, target_.DirName().value().c_str(), kCFStringEncodingMacHFS));
-  CFStringRef paths_array[] = { cf_path.get(), cf_dir_path.get() };
-  base::mac::ScopedCFTypeRef<CFArrayRef> watched_paths(CFArrayCreate(
-      NULL, reinterpret_cast<const void**>(paths_array), arraysize(paths_array),
-      &kCFTypeArrayCallBacks));
-
-  FSEventStreamContext context;
-  context.version = 0;
-  context.info = this;
-  context.retain = NULL;
-  context.release = NULL;
-  context.copyDescription = NULL;
-
-  fsevent_stream_ = FSEventStreamCreate(NULL, &FSEventsCallback, &context,
-                                        watched_paths,
-                                        start_event,
-                                        kEventLatencySeconds,
-                                        kFSEventStreamCreateFlagWatchRoot);
-  FSEventStreamScheduleWithRunLoop(fsevent_stream_, CFRunLoopGetCurrent(),
-                                   kCFRunLoopDefaultMode);
-  if (!FSEventStreamStart(fsevent_stream_)) {
-    message_loop()->PostTask(FROM_HERE,
-        NewRunnableMethod(delegate_.get(),
-                          &FilePathWatcher::Delegate::OnError));
-  }
-}
-
-void FilePathWatcherImpl::DestroyEventStream() {
-  FSEventStreamStop(fsevent_stream_);
-  FSEventStreamUnscheduleFromRunLoop(fsevent_stream_, CFRunLoopGetCurrent(),
-                                     kCFRunLoopDefaultMode);
-  FSEventStreamRelease(fsevent_stream_);
-  fsevent_stream_ = NULL;
-}
-
 }  // namespace
 
 FilePathWatcher::FilePathWatcher() {
   impl_ = new FilePathWatcherImpl();
 }