@@ -1,8 +1,7 @@

 Package: Streamer

 Type: Package

 Title: Enabling stream processing of large files

-Version: 1.3.5

-Date: 2010-10-13

+Version: 1.3.6

 Author: Martin Morgan, Nishant Gopalakrishnan

 Maintainer: Martin Morgan <mtmorgan@fhcrc.org>

 Description: Large data files can be difficult to work with in R,

@@ -10,31 +10,47 @@ setMethod(stream, "Producer",

 })

 setMethod(lapply, "Producer",

-    function(X, FUN, ..., .env=parent.frame())

+    function(X, FUN, ...)

 {

     FUN <- match.fun(FUN)

-    fun <- function(yield) {

-        y <- tryCatch(yield(), error=function(err) {

-            stop("'lapply,Producer-method' yield() failed: ",

-                 conditionMessage(err))

-        })

+    YIELD <- selectMethod("yield", class(X)) # avoid S4 dispatch

+

+    it <- 0L

+    result <- vector("list", 4096L)     # pre-allocate

+    .partialResult <- function(err) {

+        if (is(err, "simpleError")) {

+            length(result) <- it

+            err$message <- paste0("yield(): ", conditionMessage(err))

+            err$partialResult <- result

+            class(err) <- c("partialResult", class(err))

+        }

+        stop(err)

+    }

+

+    repeat {

+        y <- tryCatch(YIELD(X), error = .partialResult)

         if (!length(y))

-            return(y)

-        tryCatch(eval(FUN(y, ...), .env), error=function(err) {

-            stop("'lapply,Producer-method' FUN() failed: ",

-                 conditionMessage(err))

-        })

+            break;

+        y <- tryCatch(FUN(y, ...), error = .partialResult)

+        it <- it + 1L

+        if (it == length(result))       # grow

+            length(result) <- 1.6 * length(result)

+        result[[it]] <- y

     }

-    ## avoid S4 dispatch on yield(X)

-    .Call(.lapply_Producer, fun, X$yield, environment())

+    length(result) <- it

+    result

 })

 setMethod(sapply, "Producer",

     function(X, FUN, ..., simplify=TRUE, USE.NAMES=TRUE)

 {

     FUN <- match.fun(FUN)

-    .env <- parent.frame()

-    answer <- lapply(X = X, FUN = FUN, ..., .env=.env)

+    answer <- tryCatch(lapply(X = X, FUN = FUN, ...), error=function(err) {

+        if (is(err, "partialResult"))

+            err$partialResult <- simplify2array(err$partialResult,

+                                                higher = (simplify == "array"))

+        stop(err)

+    })

     if (!identical(simplify, FALSE) && length(answer))

         simplify2array(answer, higher = (simplify == "array"))

     else answer

@@ -24,6 +24,12 @@ test_lapply_Producer <- function()

     fun <- function(x) if (x == 3) stop("x: ", x) else x

     checkException(lapply(Seq(to=5), fun), silent=TRUE)

+    ## error partial results

+    res <- tryCatch(sapply(Seq(to=5), fun), error = function(err) {

+        err$partialResult

+    })

+    checkIdentical(c(1, 2), res)

+

     ## trigger re-allocation

     ## res <- lapply(Seq(to=4096*4),

     ##               function(x) { if (x %% 1000 == 0) message(x); x })

@@ -21,7 +21,7 @@

 }

 \usage{

-\S4method{lapply}{Producer}(X, FUN, ..., .env=parent.frame())

+\S4method{lapply}{Producer}(X, FUN, ...)

 \S4method{sapply}{Producer}(X, FUN, ..., simplify=TRUE, USE.NAMES=TRUE)

 }

@@ -44,8 +44,9 @@

 \section{Methods}{

   \code{lapply} and \code{sapply} apply \code{FUN} to each result

-  applied to \code{yield()}. Infite producers will of course exhaust

-  memory.

+  applied to \code{yield()}. Partial results on error can be recovered

+  using \code{\link{tryCatch}}, as illustrated below. Infinite producers

+  will of course exhaust memory.

   Inherited methods defined on this class include:

   \describe{

@@ -99,6 +100,15 @@ showMethods(class="Producer", where="package:Streamer")

 sapply(Seq(to=47, length.out=7), function(elt) {

     c(n = length(elt), xbar = mean(elt))

 })

+

+## recover partial results

+fun = function(i) if (i == 5) stop("oops, i == 5") else i

+res <- tryCatch(sapply(Seq(to=10), fun), error=function(err) {

+    warning(conditionMessage(err),

+            "\n  only partial results available")

+    simplify2array(err$partialResult)

+})

+res

 }

 \keyword{classes}

@@ -1,13 +1,10 @@

 #include <R_ext/Rdynload.h>

 #include "raw_input.h"

-#include "lapply.h"

 static const R_CallMethodDef callMethods[] = {

   /* raw_parse */

   {".raw_parse_count_records", (DL_FUNC) &raw_parse_count_records, 2},

   {".raw_parse", (DL_FUNC) &raw_parse, 3},

-  /* lapply */

-  {".lapply_Producer", (DL_FUNC) &lapply_Producer, 3},

   {NULL, NULL, 0}

 };

deleted file mode 100644

@@ -1,35 +0,0 @@

-#include "lapply.h"

-

-SEXP

-lapply_Producer(SEXP fun, SEXP X, SEXP rho)

-{

-    SEXP call1, ans1, ans;

-    int iter = 0;

-    R_len_t curr_size = 4096;

-    PROTECT_INDEX px;

-

-    /* FIXME: type checks */

-

-    PROTECT(call1 = lang2(fun, X));

-    PROTECT_WITH_INDEX(ans = Rf_allocVector(VECSXP, curr_size), &px);

-    while (1) {

-	PROTECT(ans1 = Rf_eval(call1, rho));

-	if (0 == Rf_length(ans1)) {

-	    UNPROTECT(1);

-	    break;

-	}

-	if (iter == curr_size) {

-	    if (curr_size == R_LEN_T_MAX)

-		Rf_error("%s cannot create %d-element vector",

-			 "'lapply,Producer-method'", curr_size);

-	    curr_size *= 1.6;

-	    if (curr_size > R_LEN_T_MAX)

-		curr_size = R_LEN_T_MAX;

-	    REPROTECT(ans = Rf_lengthgets(ans, curr_size), px);

-	}

-	SET_VECTOR_ELT(ans, iter++, ans1);

-	UNPROTECT(1);

-    }

-    UNPROTECT(2);

-    return Rf_lengthgets(ans, iter);

-}

deleted file mode 100644

@@ -1,10 +0,0 @@

-#ifndef LAPPLY_H_

-#define LAPPLY_H

-

-#include <Rdefines.h>

-

-SEXP

-lapply_Producer(SEXP fun, SEXP X, SEXP rho);

-

-#endif

-