Serialization support

NAMESPACE

@@ -35,4 +35,6 @@ export(processx_conn_read_chars)
 export(processx_conn_read_lines)
 export(processx_conn_write)
 export(run)
+export(serialize_to_raw)
+export(unserialize_from_raw)
 useDynLib(processx, .registration = TRUE, .fixes = "c_")

R/on-load.R

@@ -1,6 +1,8 @@
 
 ## nocov start
 
+serialization_version <- NULL
+
 .onLoad <- function(libname, pkgname) {
   ## This is to circumvent a ps bug
   if (ps::ps_is_supported()) ps::ps_handle()
@@ -9,6 +11,7 @@
       requireNamespace("debugme", quietly = TRUE)) {
     debugme::debugme()
   }
+  serialization_version <<- if (getRversion() >= "3.5.0") 3L else 2L
 }
 
 .onUnload <- function(libpath) {

R/serialize.R

@@ -0,0 +1,35 @@
+
+#' Serialize an R object into a raw vector
+#'
+#' Compared to [serialize()], it uses `xdr = FALSE`, `ascii = FALSE`, and
+#' `version` is 3 for R >= 3.5.0 and 2 otherwise.
+#
+#' @param x Object to serialize.
+#' @return Raw vector, serialized `x`.
+#'
+#' @export
+#' @family serialization
+#' @examples
+#' x <- 1:10
+#' serialize_to_raw(x)
+#' unserialize_from_raw(serialize_to_raw(x))
+
+serialize_to_raw <- function(x) {
+  rethrow_call(c_processx_serialize_to_raw, x, serialization_version)
+}
+
+#' Unserialize an R object from a raw vector
+#'
+#' @param x Raw vector.
+#' @return Unserialized object.
+#'
+#' @export
+#' @family serialization
+#' @examples
+#' x <- 1:10
+#' serialize_to_raw(x)
+#' unserialize_from_raw(serialize_to_raw(x))
+
+unserialize_from_raw <- function(x) {
+  rethrow_call(c_processx_unserialize_from_raw, x)
+}

src/Makevars

@@ -2,6 +2,7 @@
 
 OBJECTS = init.o poll.o errors.o processx-connection.o   \
           processx-vector.o create-time.o base64.o       \
+	  serialization.o                                \
 	  unix/childlist.o unix/connection.o             \
           unix/processx.o unix/sigchld.o unix/utils.o    \
 	  unix/named_pipe.o

src/Makevars.win

@@ -2,6 +2,7 @@
 
 OBJECTS = init.o poll.o errors.o processx-connection.o		     \
           processx-vector.o create-time.o base64.o                   \
+	  serialization.o                                            \
           win/processx.o win/stdio.o win/named_pipe.o                \
 	  win/utils.o win/thread.o
 

src/init.c

@@ -10,6 +10,9 @@ SEXP processx__unload_cleanup();
 SEXP run_testthat_tests();
 SEXP processx__echo_on();
 SEXP processx__echo_off();
+SEXP processx_serialize_to_raw(SEXP );
+SEXP processx_unserialize_from_raw(SEXP);
+
 
 static const R_CallMethodDef callMethods[]  = {
   { "processx_exec",               (DL_FUNC) &processx_exec,              16 },
@@ -55,6 +58,9 @@ static const R_CallMethodDef callMethods[]  = {
   { "processx__echo_on", (DL_FUNC) &processx__echo_on, 0 },
   { "processx__echo_off", (DL_FUNC) &processx__echo_off, 0 },
 
+  { "processx_serialize_to_raw", (DL_FUNC) &processx_serialize_to_raw, 2 },
+  { "processx_unserialize_from_raw", (DL_FUNC) &processx_unserialize_from_raw, 1 },
+
   { NULL, NULL, 0 }
 };
 

src/serialization.c

@@ -0,0 +1,126 @@
+
+#include <Rinternals.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "errors.h"
+
+#ifdef LONG_VECTOR_SUPPORT
+typedef R_xlen_t px_len_t;
+#define PXS_MAX_LENGTH R_XLEN_T_MAX
+#else
+typedef size_t px_len_t;
+#define PXS_MAX_LENGTH SIZE_MAX
+#endif
+
+struct pxs_buffer {
+  px_len_t alloc_size;
+  px_len_t size;
+  unsigned char *data;
+};
+
+#define PXS_MINSIZE 8192 * 2
+
+static void processx__resize_buffer(struct pxs_buffer *buf,
+                                    px_len_t needed) {
+
+  if (needed > PXS_MAX_LENGTH) R_THROW_ERROR("serialized object too big");
+
+  if (needed < PXS_MINSIZE) {
+    needed = PXS_MINSIZE;
+  } else if (needed < 10 * 1024 * 1024) {
+    needed = 2 * needed;
+  } else if (needed < INT_MAX / 1.2 - 100) {
+    needed = 1.2 * (double) needed;
+  }
+
+  unsigned char *tmp = realloc(buf->data, needed);
+  if (tmp == NULL) {
+    free(buf->data);
+    buf->data = NULL;
+    R_THROW_ERROR("cannot allocate buffer");
+  }
+
+  buf->data = tmp;
+  buf->alloc_size = needed;
+}
+
+static void processx__serialize_outchar_mem(R_outpstream_t stream, int c) {
+  struct pxs_buffer *buf = stream->data;
+  if (buf->size >= buf->alloc_size) {
+    processx__resize_buffer(buf, buf->size + 1);
+  }
+  buf->data[buf->size++] = (char) c;
+}
+
+static void processx__serialize_outbytes_mem(R_outpstream_t stream,
+                                             void *newdata, int length) {
+  struct pxs_buffer *buf = stream->data;
+  px_len_t needed = buf->size + (px_len_t) length;
+#ifndef LONG_VECTOR_SUPPORT
+  /* There is a potential overflow here on 32-bit systems */
+  if ((double) buf->size + length > (double) INT_MAX) {
+    R_THROW_ERROR("serialized object too big");
+  }
+#endif
+  if (needed > buf->alloc_size) processx__resize_buffer(buf, needed);
+  memcpy(buf->data + buf->size, newdata, length);
+  buf->size = needed;
+}
+
+static SEXP processx__serialize_close_mem(R_outpstream_t stream) {
+  SEXP val;
+  struct pxs_buffer *buf = stream->data;
+  PROTECT(val = allocVector(RAWSXP, buf->size));
+  memcpy(RAW(val), buf->data, buf->size);
+  if (buf->data != NULL) free(buf->data);
+  buf->data = NULL;
+  UNPROTECT(1);
+  return val;
+}
+
+SEXP processx_serialize_to_raw(SEXP x, SEXP version) {
+  struct R_outpstream_st out = { 0 };
+  R_pstream_format_t type = R_pstream_binary_format;
+  SEXP (*hook)(SEXP, SEXP) = NULL; /* TODO: support hook */
+  struct pxs_buffer buf = { 0, 0, NULL };
+
+  R_InitOutPStream(
+    &out, (R_pstream_data_t) &buf, type, asInteger(version),
+    processx__serialize_outchar_mem, processx__serialize_outbytes_mem,
+    hook, R_NilValue);
+
+  R_Serialize(x, &out);
+
+  return processx__serialize_close_mem(&out);
+}
+
+static int processx__serialize_inchar_mem(R_inpstream_t stream) {
+  struct pxs_buffer *buf = stream->data;
+  if (buf->size >= buf->alloc_size) R_THROW_ERROR("unserialize read error");
+  return buf->data[buf->size++];
+}
+
+static void processx__serialize_inbytes_mem(R_inpstream_t stream,
+                                            void *newdata, int length) {
+  struct pxs_buffer *buf = stream->data;
+  if (buf->size + (px_len_t) length > buf->alloc_size) {
+    R_THROW_ERROR("unserialize read error");
+  }
+  memcpy(newdata, buf->data + buf->size, length);
+  buf->size += length;
+}
+
+SEXP processx_unserialize_from_raw(SEXP sx) {
+  struct R_inpstream_st in = { 0 };
+  SEXP (*hook)(SEXP, SEXP) = NULL; /* TODO: support hook */
+  px_len_t length = XLENGTH(sx);
+  struct pxs_buffer buf = { length, 0, RAW(sx) };
+
+  R_InitInPStream(
+    &in, (R_pstream_data_t) &buf, R_pstream_any_format,
+    processx__serialize_inchar_mem, processx__serialize_inbytes_mem,
+    hook, R_NilValue);
+
+  return R_Unserialize(&in);
+}