More efficient ReqAios (#50)

* minimal struct sendaio

* protect one

* use official findVarInFrame APIs

* remove leftover

* awkward interface change

Author: shikokuchuo

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: nanonext
 Type: Package
 Title: NNG (Nanomsg Next Gen) Lightweight Messaging Library
-Version: 1.2.1.9010
+Version: 1.2.1.9011
 Description: R binding for NNG (Nanomsg Next Gen), a successor to ZeroMQ. NNG is
     a socket library implementing 'Scalability Protocols', a reliable,
     high-performance standard for common communications patterns including

NEWS.md

@@ -1,4 +1,4 @@
-# nanonext 1.2.1.9010 (development)
+# nanonext 1.2.1.9011 (development)
 
 #### New Features
 

src/aio.c

@@ -163,11 +163,11 @@ static void iaio_finalizer(SEXP xptr) {
 
 SEXP rnng_aio_result(SEXP env) {
 
-  const SEXP exist = nano_findVarInFrame(env, nano_ValueSymbol);
+  const SEXP exist = FINDVARINFRAME(env, nano_ValueSymbol);
   if (exist != R_UnboundValue)
     return exist;
 
-  const SEXP aio = nano_findVarInFrame(env, nano_AioSymbol);
+  const SEXP aio = FINDVARINFRAME(env, nano_AioSymbol);
 
   nano_aio *saio = (nano_aio *) NANO_PTR(aio);
 
@@ -185,11 +185,11 @@ SEXP rnng_aio_result(SEXP env) {
 
 SEXP rnng_aio_get_msg(SEXP env) {
 
-  const SEXP exist = nano_findVarInFrame(env, nano_ValueSymbol);
+  const SEXP exist = FINDVARINFRAME(env, nano_ValueSymbol);
   if (exist != R_UnboundValue)
     return exist;
 
-  const SEXP aio = nano_findVarInFrame(env, nano_AioSymbol);
+  const SEXP aio = FINDVARINFRAME(env, nano_AioSymbol);
 
   nano_aio *raio = (nano_aio *) NANO_PTR(aio);
 
@@ -209,14 +209,7 @@ SEXP rnng_aio_get_msg(SEXP env) {
   case RECVAIOS:
   case REQAIOS:
   case IOV_RECVAIOS: ;
-    nano_cv *ncv;
-    if (raio->type == REQAIOS) {
-      nano_aio *saio = (nano_aio *) raio->next;
-      ncv = (nano_cv *) saio->next;
-    } else {
-      ncv = (nano_cv *) raio->next;
-    }
-
+    nano_cv *ncv = (nano_cv *) (raio->type == REQAIOS ? ((nano_sendaio *) raio->next)->next : raio->next);
     nng_mtx *mtx = ncv->mtx;
     nng_mtx_lock(mtx);
     res = raio->result;
@@ -259,7 +252,7 @@ SEXP rnng_aio_call(SEXP x) {
 
   switch (TYPEOF(x)) {
   case ENVSXP: ;
-    const SEXP coreaio = nano_findVarInFrame(x, nano_AioSymbol);
+    const SEXP coreaio = FINDVARINFRAME(x, nano_AioSymbol);
     if (NANO_TAG(coreaio) != nano_AioSymbol)
       return x;
 
@@ -301,7 +294,7 @@ static SEXP rnng_aio_collect_impl(SEXP x, SEXP (*const func)(SEXP)) {
 
   switch (TYPEOF(x)) {
   case ENVSXP: ;
-    out = nano_findVarInFrame(func(x), nano_ValueSymbol);
+    out = FINDVARINFRAME(func(x), nano_ValueSymbol);
     if (out == R_UnboundValue) break;
     goto resume;
   case VECSXP: ;
@@ -311,7 +304,7 @@ static SEXP rnng_aio_collect_impl(SEXP x, SEXP (*const func)(SEXP)) {
     for (R_xlen_t i = 0; i < xlen; i++) {
       env = func(NANO_VECTOR(x)[i]);
       if (TYPEOF(env) != ENVSXP) goto exit;
-      env = nano_findVarInFrame(env, nano_ValueSymbol);
+      env = FINDVARINFRAME(env, nano_ValueSymbol);
       if (env == R_UnboundValue) goto exit;
       SET_VECTOR_ELT(out, i, env);
     }
@@ -346,7 +339,7 @@ SEXP rnng_aio_stop(SEXP x) {
 
   switch (TYPEOF(x)) {
   case ENVSXP: ;
-    const SEXP coreaio = nano_findVarInFrame(x, nano_AioSymbol);
+    const SEXP coreaio = FINDVARINFRAME(x, nano_AioSymbol);
     if (NANO_TAG(coreaio) != nano_AioSymbol) break;
     nano_aio *aiop = (nano_aio *) NANO_PTR(coreaio);
     nng_aio_stop(aiop->aio);
@@ -368,7 +361,7 @@ static int rnng_unresolved_impl(SEXP x) {
   int xc;
   switch (TYPEOF(x)) {
   case ENVSXP: ;
-    const SEXP coreaio = nano_findVarInFrame(x, nano_AioSymbol);
+    const SEXP coreaio = FINDVARINFRAME(x, nano_AioSymbol);
     if (NANO_TAG(coreaio) != nano_AioSymbol) {
       xc = 0; break;
     }
@@ -420,7 +413,7 @@ SEXP rnng_unresolved(SEXP x) {
 static int rnng_unresolved2_impl(SEXP x) {
 
   if (TYPEOF(x) == ENVSXP) {
-    const SEXP coreaio = nano_findVarInFrame(x, nano_AioSymbol);
+    const SEXP coreaio = FINDVARINFRAME(x, nano_AioSymbol);
     if (NANO_TAG(coreaio) != nano_AioSymbol)
       return 0;
     nano_aio *aiop = (nano_aio *) NANO_PTR(coreaio);

src/core.c

@@ -176,19 +176,6 @@ inline SEXP R_mkClosure(SEXP formals, SEXP body, SEXP env) {
 
 #endif
 
-SEXP nano_findVarInFrame(const SEXP rho, const SEXP symbol) {
-
-  SEXP frame = CAR(rho);
-  while (frame != R_NilValue) {
-    if (TAG(frame) == symbol)
-      return CAR(frame); // BINDING_VALUE
-    frame = CDR(frame);
-  }
-
-  return R_UnboundValue;
-
-}
-
 inline SEXP nano_PreserveObject(const SEXP x) {
 
   SEXP node = Rf_cons(nano_precious, CDR(nano_precious));

src/nanonext.h

@@ -102,6 +102,11 @@ typedef struct nano_handle_s {
 #define NANO_INTEGER(x) *(int *) DATAPTR_RO(x)
 #define NANO_ERROR(x) { Rf_error(x); return R_NilValue; }
 
+#if R_VERSION < R_Version(4, 5, 0)
+#define FINDVARINFRAME(env, sym) Rf_findVarInFrame(env, sym)
+#else
+#define FINDVARINFRAME(env, sym) R_getVarEx(sym, env, FALSE, R_UnboundValue)
+#endif
 #define ERROR_OUT(xc) Rf_error("%d | %s", xc, nng_strerror(xc))
 #define ERROR_RET(xc) { Rf_warning("%d | %s", xc, nng_strerror(xc)); return mk_error(xc); }
 #define NANONEXT_INIT_BUFSIZE 8192
@@ -179,6 +184,11 @@ typedef struct nano_aio_s {
   nano_aio_typ type;
 } nano_aio;
 
+typedef struct nano_sendaio_s {
+  nng_aio *aio;
+  void *next;
+} nano_sendaio;
+
 typedef struct nano_cv_s {
   int condition;
   int flag;
@@ -232,7 +242,6 @@ void nano_encode(nano_buf *, const SEXP);
 int nano_encodes(const SEXP);
 int nano_matcharg(const SEXP);
 int nano_matchargs(const SEXP);
-SEXP nano_findVarInFrame(const SEXP, const SEXP);
 SEXP nano_PreserveObject(const SEXP);
 void nano_ReleaseObject(SEXP);
 void raio_invoke_cb(void *);

src/ncurl.c

@@ -38,12 +38,12 @@ static SEXP mk_error_ncurl(const int xc) {
 
   const char *names[] = {"status", "headers", "data", ""};
   SEXP out = PROTECT(Rf_mkNamed(VECSXP, names));
-  SEXP err = Rf_ScalarInteger(xc);
+  SEXP err = PROTECT(Rf_ScalarInteger(xc));
   Rf_classgets(err, nano_error);
   SET_VECTOR_ELT(out, 0, err);
   SET_VECTOR_ELT(out, 1, err);
   SET_VECTOR_ELT(out, 2, err);
-  UNPROTECT(1);
+  UNPROTECT(2);
   return out;
 
 }
@@ -452,14 +452,14 @@ static SEXP rnng_aio_http_impl(SEXP env, const int typ) {
 
   SEXP exist;
   switch (typ) {
-  case 0: exist = nano_findVarInFrame(env, nano_ResultSymbol); break;
-  case 1: exist = nano_findVarInFrame(env, nano_ProtocolSymbol); break;
-  default: exist = nano_findVarInFrame(env, nano_ValueSymbol); break;
+  case 0: exist = FINDVARINFRAME(env, nano_ResultSymbol); break;
+  case 1: exist = FINDVARINFRAME(env, nano_ProtocolSymbol); break;
+  default: exist = FINDVARINFRAME(env, nano_ValueSymbol); break;
   }
   if (exist != R_UnboundValue)
     return exist;
 
-  const SEXP aio = nano_findVarInFrame(env, nano_AioSymbol);
+  const SEXP aio = FINDVARINFRAME(env, nano_AioSymbol);
 
   nano_aio *haio = (nano_aio *) NANO_PTR(aio);
 
@@ -474,7 +474,7 @@ static SEXP rnng_aio_http_impl(SEXP env, const int typ) {
   SEXP out, vec, rvec, response;
   nano_handle *handle = (nano_handle *) haio->next;
 
-  PROTECT(response = nano_findVarInFrame(env, nano_ResponseSymbol));
+  PROTECT(response = FINDVARINFRAME(env, nano_ResponseSymbol));
   int chk_resp = response != R_NilValue && TYPEOF(response) == STRSXP;
   const uint16_t code = nng_http_res_get_status(handle->res), relo = code >= 300 && code < 400;
   Rf_defineVar(nano_ResultSymbol, Rf_ScalarInteger(code), env);
@@ -520,9 +520,9 @@ static SEXP rnng_aio_http_impl(SEXP env, const int typ) {
   Rf_defineVar(nano_AioSymbol, R_NilValue, env);
 
   switch (typ) {
-  case 0: out = nano_findVarInFrame(env, nano_ResultSymbol); break;
-  case 1: out = nano_findVarInFrame(env, nano_ProtocolSymbol); break;
-  default: out = nano_findVarInFrame(env, nano_ValueSymbol); break;
+  case 0: out = FINDVARINFRAME(env, nano_ResultSymbol); break;
+  case 1: out = FINDVARINFRAME(env, nano_ProtocolSymbol); break;
+  default: out = FINDVARINFRAME(env, nano_ValueSymbol); break;
   }
   return out;
 

src/proto.c

@@ -223,7 +223,7 @@ SEXP rnng_aio_collect_pipe(SEXP aio) {
 
   if (TYPEOF(aio) != ENVSXP)
     goto exitlevel1;
-  const SEXP coreaio = nano_findVarInFrame(aio, nano_AioSymbol);
+  const SEXP coreaio = FINDVARINFRAME(aio, nano_AioSymbol);
   if (NANO_TAG(coreaio) != nano_AioSymbol)
     goto exitlevel1;
 

src/sync.c

@@ -54,8 +54,7 @@ static void request_complete_dropcon(void *arg) {
 static void request_complete_signal(void *arg) {
 
   nano_aio *raio = (nano_aio *) arg;
-  nano_aio *saio = (nano_aio *) raio->next;
-  nano_cv *ncv = (nano_cv *) saio->next;
+  nano_cv *ncv = (nano_cv *) ((nano_sendaio *) raio->next)->next;
   nng_cv *cv = ncv->cv;
   nng_mtx *mtx = ncv->mtx;
 
@@ -181,7 +180,7 @@ static void request_finalizer(SEXP xptr) {
 
   if (NANO_PTR(xptr) == NULL) return;
   nano_aio *xp = (nano_aio *) NANO_PTR(xptr);
-  nano_aio *saio = (nano_aio *) xp->next;
+  nano_sendaio *saio = (nano_sendaio *) xp->next;
   nng_aio_free(saio->aio);
   nng_aio_free(xp->aio);
   if (xp->data != NULL)
@@ -439,12 +438,13 @@ SEXP rnng_request(SEXP con, SEXP data, SEXP sendmode, SEXP recvmode, SEXP timeou
 
   SEXP aio, env, fun;
   nano_buf buf;
-  nano_aio *saio, *raio;
+  nano_sendaio *saio;
+  nano_aio *raio;
   nng_msg *msg;
   int xc;
 
   nano_encodes(sendmode) == 2 ? nano_encode(&buf, data) : nano_serialize(&buf, data, NANO_PROT(con));
-  saio = R_Calloc(1, nano_aio);
+  saio = R_Calloc(1, nano_sendaio);
   saio->next = ncv;
 
   if ((xc = nng_msg_alloc(&msg, 0)))
@@ -500,7 +500,7 @@ SEXP rnng_create_promise(SEXP x, SEXP ctx) {
   if (TYPEOF(x) != ENVSXP)
     return R_NilValue;
 
-  SEXP aio = nano_findVarInFrame(x, nano_AioSymbol);
+  SEXP aio = FINDVARINFRAME(x, nano_AioSymbol);
   if (NANO_TAG(aio) != nano_AioSymbol)
     return R_NilValue;
 

src/thread.c

@@ -234,7 +234,7 @@ SEXP rnng_wait_thread_create(SEXP x) {
   const SEXPTYPE typ = TYPEOF(x);
   if (typ == ENVSXP) {
 
-    const SEXP coreaio = nano_findVarInFrame(x, nano_AioSymbol);
+    const SEXP coreaio = FINDVARINFRAME(x, nano_AioSymbol);
     if (NANO_TAG(coreaio) != nano_AioSymbol)
       return x;