Implements delayed pipe_notify(flag = tools::SIGTERM)

DESCRIPTION

@@ -1,7 +1,7 @@
 Type: Package
 Package: nanonext
 Title: NNG (Nanomsg Next Gen) Lightweight Messaging Library
-Version: 1.7.1.9000
+Version: 1.7.1.9001
 Authors@R: c(
     person("Charlie", "Gao", , "charlie.gao@posit.co", role = c("aut", "cre"),
            comment = c(ORCID = "0000-0002-0750-061X")),

NEWS.md

@@ -2,6 +2,7 @@
 
 #### Updates
 
+* `pipe_notify(flag = tools::SIGTERM)` will now raise the signal with a 100ms grace period to allow a process to exit normally (#212).
 * `parse_url()` drops 'rawurl', 'host' and 'requri' from the output vector as these can be derived from the other parts.
 
 # nanonext 1.7.1

R/aio.R

@@ -303,7 +303,7 @@ stop_request <- function(x) invisible(.Call(rnng_request_stop, x))
 #' while (unresolved(aio)) {
 #'   # do stuff before checking resolution again
 #'   cat("unresolved\n")
-#'   msleep(20)
+#'   msleep(100)
 #' }
 #'
 #' unresolved(aio)

R/sync.R

@@ -167,7 +167,9 @@ cv_signal <- function(cv) invisible(.Call(rnng_cv_signal, cv))
 #'   signal, and causes any subsequent [wait()] to return FALSE instead of TRUE.
 #'   If a signal from the \pkg{tools} package, e.g. `tools::SIGINT`, or an
 #'   equivalent integer value is supplied, this sets a flag and additionally
-#'   raises this signal upon the flag being set.
+#'   raises this signal upon the flag being set. For `tools::SIGTERM`, the
+#'   signal is raised with a 100ms grace period to allow a process to exit
+#'   normally.
 #'
 #' @return Invisibly, zero on success (will otherwise error).
 #'

src/nanonext.h

@@ -124,6 +124,7 @@ extern int R_interrupts_pending;
 #define NANONEXT_CHUNK_SIZE 67108864 // must be <= INT_MAX
 #define NANONEXT_STR_SIZE 40
 #define NANONEXT_WAIT_DUR 1000
+#define NANONEXT_SLEEP_DUR 100
 #define NANO_ALLOC(x, sz)                                      \
   (x)->buf = calloc(sz, sizeof(unsigned char));                \
   if ((x)->buf == NULL) Rf_error("memory allocation failed");  \

src/sync.c

@@ -120,6 +120,16 @@ static void request_complete(void *arg) {
 
 }
 
+void delayed_sigterm(void *arg) {
+  (void) arg;
+  nng_msleep(NANONEXT_SLEEP_DUR);
+#ifdef _WIN32
+  raise(SIGTERM);
+#else
+  kill(getpid(), SIGTERM);
+#endif
+}
+
 void pipe_cb_signal(nng_pipe p, nng_pipe_ev ev, void *arg) {
 
   int sig;
@@ -134,16 +144,20 @@ void pipe_cb_signal(nng_pipe p, nng_pipe_ev ev, void *arg) {
   nng_cv_wake(cv);
   nng_mtx_unlock(mtx);
   if (sig > 1) {
+    if (sig == SIGTERM) {
+      nng_thread *thr;
+      nng_thread_create(&thr, delayed_sigterm, NULL);
+    } else {
 #ifdef _WIN32
-    if (sig == SIGINT)
-      UserBreak = 1;
-    raise(sig);
+      if (sig == SIGINT)
+        UserBreak = 1;
+      raise(sig);
 #else
-    if (sig == SIGINT)
-      R_interrupts_pending = 1;
-    kill(getpid(), sig);
+      if (sig == SIGINT)
+        R_interrupts_pending = 1;
+      kill(getpid(), sig);
 #endif
-
+    }
   }
 
 }