improved non-blocking call_aio(), remove experimental tag

Author: shikokuchuo

R/aio.R

@@ -7,7 +7,7 @@
 #'
 #' @param aio An Aio (object of class 'sendAio' or 'recvAio').
 #' @param block [default TRUE] whether to wait for completion of the AIO
-#'     operation (blocking) or return immediately. [experimental]
+#'     operation (blocking) or return immediately.
 #'
 #' @return The passed Aio object (invisibly), or NULL if non-blocking and the
 #'     Aio has yet to resolve.
@@ -36,13 +36,11 @@
 #'     (invisibly) instead of NULL. The data may then be extracted from the Aio
 #'     using \code{$result}, \code{$raw} or \code{$data} as the case may be.
 #'
-#'     It is not advisable to try to extract the data in one step in the
-#'     non-blocking case as NULL$result is also NULL, hence it would be impossible
-#'     to distinguish between an unresolved Aio and a NULL return value.
-#'
-#'     This argument has the tag [experimental], which indicates that it remains
-#'     under development. Please note that the final implementation may differ
-#'     from the current version.
+#'     It is not advisable to try to extract the data from a 'recvAio' in one
+#'     step using something like \code{call_aio(x)$data} in the non-blocking case.
+#'     This is as \code{call_aio()} will return NULL if the Aio is unresolved and
+#'     \code{NULL$data} is also \code{NULL}, hence it would be impossible to
+#'     distinguish between an unresolved Aio and a NULL return value.
 #'
 #' @examples
 #' s1 <- socket("pair", listen = "inproc://nanonext")
@@ -68,15 +66,16 @@ call_aio <- function(aio, block = TRUE) {
 
   if (length(.subset2(aio, "aio"))) {
 
-    if (!missing(block) && !isTRUE(block)) {
-      .Call(rnng_aio_check, .subset2(aio, "aio")) || return()
-    }
-
     if (inherits(aio, "recvAio")) {
 
+      if (!missing(block) && !isTRUE(block)) {
+        res <- .Call(rnng_aio_get_msg, .subset2(aio, "aio"))
+        missing(res) && return()
+      } else {
+        res <- .Call(rnng_aio_wait_get_msg, .subset2(aio, "aio"))
+      }
       mode <- .subset2(aio, "callparams")[[1L]]
       keep.raw <- .subset2(aio, "callparams")[[2L]]
-      res <- .Call(rnng_aio_get_msg, .subset2(aio, "aio"))
       if (keep.raw) aio[["raw"]] <- res
       is.integer(res) && {
         message(res, " : ", nng_error(res))
@@ -99,7 +98,13 @@ call_aio <- function(aio, block = TRUE) {
       rm("callparams", envir = aio)
 
     } else if (inherits(aio, "sendAio")) {
-      res <- .Call(rnng_aio_result, .subset2(aio, "aio"))
+
+      if (!missing(block) && !isTRUE(block)) {
+        res <- .Call(rnng_aio_result, .subset2(aio, "aio"))
+        missing(res) && return()
+      } else {
+        res <- .Call(rnng_aio_wait_result, .subset2(aio, "aio"))
+      }
       aio[["result"]] <- res
       rm("aio", envir = aio)
       if (res) {

man/call_aio.Rd

@@ -171,29 +171,49 @@ SEXP rnng_send_aio(SEXP socket, SEXP data, SEXP timeout) {
 
 }
 
-SEXP rnng_aio_check(SEXP aio) {
+SEXP rnng_aio_get_msg(SEXP aio) {
 
   if (R_ExternalPtrTag(aio) != nano_AioSymbol)
     error_return("'aio' is not a valid Aio");
   if (R_ExternalPtrAddr(aio) == NULL)
     error_return("'aio' is not an active Aio");
-  int res;
+
+  nng_aio *aiop = (nng_aio *) R_ExternalPtrAddr(aio);
   int_mtx *mutex = (int_mtx *) R_ExternalPtrAddr(Rf_getAttrib(aio, nano_StateSymbol));
   nng_mtx_lock(mutex->mtx);
-  res = mutex->state;
+  int resolv = mutex->state;
   nng_mtx_unlock(mutex->mtx);
+  if (!resolv)
+    return R_MissingArg;
+
+  int xc = nng_aio_result(aiop);
+  if (xc) {
+    nng_aio_free(aiop);
+    R_ClearExternalPtr(aio);
+    return Rf_ScalarInteger(xc);
+  }
+  nng_msg *msgp = nng_aio_get_msg(aiop);
+  size_t sz = nng_msg_len(msgp);
+  SEXP res = PROTECT(Rf_allocVector(RAWSXP, sz));
+  unsigned char *rp = RAW(res);
+  memcpy(rp, nng_msg_body(msgp), sz);
+  nng_msg_free(msgp);
+  nng_aio_free(aiop);
+  R_ClearExternalPtr(aio);
+
+  UNPROTECT(1);
+  return res;
 
-  return Rf_ScalarInteger(res);
 }
 
-SEXP rnng_aio_get_msg(SEXP aio) {
+SEXP rnng_aio_wait_get_msg(SEXP aio) {
 
   if (R_ExternalPtrTag(aio) != nano_AioSymbol)
     error_return("'aio' is not a valid Aio");
   if (R_ExternalPtrAddr(aio) == NULL)
     error_return("'aio' is not an active Aio");
-  nng_aio *aiop = (nng_aio *) R_ExternalPtrAddr(aio);
 
+  nng_aio *aiop = (nng_aio *) R_ExternalPtrAddr(aio);
   nng_aio_wait(aiop);
   int xc = nng_aio_result(aiop);
   if (xc) {
@@ -221,6 +241,30 @@ SEXP rnng_aio_result(SEXP aio) {
     error_return("'aio' is not a valid Aio");
   if (R_ExternalPtrAddr(aio) == NULL)
     error_return("'aio' is not an active Aio");
+
+  nng_aio *aiop = (nng_aio *) R_ExternalPtrAddr(aio);
+  int_mtx *mutex = (int_mtx *) R_ExternalPtrAddr(Rf_getAttrib(aio, nano_StateSymbol));
+  nng_mtx_lock(mutex->mtx);
+  int resolv = mutex->state;
+  nng_mtx_unlock(mutex->mtx);
+  if (!resolv)
+    return R_MissingArg;
+
+  int xc = nng_aio_result(aiop);
+
+  nng_aio_free(aiop);
+  R_ClearExternalPtr(aio);
+  return Rf_ScalarInteger(xc);
+
+}
+
+SEXP rnng_aio_wait_result(SEXP aio) {
+
+  if (R_ExternalPtrTag(aio) != nano_AioSymbol)
+    error_return("'aio' is not a valid Aio");
+  if (R_ExternalPtrAddr(aio) == NULL)
+    error_return("'aio' is not an active Aio");
+
   nng_aio *aiop = (nng_aio *) R_ExternalPtrAddr(aio);
   nng_aio_wait(aiop);
   int xc = nng_aio_result(aiop);
@@ -237,8 +281,10 @@ SEXP rnng_aio_stop(SEXP aio) {
     error_return("'aio' is not a valid Aio");
   if (R_ExternalPtrAddr(aio) == NULL)
     error_return("'aio' is not an active Aio");
+
   nng_aio *aiop = (nng_aio *) R_ExternalPtrAddr(aio);
   nng_aio_stop(aiop);
+
   nng_aio_free(aiop);
   R_ClearExternalPtr(aio);
   return R_NilValue;

src/aio.c

@@ -26,10 +26,11 @@ static void RegisterSymbols(void) {
 }
 
 static const R_CallMethodDef CallEntries[] = {
-  {"rnng_aio_check", (DL_FUNC) &rnng_aio_check, 1},
   {"rnng_aio_get_msg", (DL_FUNC) &rnng_aio_get_msg, 1},
   {"rnng_aio_result", (DL_FUNC) &rnng_aio_result, 1},
   {"rnng_aio_stop", (DL_FUNC) &rnng_aio_stop, 1},
+  {"rnng_aio_wait_get_msg", (DL_FUNC) &rnng_aio_wait_get_msg, 1},
+  {"rnng_aio_wait_result", (DL_FUNC) &rnng_aio_wait_result, 1},
   {"rnng_close", (DL_FUNC) &rnng_close, 1},
   {"rnng_ctx_close", (DL_FUNC) &rnng_ctx_close, 1},
   {"rnng_ctx_open", (DL_FUNC) &rnng_ctx_open, 1},

src/init.c

@@ -19,10 +19,11 @@ extern SEXP nano_UrlSymbol;
 extern SEXP nano_AioSymbol;
 
 /* define functions */
-extern SEXP rnng_aio_check(SEXP);
 extern SEXP rnng_aio_get_msg(SEXP);
 extern SEXP rnng_aio_result(SEXP);
 extern SEXP rnng_aio_stop(SEXP);
+extern SEXP rnng_aio_wait_get_msg(SEXP);
+extern SEXP rnng_aio_wait_result(SEXP);
 extern SEXP rnng_close(SEXP);
 extern SEXP rnng_ctx_close(SEXP);
 extern SEXP rnng_ctx_open(SEXP);