fix(share): non-destructive multi-reader getData

NAMESPACE

@@ -5,5 +5,5 @@ importFrom(utils, packageVersion)
 export(
   charToFact, count, countNA, countOccur, fduplicated, fpmin, fpmax, fpos, funique, iif, nif, nswitch,
   pall, pallNA, pallv, pany, panyNA, panyv, pcount, pcountNA, pfirst, plast, pmean, pprod, prange, psum, setlevels, topn, uniqLen, vswitch, psort,
-  getData, shareData, clearData
+  getData, shareData, clearData, clearShared
 )

R/call.R

@@ -1,6 +1,22 @@
 # Function calls
+checkMapName = function(map_name) {
+  if (!is.character(map_name) || length(map_name) != 1L || is.na(map_name) || !nzchar(map_name))
+    stop("Argument 'map_name' must be a non-empty string of length 1.")
+  shmName(map_name)
+}
+
+checkVerbose = function(verbose) {
+  if (!is.logical(verbose) || length(verbose) != 1L || is.na(verbose))
+    stop("Argument 'verbose' must be TRUE or FALSE.")
+  verbose
+}
+
 charToFact  = function(x, decreasing=FALSE, addNA=TRUE, nThread=getOption("kit.nThread")) .Call(CcharToFactR, x, decreasing, nThread, NA, parent.frame(), addNA)
-clearData   = function(x, verbose=FALSE) .Call("CclearMappingObjectR", x, verbose)
+clearData   = function(x, verbose=FALSE) .Call("CclearMappingObjectR", x, checkVerbose(verbose))
+clearShared = function(map_name, verbose=FALSE) {
+  map_name = checkMapName(map_name)
+  .Call("CunlinkMappingObjectR", map_name, paste0(map_name,"_key"), checkVerbose(verbose))
+}
 count       = function(x, value) .Call(CcountR, x, value)
 countNA     = function(x) .Call(CcountNAR, x)
 countOccur  = function(x) .Call(CcountOccurR, x)
@@ -64,23 +80,22 @@ psort = function(x, decreasing = FALSE, na.last = NA, nThread=getOption("kit.nTh
 shmName = function(map_name) sub("^/*", "/", map_name)
 
 shareData = function(data, map_name, verbose=FALSE) {
+  map_name = checkMapName(map_name)
+  verbose = checkVerbose(verbose)
   conn = rawConnection(raw(0L), "w")
+  on.exit(close(conn), add = TRUE)
   serialize(data, conn)
-  seek(conn, 0L)
-  map_name = shmName(map_name)
-  x = .Call(
+  .Call(
     "CcreateMappingObjectR", map_name, paste0(map_name,"_key"),
     rawConnectionValue(conn), verbose
   )
-  close(conn)
-  x
 }
 
 getData = function(map_name, verbose=FALSE) {
-  map_name = shmName(map_name)
+  map_name = checkMapName(map_name)
+  verbose = checkVerbose(verbose)
   output = .Call("CgetMappingObjectR", map_name, paste0(map_name,"_key"), verbose)
   conn = rawConnection(output,"r")
-  obj = unserialize(conn)
-  close(conn)
-  obj
+  on.exit(close(conn), add = TRUE)
+  unserialize(conn)
 }

_pkgdown.yml

@@ -73,7 +73,7 @@ reference:
   desc: "Find a matrix position inside a larger matrix and share data between R sessions."
 - contents:
   - fpos
-  - shareData/getData/clearData
+  - shareData/getData/clearData/clearShared
 
 articles:
 - title: "Introduction to kit"

man/shareData.Rd

@@ -1,35 +1,60 @@
-\name{shareData/getData/clearData}
+\name{shareData/getData/clearData/clearShared}
 \alias{shareData}
 \alias{getData}
 \alias{clearData}
-\title{ Share Data between R Sessions}
+\alias{clearShared}
+\title{Share Data between R Sessions}
 \description{
-Experimental functions that enable the user to share a R object between 2 \R sessions.
+Experimental functions that enable the user to share an R object between R sessions.
+The object is serialized once by \code{shareData} into POSIX shared memory
+(or a Windows file mapping). It can then be read multiple times, from the
+current session and/or from other processes running as the same user, via
+\code{getData}. Reads are non-destructive: the segment lives until the owner
+handle returned by \code{shareData} is cleared with \code{clearData} (or
+garbage collected). Keep the owner object alive while readers are active,
+e.g. with \code{on.exit(clearData(x))}, and clear only after all readers
+(including background workers) are done. Exactly one writer may publish a
+name at a time; do not re-assign the owner variable without clearing first
+(\code{clearData} the old handle before re-sharing its name), and do not
+re-share a name at a different size while readers are active.
 }
 \usage{
 shareData(data, map_name, verbose=FALSE)
 getData(map_name, verbose=FALSE)
 clearData(x, verbose=FALSE)
+clearShared(map_name, verbose=FALSE)
 }
 \arguments{
   \item{data}{ A \R object like a vector or a \code{data.frame}.}
-  \item{map_name}{ A character. A name for the memory map location where to store the data.}
+  \item{map_name}{ A character. A name for the memory map location where to store the data. Concurrent writers must use unique names: re-sharing a name while readers are active fails those reads loudly instead of returning mixed data. On Windows a live name cannot grow: clear it before re-sharing a larger object.}
   \item{x}{ An external pointer like the one returned by function \code{shareData}.}
   \item{verbose}{ A logical value \code{TRUE} or \code{FALSE} to provide or not information to the user.}
 }
 \value{
 \code{shareData} returns a external pointer.
-\code{getData} returns an \R object stored in the memory location \code{map_name}.
+\code{getData} returns an \R object stored in the memory location \code{map_name}. Reads do not consume the segment, so \code{getData} may be called repeatedly and from multiple processes. A read racing a concurrent write errors (\code{please retry}) instead of returning torn data.
 \code{clearData} returns \code{TRUE} or \code{FALSE} depending on whether the data have been cleared in memory.
+\code{clearShared} unlinks a segment by name, without needing the owner handle, and returns \code{TRUE} if anything was removed. Use it to reap orphaned segments (e.g. after a crashed session); on Windows it is a harmless no-op because mappings vanish with the last open handle, and always returns \code{FALSE} there.
 }
 \author{Morgan Jacob}
 \examples{
 # In R session 1: share data in memory
 # > x = shareData(mtcars,"share1")
 #
-# In R session 2: get data from session 1
+# In R session 2: get data from session 1 (repeatable, also in parallel)
+# > getData("share1")
 # > getData("share1")
 #
-# In R session 1: clear data in memory
+# In R session 1: clear data in memory once all readers are done
 # > clearData(x)
+#
+# Sharing large immutable globals with background workers (owner stays alive):
+# > x = shareData(list(graph = g, dist = D), "/mydata")
+# > # workers only receive the short map name and attach via shared memory
+# > # mirai::daemons(4)
+# > # mirai::mirai_map(ids, function(i) fun(i, kit::getData("/mydata")))
+# > # mirai::daemons(0); clearData(x)
+#
+# Reap a leftover segment by name (POSIX shared memory survives crashes):
+# > clearShared("/mydata")
 }

src/init.c

@@ -27,9 +27,10 @@ static const R_CallMethodDef CallEntries[] = {
   {"CsetlevelsR",              (DL_FUNC) &setlevelsR,              -1},
   {"CtopnR",                   (DL_FUNC) &topnR,                   -1},
   {"CvswitchR",                (DL_FUNC) &vswitchR,                -1},
-  {"CcreateMappingObjectR",    (DL_FUNC) &createMappingObjectR,    -1},
-  {"CgetMappingObjectR",       (DL_FUNC) &getMappingObjectR,       -1},
-  {"CclearMappingObjectR",     (DL_FUNC) &clearMappingObjectR,     -1},
+  {"CcreateMappingObjectR",    (DL_FUNC) &createMappingObjectR,    4},
+  {"CgetMappingObjectR",       (DL_FUNC) &getMappingObjectR,       3},
+  {"CunlinkMappingObjectR",    (DL_FUNC) &unlinkMappingObjectR,    3},
+  {"CclearMappingObjectR",     (DL_FUNC) &clearMappingObjectR,     2},
   {NULL,                       NULL,                               -1}
 };
 
@@ -63,5 +64,6 @@ void R_init_kit(DllInfo *dll) {
   R_RegisterCCallable("kit", "CvswitchR",                (DL_FUNC) &vswitchR);
   R_RegisterCCallable("kit", "CcreateMappingObjectR",    (DL_FUNC) &createMappingObjectR);
   R_RegisterCCallable("kit", "CgetMappingObjectR",       (DL_FUNC) &getMappingObjectR);
+  R_RegisterCCallable("kit", "CunlinkMappingObjectR",    (DL_FUNC) &unlinkMappingObjectR);
   R_RegisterCCallable("kit", "CclearMappingObjectR",     (DL_FUNC) &clearMappingObjectR);
 }

src/kit.h

@@ -117,6 +117,7 @@ extern SEXP vswitchR(SEXP x, SEXP values, SEXP outputs, SEXP na, SEXP nthreads,
 
 extern SEXP createMappingObjectR(SEXP MapName, SEXP MapLength, SEXP DataObject, SEXP verboseArg);
 extern SEXP getMappingObjectR(SEXP MapName, SEXP MapLength, SEXP verboseArg);
+extern SEXP unlinkMappingObjectR(SEXP MapName, SEXP MapLength, SEXP verboseArg);
 extern SEXP clearMappingObjectR(SEXP ext, SEXP verboseArg);
 
 union uno { double d; unsigned int u[2]; };