Re-implement serialization header without global variable

Author: shikokuchuo

NAMESPACE

@@ -44,7 +44,6 @@ export("%~>%")
 export("opt<-")
 export(.advance)
 export(.context)
-export(.header)
 export(.interrupt)
 export(.keep)
 export(.mark)

R/utils.R

@@ -368,23 +368,11 @@ ip_addr <- function() .Call(rnng_ip_addr)
 #'
 #' Internal package functions.
 #'
-#' @param value integer value.
-#'
-#' @return For `.header()`: the integer `value` supplied.
-#'
-#' @keywords internal
-#' @export
-#'
-.header <- function(value = 0L) .Call(rnng_header_set, value)
-
-#' Read Serialization Header
-#'
 #' @param x raw vector.
 #'
 #' @return For `.read_header()`: integer value.
 #'
 #' @keywords internal
-#' @rdname dot-header
 #' @export
 #'
 .read_header <- function(x) .Call(rnng_header_read, x)
@@ -396,7 +384,7 @@ ip_addr <- function() .Call(rnng_ip_addr)
 #' @return For `.mark()`: the logical `bool` supplied.
 #'
 #' @keywords internal
-#' @rdname dot-header
+#' @rdname dot-read_header
 #' @export
 #'
 .mark <- function(bool = TRUE) .Call(rnng_marker_set, bool)
@@ -408,7 +396,7 @@ ip_addr <- function() .Call(rnng_ip_addr)
 #' @return For `.read_marker()`: logical value `TRUE` or `FALSE`.
 #'
 #' @keywords internal
-#' @rdname dot-header
+#' @rdname dot-read_header
 #' @export
 #'
 .read_marker <- function(x) .Call(rnng_marker_read, x)

man/dot-header.Rd renamed to man/dot-read_header.Rd

@@ -522,7 +522,7 @@ SEXP rnng_send_aio(SEXP con, SEXP data, SEXP mode, SEXP timeout, SEXP pipe, SEXP
     if (raw) {
       nano_encode(&buf, data);
     } else {
-      nano_serialize(&buf, data, NANO_PROT(con));
+      nano_serialize(&buf, data, NANO_PROT(con), 0);
     }
     nng_msg *msg = NULL;
 

man/nanonext-package.Rd

@@ -344,7 +344,7 @@ SEXP rnng_send(SEXP con, SEXP data, SEXP mode, SEXP block, SEXP pipe) {
     if (raw) {
       nano_encode(&buf, data);
     } else {
-      nano_serialize(&buf, data, NANO_PROT(con));
+      nano_serialize(&buf, data, NANO_PROT(con), 0);
     }
     nng_msg *msgp = NULL;
 

src/aio.c

@@ -4,7 +4,6 @@
 
 // internals -------------------------------------------------------------------
 
-int special_header = 0;
 static int special_marker = 0;
 static nano_serial_bundle nano_bundle;
 static SEXP nano_eval_res;
@@ -316,16 +315,16 @@ SEXP nano_raw_char(const unsigned char *buf, const size_t sz) {
 
 }
 
-void nano_serialize(nano_buf *buf, SEXP object, SEXP hook) {
+void nano_serialize(nano_buf *buf, SEXP object, SEXP hook, int header) {
 
   NANO_ALLOC(buf, NANONEXT_INIT_BUFSIZE);
   struct R_outpstream_st output_stream;
 
-  if (special_header || special_marker) {
+  if (header || special_marker) {
     buf->buf[0] = 0x7;
     buf->buf[3] = (uint8_t) special_marker;
-    if (special_header)
-      memcpy(buf->buf + 4, &special_header, sizeof(int));
+    if (header)
+      memcpy(buf->buf + 4, &header, sizeof(int));
     buf->cur += 8;
   }
 
@@ -616,13 +615,6 @@ SEXP rnng_marker_read(SEXP x) {
 
 }
 
-SEXP rnng_header_set(SEXP x) {
-
-  special_header = NANO_INTEGER(x);
-  return x;
-
-}
-
 SEXP rnng_header_read(SEXP x) {
 
   int res = 0;

src/comms.c

@@ -129,7 +129,6 @@ static const R_CallMethodDef callMethods[] = {
   {"rnng_fini_priors", (DL_FUNC) &rnng_fini_priors, 0},
   {"rnng_get_opt", (DL_FUNC) &rnng_get_opt, 2},
   {"rnng_header_read", (DL_FUNC) &rnng_header_read, 1},
-  {"rnng_header_set", (DL_FUNC) &rnng_header_set, 1},
   {"rnng_interrupt_switch", (DL_FUNC) &rnng_interrupt_switch, 1},
   {"rnng_ip_addr", (DL_FUNC) &rnng_ip_addr, 0},
   {"rnng_is_error_value", (DL_FUNC) &rnng_is_error_value, 1},

src/core.c

@@ -244,7 +244,6 @@ typedef struct nano_node_s {
   struct nano_node_s *next;
 } nano_node;
 
-extern int special_header;
 extern void (*eln2)(void (*)(void *), void *, double, int);
 
 extern SEXP nano_AioSymbol;
@@ -297,7 +296,7 @@ int nano_integer(const SEXP);
 SEXP mk_error(const int);
 SEXP mk_error_data(const int);
 SEXP nano_raw_char(const unsigned char *, const size_t);
-void nano_serialize(nano_buf *, const SEXP, SEXP);
+void nano_serialize(nano_buf *, const SEXP, SEXP, int);
 SEXP nano_unserialize(unsigned char *, const size_t, SEXP);
 SEXP nano_decode(unsigned char *, const size_t, const uint8_t, SEXP);
 void nano_encode(nano_buf *, const SEXP);
@@ -341,7 +340,6 @@ SEXP rnng_fini(void);
 SEXP rnng_fini_priors(void);
 SEXP rnng_get_opt(SEXP, SEXP);
 SEXP rnng_header_read(SEXP);
-SEXP rnng_header_set(SEXP);
 SEXP rnng_interrupt_switch(SEXP);
 SEXP rnng_ip_addr(void);
 SEXP rnng_is_error_value(SEXP);

src/init.c

@@ -429,8 +429,7 @@ SEXP rnng_request(SEXP con, SEXP data, SEXP sendmode, SEXP recvmode, SEXP timeou
   if (raw) {
     nano_encode(&buf, data);
   } else {
-    special_header = id;
-    nano_serialize(&buf, data, NANO_PROT(con));
+    nano_serialize(&buf, data, NANO_PROT(con), id);
   }
 
   saio = calloc(1, sizeof(nano_saio));

src/nanonext.h

@@ -213,18 +213,15 @@ test_error(req$opt("false", list()), "type")
 
 test_class("nanoContext", ctx <- context(rep))
 test_print(ctx)
-test_equal(.header(12345L), 12345L)
 test_true(.mark())
 test_class("sendAio", csaio <- req$send_aio(data.frame(), mode = 1L, timeout = 500))
 test_zero(call_aio_(csaio)$result)
 test_class("recvAio", craio <- recv_aio(ctx, mode = 8L, timeout = 500))
 test_type("raw", res <- collect_aio(craio))
 test_true(.read_marker(res))
 test_true(!.read_marker("not"))
-test_equal(.read_header(res), 12345L)
 test_equal(.read_header("not"), 0L)
 test_type("list", unserialize(res[9:length(res)]))
-test_equal(.header(0L), 0L)
 test_true(!.mark(FALSE))
 test_zero(req$send("context test", mode ="raw", block = 500))
 test_equal(recv(ctx, mode = "string", block = 500), "context test")
@@ -234,7 +231,6 @@ test_type("logical", .unresolved(msg))
 test_type("logical", unresolved(msg))
 test_class("data.frame", call_aio(msg)$data)
 test_true(!unresolved(msg))
-test_equal(.header(2025250L), 2025250L)
 test_zero(req$send(c(TRUE, FALSE, TRUE), mode = 2L, block = 500))
 test_class("recvAio", msg <- recv_aio(ctx, mode = 6L, timeout = 500))
 test_type("logical", msg[])
@@ -260,7 +256,6 @@ test_class("recvAio", rek <- request(req$context, c(1+3i, 4+2i), send_mode = "se
 test_zero(reply(ctx, execute = identity, recv_mode = 1L, send_mode = 1L, timeout = 500))
 test_type("complex", call_aio(rek)[["data"]])
 test_type("integer", rek[["aio"]])
-test_equal(.header(0L), 0L)
 
 test_type("list", cfg <- serial_config(class = c("invalid", "custom"), sfunc = list(identity, function(x) raw(1L)), ufunc = list(identity, as.integer)))
 opt(req$socket, "serial") <- cfg
@@ -294,9 +289,9 @@ test_equal(call_aio(cr)$data, "test")
 test_type("integer", cr$aio)
 test_type("integer", send(ctxn, TRUE, mode = 0L, block = FALSE))
 test_type("externalptr", ctxn <- .context(rep))
-test_class("recvAio", cs <- request(.context(req$socket), data = TRUE, cv = NA))
+test_class("recvAio", cs <- request(.context(req$socket), data = TRUE, cv = NA, id = 12345L))
 test_notnull(cs$data)
-test_true(recv(ctxn, block = 500))
+test_equal(.read_header(recv(ctxn, mode = 8L, block = 500)), 12345L)
 test_zero(send(ctxn, TRUE, mode = 1L, block = 500))
 test_class("recvAio", cs <- request(.context(req$socket), data = TRUE, timeout = 5, id = TRUE))
 test_zero(reap(ctxn))