@@ -1,12 +1,14 @@

-setGeneric("stream",

-    function(x, ..., verbose=FALSE) standardGeneric("stream"),

-    signature="x")

-

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

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

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

+setGeneric("Stream",

+    function(x, ..., verbose=FALSE) standardGeneric("Stream"),

+    signature="x")

+

+setGeneric("DAGParam", function(x, ...) standardGeneric("DAGParam"))

+

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

            signature = "param")

new file mode 100644

@@ -0,0 +1,13 @@

+.DAGParam <- setRefClass("DAGParam",

+    fields = list( dag = "graphNEL" ),

+    contains = "Streamer",

+    methods = list(

+      nodes = function() graph::nodes(dag),

+      edges = function() graph::edges(dag),

+      inEdges = function(node) graph::inEdges(dag)[[node]],

+      show = function() {

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

+          cat("DAG nodes:", numNodes(dag), "edges:", numEdges(dag),

+              "\n")

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

+      }))

new file mode 100644

@@ -0,0 +1,31 @@

+setMethod(DAGParam, "missing",

+    function(x, ...)

+{

+    .DAGParam$new(...)

+})

+

+setMethod(DAGParam, "graphNEL",

+    function(x, ...)

+{

+    topology <- suppressWarnings(tsort(x))

+    if (nodes(x) > 1L && !length(topology))

+        stop("'x' is not a directed acyclic graph")

+    .DAGParam$new(dag=x, ...)

+})

+

+setMethod(DAGParam, "matrix",

+    function(x, W = NULL, V = NULL, ...)

+{

+    dag <- ftM2graphNEL(x, W=W, V=V, edgemode="directed")

+    DAGParam(dag, ...)

+})

+

+setMethod(DAGParam, "data.frame",

+    function(x, ...)

+{

+    nms <- c("From", "To", "W", "V")

+    if (!all(c("From", "To") %in% names(x)))

+        stop("'x' must have columns 'From', 'To'")

+    ft <- cbind(x$From, x$To)

+    DAGParam(ft, W=x$W, V=x$V, ...)

+})

new file mode 100644

@@ -0,0 +1,41 @@

+.DAGTeam <- setRefClass("DAGTeam",

+    fields = list(

+      tbuf = "list",                    # results, in topological order

+      dagParam = "DAGParam",

+      teamParam = "ParallelParam"),

+    contains = "Consumer",

+    methods = list(

+      reset = function() {

+          callSuper()

+          for (elt in tbuf)

+              elt$reset()

+      },

+      yield = function() {

+          values <- setNames(vector("list", length(tbuf)), names(tbuf))

+          ## input value

+          args <- callSuper()

+          values[[1]] <- do.call(tbuf[[1]]$FUN, list(args))

+          ## subsequent values

+          for (nm in names(values)[-1]) {

+              args <-values[dagParam$inEdges(nm)]

+              values[[nm]] <- do.call(tbuf[[nm]]$FUN, args)

+          }

+          values[[length(values)]]

+      }))

+

+DAGTeam <-

+    function(..., dagParam=DAGParam(), teamParam=MulticoreParam(1L))

+{

+    consumers <- list(...)

+    if (!setequal(names(consumers), dagParam$nodes()))

+        stop("'names(...)' and 'dagParam' nodes not equal")

+    ok <- sapply(consumers, is, "FunctionConsumer")

+    if (!all(ok))

+        stop("'...' must be 'FunctionConsumer' instances, failing: '",

+             paste(consumers[!ok], collapse="', '"), "'")

+

+    ## re-order consumers into topological order

+    tbuf <- consumers[tsort(dagParam$dag)]

+

+    .DAGTeam$new(tbuf = tbuf, dagParam=dagParam, teamParam=teamParam)

+}

new file mode 100644

@@ -0,0 +1,12 @@

+plot.DAGParam <-

+    function(x, y, ...)

+{

+    require(Rgraphviz)

+    plot(x$dag, ...)

+}

+

+plot.DAGTeam <-

+    function(x, y, ...)

+{

+    plot(x$dagParam, ...)

+}

new file mode 100644

@@ -0,0 +1,19 @@

+test_DAGTeam_serial <- function()

+{

+    df <- data.frame(From = c("A", "A", "B", "C"),

+                     To   = c("B", "C", "D", "D"),

+                     stringsAsFactors=FALSE)

+    dagParam <- DAGParam(df)

+    dteam <- DAGTeam(A=FunctionConsumer(function(y) y),

+                     B=FunctionConsumer(function(A) -A),

+                     C=FunctionConsumer(function(A) 1 / A),

+                     D=FunctionConsumer(function(B, C) B + C),

+                     dagParam=dagParam)

+    strm <- Stream(Seq(to=10), dteam)

+    

+    a <- 1:10

+    checkIdentical(-a + 1 / a, sapply(strm, c))

+    checkIdentical(numeric(0), yield(strm))

+    reset(strm)

+    checkIdentical(-a + 1 / a, sapply(strm, c))

+}

new file mode 100644

@@ -0,0 +1,112 @@

+\name{DAGTeam-class}

+\Rdversion{1.1}

+\docType{class}

+\alias{DAGTeam-class}

+\alias{DAGParam-class}

+% *Team-methods

+\alias{DAGTeam}

+\alias{DAGParam}

+\alias{DAGParam,missing-method}

+\alias{DAGParam,graphNEL-method}

+\alias{DAGParam,matrix-method}

+\alias{DAGParam,data.frame-method}

+\alias{plot.DAGParam}

+\alias{plot.DAGTeam}

+

+\title{Consumer classes for directed acyclic graph evaluation}

+

+\description{

+

+  A \code{\linkS4class{Consumer}} to route incoming tasks through nodes

+  connected as a directed acyclic graph.

+

+}

+

+\usage{

+DAGParam(x, ...)

+

+DAGTeam(..., dagParam = DAGParam(), teamParam = MulticoreParam(1L))

+

+\S3method{plot}{DAGTeam}(x, y, ...)

+}

+

+\arguments{

+

+  \item{x}{A matrix or data.frame with columns \sQuote{From},

+    \sQuote{To}, or a \code{graphNEL} object (from the graph package)

+    describing a directed acyclic graph.}

+

+  \item{...}{For \code{DAGTeam}, named \code{\link{FunctionConsumer}}

+    instances, one for each node in the graph. The

+    \code{FunctionConsumer} corresponding to the first node in the graph

+    must accept one argument; remaining \code{FunctionConsumer}

+    instances must have as input arguments the names of the nodes from

+    which the inputs derive, as in the example below.

+

+    For \code{DAGParam} when \code{x} is a data.frame or matrix,

+    data.frame columns \code{W}, \code{V} or additional arguments

+    \code{W}, \code{V} as described in \code{\link{ftM2graphNEL}}.

+

+  }

+

+  \item{dagParam}{A \code{DAGParam} instance, with all nodes referenced

+    in the graph represented by \code{\link{FunctionConsumer}} instances

+    in \code{...}.}

+

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

+    \code{MulticoreParam()}. Currently ignored (all calculations are

+    performed on a single thread).}

+

+  \item{y}{Unused.}

+

+}

+

+\section{Constructors}{

+

+  Use \code{DAGParam} and \code{DAGTeam} to construct instances of these

+  classes, with \code{ParallelParam} instances created by, e.g.,

+  \code{MulticoreParam}.

+

+}

+

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

+

+\section{Internal Class Fields and Methods}{

+

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

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

+

+  Internal methods of this class are described with

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

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

+

+}

+

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

+

+\seealso{

+

+  \code{\link{Team}} applies a single function across multiple threads..

+

+}

+

+\examples{

+df <- data.frame(From = c("A", "A", "B", "C"),

+                 To   = c("B", "C", "D", "D"),

+                 stringsAsFactors=FALSE)

+dagParam <- DAGParam(df)

+dteam <- DAGTeam(A=FunctionConsumer(function(y) y),

+                 B=FunctionConsumer(function(A) -A),

+                 C=FunctionConsumer(function(A) 1 / A),

+                 D=FunctionConsumer(function(B, C) B + C),

+                 dagParam=dagParam)

+

+plot(dteam)

+

+strm <- Stream(Seq(to=10), dteam)

+sapply(strm, c)

+reset(strm)

+

+}

+

+\keyword{classes}

@@ -70,8 +70,8 @@ register(param)

 \seealso{

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

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

-  of a stream.

+  \code{\link{DAGTeam}} to evaluate functions whose dependencies are

+  represented as directed acyclic graphs.

 }