added fmap2

Author: dipterix

NAMESPACE

@@ -21,6 +21,7 @@ export(filearray_create)
 export(filearray_load)
 export(filearray_threads)
 export(fmap)
+export(fmap2)
 export(fmap_element_wise)
 export(fwhich)
 export(mapreduce)

R/RcppExports.R

@@ -81,6 +81,10 @@ FARR_buffer_map <- function(input_filebases, output_filebase, map, buffer_nelems
     .Call(`_filearray_FARR_buffer_map`, input_filebases, output_filebase, map, buffer_nelems, result_nelems)
 }
 
+FARR_buffer_map2 <- function(input_filebases, map, buffer_nelems) {
+    .Call(`_filearray_FARR_buffer_map2`, input_filebases, map, buffer_nelems)
+}
+
 FARR_buffer_mapreduce <- function(filebase, map, reduce, buffer_nelems) {
     .Call(`_filearray_FARR_buffer_mapreduce`, filebase, map, reduce, buffer_nelems)
 }

R/map.R

@@ -1,5 +1,7 @@
 #' @title Map multiple file arrays and save results
-#' @description Advanced mapping function for multiple file arrays. This 
+#' @description Advanced mapping function for multiple file arrays. \code{fmap}
+#' runs the mapping functions and stores the results in file arrays. 
+#' \code{fmap2} stores results in memory. This 
 #' feature is experimental. There are several constraints to the input. 
 #' Failure to meet these constraints may result in undefined results, or 
 #' even crashes. Please read Section 'Details' carefully before using 
@@ -10,6 +12,8 @@
 #' @param .y a file array object, used to save results
 #' @param .input_size number of elements to read from each array of \code{x}
 #' @param .output_size \code{fun} output vector length
+#' @param .simplify whether to apply \code{\link[base]{simplify2array}} to 
+#' the result
 #' @param ... other arguments passing to \code{fun}
 #' @return File array instance \code{.y}
 #' @details 
@@ -190,6 +194,56 @@ fmap <- function(x, fun, .y, .input_size = NA, .output_size = NA, ...){
     .y
 }
 
+#' @rdname fmap
+#' @export
+fmap2 <- function(x, fun, .input_size = NA, .simplify = TRUE, ...){
+    if(!length(x)){
+        stop("`x` must be a list of file arrays")
+    }
+    
+    if(inherits(x, "FileArray")){
+        x <- list(x)
+    }
+    
+    dims <- sapply(x, dim)
+    dim <- dims[,1, drop = TRUE]
+    
+    if(any(dims - dim != 0)){
+        stop("Input `x` array dimensions must match")
+    }
+    
+    fbases <- sapply(x, function(el){
+        if( !is_filearray(el) ){
+            stop("Input `x` must only contains file arrays")
+        }
+        el$initialize_partition()
+        el$.filebase
+    })
+    
+    if(is.na(.input_size)){
+        .input_size <- get_buffer_size() / 8L
+    }
+    if(.input_size <= 0){
+        stop("`.input_size` must be postive")
+    }
+    .input_size <- as.integer(.input_size)
+    
+    args <- list(quote(input), ...)
+    map <- function(input){
+        do.call(fun, args)
+    }
+    
+    res <- FARR_buffer_map2(
+        input_filebases = fbases,
+        map = map,
+        buffer_nelems = .input_size
+    )
+    if(.simplify){
+        res <- simplify2array(res)
+    }
+    res
+}
+
 is_filearray <- function(object){
     if(!isS4(object)){ return(FALSE) }
     cls <- class(object)

man/fmap.Rd

@@ -274,6 +274,19 @@ BEGIN_RCPP
     return rcpp_result_gen;
 END_RCPP
 }
+// FARR_buffer_map2
+SEXP FARR_buffer_map2(std::vector<std::string>& input_filebases, const Function& map, const int& buffer_nelems);
+RcppExport SEXP _filearray_FARR_buffer_map2(SEXP input_filebasesSEXP, SEXP mapSEXP, SEXP buffer_nelemsSEXP) {
+BEGIN_RCPP
+    Rcpp::RObject rcpp_result_gen;
+    Rcpp::RNGScope rcpp_rngScope_gen;
+    Rcpp::traits::input_parameter< std::vector<std::string>& >::type input_filebases(input_filebasesSEXP);
+    Rcpp::traits::input_parameter< const Function& >::type map(mapSEXP);
+    Rcpp::traits::input_parameter< const int& >::type buffer_nelems(buffer_nelemsSEXP);
+    rcpp_result_gen = Rcpp::wrap(FARR_buffer_map2(input_filebases, map, buffer_nelems));
+    return rcpp_result_gen;
+END_RCPP
+}
 // FARR_buffer_mapreduce
 SEXP FARR_buffer_mapreduce(const std::string& filebase, const Function map, const Nullable<Function> reduce, const int& buffer_nelems);
 RcppExport SEXP _filearray_FARR_buffer_mapreduce(SEXP filebaseSEXP, SEXP mapSEXP, SEXP reduceSEXP, SEXP buffer_nelemsSEXP) {
@@ -411,6 +424,7 @@ static const R_CallMethodDef CallEntries[] = {
     {"_filearray_FARR_subset_sequential", (DL_FUNC) &_filearray_FARR_subset_sequential, 7},
     {"_filearray_FARR_subset2", (DL_FUNC) &_filearray_FARR_subset2, 8},
     {"_filearray_FARR_buffer_map", (DL_FUNC) &_filearray_FARR_buffer_map, 5},
+    {"_filearray_FARR_buffer_map2", (DL_FUNC) &_filearray_FARR_buffer_map2, 3},
     {"_filearray_FARR_buffer_mapreduce", (DL_FUNC) &_filearray_FARR_buffer_mapreduce, 4},
     {"_filearray_getThreads", (DL_FUNC) &_filearray_getThreads, 1},
     {"_filearray_setThreads", (DL_FUNC) &_filearray_setThreads, 2},

src/RcppExports.cpp

@@ -164,6 +164,106 @@ SEXP FARR_buffer_map(
     return(R_NilValue);
 }
 
+// [[Rcpp::export]]
+SEXP FARR_buffer_map2(
+        std::vector<std::string>& input_filebases,
+        const Function& map,
+        const int& buffer_nelems
+){
+    // prepare inputs
+    int narrays = input_filebases.size();
+    std::vector<List> metas(narrays);
+    std::vector<SEXPTYPE> arr_types(narrays);
+    std::vector<SEXPTYPE> file_buffer_types(narrays);
+    std::vector<SEXPTYPE> memory_buffer_types(narrays);
+    
+    std::vector<SEXP> cumparts(narrays);
+    std::vector<int64_t> part_lengths(narrays);
+    
+    SEXP in_dim = R_NilValue;
+    
+    for(int ii = 0; ii < narrays; ii++){
+        std::string fbase = correct_filebase(input_filebases[ii]);
+        input_filebases[ii] = fbase;
+        List meta = FARR_meta(fbase);
+        metas[ii] = meta;
+        arr_types[ii] = meta["sexp_type"];
+        file_buffer_types[ii] = file_buffer_sxptype(arr_types[ii]);
+        memory_buffer_types[ii] = array_memory_sxptype(arr_types[ii]);
+        cumparts[ii] = realToInt64_inplace(meta["cumsum_part_sizes"]);
+        if( in_dim == R_NilValue ){
+            in_dim = meta["dimension"];
+            realToInt64_inplace(in_dim);
+        }
+    }
+    
+    if( in_dim == R_NilValue ){
+        stop("Cannot obtain input dimensions");
+    }
+    
+    R_xlen_t in_ndims = Rf_length(in_dim);
+    int64_t* in_dimptr = INTEGER64(in_dim);
+    int64_t in_unit_partlen = 1;
+    for(R_xlen_t jj = 0; jj <in_ndims - 1; jj++, in_dimptr++){
+        in_unit_partlen *= *in_dimptr;
+    }
+    int64_t in_array_length = in_unit_partlen * *(INTEGER64(in_dim) + (in_ndims - 1));
+    
+    
+    // allocate buffers
+    SEXP argbuffers = PROTECT(Rf_allocVector(VECSXP, narrays));
+    for(int ii = 0; ii < narrays; ii++){
+        SET_VECTOR_ELT(argbuffers, ii, PROTECT(Rf_allocVector(memory_buffer_types[ii], buffer_nelems)));
+    }
+    
+    int64_t current_pos = 0;
+    
+    int ncores = getThreads();
+    if( ncores > narrays ){
+        ncores = narrays;
+    }
+    
+    R_xlen_t niters = in_array_length / buffer_nelems;
+    if( niters * buffer_nelems < in_array_length ){
+        niters++;
+    }
+    SEXP ret = PROTECT(Rf_allocVector(VECSXP, niters));
+    R_xlen_t iter = 0;
+    
+    for( ; current_pos < in_array_length; current_pos += buffer_nelems, iter++ ){
+        
+#pragma omp parallel num_threads(ncores)
+{
+#pragma omp for schedule(static, 1) nowait
+        for(int ii = 0; ii < narrays; ii++){
+            FARR_subset_sequential(
+            input_filebases[ii],
+            in_unit_partlen,
+            cumparts[ii],
+            arr_types[ii],
+            VECTOR_ELT(argbuffers, ii),
+            current_pos, buffer_nelems
+            );
+        }
+}   
+
+        try{
+            SET_VECTOR_ELT(ret, iter, Shield<SEXP>(map(argbuffers)));
+        } catch(std::exception &ex){
+            UNPROTECT(2 + narrays);
+            forward_exception_to_r(ex);
+        } catch(...){
+            UNPROTECT(2 + narrays);
+            stop("Unknown error.");
+        }
+
+
+    }
+    
+    UNPROTECT(2 + narrays);
+    
+    return(ret);
+}
 
 /*** R
 # devtools::load_all()
@@ -190,6 +290,15 @@ FARR_buffer_map(
     3L,
     1L
 )
-
+res <- FARR_buffer_map2(
+    fbases,
+    function(x){
+        print(c(x[[1]], sum(x[[1]])))
+        sum(x[[1]])
+    },
+    3L,
+    1L
+)
+# y[] - simplify2array(res)
 
 */

src/map.cpp

@@ -51,6 +51,10 @@ test_that("map arrays", {
 
     expect_equal(output[], b)
 
+    
+    d <- fmap2(x, f, .input_size = 842800, .simplify = TRUE)
+    expect_equal(d, b)
+    
     x$delete()
     output$delete()
 

tests/testthat/test-map.R

@@ -138,7 +138,7 @@ In the current version, converting from `double` to `float` introduces overhead
 
 Instead of writing one slice at a time along each margin, we write `100x100x100x5` (10 slices) each time.
 
-1.  partition margin
+1.  Write blocks of data along the partition margin
 
 ```{r}
 microbenchmark::microbenchmark(
@@ -162,7 +162,7 @@ microbenchmark::microbenchmark(
 #>   float 0.626 0.662 0.783  0.698 0.861 1.02     3
 ```
 
-2. Write along fast margin
+2. Write blocks of data along the fast margin
 
 ```{r}
 microbenchmark::microbenchmark(
@@ -186,7 +186,7 @@ microbenchmark::microbenchmark(
 #>   float 0.625 0.652 0.732  0.679 0.786 0.893     3
 ```
 
-3. Writing along slow margin
+3. Write blocks of data along slow margin
 
 ```{r}
 microbenchmark::microbenchmark(
@@ -209,9 +209,6 @@ microbenchmark::microbenchmark(
 #>   float 2.64 2.70 2.73   2.77 2.78 2.79     3
 ```
 
-Note: always avoid writing along the first margin (like `x[i,,,] <- ...`). It's slow.
-
-
 
 ## Read 
 
@@ -292,6 +289,8 @@ Collapse calculates the margin sum/mean. Collapse function in `filearray` uses s
 
 ```{r}
 keep <- c(2, 4)
+output <- filearray_create(tempfile(), dim(x_dbl)[keep])
+output$initialize_partition()
 microbenchmark::microbenchmark(
   farr_double = { x_dbl$collapse(keep = keep, method = "sum") },
   farr_float = { x_flt$collapse(keep = keep, method = "sum") },
@@ -310,3 +309,4 @@ microbenchmark::microbenchmark(
 
 The `dipsaus` package uses multiple threads to collapse arrays in-memory. It is `7~8x` as fast as base R. File array is `1~2x` as fast as base R. Both `dipsaus` and `filearray` have little memory over-heads.
 
+