resource fixes

Author: shikokuchuo

R/watch.R

@@ -15,7 +15,8 @@
 #'   `NULL`, causes event paths and types to be written to `stdout` instead.
 #'
 #' @return A 'Watcher' R6 class object. Start and stop background monitoring
-#'   using the `$start()` and `$stop()` methods.
+#'   using the `$start()` and `$stop()` methods - these return a logical value
+#'   whether or not they have succeeded.
 #'
 #' @examples
 #' w <- watcher(tempdir())
@@ -47,13 +48,19 @@ Watcher <- R6Class(
       invisible(self)
     },
     start = function() {
-      res <- .Call(watcher_start_monitor, private$watch)
-      if (res) self$active <- TRUE
-      invisible(res)
+      res <- self$active
+      if (!res) {
+        self$active <- .Call(watcher_start_monitor, private$watch)
+        res <- !self$active
+      }
+      invisible(!res)
     },
     stop = function() {
-      res <- invisible(.Call(watcher_stop_monitor, private$watch))
-      if (res) self$active <- FALSE
+      res <- self$active
+      if (res) {
+        res <- .Call(watcher_stop_monitor, private$watch)
+        self$active <- !res
+      }
       invisible(res)
     }
   ),

configure.win

@@ -1,5 +1,3 @@
-#!/bin/sh
-
 LIB_VER="1.18.2"
 
 # Find compiler and export flags

man/watcher-package.Rd

@@ -1,6 +1,6 @@
 #include "watcher.h"
 
-static void Wprintf(const int err, const char *fmt, ...) {
+static void Wprintf(const char *fmt, ...) {
 
   char buf[WATCHER_BUFSIZE];
   va_list arg_ptr;
@@ -9,11 +9,11 @@ static void Wprintf(const int err, const char *fmt, ...) {
   int bytes = vsnprintf(buf, WATCHER_BUFSIZE, fmt, arg_ptr);
   va_end(arg_ptr);
 
-  if (write(err ? STDERR_FILENO : STDOUT_FILENO, buf, (size_t) bytes)) {};
+  if (write(STDOUT_FILENO, buf, (size_t) bytes)) {};
 
 }
 
-void session_finalizer(SEXP xptr) {
+static void session_finalizer(SEXP xptr) {
   if (R_ExternalPtrAddr(xptr) == NULL) return;
   FSW_HANDLE handle = (FSW_HANDLE) R_ExternalPtrAddr(xptr);
   fsw_stop_monitor(handle);
@@ -22,38 +22,53 @@ void session_finalizer(SEXP xptr) {
 
 void (*eln2)(void (*)(void *), void *, double, int) = NULL;
 
-void load_later_safe(void *data) {
-  (void) data;
-  SEXP fn, call;
-  fn = Rf_install("loadNamespace");
+static void load_later_safe(void *data) {
+  SEXP call, fn = (SEXP) data;
   PROTECT(call = Rf_lang2(fn, Rf_mkString("later")));
   Rf_eval(call, R_GlobalEnv);
   UNPROTECT(1);
 }
 
-void exec_later(void *data) {
+static void exec_later(void *data) {
   SEXP call, fn = (SEXP) data;
   PROTECT(call = Rf_lcons(fn, R_NilValue));
   Rf_eval(call, R_GlobalEnv);
   UNPROTECT(1);
 }
 
-void process_events(fsw_cevent const *const events, const unsigned int event_num, void *data) {
+// custom non-allocating version of fsw_get_event_flag_name() handling main types
+static void get_event_flag_name(const int flag, char *buf) {
+  const char *name;
+  switch(flag) {
+  case 1 << 1: name = "Created"; break;
+  case 1 << 2: name = "Updated"; break;
+  case 1 << 3: name = "Removed"; break;
+  case 1 << 4: name = "Renamed"; break;
+  default: name = "Unknown"; break;
+  }
+  memcpy(buf, name, strlen(name));
+}
+
+static void process_events(fsw_cevent const *const events, const unsigned int event_num, void *data) {
   if (data != R_NilValue && eln2 != NULL) {
     eln2(exec_later, data, 0, 0);
   } else {
-    for (unsigned int i = 0; i < event_num; i++)
-      for (unsigned int j = 0; j < events[i].flags_num; j++)
-        Wprintf(0, "Event: %s\n Flag: %s\n", events[i].path, fsw_get_event_flag_name(events[i].flags[j]));
+    char buf[8]; // large enough for subset of events handled by get_event_flag_name()
+    memset(buf, 0, sizeof(buf));
+    for (unsigned int i = 0; i < event_num; i++) {
+      for (unsigned int j = 0; j < events[i].flags_num; j++) {
+        get_event_flag_name(events[i].flags[j], buf);
+        Wprintf("Event: %s\n Flag: %s\n", events[i].path, buf);
+      }
+    }
   }
 }
 
-void* watcher_thread(void *args) {
+static void * watcher_thread(void *args) {
 
-  pthread_detach(pthread_self());
   FSW_HANDLE handle = (FSW_HANDLE) args;
   fsw_start_monitor(handle);
-  return (void *) NULL;
+  return NULL;
 
 }
 
@@ -84,6 +99,7 @@ SEXP watcher_create(SEXP path, SEXP recursive, SEXP callback) {
     Rf_error("Failed to set watch callback");
   }
 
+  // filter only for main event types: Created, Updated, Removed, Renamed
   fsw_event_type_filter filter;
   for (int flag = Created; flag <= Renamed; flag = flag << 1) {
     filter.flag = flag;
@@ -106,12 +122,18 @@ SEXP watcher_start_monitor(SEXP session) {
 
   FSW_HANDLE handle = (FSW_HANDLE) R_ExternalPtrAddr(session);
   pthread_t thr;
+  pthread_attr_t attr;
+
+  pthread_attr_init(&attr);
+  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 
-  if (eln2 == NULL && R_ToplevelExec(load_later_safe, NULL)) {
+  if (eln2 == NULL && R_ToplevelExec(load_later_safe, (void *) Rf_install("loadNamespace"))) {
     eln2 = (void (*)(void (*)(void *), void *, double, int)) R_GetCCallable("later", "execLaterNative2");
   }
+  const int ret = pthread_create(&thr, &attr, &watcher_thread, handle);
+  pthread_attr_destroy(&attr);
 
-  return Rf_ScalarLogical(pthread_create(&thr, NULL, &watcher_thread, handle) == 0);
+  return Rf_ScalarLogical(ret == 0);
 
 }
 

man/watcher.Rd

@@ -1,20 +1,30 @@
-test_that("watcher() watches", {
-  x <- 0L
-  dir <- file.path(tempdir(), "watcher-test")
-  subdir <- file.path(dir, "subdir")
-  dir.create(dir)
-  dir.create(subdir)
+dir <- file.path(tempdir(), "watcher-test")
+subdir <- file.path(dir, "subdir")
+dir.create(dir)
+dir.create(subdir)
+
+test_that("watcher() logs", {
   w <- watcher(dir, recursive = FALSE, callback = NULL)
   expect_s3_class(w, "Watcher")
   expect_false(w$recursive)
   expect_false(w$active)
+  expect_false(w$stop())
   expect_true(w$start())
   expect_true(w$active)
-  Sys.sleep(1L)
+  expect_type(w$path, "character")
+  Sys.sleep(1)
   file.create(file.path(dir, "testfile"))
-  Sys.sleep(1L)
+  file.remove(file.path(dir, "testfile"))
+  Sys.sleep(1)
+  expect_false(w$start())
   expect_true(w$stop())
+  expect_false(w$stop())
   expect_false(w$active)
+  rm(w)
+})
+
+test_that("watcher() callbacks", {
+  x <- 0L
   w <- watcher(dir, recursive = TRUE, callback = function() x <<- x + 1L)
   expect_output(print(w))
   expect_s3_class(w, "Watcher")
@@ -23,27 +33,32 @@ test_that("watcher() watches", {
   skip_if_not_installed("later")
   expect_true(w$start())
   expect_true(w$active)
-  Sys.sleep(1L)
+  Sys.sleep(1)
   file.create(file.path(subdir, "oldfile"))
-  later::run_now(2L)
+  later::run_now(2)
   expect_equal(x, 1L)
   file.rename(file.path(subdir, "oldfile"), file.path(subdir, "newfile"))
-  later::run_now(2L)
+  later::run_now(2)
   expect_equal(x, 2L)
   file.remove(file.path(subdir, "newfile"))
-  later::run_now(2L)
+  later::run_now(2)
   expect_equal(x, 3L)
   file.create(file.path(dir, "oldfile"))
-  later::run_now(2L)
+  later::run_now(2)
   expect_equal(x, 4L)
   file.rename(file.path(dir, "oldfile"), file.path(dir, "newfile"))
-  later::run_now(2L)
+  later::run_now(2)
   expect_equal(x, 5L)
   file.remove(file.path(dir, "newfile"))
-  later::run_now(2L)
+  later::run_now(2)
   expect_equal(x, 6L)
   expect_true(w$stop())
   expect_false(w$active)
-  unlink(dir, recursive = TRUE, force = TRUE)
+  rm(w)
+})
+
+test_that("watcher() validation", {
   expect_error(watcher(callback = "callback"), "must be a function")
 })
+
+unlink(dir, recursive = TRUE, force = TRUE)