Create callback based on file descriptor monitoring

.github/workflows/R-CMD-check.yaml

@@ -4,11 +4,7 @@
 # For most R packages it is better to use https://github.com/r-lib/actions
 on:
   push:
-    branches: [main, rc-**]
   pull_request:
-    branches: [main]
-  schedule:
-    - cron:  '0 5 * * 1' # every monday
 
 name: Package checks
 

DESCRIPTION

@@ -20,9 +20,10 @@ Imports:
     rlang
 LinkingTo: Rcpp
 Roxygen: list(markdown = TRUE)
-RoxygenNote: 7.3.1
+RoxygenNote: 7.3.2
 Suggests:
     knitr,
+    nanonext,
     rmarkdown,
     testthat (>= 2.1.0)
 VignetteBuilder: knitr

NAMESPACE

@@ -8,6 +8,7 @@ export(destroy_loop)
 export(exists_loop)
 export(global_loop)
 export(later)
+export(later_fd)
 export(loop_empty)
 export(next_op_secs)
 export(run_now)

R/RcppExports.R

@@ -13,6 +13,10 @@ using_ubsan <- function() {
     .Call(`_later_using_ubsan`)
 }
 
+execLater_fd <- function(callback, fds, timeoutSecs, loop_id) {
+    .Call(`_later_execLater_fd`, callback, fds, timeoutSecs, loop_id)
+}
+
 setCurrentRegistryId <- function(id) {
     invisible(.Call(`_later_setCurrentRegistryId`, id))
 }

R/later.R

@@ -269,6 +269,68 @@ later <- function(func, delay = 0, loop = current_loop()) {
   invisible(create_canceller(id, loop$id))
 }
 
+#' Executes a function when a file descriptor is ready
+#'
+#' @param func A function that takes a single argument, a logical vector that
+#'   indicates which file descriptors are ready. This may be all `FALSE` if the
+#'   `timeout` argument is non-`Inf`, or all `NA` if one or more file
+#'   descriptors are invalid.
+#' @param fd_set Integer vector of file descriptors to monitor. Or on Windows,
+#'   `SOCKET` values.
+#' @param timeout Number of seconds to wait before giving up, and calling `func`
+#'   with all `FALSE`.
+#' @param loop A handle to an event loop. Defaults to the currently-active loop.
+#'
+#' @return A function, which, if invoked, will cancel the callback. The
+#'   function will return \code{TRUE} if the callback was successfully
+#'   cancelled and \code{FALSE} if not (this occurs if the callback has
+#'   executed or has been cancelled already).
+#'
+#' @examplesIf requireNamespace("nanonext", quietly = TRUE)
+#' # create nanonext socket
+#' s1 <- nanonext::socket(listen = "inproc://nano")
+#' s2 <- nanonext::socket(dial = "inproc://nano")
+#' fd1 <- nanonext::opt(s1, "recv-fd")
+#' fd2 <- nanonext::opt(s2, "recv-fd")
+#'
+#' # 1. timeout: prints FALSE, FALSE
+#' later_fd(print, c(fd1, fd2), 0.1)
+#' Sys.sleep(0.2)
+#' run_now()
+#'
+#' # 2. fd1 active: prints TRUE, FALSE
+#' later_fd(print, c(fd1, fd2), 1)
+#' res <- nanonext::send(s2, "msg")
+#' Sys.sleep(0.1)
+#' run_now()
+#'
+#' # 3. both active: prints TRUE, TRUE
+#' res <- nanonext::send(s1, "msg")
+#' later_fd(print, c(fd1, fd2), 1)
+#' Sys.sleep(0.1)
+#' run_now()
+#'
+#' # 4. fd2 active: prints FALSE, TRUE
+#' res <- nanonext::recv(s1)
+#' later_fd(print, c(fd1, fd2), 1)
+#' Sys.sleep(0.1)
+#' run_now()
+#'
+#' # 5. fds invalid: prints NA, NA
+#' close(s2)
+#' close(s1)
+#' later_fd(print, c(fd1, fd2), 1)
+#' Sys.sleep(0.1)
+#' run_now()
+#'
+#' @export
+later_fd <- function(func, fd_set, timeout = Inf, loop = current_loop()) {
+  if (!is.function(func)) {
+    func <- rlang::as_function(func)
+  }
+  invisible(execLater_fd(func, fd_set, timeout, loop$id))
+}
+
 # Returns a function that will cancel a callback with the given ID. If the
 # callback has already been executed or canceled, then the function has no
 # effect.

src/Makevars.win

@@ -1,4 +1,5 @@
 PKG_CPPFLAGS = -DSTRICT_R_HEADERS
+PKG_LIBS = -lWs2_32
 
 #### Debugging flags ####
 # Uncomment to enable thread assertions

src/RcppExports.cpp

@@ -41,6 +41,20 @@ BEGIN_RCPP
     return rcpp_result_gen;
 END_RCPP
 }
+// execLater_fd
+Rcpp::LogicalVector execLater_fd(Rcpp::Function callback, Rcpp::IntegerVector fds, Rcpp::NumericVector timeoutSecs, Rcpp::IntegerVector loop_id);
+RcppExport SEXP _later_execLater_fd(SEXP callbackSEXP, SEXP fdsSEXP, SEXP timeoutSecsSEXP, SEXP loop_idSEXP) {
+BEGIN_RCPP
+    Rcpp::RObject rcpp_result_gen;
+    Rcpp::RNGScope rcpp_rngScope_gen;
+    Rcpp::traits::input_parameter< Rcpp::Function >::type callback(callbackSEXP);
+    Rcpp::traits::input_parameter< Rcpp::IntegerVector >::type fds(fdsSEXP);
+    Rcpp::traits::input_parameter< Rcpp::NumericVector >::type timeoutSecs(timeoutSecsSEXP);
+    Rcpp::traits::input_parameter< Rcpp::IntegerVector >::type loop_id(loop_idSEXP);
+    rcpp_result_gen = Rcpp::wrap(execLater_fd(callback, fds, timeoutSecs, loop_id));
+    return rcpp_result_gen;
+END_RCPP
+}
 // setCurrentRegistryId
 void setCurrentRegistryId(int id);
 RcppExport SEXP _later_setCurrentRegistryId(SEXP idSEXP) {

src/fd.cpp

@@ -0,0 +1,147 @@
+#ifdef _WIN32
+#include <winsock2.h>
+#endif
+#include <Rcpp.h>
+#include <unistd.h>
+#include <cstdlib>
+#include "later.h"
+#include "callback_registry_table.h"
+
+extern CallbackRegistryTable callbackRegistryTable;
+
+typedef struct ThreadArgs_s {
+  SEXP callback;
+  R_xlen_t num_fds;
+  std::unique_ptr<std::vector<int>> fds;
+  fd_set read_fds;
+  struct timeval tv;
+  int max_fd;
+  int loop;
+  bool flag;
+} ThreadArgs;
+
+static void later_callback(void *arg) {
+
+  std::unique_ptr<std::shared_ptr<ThreadArgs>> argsptr(static_cast<std::shared_ptr<ThreadArgs>*>(arg));
+  std::shared_ptr<ThreadArgs> args = *argsptr;
+
+  SEXP results, call;
+  // TODO: actually pre-allocate call on main thread, values can be safely updated from background thread
+  // only call Rf_eval() here to minimise runtime in later callback.
+  PROTECT(results = Rf_allocVector(LGLSXP, args->num_fds));
+  std::memcpy((void *) DATAPTR_RO(results), args->fds->data(), args->num_fds * sizeof(int));
+  PROTECT(call = Rf_lcons(args->callback, Rf_cons(results, R_NilValue)));
+  Rf_eval(call, R_GlobalEnv);
+  UNPROTECT(2);
+  R_ReleaseObject(args->callback);
+
+}
+
+// CONSIDER: upgrade select() to poll() / WSAPoll().
+// CONSIDER: add method for HANDLES on Windows. Assuming we only accept integer
+// values for both, we could use heuristics: check if it is a valid SOCKET or
+// else assume HANDLE - but this is not fool-proof.
+// Otherwise would require an interface for user to specify type.
+static void *select_thread(void *arg) {
+
+  std::unique_ptr<std::shared_ptr<ThreadArgs>> argsptr(static_cast<std::shared_ptr<ThreadArgs>*>(arg));
+  std::shared_ptr<ThreadArgs> args = *argsptr;
+
+  // CONSIDER: if we should be checking more than read activity, i.e. write and error.
+  // Could check all types for all fds, which would be inefficient but
+  // otherwise would require user interface to specify events for each fd
+  int ready = select(args->max_fd + 1, &args->read_fds, NULL, NULL, args->flag ? &args->tv : NULL);
+
+  args->flag = ready < 0;
+
+  for (R_xlen_t i = 0; i < args->num_fds; i++) {
+    (*args->fds)[i] = args->flag ? R_NaInt : FD_ISSET((*args->fds)[i], &args->read_fds) != 0;
+  }
+
+  callbackRegistryTable.scheduleCallback(later_callback, static_cast<void *>(argsptr.release()), 0, args->loop);
+
+  return nullptr;
+
+}
+
+#ifdef _WIN32
+static DWORD WINAPI select_thread_win(LPVOID lpParameter) {
+  select_thread(lpParameter);
+  return 1;
+}
+#endif
+
+// [[Rcpp::export]]
+Rcpp::LogicalVector execLater_fd(Rcpp::Function callback, Rcpp::IntegerVector fds, Rcpp::NumericVector timeoutSecs, Rcpp::IntegerVector loop_id) {
+
+  R_xlen_t num_fds = fds.size();
+  double timeout = timeoutSecs[0];
+  int loop = loop_id[0];
+  int max_fd = -1;
+
+  std::shared_ptr<ThreadArgs> args = std::make_shared<ThreadArgs>();
+  std::unique_ptr<std::vector<int>> fdvals(new std::vector<int>(num_fds));
+  args->num_fds = num_fds;
+  R_PreserveObject(callback);
+  args->callback = callback;
+  args->loop = loop;
+
+  FD_ZERO(&args->read_fds);
+
+  if (num_fds)
+    std::memcpy(fdvals->data(), fds.begin(), num_fds * sizeof(int));
+  for (R_xlen_t i = 0; i < num_fds; i++) {
+    FD_SET((*fdvals)[i], &args->read_fds);
+    max_fd = std::max(max_fd, (*fdvals)[i]);
+  }
+  args->fds = std::move(fdvals);
+  args->max_fd = max_fd;
+  args->flag = timeout != R_PosInf;
+
+  // handle -1 returned by curl_multi_timeout() with default of 1s
+  if (args->flag) {
+    args->tv.tv_sec = timeout < 0 ? 1 : (int) timeoutSecs[0];
+    args->tv.tv_usec = timeout < 0 ? 0 : ((int) (timeoutSecs[0] * 1000)) % 1000 * 1000;
+  }
+
+  std::unique_ptr<std::shared_ptr<ThreadArgs>> argsptr(new std::shared_ptr<ThreadArgs>(args));
+
+#ifdef _WIN32
+
+  HANDLE hThread = CreateThread(NULL, 0, select_thread_win, static_cast<LPVOID>(argsptr.release()), 0, NULL);
+
+  if (hThread == NULL) {
+    Rcpp::stop("thread creation error: " + std::to_string(GetLastError()));
+  }
+
+  CloseHandle(hThread);
+
+#else
+
+  pthread_attr_t attr;
+  pthread_t t;
+
+  // TODO: actually allocate global pthread_attr in .onLoad() and re-use, destroy in .onUnload()
+  if (pthread_attr_init(&attr))
+    Rcpp::stop("thread creation error: " + std::string(strerror(errno)));
+
+  if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) ||
+      pthread_create(&t, &attr, select_thread, static_cast<void *>(argsptr.release()))) {
+    pthread_attr_destroy(&attr);
+    Rcpp::stop("thread creation error: " + std::string(strerror(errno)));
+  }
+
+  if (pthread_attr_destroy(&attr)) {
+    Rcpp::stop("thread creation error: " + std::string(strerror(errno)));
+  }
+
+#endif
+
+  // TODO: Add cancellation: clean way is to insert something that is also
+  // waited on that we can signal to return. But due to the setup overhead
+  // perhaps have this be optional. Otherwise can retain a reference to the
+  // thead to forcefully terminate it - but am wary of potential edge cases.
+
+  return true;
+
+}

src/init.c

@@ -13,6 +13,7 @@ SEXP _later_ensureInitialized(void);
 SEXP _later_execCallbacks(SEXP, SEXP, SEXP);
 SEXP _later_idle(SEXP);
 SEXP _later_execLater(SEXP, SEXP, SEXP);
+SEXP _later_execLater_fd(SEXP, SEXP, SEXP, SEXP);
 SEXP _later_cancel(SEXP, SEXP);
 SEXP _later_nextOpSecs(SEXP);
 SEXP _later_testCallbackOrdering(void);
@@ -33,6 +34,7 @@ static const R_CallMethodDef CallEntries[] = {
   {"_later_execCallbacks",          (DL_FUNC) &_later_execCallbacks,          3},
   {"_later_idle",                   (DL_FUNC) &_later_idle,                   1},
   {"_later_execLater",              (DL_FUNC) &_later_execLater,              3},
+  {"_later_execLater_fd",           (DL_FUNC) &_later_execLater_fd,           4},
   {"_later_cancel",                 (DL_FUNC) &_later_cancel,                 2},
   {"_later_nextOpSecs",             (DL_FUNC) &_later_nextOpSecs,             1},
   {"_later_testCallbackOrdering",   (DL_FUNC) &_later_testCallbackOrdering,   0},

src/later.cpp

@@ -16,7 +16,7 @@ using std::shared_ptr;
 
 static size_t exec_callbacks_reentrancy_count = 0;
 
-static CallbackRegistryTable callbackRegistryTable;
+CallbackRegistryTable callbackRegistryTable;
 
 
 class ProtectCallbacks {