Ticket #514: osx_dir_watch.patch

source/lib/sysdep/os/osx/dir_watch.cpp

@@ -19 +19 @@
  * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
  * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  */
+#include "precompiled.h"
+#include "ps/Filesystem.h"
+#include <CoreFoundation/CoreFoundation.h>
+#include <CoreServices/CoreServices.h>
+#include <sys/ioctl.h>
+#include <libgen.h>
 
-#include "precompiled.h"
 #include "lib/sysdep/dir_watch.h"
+#include "lib/file/vfs/vfs_util.h"
 
 // stub implementations
+static pthread_t g_event_loop_thread;
+static pthread_mutex_t g_mutex_notifications = PTHREAD_MUTEX_INITIALIZER;
+static pthread_mutex_t g_mutex_timestamps = PTHREAD_MUTEX_INITIALIZER;
+static std::vector<DirWatchNotification> g_notifications;
+static CFRunLoopRef loop = NULL;
+static int initialized = 0;
+static CFMutableArrayRef     g_paths;
+static volatile bool done = false;
+static int pipes[2];
 
-Status dir_watch_Add(const OsPath& UNUSED(path), PDirWatch& UNUSED(dirWatch))
+struct dentry
 {
+    struct timespec ts;
+    off_t size;
+    bool is_dir;
+};
+typedef std::map< OsPath, std::map<OsPath, struct dentry> > t_timestamp_map;
+typedef std::map<OsPath, struct dentry> t_timestamp_inner_map;
+typedef std::map< OsPath, std::map<OsPath, struct dentry> >::iterator t_timestamp_map_iterator;
+typedef std::map<OsPath, struct dentry>::iterator t_timestamp_inner_map_iterator;
+static t_timestamp_map g_timestamps;
+static wchar_t        path_buff[PATH_MAX];
+
+static Status record_timestamps(const VfsPath& pathname, const FileInfo& fileInfo, uintptr_t cbData);
+static Status remove_timestamps(const VfsPath& pathname, const FileInfo& UNUSED(fileInfo), const uintptr_t cbData);
+static void fsevents_callback(FSEventStreamRef streamRef, void *clientCallBackInfo,
+                  int numEvents,
+                  const char *const eventPaths[],
+                  const FSEventStreamEventFlags *eventFlags,
+                  const uint64_t *eventIDs);
+Status dir_watch_Delete(const OsPath&);
+
+struct DirWatch
+{
+    DirWatch()
+    {
+    }
+    ~DirWatch()
+    {
+        dir_watch_Delete(path);
+    }
+    OsPath path;
+};
+
+static CFFileDescriptorRef fdref = NULL;
+static CFRunLoopSourceRef fd_rl_src = NULL;
+
+static void pipe_callback(CFFileDescriptorRef UNUSED(kq_cffd), CFOptionFlags UNUSED(callBackTypes), void* UNUSED(info))
+{
+    char c;
+    while (read(pipes[0], &c, 1) > 0) ;
+    CFRunLoopStop(loop);
+    CFFileDescriptorEnableCallBacks(fdref, kCFFileDescriptorReadCallBack);
+}
+
+static Status setup_run_loop_pipe_read_handler()
+{
+    fdref = CFFileDescriptorCreate(NULL, pipes[0], 1, pipe_callback, NULL);
+    if (fdref == NULL)
+    {
+        return ERR::FAIL;
+    }
+
+    fd_rl_src = CFFileDescriptorCreateRunLoopSource(NULL, fdref, (CFIndex)0);
+    if (fd_rl_src == NULL)
+    {
+        CFFileDescriptorInvalidate(fdref);
+        CFRelease(fdref);
+        fdref = NULL;
+        return ERR::FAIL;
+    }
+
+    CFRunLoopAddSource(loop, fd_rl_src, kCFRunLoopDefaultMode);
+    CFFileDescriptorEnableCallBacks(fdref, kCFFileDescriptorReadCallBack);
+
+    return INFO::OK;
+}
+
+static void cleanup_run_loop_pipe_read_handler()
+{
+    CFRunLoopRemoveSource(loop, fd_rl_src, kCFRunLoopDefaultMode);
+    CFFileDescriptorInvalidate(fdref);
+    CFRelease(fd_rl_src);
+    CFRelease(fdref);
+
+    fd_rl_src = NULL;
+    fdref   = NULL;
+}
+
+static void fam_deinit()
+{
+    done = true;
+    cleanup_run_loop_pipe_read_handler();
+    CFRunLoopStop(loop);
+
+    // Wait for the thread to finish
+    pthread_join(g_event_loop_thread, NULL);
+}
+
+static void* event_loop(void*)
+{
+    loop = CFRunLoopGetCurrent();
+    FSEventStreamRef      stream_ref = NULL;
+    FSEventStreamEventId sinceWhen = kFSEventStreamEventIdSinceNow;
+    setup_run_loop_pipe_read_handler();
+
+    do {
+        stream_ref = FSEventStreamCreate(kCFAllocatorDefault,
+                                    (FSEventStreamCallback)&fsevents_callback,
+                                    NULL,
+                                    g_paths,
+                                    sinceWhen,
+                                    0.3,
+                                    kFSEventStreamCreateFlagNone);
+        FSEventStreamScheduleWithRunLoop(stream_ref, loop, kCFRunLoopDefaultMode);
+        if (!FSEventStreamStart(stream_ref))
+        {
+            debug_warn(L"event_loop FSEventStreamStart failed!");
+            return NULL;
+        }
+        CFRunLoopRun();
+        FSEventStreamFlushSync(stream_ref);
+        FSEventStreamStop(stream_ref);
+        FSEventStreamUnscheduleFromRunLoop(stream_ref, loop, kCFRunLoopDefaultMode);
+        FSEventStreamInvalidate(stream_ref);
+        FSEventStreamRelease(stream_ref);
+        sinceWhen = FSEventsGetCurrentEventId();
+    }
+    while (!done);
+    return NULL;
+}
+
+static inline void add_to_g_paths(const OsPath& path)
+{
+    pthread_mutex_lock(&g_mutex_timestamps);
+    vfs::ForEachFile(g_VFS, path, record_timestamps, (uintptr_t)&g_timestamps, L"*", vfs::DIR_RECURSIVE);
+    pthread_mutex_unlock(&g_mutex_timestamps);
+
+    CFStringEncoding encoding = (CFByteOrderLittleEndian == CFByteOrderGetCurrent()) ? 
+                                      kCFStringEncodingUTF32LE : kCFStringEncodingUTF32BE;
+    CFStringRef macpath = CFStringCreateWithBytes(
+        NULL,
+        (const UInt8*)path.string().c_str(),
+        wcslen(path.string().c_str()) * sizeof(wchar_t),
+        encoding,
+        false
+    );
+
+    if (macpath == NULL) 
+    {
+        debug_warn(L"add_to_g_paths CFStringCreateWithCString failed!");
+        return;
+    }
+    CFArrayAppendValue(g_paths, macpath);
+    CFArraySortValues(g_paths, CFRangeMake(0, CFArrayGetCount(g_paths)), (CFComparatorFunction)CFStringCompare, NULL);
+    CFRelease(macpath);
+}
+
+static Status compare_timestamps(t_timestamp_map& old_map, t_timestamp_map& new_map)
+{
+    #define time_equal(a, b) (a.tv_sec == b.tv_sec && a.tv_nsec == b.tv_nsec)
+
+    t_timestamp_map_iterator it_old;
+    t_timestamp_map_iterator it_new;
+    t_timestamp_inner_map_iterator it_old_inner;
+    t_timestamp_inner_map_iterator it_new_inner;
+    t_timestamp_inner_map inner_map;
+    for (it_old = old_map.begin(); it_old != old_map.end(); it_old++)
+    {
+        inner_map = it_old->second;
+        it_new = new_map.find(it_old->first);
+        for (it_old_inner = inner_map.begin(); it_old_inner != inner_map.end(); it_old_inner++)
+        {
+            if ((it_new == new_map.end()) || ((it_new_inner = it_new->second.find(it_old_inner->first)) == it_new->second.end()))
+            {
+                // deleted
+                DirWatchNotification dn(it_old_inner->first, DirWatchNotification::Deleted);
+                pthread_mutex_lock(&g_mutex_notifications);
+                g_notifications.push_back(dn);
+                pthread_mutex_unlock(&g_mutex_notifications);
+            }
+            else if (!time_equal(it_new_inner->second.ts, it_old_inner->second.ts) || (it_new_inner->second.size != it_old_inner->second.size))
+            {
+                // updated
+                DirWatchNotification dn(it_old_inner->first, DirWatchNotification::Changed);
+                pthread_mutex_lock(&g_mutex_notifications);
+                g_notifications.push_back(dn);
+                pthread_mutex_unlock(&g_mutex_notifications);
+            }
+        }
+    }
+    for (it_new = new_map.begin(); it_new != new_map.end(); it_new++)
+    {
+        inner_map = it_new->second;
+        it_old = old_map.find(it_new->first);
+        for (it_new_inner = inner_map.begin(); it_new_inner != inner_map.end(); it_new_inner++)
+        {
+            if ((it_old == old_map.end()) || ((it_old_inner = it_old->second.find(it_new_inner->first)) == it_old->second.end()))
+            {
+                // new
+                DirWatchNotification dn(it_new_inner->first, DirWatchNotification::Created);
+                pthread_mutex_lock(&g_mutex_notifications);
+                g_notifications.push_back(dn);
+                pthread_mutex_unlock(&g_mutex_notifications);
+            }
+        }
+    }
+
+    #undef time_equal
+    return INFO::OK;
+}
+
+static void fsevents_callback(FSEventStreamRef UNUSED(streamRef), void* UNUSED(clientCallBackInfo),
+                  int numEvents,
+                  const char *const eventPaths[],
+                  const FSEventStreamEventFlags *eventFlags,
+                  const uint64_t* UNUSED(eventIDs))
+{
+    for (int i=0; i < numEvents; i++)
+    {
+        size_t flags = 0;
+        if (eventFlags[i] & kFSEventStreamEventFlagHistoryDone) 
+        {
+            continue;
+        }
+        else if (eventFlags[i] & kFSEventStreamEventFlagRootChanged) 
+        {
+            // do not handle this flag temporarily
+            continue;
+        }
+        else if (eventFlags[i] & kFSEventStreamEventFlagMustScanSubDirs)
+        {
+            flags |= vfs::DIR_RECURSIVE;
+        }
+        t_timestamp_map new_fs;
+        vfs::ForEachFile(g_VFS, eventPaths[i], record_timestamps, (uintptr_t)&new_fs, L"*", flags);
+        t_timestamp_map_iterator it_new;
+        t_timestamp_map_iterator it_g;
+        t_timestamp_map old_fs;
+        pthread_mutex_lock(&g_mutex_timestamps);
+        for (it_new = new_fs.begin(); it_new != new_fs.end(); it_new++)
+        {
+            it_g = g_timestamps.find(it_new->first);
+            if (it_g != g_timestamps.end())
+            {
+                old_fs.insert(std::pair<OsPath, t_timestamp_inner_map>(it_g->first, it_g->second));
+            }
+        }
+        for (it_new = new_fs.begin(); it_new != new_fs.end(); it_new++)
+        {
+            g_timestamps[it_new->first] = it_new->second;
+        }
+        pthread_mutex_unlock(&g_mutex_timestamps);
+        compare_timestamps(old_fs, new_fs);
+    }
+}
+
+static Status record_timestamps(const VfsPath& pathname, const FileInfo& UNUSED(fileInfo), const uintptr_t cbData)
+{
+    struct stat st;
+    if (wstat(pathname.string().c_str(), &st) == 0)
+    {
+        t_timestamp_map* pmap = (t_timestamp_map*)cbData;
+        wcscpy(path_buff, pathname.string().c_str());
+        char* dir = dirname(path_buff);
+        t_timestamp_map_iterator it = pmap->find(dir);
+        struct dentry d = {ts:st.st_mtimespec, size:st.st_size, is_dir: S_ISDIR(st.st_mode)};
+        if (it != pmap->end())
+        {
+            it->second.insert(std::pair<OsPath, struct dentry>(pathname, d));
+        }
+        else
+        {
+            t_timestamp_inner_map inner_map;
+            inner_map.insert(std::pair<OsPath, struct dentry>(pathname, d));
+            pmap->insert(std::pair<OsPath, t_timestamp_inner_map>(dir, inner_map));
+        }
+    }
+    return INFO::OK;
+}
+
+static Status remove_timestamps(const VfsPath& pathname, const FileInfo& UNUSED(fileInfo), const uintptr_t cbData)
+{
+    struct stat st;
+    if (wstat(pathname.string().c_str(), &st) == 0)
+    {
+        t_timestamp_map* pmap = (t_timestamp_map*)cbData;
+        wcscpy(path_buff, pathname.string().c_str());
+        char* dir = dirname(path_buff);
+        t_timestamp_map_iterator it = pmap->find(dir);
+        if (it != pmap->end())
+        {
+            t_timestamp_inner_map_iterator it_inner = it->second.find((const char*)(pathname.string().c_str()));
+            if (it_inner != it->second.end())
+            {
+                it->second.erase(it_inner);
+            }
+        }
+    }
+    return INFO::OK;
+}
+
+Status dir_watch_Delete(const OsPath& path)
+{
+    CFStringEncoding encoding = (CFByteOrderLittleEndian == CFByteOrderGetCurrent()) ? 
+                                      kCFStringEncodingUTF32LE : kCFStringEncodingUTF32BE;
+    CFStringRef macpath = CFStringCreateWithBytes(
+        NULL,
+        (const UInt8*)path.string().c_str(),
+        wcslen(path.string().c_str()) * sizeof(wchar_t),
+        encoding,
+        false
+    );
+    if (macpath == NULL)
+    {
+        debug_warn(L"dir_watch_Delete CFStringCreateWithCString failed!");
+        return ERR::FAIL;
+    }
+    CFIndex idx = CFArrayBSearchValues(g_paths, CFRangeMake(0, CFArrayGetCount(g_paths)), macpath, (CFComparatorFunction)CFStringCompare, NULL);
+    CFRelease(macpath);
+    CFArrayRemoveValueAtIndex(g_paths, idx);
+    pthread_mutex_lock(&g_mutex_timestamps);
+    vfs::ForEachFile(g_VFS, path, remove_timestamps, (uintptr_t)&g_timestamps, L"*", vfs::DIR_RECURSIVE);
+    pthread_mutex_unlock(&g_mutex_timestamps);
+    write(pipes[1], "1", 1);
+    return INFO::OK;
+}
+
+Status dir_watch_Add(const OsPath& path, PDirWatch& dirWatch)
+{
+    if (!initialized)
+    {
+        if (0 != pipe(pipes))
+        {
+            debug_warn(L"dir_watch_Add pipe failed!");
+            return ERR::FAIL;
+        }
+        unsigned long nb = 1;
+        if (-1 == ioctl(pipes[0], FIONBIO, &nb))
+        {
+            debug_warn(L"dir_watch_Add ioctl failed!");
+            return ERR::FAIL;
+        }
+        g_paths = CFArrayCreateMutable(kCFAllocatorDefault, 0, &kCFTypeArrayCallBacks);
+        if (g_paths == NULL)
+        {
+            debug_warn(L"dir_watch_Add CFArrayCreateMutable failed!");
+            return ERR::FAIL;
+        }
+        add_to_g_paths(path);
+
+        if (pthread_create(&g_event_loop_thread, NULL, &event_loop, NULL))
+        {
+            debug_warn(L"dir_watch_Add pthread_create failed!");
+            return ERR::FAIL;
+        }
+        atexit(fam_deinit);
+        initialized = 1;
+    }
+    else
+    {
+        add_to_g_paths(path);
+        write(pipes[1], "1", 1);
+    }
+
+    PDirWatch tmpDirWatch(new DirWatch);
+    dirWatch.swap(tmpDirWatch);
+    dirWatch->path = path;
+
     return INFO::OK;
 }
 
-Status dir_watch_Poll(DirWatchNotifications& UNUSED(notifications))
+Status dir_watch_Poll(DirWatchNotifications& notifications)
 {
+    if (1 != initialized)
+        return ERR::FAIL;
+    std::vector<DirWatchNotification> polled_notifications;
+
+    pthread_mutex_lock(&g_mutex_notifications);
+    g_notifications.swap(polled_notifications);
+    pthread_mutex_unlock(&g_mutex_notifications);
+
+    for (size_t i = 0; i < polled_notifications.size(); ++i)
+    {
+        notifications.push_back(polled_notifications[i]);
+    }
     return INFO::OK;
 }
+