Use FSEvents for recursive file watch on Mac

base/files/file_path_watcher_fsevents.cc

Index: base/files/file_path_watcher_fsevents.cc
diff --git a/base/files/file_path_watcher_fsevents.cc b/base/files/file_path_watcher_fsevents.cc
new file mode 100644
index 0000000000000000000000000000000000000000..d7fa65733a315b3d596d865e766e47dc7c916179
--- /dev/null
+++ b/base/files/file_path_watcher_fsevents.cc
@@ -0,0 +1,263 @@
+// Copyright 2014 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 "base/files/file_path_watcher_fsevents.h"
+
+#include <list>
+
+#include "base/bind.h"
+#include "base/file_util.h"
+#include "base/lazy_instance.h"
+#include "base/logging.h"
+#include "base/mac/libdispatch_task_runner.h"
+#include "base/mac/scoped_cftyperef.h"
+#include "base/message_loop/message_loop.h"
+
+namespace base {
+
+namespace {
+
+// The latency parameter passed to FSEventsStreamCreate().
+const CFAbsoluteTime kEventLatencySeconds = 0.3;
+
+class FSEventsTaskRunner : public mac::LibDispatchTaskRunner {
+ public:
+   FSEventsTaskRunner()
+       : mac::LibDispatchTaskRunner("chromium.org.FilePathWatcherFSEvents") {

[Mark Mentovai, 2014/06/08 15:29:02]

These should be reverse DNS: org.chromium.whatever.

[Mark Mentovai, 2014/06/08 15:29:02]

Do we really need another thread for this? The UI thread should be capable of
receiving FSEvent notifications. This shouldn’t be considered dangerous because
there won’t be any blocking involved. The UI thread can post tasks to the IO
thread or wherever you actually want things to be handled (as they would be on
other platforms).

[vandebo (ex-Chrome), 2014/06/10 00:52:05]

There was a thread "(OSX) Message Loops in Chrome" that you were all, the
conclusion of which was to use a dispatch queue.  I am not a Mac expert and
relied on the conclusion of that thread.

As I understand things, dispatch queues don't necessarily make a new thread, but
there is a thread pool which tasks are dispatched to - like the Chrome worker
pull.

I don't think CFRunLoopGetMain() is a complete solution for the UI thread, since
I would need to posts tasks to the UI thread. Therefore, I'd need to augment the
interface to get a reference to the UI message loop, either by changing the
Watch() method or adding a static method to set it during start up.

[vandebo (ex-Chrome), 2014/06/10 00:52:06]

Done.

[Robert Sesek, 2014/06/10 21:34:05]

This is a correct understanding. Libdispatch dynamically manages a thread pool
and guarantees execution semantics for queues, though queues can be scheduled on
arbitrary threads. We already will have Libdispatch threads running in the
browser, so this shouldn't be too bad.
I think this an appropriate use of LibDispatchTaskRunner. On first glance, I
thought it would maybe be possible to just use Libdispatch/dispatch_async()
directly, even for what is currently being PostTask()ed. But because the target
object of base::Bind() is C++ ref-counted, using blocks exclusively (without
LibDispatchTaskRunner) would mess up (or make more clunky) the ref-counting of
that object. I think this is a good use of LibDispatchTaskRunner.

+   }
+
+ protected:
+   virtual ~FSEventsTaskRunner() {}
+};
+
+static LazyInstance<FSEventsTaskRunner>::Leaky g_task_runner =
+    LAZY_INSTANCE_INITIALIZER;
+
+// Resolve any symlinks in the path.
+FilePath ResolvePath(const FilePath& path) {

[Mark Mentovai, 2014/06/08 15:29:02]

Can you just call realpath()?

[vandebo (ex-Chrome), 2014/06/10 00:52:06]

Hmm, it seems so.  Thanks.

[vandebo (ex-Chrome), 2014/06/10 17:19:30]

Actually no - I must have missed a compile or something when I tested this. 
realpath() assumes that the entire path exists.  The recursive watch tests
assume that the full path may not exist and tests that creating the final
directory triggers a watch.

+  const unsigned kMaxLinksToResolve = 255;
+
+  std::vector<FilePath::StringType> component_vector;
+  path.GetComponents(&component_vector);
+  std::list<FilePath::StringType>
+      components(component_vector.begin(), component_vector.end());
+
+  FilePath result;
+  unsigned resolve_count = 0;
+  while (resolve_count < kMaxLinksToResolve && !components.empty()) {
+    FilePath component(*components.begin());
+    components.pop_front();
+
+    FilePath current;
+    if (component.IsAbsolute()) {
+      current = component;
+    } else {
+      current = result.Append(component);
+    }
+
+    FilePath target;
+    if (ReadSymbolicLink(current, &target)) {
+      if (target.IsAbsolute())
+        result.clear();
+      std::vector<FilePath::StringType> target_components;
+      target.GetComponents(&target_components);
+      components.insert(components.begin(), target_components.begin(),
+                        target_components.end());
+      resolve_count++;
+    } else {
+      result = current;
+    }
+  }
+
+  if (resolve_count >= kMaxLinksToResolve)
+    result.clear();
+  return result;
+}
+
+// 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[]) {
+  FilePathWatcherFSEvents* watcher =
+      reinterpret_cast<FilePathWatcherFSEvents*>(event_watcher);
+  DCHECK(g_task_runner.Get().RunsTasksOnCurrentThread());
+
+  bool root_changed = watcher->ResolveTargetPath();
+  std::vector<FilePath> paths;
+  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]);
+    paths.push_back(FilePath(
+        reinterpret_cast<char**>(event_paths)[i]).StripTrailingSeparators());
+  }
+
+  // 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.
+    g_task_runner.Get().PostTask(
+        FROM_HERE,
+        Bind(&FilePathWatcherFSEvents::UpdateEventStream, watcher,
+             root_change_at));
+  }
+
+  watcher->OnFilePathsChanged(paths);

[Mark Mentovai, 2014/06/08 15:29:02]

UpdateEventStream happens in the future. OnFilePathsChanged happens right here,
right now. If some changes to the filesystem occur in between these two events,
they’ll be missed.

[vandebo (ex-Chrome), 2014/06/10 00:52:06]

My understanding is that the event id prevents that race.  When
UpdateEventStream() calls FSEVETNStreamCreate(), it passes |root_change_at| as
the starting point for notifications, so it doesn't matter how much wall clock
time passes.

+}
+
+}  // namespace
+
+FilePathWatcherFSEvents::FilePathWatcherFSEvents() : fsevent_stream_(NULL) {
+}
+
+void FilePathWatcherFSEvents::OnFilePathsChanged(
+    const std::vector<FilePath>& paths) {
+  if (!message_loop()->BelongsToCurrentThread()) {
+    message_loop()->PostTask(
+        FROM_HERE,
+        Bind(&FilePathWatcherFSEvents::OnFilePathsChanged, this, paths));
+    return;
+  }
+
+  DCHECK(message_loop()->BelongsToCurrentThread());
+  if (resolved_target_.empty())
+    return;
+
+  for (size_t i = 0; i < paths.size(); i++) {
+    if (resolved_target_.IsParent(paths[i]) || resolved_target_ == paths[i]) {
+      callback_.Run(target_, false);
+      return;
+    }
+  }
+}
+
+bool FilePathWatcherFSEvents::Watch(const FilePath& path,
+                                    bool recursive,
+                                    const FilePathWatcher::Callback& callback) {
+  DCHECK(resolved_target_.empty());
+  DCHECK(MessageLoopForIO::current());
+  DCHECK(!callback.is_null());
+
+  // This class could support non-recursive watches, but that is currently
+  // left to FilePathWatcherKQueue.
+  if (!recursive)
+    return false;
+
+  set_message_loop(MessageLoopProxy::current());
+  callback_ = callback;
+  target_ = path;
+
+  FSEventStreamEventId start_event = FSEventsGetCurrentEventId();
+  g_task_runner.Get().PostTask(
+      FROM_HERE,
+      Bind(&FilePathWatcherFSEvents::StartEventStream, this, start_event));
+  return true;
+}
+
+void FilePathWatcherFSEvents::Cancel() {
+  if (callback_.is_null()) {
+    // Watch was never called, so exit.
+    set_cancelled();
+    return;
+  }
+
+  // Switch to the dispatch queue thread if necessary, so we can tear down
+  // the event stream.
+  if (!g_task_runner.Get().RunsTasksOnCurrentThread()) {
+    g_task_runner.Get().PostTask(
+        FROM_HERE,
+        Bind(&FilePathWatcherFSEvents::CancelOnMessageLoopThread, this));
+  } else {
+    CancelOnMessageLoopThread();
+  }
+}
+
+void FilePathWatcherFSEvents::CancelOnMessageLoopThread() {
+  // For all other implementations, the "message loop thread" is the IO thread,
+  // as returned by message_loop(). This implementation, however, needs to
+  // cancel pending work on the Dipatch Queue thread.
+  DCHECK(g_task_runner.Get().RunsTasksOnCurrentThread());
+
+  set_cancelled();
+  if (fsevent_stream_) {
+    DestroyEventStream();
+    callback_.Reset();
+    target_.clear();
+    resolved_target_.clear();
+  }
+}
+
+void FilePathWatcherFSEvents::UpdateEventStream(
+    FSEventStreamEventId start_event) {
+  DCHECK(g_task_runner.Get().RunsTasksOnCurrentThread());
+
+  // 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() || resolved_target_.empty())
+    return;
+
+  if (fsevent_stream_)
+    DestroyEventStream();
+
+  ScopedCFTypeRef<CFStringRef> cf_path(CFStringCreateWithCString(
+      NULL, resolved_target_.value().c_str(), kCFStringEncodingMacHFS));
+  ScopedCFTypeRef<CFStringRef> cf_dir_path(CFStringCreateWithCString(
+      NULL, resolved_target_.DirName().value().c_str(),
+      kCFStringEncodingMacHFS));
+  CFStringRef paths_array[] = { cf_path.get(), cf_dir_path.get() };
+  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);
+  FSEventStreamSetDispatchQueue(fsevent_stream_,
+                                g_task_runner.Get().GetDispatchQueue());
+
+  if (!FSEventStreamStart(fsevent_stream_))
+    message_loop()->PostTask(FROM_HERE, Bind(callback_, target_, true));
+}
+
+bool FilePathWatcherFSEvents::ResolveTargetPath() {
+  DCHECK(g_task_runner.Get().RunsTasksOnCurrentThread());
+  FilePath resolved = ResolvePath(target_).StripTrailingSeparators();
+  bool changed = resolved != resolved_target_;
+  resolved_target_ = resolved;
+  if (resolved_target_.empty())
+    message_loop()->PostTask(FROM_HERE, Bind(callback_, target_, true));
+  return changed;
+}
+
+void FilePathWatcherFSEvents::DestroyEventStream() {
+  FSEventStreamStop(fsevent_stream_);
+  FSEventStreamInvalidate(fsevent_stream_);
+  FSEventStreamRelease(fsevent_stream_);
+  fsevent_stream_ = NULL;
+}
+
+void FilePathWatcherFSEvents::StartEventStream(
+    FSEventStreamEventId start_event) {
+  DCHECK(g_task_runner.Get().RunsTasksOnCurrentThread());
+  ResolveTargetPath();
+  UpdateEventStream(start_event);
+}
+
+FilePathWatcherFSEvents::~FilePathWatcherFSEvents() {}
+
+}  // namespace base