R/clonly.R

In bamonroe/ctools: Unified Syntax for Parallel Operations in R

# Functions that only have use with cluster operations

# Repetitive function used in c.export
list.agg <- function(obj,to.export) {

		for (i in 1:length(obj)) {
			# Check for character vectors and existance
			is.char <- ifelse(is.character(obj[[i]]), T, F)
			it.exists <- ifelse( is.char == T && exists(obj[[i]]), T, F )
			if ( !is.char ) 	 stop(paste("Argument", i, "is not a string"))
			if ( !it.exists ) stop(paste("Argument", i, "does not refer to an existing object"))

			# Add to the export list
			te.len <- length(to.export) + 1
			to.export[[te.len]] <- obj[[i]]
		}

		return(to.export)

}

#' Export function for cluster parallel
#'
#' Wrapper function for parExport() and Rhpc_Export().
#' Note: This is effectively necessary for PSOCKS and MPI portability.
#' @param ... string objects containing variables to export. 
#' @param push logical of whether or not to push the export list, defaults to TRUE
#' @param clear logical of whether or not to erase the export list and add the included options to it, defaults to FALSE
#' @export c.export
#' @examples
#' vec <- 1:10
#' c.export("vec", push=FALSE)    # adds object 'vec' to list of things to export, but doesn't export it
#' c.export(push=T)               # export all objects in the to.export list
#' c.export("vec")                # add "vec" to the to.export list, then export all items in list
c.export <- function(..., push=TRUE, clear=FALSE) {

	if (! get("use.cluster", envir = cluster)) {
		return(not.useful("c.export"))
	}

	obj <- list(...)

	to.export <- get("export.list", envir = cluster)
	exporter  <- get("cluster.exporter", envir = cluster)

	if (clear == TRUE) {
		to.export <- list()
	}

	if (length(obj) > 0) {
		to.export <- list.agg(obj, to.export)
		assign("export.list", to.export, envir = cluster)
	}

	if (push == TRUE) {
		# Use the exporter
		exporter(get("cluster.object", envir = cluster), to.export)
		# Flush the list after all its elements have been exported
		to.export <- list()
		assign("export.list", to.export, envir = cluster)
	}

}

#' Evaluate expression on all cluster nodes
#'
#' Wrapper function for parEvalQ() and Rhpc_EvalQ().
#' @param expr an expression to be evaluate on every cluster node
#' @export c.eval
#' @examples
#' c.eval(vec <- 1:10)		# evaluate the assignment of sequence 1:10 to object vec
c.eval <- function(expr, envir = parent.frame()) {

	if (get("use.cluster", envir = cluster)) {
		cl.eval <- get("cluster.eval", envir = cluster)
		cl <- get("cluster.object", envir = cluster)
		cl.eval(cl = cl, expr = expr)
	} else{
		eval(expr = expr, envir = envir)
	}
}