Added wakeup sources to cf-reactor

cf-reactor/Makefile.am

@@ -48,7 +48,10 @@ libcf_reactor_la_SOURCES = \
 	stoppable_thread.c stoppable_thread.h \
 	user_watcher.c user_watcher.h \
 	wakeup_channel.c wakeup_channel.h \
-	watcher.c watcher.h
+	watch_backends.c watch_backends.h \
+	watch_location.c watch_location.h \
+	watcher.c watcher.h \
+	watcher_report.c watcher_report.h
 
 if !BUILTIN_EXTENSIONS
  bin_PROGRAMS = cf-reactor

cf-reactor/file_watcher.c

@@ -26,12 +26,14 @@
 #include <watcher.h>
 #include <logging.h>
 #include <alloc.h>
-#include <file_lib.h>           /* safe_fopen() */
+#include <file_lib.h>           /* safe_open() */
 #include <hash.h>               /* HashDigestFromId() */
 #include <conversion.h>         /* IntFromString() */
 #include <prototypes3.h>        /* cf_strtimestamp_local() */
 #include <sys/stat.h>
+#include <fcntl.h>
 #include <errno.h>
+#include <string.h>             /* strerror() */
 #include <openssl/evp.h>
 
 #ifdef __MINGW32__
@@ -44,6 +46,16 @@
 
 #define CONTENT_HASH_METHOD HASH_METHOD_SHA256
 
+/* What changes when a file's content may have changed, see StatSignatureOf() */
+typedef struct
+{
+    dev_t dev;
+    ino_t ino;
+    off_t size;
+    struct timespec mtime;
+    struct timespec ctime;
+} StatSignature;
+
 typedef struct
 {
     char *path;
@@ -64,6 +76,7 @@ typedef struct
         {
             bool hashed;      /* Whether it was a regular file, that was hashed */
             unsigned char digest[EVP_MAX_MD_SIZE + 1];
+            StatSignature signature; /* of the file when it was hashed */
         } content;
 
         /* FileWatcherCheckFilePermissionsChanged() */
@@ -92,26 +105,57 @@ typedef struct
     } last;
 } FileWatcherState;
 
+/**
+ * Copy of path in a form stat() accepts. On Windows, msvcrt's stat() fails
+ * with ENOENT for a path with a trailing separator, or with the \\?\ (or
+ * //?/) prefix, so a watcher with such a path would never see its file.
+ */
+static char *WatchPathCopy(const char *path)
+{
+    assert(path != NULL);
+
+    char *copy = xstrdup(path);
+#ifdef __MINGW32__
+    if ((copy[0] == '\\' || copy[0] == '/') && (copy[1] == '\\' || copy[1] == '/') &&
+        copy[2] == '?' && (copy[3] == '\\' || copy[3] == '/'))
+    {
+        memmove(copy, copy + 4, strlen(copy + 4) + 1);
+    }
+
+    /* Keeps the separator of a drive's root, e.g. C:/ */
+    size_t length = strlen(copy);
+    while (length > 1 && (copy[length - 1] == '\\' || copy[length - 1] == '/') &&
+           !(length == 3 && copy[1] == ':'))
+    {
+        copy[--length] = '\0';
+    }
+#endif
+    return copy;
+}
 
 void *FileWatcherStateNew(ARG_UNUSED const EvalContext *ctx, const Rlist *args)
 {
     assert(args != NULL);
     const char *path = RlistScalarValue(args);
 
     FileWatcherState *fws = (FileWatcherState *) xcalloc(1, sizeof(FileWatcherState));
-    fws->path = xstrdup(path);
+    fws->path = WatchPathCopy(path);
 
     return fws;
 }
 
 void *FileWatcherSizeStateNew(ARG_UNUSED const EvalContext *ctx, const Rlist *args)
 {
     assert(args != NULL && args->next != NULL);
+    if (args == NULL || args->next == NULL)
+    {
+        return NULL;
+    }
     const char *path = RlistScalarValue(args);
     const long size_threshold = IntFromString(RlistScalarValue(args->next));
 
     FileWatcherState *fws = (FileWatcherState *) xcalloc(1, sizeof(FileWatcherState));
-    fws->path = xstrdup(path);
+    fws->path = WatchPathCopy(path);
     fws->last.file_size.threshold = size_threshold;
 
     return fws;
@@ -170,6 +214,19 @@ static bool FileLstat(FileWatcherState *fws, bool *exists, struct stat *sb)
         return true;
     }
 
+#ifdef __MINGW32__
+    /* msvcrt's stat() also fails with ENOENT while the file's directory is
+     * open without sharing (FindFirstFile() gets a sharing violation), when
+     * the file is still there: then whether it changed is unknown */
+    WIN32_FILE_ATTRIBUTE_DATA attributes;
+    if (errno == ENOENT &&
+        GetFileAttributesExA(fws->path, GetFileExInfoStandard, &attributes))
+    {
+        LogCheckError(fws, "lstat", strerror(EBUSY));
+        return false;
+    }
+#endif
+
     if (errno == ENOENT || errno == ENOTDIR)
     {
         *exists = false;
@@ -295,15 +352,103 @@ WatcherCheckResult FileWatcherCheckFileModified(void *state)
 
 // =========== code for the file_content_modified event ===========
 
+static struct timespec StatCtime(const struct stat *sb)
+{
+    assert(sb != NULL);
+
+#if defined(HAVE_STRUCT_STAT_ST_MTIM)
+    return sb->st_ctim;
+#elif defined(HAVE_STRUCT_STAT_ST_MTIMESPEC)
+    return sb->st_ctimespec;
+#else
+    return (struct timespec) { .tv_sec = sb->st_ctime, .tv_nsec = 0 };
+#endif
+}
+
+/** Content can't change without one of these changing too: writing updates
+ * mtime and ctime, replacing the file changes the inode, and setting mtime
+ * back still updates ctime. So the file is only hashed again when they
+ * differ, which keeps frequent checks of many files cheap. */
+static StatSignature StatSignatureOf(const struct stat *sb)
+{
+    assert(sb != NULL);
+
+    StatSignature signature;
+    memset(&signature, 0, sizeof(signature)); /* no padding garbage for memcmp() */
+    signature.dev = sb->st_dev;
+    signature.ino = sb->st_ino;
+    signature.size = sb->st_size;
+    signature.mtime = StatMtime(sb);
+    signature.ctime = StatCtime(sb);
+    return signature;
+}
+
+/** Opens the regular file at path for reading. Fails with EINVAL if it isn't
+ * one after all: the caller lstat()ed it, but it may have been swapped for a
+ * FIFO since, and opening that for reading would block forever. */
+static int OpenRegularFile(const char *path)
+{
+    assert(path != NULL);
+
+    int flags = O_RDONLY;
+#ifdef O_NONBLOCK
+    flags |= O_NONBLOCK;
+#endif
+#ifdef O_NOFOLLOW
+    flags |= O_NOFOLLOW;
+#endif
+#ifdef O_BINARY
+    flags |= O_BINARY;
+#endif
+    const int fd = safe_open(path, flags);
+    if (fd < 0)
+    {
+        return -1;
+    }
+
+    struct stat sb;
+    if (fstat(fd, &sb) != 0)
+    {
+        const int err = errno;
+        close(fd);
+        errno = err;
+        return -1;
+    }
+    if (!S_ISREG(sb.st_mode))
+    {
+        close(fd);
+        errno = EINVAL;
+        return -1;
+    }
+
+#ifndef __MINGW32__
+    /* Regular files never block, so this is only to leave the fd as expected */
+    const int fd_flags = fcntl(fd, F_GETFL);
+    if (fd_flags != -1)
+    {
+        fcntl(fd, F_SETFL, fd_flags & ~O_NONBLOCK);
+    }
+#endif
+    return fd;
+}
+
 /* Hashes the content of the file into digest. Returns false and sets errno
  * on failure. */
 static bool HashFileContent(const char *path, unsigned char digest[EVP_MAX_MD_SIZE + 1])
 {
     assert(path != NULL);
 
-    FILE *file = safe_fopen(path, "rb");
+    const int fd = OpenRegularFile(path);
+    if (fd < 0)
+    {
+        return false;
+    }
+    FILE *file = fdopen(fd, "rb");
     if (file == NULL)
     {
+        const int err = errno;
+        close(fd);
+        errno = err;
         return false;
     }
 
@@ -350,7 +495,15 @@ WatcherCheckResult FileWatcherCheckFileContentModified(void *state)
 
     bool hashed = false;
     unsigned char digest[EVP_MAX_MD_SIZE + 1] = { 0 };
-    if (exists && S_ISREG(sb.st_mode))
+    const StatSignature signature = exists ? StatSignatureOf(&sb) : (StatSignature) { 0 };
+    if (exists && S_ISREG(sb.st_mode) && fws->last.content.hashed &&
+        memcmp(&fws->last.content.signature, &signature, sizeof(signature)) == 0)
+    {
+        /* Unchanged since hashed, see StatSignatureOf() */
+        hashed = true;
+        memcpy(digest, fws->last.content.digest, sizeof(digest));
+    }
+    else if (exists && S_ISREG(sb.st_mode))
     {
         if (HashFileContent(fws->path, digest))
         {
@@ -369,6 +522,7 @@ WatcherCheckResult FileWatcherCheckFileContentModified(void *state)
 
     fws->last.content.hashed = hashed;
     memcpy(fws->last.content.digest, digest, sizeof(digest));
+    fws->last.content.signature = signature;
     return FileWatcherRecordObservation(fws, exists, modified, "content modification");
 }
 

cf-reactor/package_watcher.c

@@ -300,6 +300,9 @@ static void ListingThreadMain(StoppableThread *thread, void *arg)
     while (!StoppableThreadShouldStop(thread))
     {
         const time_t start = time(NULL);
+        /* Scheduled on the monotonic clock, so that a step back of the wall
+         * clock doesn't stop the listings until it catches up again */
+        const time_t start_monotonic = MonotonicClockNow();
         if (wrapper == NULL)
         {
             wrapper =
@@ -330,10 +333,10 @@ static void ListingThreadMain(StoppableThread *thread, void *arg)
 
         /* Counted from the start, so that a slow listing doesn't delay the
          * next one. Sleeps in a loop, since a sleep can end early. */
-        const time_t next = start + wait_secs;
+        const time_t next = start_monotonic + wait_secs;
         time_t now;
         while (!StoppableThreadShouldStop(thread)
-               && ((now = time(NULL)) < next))
+               && ((now = MonotonicClockNow()) < next))
         {
             StoppableThreadSleep(thread, next - now);
         }