Implement stop_request() to request cancellation when a request is stopped (#185)

* request_stop() concept

* Make safe for all return types

* More tests

* Update docs

Author: shikokuchuo

DESCRIPTION

@@ -1,7 +1,7 @@
 Type: Package
 Package: nanonext
 Title: NNG (Nanomsg Next Gen) Lightweight Messaging Library
-Version: 1.6.2.9002
+Version: 1.6.2.9003
 Authors@R: c(
     person("Charlie", "Gao", , "[email protected]", role = c("aut", "cre"),
            comment = c(ORCID = "0000-0002-0750-061X")),

NAMESPACE

@@ -96,6 +96,7 @@ export(socket)
 export(stat)
 export(status_code)
 export(stop_aio)
+export(stop_request)
 export(stream)
 export(subscribe)
 export(survey_time)

R/aio.R

@@ -264,6 +264,23 @@ collect_aio_ <- function(x) .Call(rnng_aio_collect_safe, x)
 #'
 stop_aio <- function(x) invisible(.Call(rnng_aio_stop, x))
 
+#' Stop Request Operation
+#'
+#' Stop an asynchronous Aio operation, or a list of Aio operations, created by
+#' [request()]. This is an augmented version of [stop_aio()] that additionally
+#' requests cancellation by sending an integer zero followed by the context ID
+#' over the context, and waiting for the response.
+#'
+#' @param x an Aio or list of Aios (objects of class 'recvAio' returned by
+#'   [request()]).
+#'
+#' @return Invisibly, a logical vector.
+#'
+#' @keywords internal
+#' @export
+#'
+stop_request <- function(x) invisible(.Call(rnng_request_stop, x))
+
 #' Query if an Aio is Unresolved
 #'
 #' Query whether an Aio, Aio value or list of Aios remains unresolved. Unlike

man/stop_request.Rd

@@ -469,6 +469,61 @@ SEXP rnng_aio_stop(SEXP x) {
 
 }
 
+SEXP rnng_request_stop(SEXP x) {
+
+  SEXP out;
+  switch (TYPEOF(x)) {
+  case ENVSXP: ;
+    SEXP coreaio;
+    nng_msg *msgp = NULL;
+    int res = 0;
+    PROTECT(coreaio = Rf_findVarInFrame(x, nano_AioSymbol));
+    if (NANO_PTR_CHECK(coreaio, nano_AioSymbol)) goto fail;
+    nano_aio *aiop = (nano_aio *) NANO_PTR(coreaio);
+    if (aiop->type != REQAIOS && aiop->type != REQAIO) goto fail;
+    nng_aio_stop(aiop->aio);
+    nano_saio *saio = (nano_saio *) aiop->cb;
+    if (saio->id == 0) goto fail;
+
+    const SEXP context = Rf_getAttrib(coreaio, nano_ContextSymbol);
+    if (NANO_PTR_CHECK(context, nano_ContextSymbol)) goto fail;
+    nng_ctx *ctx = (nng_ctx *) NANO_PTR(context);
+    const nng_duration dur = NANONEXT_WAIT_DUR;
+    if (nng_ctx_set_ms(*ctx, "send-timeout", dur) ||
+        nng_ctx_set_ms(*ctx, "recv-timeout", dur) ||
+        nng_msg_alloc(&msgp, 0) ||
+        nng_msg_append_u32(msgp, 0) ||
+        nng_msg_append(msgp, &saio->id, sizeof(int)) ||
+        nng_ctx_sendmsg(*ctx, msgp, 0)) {
+      goto fail;
+    }
+    msgp = NULL;
+    if (nng_ctx_recvmsg(*ctx, &msgp, 0))
+      goto fail;
+    memcpy(&res, nng_msg_body(msgp), sizeof(int));
+
+    fail:
+    nng_msg_free(msgp);
+    UNPROTECT(1);
+    out = Rf_ScalarLogical(res != 0);
+    break;
+  case VECSXP: ;
+    const R_xlen_t xlen = Rf_xlength(x);
+    PROTECT(out = Rf_allocVector(LGLSXP, xlen));
+    for (R_xlen_t i = xlen - 1; i >= 0; i--) {
+      SEXP item = rnng_request_stop(NANO_VECTOR(x)[i]);
+      SET_LOGICAL_ELT(out, i, NANO_INTEGER(item));
+    }
+    UNPROTECT(1);
+    break;
+  default:
+    out = Rf_ScalarLogical(0);
+  }
+
+  return out;
+
+}
+
 static int rnng_unresolved_impl(SEXP x) {
 
   int xc;

src/aio.c

@@ -153,6 +153,7 @@ static const R_CallMethodDef callMethods[] = {
   {"rnng_recv", (DL_FUNC) &rnng_recv, 4},
   {"rnng_recv_aio", (DL_FUNC) &rnng_recv_aio, 6},
   {"rnng_request", (DL_FUNC) &rnng_request, 8},
+  {"rnng_request_stop", (DL_FUNC) &rnng_request_stop, 1},
   {"rnng_send", (DL_FUNC) &rnng_send, 5},
   {"rnng_send_aio", (DL_FUNC) &rnng_send_aio, 6},
   {"rnng_serial_config", (DL_FUNC) &rnng_serial_config, 3},

src/init.c

@@ -123,6 +123,7 @@ extern int R_interrupts_pending;
 #define NANONEXT_SERIAL_THR 134217728
 #define NANONEXT_CHUNK_SIZE INT_MAX // must be <= INT_MAX
 #define NANONEXT_STR_SIZE 40
+#define NANONEXT_WAIT_DUR 1000
 #define NANO_ALLOC(x, sz)                                      \
   (x)->buf = calloc(sz, sizeof(unsigned char));                \
   if ((x)->buf == NULL) Rf_error("memory allocation failed");  \
@@ -390,6 +391,7 @@ SEXP rnng_reap(SEXP);
 SEXP rnng_recv(SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_recv_aio(SEXP, SEXP, SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_request(SEXP, SEXP, SEXP, SEXP, SEXP, SEXP, SEXP, SEXP);
+SEXP rnng_request_stop(SEXP);
 SEXP rnng_send(SEXP, SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_send_aio(SEXP, SEXP, SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_serial_config(SEXP, SEXP, SEXP);

src/nanonext.h

@@ -114,6 +114,7 @@ test_identical(call_aio(msg), msg)
 test_class("recvAio", msg <- n1$recv_aio(mode = "complex", timeout = 500))
 test_null(stop_aio(msg))
 test_null(stop_aio(n))
+test_false(stop_request(n))
 test_identical(call_aio(msg), msg)
 test_class("errorValue", msg$data)
 test_identical(call_aio(n), n)
@@ -300,6 +301,7 @@ test_notnull(cs$data)
 test_true(recv(ctxn, block = 500))
 test_zero(send(ctxn, TRUE, mode = 1L, block = 500))
 test_class("recvAio", cs <- request(.context(req$socket), data = TRUE, timeout = 5, id = TRUE))
+test_false(stop_request(cs))
 test_zero(reap(ctxn))
 test_equal(reap(ctxn), 7L)
 test_zero(pipe_notify(rep, cv, add = TRUE, flag = TRUE))
@@ -450,6 +452,7 @@ test_type("integer", call_aio_(put1)$status)
 if (put1$status == 200L) test_notnull(put1$headers)
 if (put1$status == 200L) test_notnull(put1$data)
 test_null(stop_aio(put1))
+test_false(stop_request(put1))
 test_class("ncurlAio", haio <- ncurl_aio("https://i.i"))
 test_class("errorValue", call_aio(haio)$data)
 test_print(haio$data)
@@ -588,8 +591,9 @@ test_error(collect_aio_("a"), "object is not an Aio or list of Aios")
 test_error(collect_aio_(list("a")), "object is not an Aio or list of Aios")
 test_error(collect_aio(list(fakesock)), "object is not an Aio or list of Aios")
 test_null(stop_aio("a"))
+test_false(stop_request("a"))
 test_null(stop_aio(list("a")))
-test_null(stop_aio(list("a")))
+test_false(stop_request(list("a")))
 test_null(.keep(NULL, new.env()))
 test_null(.keep(new.env(), new.env()))