Added wakeup sources to cf-reactor

cf-reactor/Makefile.am

@@ -47,7 +47,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>
 
 /* Minimum time between two logs of a persisting failure to check the file,
@@ -40,6 +42,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;
@@ -60,6 +72,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() */
@@ -81,26 +94,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;
@@ -129,8 +173,10 @@ static void LogCheckError(FileWatcherState *fws, const char *op, int err)
     if (fws->last_error_logged == 0 ||
         now - fws->last_error_logged >= CHECK_ERROR_LOG_INTERVAL_SECS)
     {
+        /* strerror(), as err is an errno value: GetErrorStrFromCode()
+         * takes a Windows error code on Windows */
         Log(LOG_LEVEL_ERR, "Unable to %s '%s' while checking for file events: %s",
-            op, fws->path, GetErrorStrFromCode(err));
+            op, fws->path, strerror(err));
         fws->last_error_logged = now;
     }
 }
@@ -155,6 +201,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", EBUSY);
+        return false;
+    }
+#endif
+
     if (errno == ENOENT || errno == ENOTDIR)
     {
         *exists = false;
@@ -280,15 +339,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;
     }
 
@@ -335,7 +482,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))
         {
@@ -354,6 +509,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/reactor_transform.c

@@ -150,6 +150,7 @@ typedef struct
     WatcherStateNewFn new_state;
     WatcherStateDestroyFn destroy_state;
     WatcherCheckFn check;
+    ReactorWatchTargetType target; /* what the 'name' constraint is about */
 } EventType;
 
 static const EventArg NO_EXTRA_ARGS[] =
@@ -163,17 +164,20 @@ static const EventArg FILE_SIZE_EXCEEDS_ARGS[] =
     { NULL, RVAL_TYPE_NOPROMISEE, false }
 };
 
-#define EventTypeNew(name, type, args, new_state, destroy_state, check) \
-    { name, type, args, new_state, destroy_state, check }
-#define EventTypeNewNull() EventTypeNew(NULL, RVAL_TYPE_NOPROMISEE, NULL, NULL, NULL, NULL)
+#define EventTypeNew(name, type, args, new_state, destroy_state, check, target) \
+    { name, type, args, new_state, destroy_state, check, target }
+#define EventTypeNewNull() \
+    EventTypeNew(NULL, RVAL_TYPE_NOPROMISEE, NULL, NULL, NULL, NULL, REACTOR_WATCH_TARGET_FILE)
 
 /* A file event, about the file at the path of its constraint */
 #define FileEventTypeNew(name, check) \
-    EventTypeNew(name, RVAL_TYPE_SCALAR, NO_EXTRA_ARGS, FileWatcherStateNew, FileWatcherStateDestroy, check)
+    EventTypeNew(name, RVAL_TYPE_SCALAR, NO_EXTRA_ARGS, FileWatcherStateNew, FileWatcherStateDestroy, \
+                 check, REACTOR_WATCH_TARGET_FILE)
 
 /* A user event, about the user with the name of its constraint */
 #define UserEventTypeNew(name, check) \
-    EventTypeNew(name, RVAL_TYPE_SCALAR, NO_EXTRA_ARGS, UserWatcherStateNew, UserWatcherStateDestroy, check)
+    EventTypeNew(name, RVAL_TYPE_SCALAR, NO_EXTRA_ARGS, UserWatcherStateNew, UserWatcherStateDestroy, \
+                 check, REACTOR_WATCH_TARGET_USERS)
 
 static const EventType EVENT_TABLE[] =
 {
@@ -184,7 +188,8 @@ static const EventType EVENT_TABLE[] =
     FileEventTypeNew(REACTOR_FILE_PERMISSIONS_CHANGED, FileWatcherCheckFilePermissionsChanged),
     FileEventTypeNew(REACTOR_FILE_OWNER_CHANGED, FileWatcherCheckFileOwnerChanged),
     EventTypeNew(REACTOR_FILE_SIZE_EXCEEDS, RVAL_TYPE_SCALAR, FILE_SIZE_EXCEEDS_ARGS,
-                 FileWatcherSizeStateNew, FileWatcherStateDestroy, FileWatcherCheckFileSizeExceeds),
+                 FileWatcherSizeStateNew, FileWatcherStateDestroy, FileWatcherCheckFileSizeExceeds,
+                 REACTOR_WATCH_TARGET_FILE),
     UserEventTypeNew(REACTOR_USER_ADDED, UserWatcherCheckUserAdded),
     UserEventTypeNew(REACTOR_USER_REMOVED, UserWatcherCheckUserRemoved),
     EventTypeNewNull()
@@ -273,6 +278,31 @@ static bool EventsPromiseGetInterval(const Promise *pp, time_t *interval)
     return true;
 }
 
+/** Sets *methods to the 'wakeup_sources' of the (expanded) events promise, NULL
+ * if it has none. The method names were checked against the syntax already,
+ * on expansion. */
+static bool EventsPromiseGetWakeupSources(const Promise *pp, const Rlist **methods)
+{
+    assert(pp != NULL);
+    assert(methods != NULL);
+
+    *methods = NULL;
+    const Constraint *con = PromiseGetConstraint(pp, REACTOR_WAKEUP_SOURCES);
+    if (con == NULL)
+    {
+        return true;
+    }
+    if (con->rval.type != RVAL_TYPE_LIST)
+    {
+        Log(LOG_LEVEL_ERR,
+            "Reactor events promise '%s' has an invalid '%s', expected a list of wakeup sources",
+            pp->promiser, REACTOR_WAKEUP_SOURCES);
+        return false;
+    }
+    *methods = RvalRlistValue(con->rval);
+    return true;
+}
+
 static bool ResolveWhenBody(const EvalContext *ctx, const Promise *pp, WatcherOptions *opt)
 {
     assert(ctx != NULL);
@@ -331,6 +361,7 @@ static bool ResolveWhenBody(const EvalContext *ctx, const Promise *pp, WatcherOp
         if (BodySyntaxGetConstraintSyntax(when_syntax, cp->lval) == NULL || /* not a 'when' constraint */
             StringEqual(cp->lval, event->name) ||
             StringEqual(cp->lval, "interval") ||
+            StringEqual(cp->lval, REACTOR_WAKEUP_SOURCES) ||
             StringEqual(cp->lval, "meta"))                                 /* also a promise attribute */
         {
             continue;
@@ -351,12 +382,19 @@ static bool ResolveWhenBody(const EvalContext *ctx, const Promise *pp, WatcherOp
         }
     }
 
-    if (!EventsPromiseGetInterval(pp, &opt->interval))
+    if (!EventsPromiseGetInterval(pp, &opt->interval) ||
+        !EventsPromiseGetWakeupSources(pp, &opt->methods))
     {
         RlistDestroy(args);
         return false;
     }
 
+    /* The path is the value of the event's constraint, which lives as long
+     * as pp, unlike args */
+    opt->target.type = event->target;
+    opt->target.path = (event->target == REACTOR_WATCH_TARGET_FILE) ?
+        RvalScalarValue(PromiseGetConstraint(pp, event->name)->rval) : NULL;
+
     opt->state = event->new_state(ctx, args);
     RlistDestroy(args);
     if (opt->state == NULL)