R/use_options.R

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

use.MPI <- function(cores = (Rhpc::Rhpc_mpi_universe_size() - 1)) {

	# Rhpc requires initialization and worker number setting to be separate processes
	Rhpc::Rhpc_initialize()

	# Rhcp requires the total worker number, not the total CPU number
	cores <- min(max(round(cores), 1), Rhpc::Rhpc_mpi_universe_size() - 1)
	assign("cores", cores, envir = cluster)
	assign("maxcores", cores, envir = cluster)

	assign("use.cluster",      TRUE,                   envir = cluster)
	assign("cluster.exporter", Rhpc::Rhpc_Export,      envir = cluster)
	assign("cluster.apply",    Rhpc::Rhpc_apply,       envir = cluster)
	assign("cluster.applyLB",  Rhpc::Rhpc_apply,       envir = cluster)		# There is no LB for Rhpc_apply
	assign("cluster.lapply",   Rhpc::Rhpc_lapply,      envir = cluster)
	assign("cluster.lapplyLB", Rhpc::Rhpc_lapplyLB,    envir = cluster)
	assign("cluster.sapply",   Rhpc::Rhpc_sapply,      envir = cluster)
	assign("cluster.sapplyLB", Rhpc::Rhpc_sapplyLB,    envir = cluster)
	assign("cluster.done",     Rhpc::Rhpc_finalize,    envir = cluster)
	assign("cluster.eval",     Rhpc::Rhpc_EvalQ,       envir = cluster)
	assign("cluster.call",     Rhpc::Rhpc_worker_call, envir = cluster)
	assign("cluster.seed",     Rhpc::Rhpc_setupRNG,    envir = cluster)

	cluster.make <- Rhpc::Rhpc_getHandle
	assign("cluster.make", cluster.make, envir = cluster)

	cluster.common(cluster.make,cores)

	# Determine which MPI workers are on which hosts
	h.names <- c.call(system2,"hostname",stdout=T)
	h.all <- unique(h.names)
	hosts <- list()

	for ( h in 1:length(h.all)) {
		hosts[[h.all[[h]]]] <- NULL
	}
	for (h in 1:length(h.names)) {
		hosts[[h.names[[h]]]] <- append(hosts[[h.names[[h]]]], h)
	}
	assign("host.nodes", hosts, envir = cluster)

	message(paste("Using MPI:", cores, "cores"), appendLF = T)

}

use.PSOCKS <- function(cores = parallel::detectCores()) {

	cores <- min(max(round(cores), 1), parallel::detectCores())
	assign("cores", cores, envir = cluster)
	assign("maxcores", cores, envir = cluster)

	assign("use.cluster",      TRUE,                    envir = cluster)
	assign("cluster.exporter", parallel::clusterExport, envir = cluster)
	assign("cluster.apply",    parallel::clusterApply,      envir = cluster)
	assign("cluster.applyLB",  parallel::clusterApplyLB,    envir = cluster)
	assign("cluster.lapply",   parallel::parLapply,     envir = cluster)
	assign("cluster.lapplyLB", parallel::parLapplyLB,   envir = cluster)
	assign("cluster.sapply",   parallel::parSapply,     envir = cluster)
	assign("cluster.sapplyLB", parallel::parSapplyLB,   envir = cluster)
	assign("cluster.done",     parallel::stopCluster,   envir = cluster)
	assign("cluster.eval",     parallel::clusterEvalQ,  envir = cluster)
	assign("cluster.call",     parallel::clusterCall,   envir = cluster)

	cluster.make <- parallel::makePSOCKcluster
	assign("cluster.make", cluster.make , envir = cluster)

	cluster.common(cluster.make,cores)

	message(paste("Using PSOCKS:", get("cores",envir=cluster), "cores"), appendLF = T)

}

# Forking is easy
use.FORK <- function(cores = parallel::detectCores()) {

	cores <- min(max(round(cores), 1), parallel::detectCores())
	assign("cores", cores, envir = cluster)
	assign("maxcores", cores, envir = cluster)

	assign("source.list", list(), envir = cluster)
	assign("sourceCpp.list", list(), envir = cluster)

	assign("use.cluster", FALSE , envir = cluster)

	all.common()

	message(paste("Using FORK:", get("cores",envir=cluster), "cores"), appendLF = T)

}

# Some functionality is common to MPI and PSOCKS clusters
cluster.common <- function(cluster.make,cores) {

	assign("cluster.object", do.call(cluster.make, list(cores)), envir = cluster)

	assign("export.list", list(), envir = cluster)

	all.common()

}

# Some functionality is common to all types
all.common <- function() {

	assign("source.list", list(), envir = cluster)
	assign("sourceCpp.list", list(), envir = cluster)

}

# Some functionality is only useful for MPI and PSOCKS clusters
not.useful <- function(fun) {
	message(paste0("'", fun, "' function only useful when using a cluster"), appendLF = T)
	message("include to increase portability of code", appendLF = T)
}