darwin: use FSEvents to watch directory changes

config-unix.mk

@@ -60,6 +60,7 @@ CPPFLAGS += -D_DARWIN_USE_64_BIT_INODE=1
 LINKFLAGS+=-framework CoreServices
 OBJS += src/unix/darwin.o
 OBJS += src/unix/kqueue.o
+OBJS += src/unix/fsevents.o
 endif
 
 ifeq (Linux,$(uname_S))

include/uv-private/uv-darwin.h

@@ -29,10 +29,23 @@
 # define UV_PLATFORM_SEM_T semaphore_t
 #endif
 
+#define UV_PLATFORM_LOOP_FIELDS                                               \
+  uv_thread_t cf_thread;                                                      \
+  void* cf_cb;                                                                \
+  void* cf_loop;                                                              \
+  uv_mutex_t cf_mutex;                                                        \
+  uv_sem_t cf_sem;                                                            \
+  ngx_queue_t cf_signals;                                                     \
+
 #define UV_PLATFORM_FS_EVENT_FIELDS                                           \
   ev_io event_watcher;                                                        \
   int fflags;                                                                 \
   int fd;                                                                     \
+  void* cf_eventstream;                                                       \
+  uv_async_t* cf_cb;                                                          \
+  ngx_queue_t cf_events;                                                      \
+  uv_sem_t cf_sem;                                                            \
+  uv_mutex_t cf_mutex;                                                        \
 
 #define UV_STREAM_PRIVATE_PLATFORM_FIELDS                                     \
   void* select;                                                               \

src/unix/darwin.c

@@ -43,13 +43,137 @@
 
 static char *process_title;
 
+/* Forward declarations */
+void uv__cf_loop_runner(void* arg);
+void uv__cf_loop_cb(void* arg);
+
+typedef struct uv__cf_loop_signal_s uv__cf_loop_signal_t;
+struct uv__cf_loop_signal_s {
+  void* arg;
+  cf_loop_signal_cb cb;
+  ngx_queue_t member;
+};
+
 
 int uv__platform_loop_init(uv_loop_t* loop, int default_loop) {
+  CFRunLoopSourceContext ctx;
+  int r;
+
+  loop->cf_loop = NULL;
+  if ((r = uv_mutex_init(&loop->cf_mutex)))
+    return r;
+  if ((r = uv_sem_init(&loop->cf_sem, 0)))
+    return r;
+  ngx_queue_init(&loop->cf_signals);
+
+  memset(&ctx, 0, sizeof(ctx));
+  ctx.info = loop;
+  ctx.perform = uv__cf_loop_cb;
+  loop->cf_cb = CFRunLoopSourceCreate(NULL, 0, &ctx);
+
+  if ((r = uv_thread_create(&loop->cf_thread, uv__cf_loop_runner, loop)))
+    return r;
+
+  /* Synchronize threads */
+  uv_sem_wait(&loop->cf_sem);
+  assert(((volatile CFRunLoopRef) loop->cf_loop) != NULL);
+
   return 0;
 }
 
 
 void uv__platform_loop_delete(uv_loop_t* loop) {
+  ngx_queue_t* item;
+  uv__cf_loop_signal_t* s;
+
+  assert(loop->cf_loop != NULL);
+  CFRunLoopStop(loop->cf_loop);
+  uv_thread_join(&loop->cf_thread);
+  loop->cf_loop = NULL;
+
+  uv_sem_destroy(&loop->cf_sem);
+  uv_mutex_destroy(&loop->cf_mutex);
+
+  /* Free any remaining data */
+  while (!ngx_queue_empty(&loop->cf_signals)) {
+    item = ngx_queue_head(&loop->cf_signals);
+
+    s = ngx_queue_data(item, uv__cf_loop_signal_t, member);
+
+    ngx_queue_remove(item);
+    free(s);
+  }
+}
+
+
+void uv__cf_loop_runner(void* arg) {
+  uv_loop_t* loop;
+
+  loop = arg;
+
+  /* Get thread's loop */
+  *((volatile CFRunLoopRef*)&loop->cf_loop) = CFRunLoopGetCurrent();
+
+  CFRunLoopAddSource(loop->cf_loop,
+                     loop->cf_cb,
+                     kCFRunLoopDefaultMode);
+
+  uv_sem_post(&loop->cf_sem);
+
+  CFRunLoopRun();
+
+  CFRunLoopRemoveSource(loop->cf_loop,
+                        loop->cf_cb,
+                        kCFRunLoopDefaultMode);
+}
+
+
+void uv__cf_loop_cb(void* arg) {
+  uv_loop_t* loop;
+  ngx_queue_t* item;
+  ngx_queue_t split_head;
+  uv__cf_loop_signal_t* s;
+
+  loop = arg;
+
+  uv_mutex_lock(&loop->cf_mutex);
+  ngx_queue_init(&split_head);
+  if (!ngx_queue_empty(&loop->cf_signals)) {
+    ngx_queue_t* split_pos = ngx_queue_next(&loop->cf_signals);
+    ngx_queue_split(&loop->cf_signals, split_pos, &split_head);
+  }
+  uv_mutex_unlock(&loop->cf_mutex);
+
+  while (!ngx_queue_empty(&split_head)) {
+    item = ngx_queue_head(&split_head);
+
+    s = ngx_queue_data(item, uv__cf_loop_signal_t, member);
+    s->cb(s->arg);
+
+    ngx_queue_remove(item);
+    free(s);
+  }
+}
+
+
+void uv__cf_loop_signal(uv_loop_t* loop, cf_loop_signal_cb cb, void* arg) {
+  uv__cf_loop_signal_t* item;
+
+  item = malloc(sizeof(*item));
+  /* XXX: Fail */
+  if (item == NULL)
+    abort();
+
+  item->arg = arg;
+  item->cb = cb;
+
+  uv_mutex_lock(&loop->cf_mutex);
+  ngx_queue_insert_tail(&loop->cf_signals, &item->member);
+  uv_mutex_unlock(&loop->cf_mutex);
+
+  assert(loop->cf_loop != NULL);
+  CFRunLoopSourceSignal(loop->cf_cb);
+  CFRunLoopWakeUp(loop->cf_loop);
 }
 
 

src/unix/fsevents.c

@@ -0,0 +1,225 @@
+/* Copyright Joyent, Inc. and other Node contributors. All rights reserved.
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+
+#include "uv.h"
+#include "internal.h"
+
+#include <assert.h>
+#include <stdlib.h>
+#include <CoreServices/CoreServices.h>
+
+typedef struct uv__fsevents_event_s uv__fsevents_event_t;
+
+struct uv__fsevents_event_s {
+  int events;
+  ngx_queue_t member;
+  char path[1];
+};
+
+
+#define UV__FSEVENTS_WALK(handle, block)                                      \
+    {                                                                         \
+      ngx_queue_t* curr;                                                      \
+      ngx_queue_t split_head;                                                 \
+      uv__fsevents_event_t* event;                                            \
+      uv_mutex_lock(&(handle)->cf_mutex);                                     \
+      ngx_queue_init(&split_head);                                            \
+      if (!ngx_queue_empty(&(handle)->cf_events)) {                           \
+        ngx_queue_t* split_pos = ngx_queue_next(&(handle)->cf_events);        \
+        ngx_queue_split(&(handle)->cf_events, split_pos, &split_head);        \
+      }                                                                       \
+      uv_mutex_unlock(&(handle)->cf_mutex);                                   \
+      while (!ngx_queue_empty(&split_head)) {                                 \
+        curr = ngx_queue_head(&split_head);                                   \
+        /* Invoke callback */                                                 \
+        event = ngx_queue_data(curr, uv__fsevents_event_t, member);           \
+        ngx_queue_remove(curr);                                               \
+        /* Invoke block code, but only if handle wasn't closed */             \
+        if (((handle)->flags & (UV_CLOSING | UV_CLOSED)) == 0)                \
+          block                                                               \
+        /* Free allocated data */                                             \
+        free(event);                                                          \
+      }                                                                       \
+    }
+
+
+void uv__fsevents_cb(uv_async_t* cb, int status) {
+  uv_fs_event_t* handle;
+
+  handle = cb->data;
+
+  UV__FSEVENTS_WALK(handle, {
+    if (handle->fd != -1)
+      handle->cb(handle, event->path, event->events, 0);
+  })
+
+  if ((handle->flags & (UV_CLOSING | UV_CLOSED)) == 0 && handle->fd == -1)
+    uv__fsevents_close(handle);
+}
+
+
+void uv__fsevents_event_cb(ConstFSEventStreamRef streamRef,
+                           void* info,
+                           size_t numEvents,
+                           void* eventPaths,
+                           const FSEventStreamEventFlags eventFlags[],
+                           const FSEventStreamEventId eventIds[]) {
+  size_t i;
+  int len;
+  char** paths;
+  uv_fs_event_t* handle;
+  uv__fsevents_event_t* event;
+  ngx_queue_t add_list;
+
+  handle = info;
+  paths = eventPaths;
+  ngx_queue_init(&add_list);
+
+  for (i = 0; i < numEvents; i++) {
+    /* Ignore system events */
+    if (eventFlags[i] & (kFSEventStreamEventFlagUserDropped |
+                         kFSEventStreamEventFlagKernelDropped |
+                         kFSEventStreamEventFlagEventIdsWrapped |
+                         kFSEventStreamEventFlagHistoryDone |
+                         kFSEventStreamEventFlagMount |
+                         kFSEventStreamEventFlagUnmount)) {
+      continue;
+    }
+
+    /* TODO: Report errors */
+    len = strlen(paths[i]);
+    event = malloc(sizeof(*event) + len);
+    if (event == NULL)
+      break;
+
+    memcpy(event->path, paths[i], len + 1);
+
+    if (eventFlags[i] & kFSEventStreamEventFlagItemModified)
+      event->events = UV_CHANGE;
+    else
+      event->events = UV_RENAME;
+
+    ngx_queue_insert_tail(&add_list, &event->member);
+  }
+  uv_mutex_lock(&handle->cf_mutex);
+  ngx_queue_add(&handle->cf_events, &add_list);
+  uv_mutex_unlock(&handle->cf_mutex);
+
+  uv_async_send(handle->cf_cb);
+}
+
+
+void uv__fsevents_schedule(void* arg) {
+  uv_fs_event_t* handle;
+
+  handle = arg;
+  FSEventStreamScheduleWithRunLoop(handle->cf_eventstream,
+                                   handle->loop->cf_loop,
+                                   kCFRunLoopDefaultMode);
+  FSEventStreamStart(handle->cf_eventstream);
+  uv_sem_post(&handle->cf_sem);
+}
+
+
+int uv__fsevents_init(uv_fs_event_t* handle) {
+  FSEventStreamContext ctx;
+  FSEventStreamRef ref;
+  CFStringRef path;
+  CFArrayRef paths;
+  CFAbsoluteTime latency;
+  FSEventStreamCreateFlags flags;
+
+  /* Initialize context */
+  ctx.version = 0;
+  ctx.info = handle;
+  ctx.retain = NULL;
+  ctx.release = NULL;
+  ctx.copyDescription = NULL;
+
+  /* Initialize paths array */
+  path = CFStringCreateWithCString(NULL,
+                                   handle->filename,
+                                   CFStringGetSystemEncoding());
+  paths = CFArrayCreate(NULL, (const void**)&path, 1, NULL);
+
+  latency = 0.15;
+
+  /* Set appropriate flags */
+  flags = kFSEventStreamCreateFlagFileEvents;
+
+  ref = FSEventStreamCreate(NULL,
+                            &uv__fsevents_event_cb,
+                            &ctx,
+                            paths,
+                            kFSEventStreamEventIdSinceNow,
+                            latency,
+                            flags);
+  handle->cf_eventstream = ref;
+
+  /*
+   * Events will occur in other thread.
+   * Initialize callback for getting them back into event loop's thread
+   */
+  handle->cf_cb = malloc(sizeof(*handle->cf_cb));
+  if (handle->cf_cb == NULL)
+    return uv__set_sys_error(handle->loop, ENOMEM);
+
+  handle->cf_cb->data = handle;
+  uv_async_init(handle->loop, handle->cf_cb, uv__fsevents_cb);
+  handle->cf_cb->flags |= UV__HANDLE_INTERNAL;
+  uv_unref((uv_handle_t*) handle->cf_cb);
+
+  uv_mutex_init(&handle->cf_mutex);
+  uv_sem_init(&handle->cf_sem, 0);
+  ngx_queue_init(&handle->cf_events);
+
+  uv__cf_loop_signal(handle->loop, uv__fsevents_schedule, handle);
+
+  return 0;
+}
+
+
+int uv__fsevents_close(uv_fs_event_t* handle) {
+  if (handle->cf_eventstream == NULL)
+    return -1;
+
+  /* Ensure that event stream was scheduled */
+  uv_sem_wait(&handle->cf_sem);
+
+  /* Stop emitting events */
+  FSEventStreamStop(handle->cf_eventstream);
+
+  /* Release stream */
+  FSEventStreamInvalidate(handle->cf_eventstream);
+  FSEventStreamRelease(handle->cf_eventstream);
+  handle->cf_eventstream = NULL;
+
+  uv_close((uv_handle_t*) handle->cf_cb, (uv_close_cb) free);
+
+  /* Free data in queue */
+  UV__FSEVENTS_WALK(handle, {
+    /* NOP */
+  })
+
+  uv_mutex_destroy(&handle->cf_mutex);
+  uv_sem_destroy(&handle->cf_sem);
+
+  return 0;
+}

src/unix/internal.h

@@ -192,4 +192,32 @@ void uv__udp_finish_close(uv_udp_t* handle);
 int uv__make_socketpair(int fds[2], int flags);
 int uv__make_pipe(int fds[2], int flags);
 
+#if defined(__APPLE__)
+typedef void (*cf_loop_signal_cb)(void*);
+
+void uv__cf_loop_signal(uv_loop_t* loop, cf_loop_signal_cb cb, void* arg);
+
+int uv__fsevents_init(uv_fs_event_t* handle);
+int uv__fsevents_close(uv_fs_event_t* handle);
+
+/* OSX < 10.7 has no file events, polyfill them */
+#if __ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__ < 1070
+
+static const int kFSEventStreamCreateFlagFileEvents = 0x00000010;
+static const int kFSEventStreamEventFlagItemCreated = 0x00000100;
+static const int kFSEventStreamEventFlagItemRemoved = 0x00000200;
+static const int kFSEventStreamEventFlagItemInodeMetaMod = 0x00000400;
+static const int kFSEventStreamEventFlagItemRenamed = 0x00000800;
+static const int kFSEventStreamEventFlagItemModified = 0x00001000;
+static const int kFSEventStreamEventFlagItemFinderInfoMod = 0x00002000;
+static const int kFSEventStreamEventFlagItemChangeOwner = 0x00004000;
+static const int kFSEventStreamEventFlagItemXattrMod = 0x00008000;
+static const int kFSEventStreamEventFlagItemIsFile = 0x00010000;
+static const int kFSEventStreamEventFlagItemIsDir = 0x00020000;
+static const int kFSEventStreamEventFlagItemIsSymlink = 0x00040000;
+
+#endif /* __ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__ < 1070 */
+
+#endif /* defined(__APPLE__) */
+
 #endif /* UV_UNIX_INTERNAL_H_ */

src/unix/kqueue.c

@@ -89,6 +89,9 @@ int uv_fs_event_init(uv_loop_t* loop,
                      uv_fs_event_cb cb,
                      int flags) {
   int fd;
+#if defined(__APPLE__)
+  struct stat statbuf;
+#endif /* defined(__APPLE__) */
 
   /* We don't support any flags yet. */
   assert(!flags);
@@ -105,6 +108,22 @@ int uv_fs_event_init(uv_loop_t* loop,
   handle->fflags = 0;
   handle->cb = cb;
   handle->fd = fd;
+
+#if defined(__APPLE__)
+  /* Nullify field to perform checks later */
+  handle->cf_eventstream = NULL;
+
+  if (fstat(fd, &statbuf))
+    goto fallback;
+  /* FSEvents works only with directories */
+  if (!(statbuf.st_mode & S_IFDIR))
+    goto fallback;
+
+  return uv__fsevents_init(handle);
+
+fallback:
+#endif /* defined(__APPLE__) */
+
   uv__fs_event_start(handle);
 
   return 0;
@@ -112,7 +131,13 @@ int uv_fs_event_init(uv_loop_t* loop,
 
 
 void uv__fs_event_close(uv_fs_event_t* handle) {
+#if defined(__APPLE__)
+  if (uv__fsevents_close(handle))
     uv__fs_event_stop(handle);
+#else
+  uv__fs_event_stop(handle);
+#endif /* defined(__APPLE__) */
+
   uv__handle_stop(handle);
   free(handle->filename);
   close(handle->fd);

test/test-fs-event.c

@@ -98,7 +98,8 @@ static void fs_event_cb_dir(uv_fs_event_t* handle, const char* filename,
   ASSERT(handle == &fs_event);
   ASSERT(status == 0);
   ASSERT(events == UV_RENAME);
-  ASSERT(filename == NULL || strcmp(filename, "file1") == 0);
+  ASSERT(filename == NULL || strcmp(filename, "file1") == 0 ||
+         strstr(filename, "watch_dir") != NULL);
   uv_close((uv_handle_t*)handle, close_cb);
 }
 

uv.gyp

@@ -147,7 +147,7 @@
           'libraries': [ '-lm' ]
         }],
         [ 'OS=="mac"', {
-          'sources': [ 'src/unix/darwin.c' ],
+          'sources': [ 'src/unix/darwin.c', 'src/unix/fsevents.c' ],
           'direct_dependent_settings': {
             'libraries': [
               '$(SDKROOT)/System/Library/Frameworks/CoreServices.framework',