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

 Package: Streamer

 Type: Package

 Title: Enabling stream processing of large files

-Version: 1.3.8

+Version: 1.3.9

 Author: Martin Morgan, Nishant Gopalakrishnan

 Maintainer: Martin Morgan <mtmorgan@fhcrc.org>

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

@@ -21,7 +21,8 @@ Collate:

   AllGenerics.R 

   Streamer-class.R Producer-class.R Consumer-class.R Stream-class.R

   ConnectionProducer-classes.R RawInput-class.R Seq-class.R

-  Downsample-class.R Team-class.R FunctionProducerConsumer-classes.R

+  Downsample-class.R FunctionProducerConsumer-classes.R

+  ParallelParam-classes.R Team-class.R Team-methods.R

   Utility-classes.R ParallelConnector-class.R TConnector-class.R

   YConnector-class.R

   lapply-methods.R stream-methods.R

@@ -7,3 +7,6 @@ setGeneric("reset", function(x, ...) standardGeneric("reset"))

 setGeneric("yield", function(x, ...) standardGeneric("yield"))

 setGeneric("status", function(x, ...) standardGeneric("status"))

+

+setGeneric("Team", function(FUN, ..., param) standardGeneric("Team"),

+           signature = "param")

new file mode 100644

@@ -0,0 +1,62 @@

+## ParallelParam

+

+.ParallelParam <- setRefClass("ParallelParam",

+    fields = list(size = "integer"),

+    contains = "Streamer",

+    methods = list(

+      initialize = function(...) {

+          callSuper(...)

+          if (length(size) != 0 && size < 1L)

+              stop("'size' must be >= 1")

+          .self

+      },

+      show = function() {

+          cat("class:", class(.self), "\n")

+          cat("size:", size, "\n")

+          cat("verbose:", verbose, "\n")

+      }))

+

+.MulticoreParam <- setRefClass("MulticoreParam",

+    fields = list(mc.set.seed = "logical"),

+    contains = "ParallelParam",

+    methods = list(

+      show = function() {

+          callSuper()

+          cat("mc.set.seed:", mc.set.seed, "\n")

+      }))

+

+MulticoreParam <-

+    function(size = getOption("mc.cores", 2L), mc.set.seed = TRUE,

+             ...)

+{

+    .MulticoreParam$new(size=as.integer(size),

+                        mc.set.seed=mc.set.seed, ..., inUse=TRUE)

+}

+

+## ParallelRegister

+

+.parallelRegister <- setRefClass("ParallelRegister",

+    fields = list(param = "ParallelParam"),

+    methods = list(

+      register = function(param) {

+          oparam <- param

+          if (is.null(param))

+              .self$param <- .ParallelParam$new()

+          else

+              .self$param <- param

+          invisible(oparam)

+      },

+      show = function() {

+          if (param$inUse)

+              param$show()

+          else

+              cat("<empty>\n")

+      }))$new()                         # singleton

+

+register <- function(param)

+{

+    if (missing(param))

+        .parallelRegister

+    else 

+        .parallelRegister$register(param)

+}

@@ -1,3 +1,11 @@

+## Team

+

+.Team <- setRefClass("Team",

+    fields = list(parallelParam = "ParallelParam"),

+    contains = "Consumer")

+

+## MulticoreTeam

+

 .mccollect <- 

     function (jobs, wait = TRUE, timeout = 0, intermediate = FALSE)

 {

@@ -25,13 +33,12 @@

     results

 }

-.Team <- setRefClass("Team",

+.MulticoreTeam <- setRefClass("MulticoreTeam",

     fields=list(

       tasks="list",          # status: IDLE, YIELD, VALUE, ERROR, DONE

       FUN="function",

-      mc.set.seed="logical", silent="logical",

       .id="integer", .yid="integer"),

-    contains = "Consumer",

+    contains = "Team",

     methods = list(

       initialize = function(...) {

           callSuper(..., .id=1L, .yid=1L)

@@ -60,7 +67,9 @@

               task[c("name", "result", "status")] <-

                   list(.id, value, "ERROR")

           } else {

-              task <- .mc_parallel(FUN(value), .id, mc.set.seed, silent)

+              task <- .mc_parallel(FUN(value), .id,

+                                   parallelParam$mc.set.seed,

+                                   parallelParam$verbose)

               task$status <- "YIELD"

           }

           .self$tasks[[idx]] <- task

@@ -109,22 +118,8 @@

       },

       show = function() {

-          cat("mc.set.seed:", mc.set.seed, "\n")

-          cat("silent:", silent, "\n")

-          cat("tasks status:\n")

+          cat("MulticoreParam:\n")

+          parallelParam$show()

+          cat("\ntasks status:\n")

           print(noquote(setNames(status(), names())))

       }))

-

-Team <-

-    function(FUN, size=1L, mc.set.seed=TRUE, silent=TRUE, ...)

-{

-    if (.Platform$OS.type != "unix")

-        stop("'Team' not supported on platform '", .Platform$OS.type, "'")

-    require(parallel)

-    if (size < 1L)

-        stop("'size' must be >= 1")

-    tasks <- replicate(size, list(status="IDLE", name=NA_character_),

-                       simplify=FALSE)

-    .Team$new(FUN=FUN, tasks=tasks, mc.set.seed=mc.set.seed,

-              silent=silent, ...)

-}

new file mode 100644

@@ -0,0 +1,24 @@

+setMethod("Team", c(param="missing"),

+    function(FUN, ..., param)

+{

+    param <- parallelRegister()$param

+    if (!param$inUse)

+        if (.Platform$OS.type == "unix")

+            param <- MulticoreParam()

+        else

+            stop("'Team' not supported on '",

+                 .Platform$OS.type, "' operating system")

+    Team(FUN, ..., param=param)

+})

+

+setMethod("Team", c(param="MulticoreParam"),

+    function(FUN, ..., param)

+{

+    if (.Platform$OS.type != "unix")

+        stop("'Team' with 'param=MulticoreParam()' not supported on '",

+             .Platform$OS.type, "' operating system")

+    require(parallel)

+    tasks <- replicate(param$size, list(status="IDLE",

+                       name=NA_character_), simplify=FALSE)

+    .MulticoreTeam$new(FUN=FUN, tasks=tasks, ..., parallelParam=param)

+})

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

 \alias{FunctionConsumer}

 \alias{FunctionConsumer-class}

-\title{Classes to create Producers and Consumers from functions}

+\title{Classes for user-defined Producers and Consumers}

 \description{

new file mode 100644

@@ -0,0 +1,87 @@

+\name{ParallelParam-class}

+\Rdversion{1.1}

+\docType{class}

+\alias{ParallelParam-class}

+\alias{ParallelParam}

+\alias{ParallelRegister-class}

+\alias{MulticoreParam-class}

+\alias{MulticoreParam}

+\alias{register}

+

+\title{Classes to configure parallel evaluation}

+

+\description{

+

+  Configure and register parallel calculations, e.g., for

+  \code{\link{Team}} evaluation.

+

+}

+

+\usage{

+MulticoreParam(size = getOption("mc.cores", 2L),

+    mc.set.seed = TRUE, ...)

+register(param)

+}

+

+\arguments{

+

+  \item{size}{The number of members in the parallel cluster.}

+

+  \item{mc.set.seed}{\code{logical(1)}; see \code{?mcparallel} on unix

+    platforms.}

+

+  \item{param}{A \code{ParallelParam} instance, such as generated by

+    \code{MulticoreParam()}.}

+

+  \item{...}{Additional arguments, e.g., \code{verbose}, passed to the

+  \code{Streamer} class.}

+

+}

+

+\section{Constructors}{

+  Use \code{MulticoreParam} to construct instances of this class.

+}

+

+\section{Methods}{

+  \describe{

+

+    \item{register}{Invoked with an argument \code{param} stores the

+      \code{param} for use in subsequent parallel computation. Use

+      \code{NULL} to clear the register. The function returns,

+      invisibly, the previously registered parameter instance, if any.}

+

+  }

+

+}

+

+\section{Internal Class Fields and Methods}{

+

+  Internal fields of this class are are described with, e.g.,

+  \code{getRefClass("MulticoreParam")$fields}.

+

+  Internal methods of this class are described with

+  \code{getRefClass("MulticoreParam")$methods()} and

+  \code{getRefClass("MulticoreParam")$help()}.

+

+}

+

+\author{Martin Morgan \url{mtmorgan@fhcrc.org}}

+

+\seealso{

+

+  \code{\link{Team}} to apply one function in parallel,

+  \code{\link{TConnector}} to apply different functions to all elements

+  of a stream.

+

+}

+

+\examples{

+if (.Platform$OS.type != "windows") {

+    oparam <- register()       ## previous setting

+    param <- MulticoreParam()  ## default multicore settings

+    register(param)            ## register for future use, e.g,. Team

+    register(oparam)           ## reset original

+}

+}

+

+\keyword{classes}

@@ -2,7 +2,7 @@

 \alias{Streamer-package}

 \alias{Streamer}

 \docType{package}

-\title{Enable stream processing of large data}

+\title{Package to enable stream (iterative) processing of large data}

 \description{

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

@@ -2,9 +2,14 @@

 \Rdversion{1.1}

 \docType{class}

 \alias{Team-class}

+\alias{MulticoreTeam-class}

+% *Team-methods

 \alias{Team}

+\alias{Team,missing-method}

+\alias{Team,MulticoreParam-method}

-\title{Consumer class to enable parallel evaluation}

+

+\title{Consumer classes to enable parallel evaluation}

 \description{

@@ -13,7 +18,10 @@

 }

-\usage{Team(FUN, size=1L, mc.set.seed=TRUE, silent=TRUE, ...)}

+\usage{

+\S4method{Team}{missing}(FUN, ..., param)

+\S4method{Team}{MulticoreParam}(FUN, ..., param)

+}

 \arguments{

@@ -21,21 +29,23 @@

     consumer), to be applied to each element of the stream. The return

     value of the function is the value yield'ed.}

-  \item{size}{The number of members in the team. If the upstream

-    processing time is u unit of time, and \code{FUN} takes v units of

-    time, then an efficient size is v / u or the number of cores -1,

-    whichever is smaller.}

-

-  \item{mc.set.seed, silent}{\code{logical(1)}; see

-    \code{?mcparallel} on unix platforms.}

-

   \item{...}{Additional arguments (e.g., \code{verbose}, passed to the

     \code{\linkS4class{Consumer}} constructor.}

+  \item{param}{A \code{ParallelParam} instance, such as generated by

+    \code{MulticoreParam()}.}

+

 }

 \section{Constructors}{

-  Use \code{Team} to contruct instances of this class.

+  Use \code{Team} to construct instances of this class.

+

+  When \code{param} is missing, \code{Team} consults the registry (see

+  \code{\link{register}}) for a parallel parameter class. If none is

+  found and \code{.Platform$OS.type == "unix"}, a default

+  \code{\link{MulticoreParam}} instance is used. An error is signaled on

+  other operating systems (i.e., Windows)

+

 }

 \section{Methods}{See \code{\link{Consumer}} Methods.}

@@ -43,11 +53,11 @@

 \section{Internal Class Fields and Methods}{

   Internal fields of this class are are described with, e.g.,

-  \code{getRefClass("Team")$fields}.

+  \code{getRefClass("MulticoreTeam")$fields}.

   Internal methods of this class are described with

-  \code{getRefClass("Team")$methods()} and

-  \code{getRefClass("Team")$help()}.

+  \code{getRefClass("MulticoreTeam")$methods()} and

+  \code{getRefClass("MulticoreTeam")$help()}.

 }

@@ -55,18 +65,20 @@

 \seealso{

-  \code{\link{TConnector}} to apply \emph{different} functions to all

-  elements of the team.

+  \code{\link{ParallelParam}} for configuring parallel

+  environments. \code{\link{TConnector}} to apply \emph{different}

+  functions to all elements of the team.

 }

 \examples{

 if (.Platform$OS.type != "windows") {

-    team <- Team(function(x) { Sys.sleep(1); mean(x) }, size=5)

+    param <- MulticoreParam(size=5)

+    team <- Team(function(x) { Sys.sleep(1); mean(x) }, param=param)

     s <- stream(Seq(to=50, yieldSize=5), team)

     system.time({while(length(y <- yield(s)))

         print(y)

-    })

+    })  ## about 2 seconds

 }

 }

@@ -5,7 +5,7 @@

 \alias{stream,Consumer-method}

 \alias{stream,Producer-method}

-\title{Function to create a Stream from a Producer and Consumers}

+\title{Function to create a Stream from a Producer and zero or more Consumers}

 \description{

@@ -4,7 +4,7 @@

 \alias{yield-methods}

 \alias{yield,Streamer-method}

-\title{Function to yield one chunk of data}

+\title{Function to yield one task from a Stream or Producer}

 \description{