| Title: | Interface (Wrapper) to MPI (Message-Passing Interface) |
|---|---|
| Description: | An interface (wrapper) to MPI. It also provides interactive R manager and worker environment. |
| Authors: | Hao Yu [aut] |
| Maintainer: | Hao Yu <[email protected]> |
| License: | GPL (>= 2) |
| Version: | 0.7-2.1 |
| Built: | 2024-11-23 06:20:38 UTC |
| Source: | CRAN |
lamhosts finds the host name associated with its node number. Can be used
by mpi.spawn.Rslaves to spawn R slaves on selected hosts. This is
a LAM-MPI specific function.
mpi.is.master checks if it is running on master or slaves.
mpi.hostinfo finds an individual host information including rank and
size in a comm.
slave.hostinfo is executed only by master and find all master and slaves
host information in a comm.
lamhosts() mpi.is.master() mpi.hostinfo(comm = 1) slave.hostinfo(comm = 1, short=TRUE)lamhosts() mpi.is.master() mpi.hostinfo(comm = 1) slave.hostinfo(comm = 1, short=TRUE)
comm |
a communicator number |
short |
if true, a short form is printed |
lamhosts returns CPUs nodes numbers with their host names.
mpi.is.master returns TRUE if it is on master and FALSE otherwise.
mpi.hostinfo sends to stdio a host name, rank, size and comm.
slave.hostname sends to stdio a list of host, rank, size, and comm
information for all master and slaves. With short=TRUE and 8 slaves or more,
the first 3 and last 2 slaves are shown.
Hao Yu
mpi.abort makes a “best attempt" to abort all tasks in a comm.
mpi.abort(comm = 1)mpi.abort(comm = 1)
comm |
a communicator number |
1 if success. Otherwise 0.
Hao Yu
Find MPI constants: MPI_ANY_SOURCE, MPI_ANY_TAG, or MPI_PROC_NULL
mpi.any.source() mpi.any.tag() mpi.proc.null()mpi.any.source() mpi.any.tag() mpi.proc.null()
None
These constants are mainly used by
mpi.send, mpi.recv, and
mpi.probe.
Different implementation of MPI may use different
integers for MPI_ANY_SOURCE, MPI_ANY_TAG, and MPI_PROC_NULL. Hence one
should use these functions instead real integers for MPI communications.
Each function returns an integer value.
An array (length <= total number of slaves) is scattered to slaves so that the first
slave calls FUN with arguments x[[1]] and ..., the second one
calls with arguments x[[2]] and ..., and so on. mpi.iapply is a
nonblocking version of mpi.apply so that it will not consume CPU on master node.
mpi.apply(X, FUN, ..., comm=1) mpi.iapply(X, FUN, ..., comm=1, sleep=0.01)mpi.apply(X, FUN, ..., comm=1) mpi.iapply(X, FUN, ..., comm=1, sleep=0.01)
X |
an array |
FUN |
a function |
... |
optional arguments to |
comm |
a communicator number |
sleep |
a sleep interval on master node (in sec) |
A list of the results is returned. Its length is the same as that of x. In
case the call FUN with arguments x[[i]] and ... fails on ith
slave, corresponding error message will be returned in the returning list.
Hao Yu
#Assume that there are at least 5 slaves running #Otherwise run mpi.spawn.Rslaves(nslaves=5) #x=c(10,20) #mpi.apply(x,runif) #meanx=1:5 #mpi.apply(meanx,rnorm,n=2,sd=4)#Assume that there are at least 5 slaves running #Otherwise run mpi.spawn.Rslaves(nslaves=5) #x=c(10,20) #mpi.apply(x,runif) #meanx=1:5 #mpi.apply(meanx,rnorm,n=2,sd=4)
(Load balancing) parallellapply and related functions.
mpi.applyLB(X, FUN, ..., apply.seq=NULL, comm=1) mpi.parApply(X, MARGIN, FUN, ..., job.num = mpi.comm.size(comm)-1, apply.seq=NULL, comm=1) mpi.parLapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1) mpi.parSapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, simplify=TRUE, USE.NAMES = TRUE, comm=1) mpi.parRapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1) mpi.parCapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1) mpi.parReplicate(n, expr, job.num=mpi.comm.size(comm)-1, apply.seq=NULL, simplify = "array", comm=1) mpi.parMM (A, B, job.num=mpi.comm.size(comm)-1, comm=1)mpi.applyLB(X, FUN, ..., apply.seq=NULL, comm=1) mpi.parApply(X, MARGIN, FUN, ..., job.num = mpi.comm.size(comm)-1, apply.seq=NULL, comm=1) mpi.parLapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1) mpi.parSapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, simplify=TRUE, USE.NAMES = TRUE, comm=1) mpi.parRapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1) mpi.parCapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1) mpi.parReplicate(n, expr, job.num=mpi.comm.size(comm)-1, apply.seq=NULL, simplify = "array", comm=1) mpi.parMM (A, B, job.num=mpi.comm.size(comm)-1, comm=1)
X |
an array or matrix. |
MARGIN |
vector specifying the dimensions to use. |
FUN |
a function. |
simplify |
logical or character string; should the result be simplified to a vector, matrix or higher dimensional array if possible? |
USE.NAMES |
logical; if |
n |
number of replications. |
A |
a matrix |
B |
a matrix |
expr |
expression to evaluate repeatedly. |
job.num |
Total job numbers. If job numbers is bigger than total slave numbers (default value), a load balancing approach is used. |
apply.seq |
if reproducing the same computation (simulation) is desirable, set it to the integer vector .mpi.applyLB generated in previous computation (simulation). |
... |
optional arguments to |
comm |
a communicator number |
Unless length of X is no more than total slave numbers (slave.num) and in this case
mpi.applyLB is the same as mpi.apply, mpi.applyLB sends a next job to a
slave who just delivered a finished job. The sequence of slaves who deliver results to master are
saved into .mpi.applyLB. It keeps track which part of results done by which slaves.
.mpi.applyLB can be used to reproduce the same simulation result if the same seed is
used and the argument apply.seq is equal to .mpi.applyLB.
With the default value of argument job.num which is slave.num, mpi.parApply,
mpi.parLapply, mpi.parSapply, mpi.parRapply, mpi.parCapply,
mpi.parSapply, and mpi.parMM are clones of snow's parApply, parLappy,
parSapply, parRapply, parCapply, parSapply, and parMM, respectively. When job.num is
bigger than slave.num, a load balancing approach is used.
When using the argument apply.seq with .mpi.applyLB, be sure all settings are the same
as before, i.e., the same data, job.num, slave.num, and seed. Otherwise a deadlock could occur.
Notice that apply.seq is useful only if job.num is bigger than slave.num.
#Assume that there are some slaves running #mpi.applyLB #x=1:7 #mpi.applyLB(x,rnorm,mean=2,sd=4) #get the same simulation #mpi.remote.exec(set.seed(111)) #mpi.applyLB(x,rnorm,mean=2,sd=4) #mpi.remote.exec(set.seed(111)) #mpi.applyLB(x,rnorm,mean=2,sd=4,apply.seq=.mpi.applyLB) #mpi.parApply #x=1:24 #dim(x)=c(2,3,4) #mpi.parApply(x, MARGIN=c(1,2), FUN=mean,job.num = 5) #mpi.parLapply #mdat <- matrix(c(1,2,3, 7,8,9), nrow = 2, ncol=3, byrow=TRUE, # dimnames = list(c("R.1", "R.2"), c("C.1", "C.2", "C.3"))) #mpi.parLapply(mdat, rnorm) #mpi.parSapply #mpi.parSapply(mdat, rnorm) #mpi.parMM #A=matrix(1:1000^2,ncol=1000) #mpi.parMM(A,A)#Assume that there are some slaves running #mpi.applyLB #x=1:7 #mpi.applyLB(x,rnorm,mean=2,sd=4) #get the same simulation #mpi.remote.exec(set.seed(111)) #mpi.applyLB(x,rnorm,mean=2,sd=4) #mpi.remote.exec(set.seed(111)) #mpi.applyLB(x,rnorm,mean=2,sd=4,apply.seq=.mpi.applyLB) #mpi.parApply #x=1:24 #dim(x)=c(2,3,4) #mpi.parApply(x, MARGIN=c(1,2), FUN=mean,job.num = 5) #mpi.parLapply #mdat <- matrix(c(1,2,3, 7,8,9), nrow = 2, ncol=3, byrow=TRUE, # dimnames = list(c("R.1", "R.2"), c("C.1", "C.2", "C.3"))) #mpi.parLapply(mdat, rnorm) #mpi.parSapply #mpi.parSapply(mdat, rnorm) #mpi.parMM #A=matrix(1:1000^2,ncol=1000) #mpi.parMM(A,A)
mpi.barrier blocks the caller until all members have called it.
mpi.barrier(comm = 1)mpi.barrier(comm = 1)
comm |
a communicator number |
1 if success. Otherwise 0.
Hao Yu
mpi.bcast is a collective call among all members in a comm. It
broadcasts a message from the specified rank to all members.
mpi.bcast(x, type, rank = 0, comm = 1, buffunit=100)mpi.bcast(x, type, rank = 0, comm = 1, buffunit=100)
x |
data to be sent or received. Must be the same type among all members. |
type |
1 for integer, 2 for double, and 3 for character. Others are not supported. |
rank |
the sender. |
comm |
a communicator number. |
buffunit |
a buffer unit number. |
mpi.bcast is a blocking call among all members in a comm, i.e,
all members have to wait until everyone calls it. All members have to
prepare the same type of messages (buffers). Hence it is relatively
difficult to use in R environment since the receivers may not know what
types of data to receive, not mention the length of data. Users should
use various extensions of mpi.bcast in R. They are
mpi.bcast.Robj, mpi.bcast.cmd, and
mpi.bcast.Robj2slave.
When type=5, MPI continuous datatype (double) is defined with unit given by
buffunit. It is used to transfer huge data where a double vector or matrix
is divided into many chunks with unit buffunit. Total
ceiling(length(obj)/buffunit) units are transferred. Due to MPI specification, both
buffunit and total units transferred cannot be over 2^31-1. Notice that the last
chunk may not have full length of data due to rounding. Special care is needed.
mpi.bcast returns the message broadcasted by the sender
(specified by the rank).
mpi.bcast.Robj,
mpi.bcast.cmd,
mpi.bcast.Robj2slave.
mpi.bcast.cmd is an extension of mpi.bcast.
It is mainly used to transmit a command from master to all R slaves
spawned by using slavedaemon.R script.
mpi.bcast.cmd(cmd=NULL, ..., rank = 0, comm = 1, nonblock=FALSE, sleep=0.1)mpi.bcast.cmd(cmd=NULL, ..., rank = 0, comm = 1, nonblock=FALSE, sleep=0.1)
cmd |
a command to be sent from master. |
... |
used as arguments to cmd (function command) for passing their (master) values to R slaves, i.e., if ‘myfun(x)’ will be executed on R slaves with ‘x’ as master variable, use mpi.bcast.cmd(cmd=myfun, x=x). |
rank |
the sender |
comm |
a communicator number |
nonblock |
logical. If TRUE, a nonblock procedure is used on all receivers so that they will consume none or little CPUs while waiting. |
sleep |
a sleep interval, used when nonblock=TRUE. Smaller sleep is, more response receivers are, more CPUs consume |
mpi.bcast.cmd is a collective call. This means all members in a communicator must
execute it at the same time. If slaves are spawned (created) by using slavedaemon.R (Rprofile script),
then they are running mpi.bcast.cmd in infinite loop (idle state). Hence master can execute
mpi.bcast.cmd alone to start computation. On the master, cmd and ... are put together
as a list which is then broadcasted (after serialization) to all slaves (using for loop with mpi.send
and mpi.recv pair). All slaves will return an expression which will be evaluated by either slavedaemon.R,
or by whatever an R script based on slavedaemon.R.
If nonblock=TRUE, then on receiving side, a nonblock procedure is used to check if there is a message. If not, it will sleep for the specied amount and repeat itself.
Please use mpi.remote.exec if you want the executed results returned from R
slaves.
mpi.bcast.cmd returns no value for the sender and an expression of the transmitted command for others.
Be caution to use mpi.bcast.cmd alone by master in the middle of comptuation. Only all slaves in idle
states (waiting instructions from master) can it be used. Othewise it may result miscommunication
with other MPI calls.
Hao Yu
mpi.bcast.Robj and mpi.bcast.Robj2slave are used to move
a general R object around among master and all slaves.
mpi.bcast.Robj(obj = NULL, rank = 0, comm = 1) mpi.bcast.Robj2slave(obj, comm = 1, all = FALSE) mpi.bcast.Rfun2slave(comm = 1) mpi.bcast.data2slave(obj, comm = 1, buffunit = 100)mpi.bcast.Robj(obj = NULL, rank = 0, comm = 1) mpi.bcast.Robj2slave(obj, comm = 1, all = FALSE) mpi.bcast.Rfun2slave(comm = 1) mpi.bcast.data2slave(obj, comm = 1, buffunit = 100)
obj |
an R object to be transmitted from the sender |
rank |
the sender. |
comm |
a communicator number. |
all |
a logical. If TRUE, all R objects on master are transmitted to slaves. |
buffunit |
a buffer unit number. |
mpi.bcast.Robj is an extension of mpi.bcast for
moving a general R object around from a sender to everyone.
mpi.bcast.Robj2slave does an R object transmission from
master to all slaves unless all=TRUE in which case, all master's objects with
the global enviroment are transmitted to all slavers.
mpi.bcast.data2slave transfers data (a double vector or a matrix)
natively without (un)serilization. It should be used with a huge vector or matrix.
It results less memory usage and faster transmission. Notice that data with
missing values (NA) are allowed.
mpi.bcast.Robj returns no value for the sender and the
transmitted one for others. mpi.bcast.Robj2slave returns no value for the master
and the transmitted R object along its name on slaves. mpi.bcast.Rfun2slave
transmits all master's functions to slaves and returns no value. mpi.bcast.data2slave
transmits a double vector or a matrix to slaves and returns no value.
Hao Yu
mpi.cart.coords translates a rank to its Cartesian topology coordinate.
mpi.cart.coords(comm=3, rank, maxdims)mpi.cart.coords(comm=3, rank, maxdims)
comm |
Communicator with Cartesian structure |
rank |
rank of a process within group |
maxdims |
length of vector coord in the calling program |
This function is the rank-to-coordinates translator. It is the inverse map of
mpi.cart.rank. maxdims is at least as big as ndims as
returned by mpi.cartdim.get.
mpi.cart.coords returns an integer array containing the Cartesian
coordinates of specified process.
Alek Hunchak and Hao Yu
#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.cart.coords(3,4,2)#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.cart.coords(3,4,2)
mpi.cart.create creates a Cartesian structure of arbitrary dimension.
mpi.cart.create(commold=1, dims, periods, reorder=FALSE, commcart=3)mpi.cart.create(commold=1, dims, periods, reorder=FALSE, commcart=3)
commold |
Input communicator |
dims |
Integery array of size ndims specifying the number of processes in each dimension |
periods |
Logical array of size ndims specifying whether the grid is periodic or not in each dimension |
reorder |
ranks may be reordered or not |
commcart |
The new communicator to which the Cartesian topology information is attached |
If reorder = false, then the rank of each process in the new group is the same as its rank in the old group. If the total size of the Cartesian grid is smaller than the size of the group of commold, then some processes are returned mpi.comm.null. The call is erroneous if it specifies a grid that is larger than the group size.
mpi.cart.create returns 1 if success and 0 otherwise.
Alek Hunchak and Hao Yu
#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T))#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T))
mpi.cart.get provides the user with information on the Cartesian topology
associated with a comm.
mpi.cart.get(comm=3, maxdims)mpi.cart.get(comm=3, maxdims)
comm |
Communicator with Cartesian structure |
maxdims |
length of vectors dims, periods, and coords in the calling program |
The coords are as given for the rank of the calling process as shown.
mpi.cart.get returns a vector containing information on the Cartesian topology
associated with comm. maxdims must be at least ndims as returned by mpi.cartdim.get.
Alek Hunchak and Hao Yu
mpi.cart.create,mpi.cartdim.get
#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.remote.exec(mpi.cart.get(3,2))#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.remote.exec(mpi.cart.get(3,2))
mpi.cart.rank translates a Cartesian topology coordinate to its rank.
mpi.cart.rank(comm=3, coords)mpi.cart.rank(comm=3, coords)
comm |
Communicator with Cartesian structure |
coords |
Specifies the Cartesian coordinates of a process |
For a process group with a Cartesian topology, this function translates the logical
process coordinates to process ranks as they are used by the point-to-point routines.
It is the inverse map of mpi.cart.coords.
mpi.cart.rank returns the rank of the specified process.
Alek Hunchak and Hao Yu
#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.cart.rank(3,c(1,0))#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.cart.rank(3,c(1,0))
mpi.cart.shift shifts the Cartesian topology in both manners, displacement
and direction.
mpi.cart.shift(comm=3, direction, disp)mpi.cart.shift(comm=3, direction, disp)
comm |
Communicator with Cartesian structure |
direction |
Coordinate dimension of the shift |
disp |
displacement (>0 for upwards or left shift, <0 for downwards or right shift) |
mpi.cart.shift provides neighbor ranks from given direction and displacement.
The direction argument indicates the dimension of the shift. direction=1 means the first dim,
direction=2 means the second dim, etc. disp=1 or -1 provides immediate neighbor ranks and disp=2
or -2 provides neighbor's neighbor ranks. Negative ranks mean out of boundary. They correspond to
mpi.proc.null.
mpi.cart.shift returns a vector containing information regarding the
rank of the source process and rank of the destination process.
Alek Hunchak and Hao Yu
#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.remote.exec(mpi.cart.shift(3,2,1))#get neighbor ranks #mpi.remote.exec(mpi.cart.shift(3,1,1))#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.remote.exec(mpi.cart.shift(3,2,1))#get neighbor ranks #mpi.remote.exec(mpi.cart.shift(3,1,1))
mpi.cartdim.get gets dim information about a Cartesian topology.
mpi.cartdim.get(comm=3)mpi.cartdim.get(comm=3)
comm |
Communicator with Cartesian structure |
Can be used to provide other functions with the correct size of arrays.
mpi.cartdim.get returns the number of dimensions of the Cartesian structure
Alek Hunchak and Hao Yu
#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.cartdim.get(comm=3)#Need at least 9 slaves #mpi.bcast.cmd(mpi.cart.create(1,c(3,3),c(F,T))) #mpi.cart.create(1,c(3,3),c(F,T)) #mpi.cartdim.get(comm=3)
mpi.comm.disconnect disconnects itself from a communicator and then
deallocates the communicator so it points to MPI_COMM_NULL.
mpi.comm.disconnect(comm=1)mpi.comm.disconnect(comm=1)
comm |
a communicator number |
When members associated with a communicator finish jobs or exit, they have to
call mpi.comm.disconnect to release resource if the communicator was
created from an intercommunicator by mpi.intercomm.merge. If
mpi.comm.free is used instead, mpi.finalize called
by slaves may cause undefined impacts on master who wishes to stay.
1 if success. Otherwise 0.
Hao Yu
mpi.comm.free deallocates a communicator so it
points to MPI_COMM_NULL.
mpi.comm.free(comm=1)mpi.comm.free(comm=1)
comm |
a communicator number |
When members associated with a communicator finish jobs or exit, they have to
call mpi.comm.free to release resource so mpi.comm.size
will return 0. If the comm was created from an intercommunicator by
mpi.intercomm.merge, use mpi.comm.disconnect instead.
1 if success. Otherwise 0.
Hao Yu
mpi.comm.get.parent is mainly used by slaves to find the
intercommunicator or the parent who spawns them. The intercommunicator is saved
in the specified comm number.
mpi.comm.remote.size is mainly used by master to find the total number of
slaves spawned.
mpi.comm.test.inter tests if a comm is an intercomm or not.
mpi.comm.get.parent(comm = 2) mpi.comm.remote.size(comm = 2) mpi.comm.test.inter(comm = 2)mpi.comm.get.parent(comm = 2) mpi.comm.remote.size(comm = 2) mpi.comm.test.inter(comm = 2)
comm |
an intercommunicator number. |
mpi.comm.get.parent and mpi.comm.test.inter return 1 if success
and 0 otherwise.
mpi.comm.remote.size returns the total number of members in the remote
group in an intercomm.
Hao Yu
mpi.comm.set.errhandler sets a communicator to MPI_ERRORS_RETURN
instead of
MPI_ERRORS_ARE_FATAL (default) which crashes R on any type of MPI errors.
Almost all MPI API calls return errcodes which can map to specific MPI error
messages. All MPI related error messages come from predefined
MPI_Error_string.
mpi.comm.set.errhandler(comm = 1)mpi.comm.set.errhandler(comm = 1)
comm |
a communicator number |
1 if success. Otherwise 0.
Hao Yu
mpi.comm.c2f converts the comm (a C communicator) and returns an integer that can be
used as the communicator in external FORTRAN code. mpi.comm.dup duplicates
(copies) a comm to a new comm. mpi.comm.rank
returns its rank in a comm. mpi.comm.size returns
the total number of members in a comm.
mpi.comm.c2f(comm=1) mpi.comm.dup(comm, newcomm) mpi.comm.rank(comm = 1) mpi.comm.size(comm = 1)mpi.comm.c2f(comm=1) mpi.comm.dup(comm, newcomm) mpi.comm.rank(comm = 1) mpi.comm.size(comm = 1)
comm |
a communicator number |
newcomm |
a new communicator number |
Hao Yu
#Assume that there are some slaves running #mpi.comm.size(comm=1) #mpi.comm.size(comm=0) #mpi.remote.exec(mpi.comm.rank(comm=1)) #mpi.remote.exec(mpi.comm.rank(comm=0)) #mpi.remote.exec(mpi.comm.size(comm=1)) #mpi.remote.exec(mpi.comm.size(comm=0)) #mpi.bcast.cmd(mpi.comm.dup(comm=1,newcomm=5)) #mpi.comm.dup(comm=1,newcomm=5)#Assume that there are some slaves running #mpi.comm.size(comm=1) #mpi.comm.size(comm=0) #mpi.remote.exec(mpi.comm.rank(comm=1)) #mpi.remote.exec(mpi.comm.rank(comm=0)) #mpi.remote.exec(mpi.comm.size(comm=1)) #mpi.remote.exec(mpi.comm.size(comm=0)) #mpi.bcast.cmd(mpi.comm.dup(comm=1,newcomm=5)) #mpi.comm.dup(comm=1,newcomm=5)
mpi.comm.spawn tries to start nslaves identical copies of
slaves, establishing communication with them and returning an
intercommunicator. The spawned slaves are referred to as children, and the
process that spawned them is called the parent (master). The children have
their own MPI_COMM_WORLD represented by comm 0. To make communication
possible among master and slaves, all slaves should use
mpi.comm.get.parent to find their parent and use
mpi.intercomm.merge to merger an intercomm to a comm.
mpi.comm.spawn(slave, slavearg = character(0), nslaves = mpi.universe.size(), info = 0, root = 0, intercomm = 2, quiet = FALSE)mpi.comm.spawn(slave, slavearg = character(0), nslaves = mpi.universe.size(), info = 0, root = 0, intercomm = 2, quiet = FALSE)
slave |
a file name to an executable program. |
slavearg |
an argument list (a char vector) to slave. |
nslaves |
number of slaves to be spawned. |
info |
an info number. |
root |
the root member who spawns slaves. |
intercomm |
an intercomm number. |
quiet |
a logical. If TRUE, do not print anything unless an error occurs. |
Unless quiet = TRUE, a message is printed to indicate how many slaves are successfully
spawned and how many failed.
Hao Yu
mpi.comm.get.parent,
mpi.intercomm.merge.
mpi.dims.create Create a Cartesian dimension used by mpi.cart.create.
mpi.dims.create(nnodes, ndims, dims=integer(ndims))mpi.dims.create(nnodes, ndims, dims=integer(ndims))
nnodes |
Number of nodes in a cluster |
ndims |
Number of dimension in a Cartesian topology |
dims |
Initial dimension numbers |
The entries in the return value are set to describe a Cartesian grid with
ndims dimensions and a total of nnodes nodes. The dimensions are set
to be as close to each other as possible, using an appropriate divisibility
algorithm. The return value can be constrained by specifying positive number(s) in
dims. Only those 0 values in dims are modified by
mpi.dims.create.
mpi.dims.create returns the dimension vector used by
that in mpi.cart.create.
Hao Yu
#What is the dim numbers of 2 dim Cartersian topology under a grid of 36 nodes #mpi.dims.create(36,2) #return c(6,6) #Constrained dim numbers #mpi.dims.create(12,2,c(0,4)) #return c(9,4)#What is the dim numbers of 2 dim Cartersian topology under a grid of 36 nodes #mpi.dims.create(36,2) #return c(6,6) #Constrained dim numbers #mpi.dims.create(12,2,c(0,4)) #return c(9,4)
mpi.exit terminates MPI execution environment and detaches the
library Rmpi. After that, you can still work on R.
mpi.quit terminates MPI execution environment and quits R.
mpi.exit() mpi.quit(save = "no")mpi.exit() mpi.quit(save = "no")
save |
the same argument as |
Normally, mpi.finalize is used to clean all MPI states.
However, it will not detach the library Rmpi. To be more safe leaving MPI,
mpi.exit not only calls mpi.finalize but also detaches the
library Rmpi. This will make reload the library Rmpi impossible.
If leaving MPI and R altogether, one simply uses mpi.quit.
mpi.exit always returns 1
Hao Yu
Terminates MPI execution environment.
mpi.finalize()mpi.finalize()
None
This routines must be called by each slave (master) before it exits. This
call cleans all MPI state. Once mpi.finalize has been called, no MPI
routine may be called. To be more safe leaving MPI, please use
mpi.exit which not only calls mpi.finalize but also
detaches the library Rmpi. This will make reload the library Rmpi impossible.
Always return 1
Hao Yu
mpi.gather and mpi.gatherv (vector variant) gather each
member's message to the member specified by the argument root.
The root member receives the messages and stores them in rank
order. mpi.allgather and mpi.allgatherv are the same as
mpi.gather and mpi.gatherv except that all members receive
the result instead of just the root.
mpi.gather(x, type, rdata, root = 0, comm = 1) mpi.gatherv(x, type, rdata, rcounts, root = 0, comm = 1) mpi.allgather(x, type, rdata, comm = 1) mpi.allgatherv(x, type, rdata, rcounts, comm = 1)mpi.gather(x, type, rdata, root = 0, comm = 1) mpi.gatherv(x, type, rdata, rcounts, root = 0, comm = 1) mpi.allgather(x, type, rdata, comm = 1) mpi.allgatherv(x, type, rdata, rcounts, comm = 1)
x |
data to be gathered. Must be the same type. |
type |
1 for integer, 2 for double, and 3 for character. Others are not supported. |
rdata |
the receive buffer. Must be the same type as the sender and big enough to include all message gathered. |
rcounts |
int vector specifying the length of each message. |
root |
rank of the receiver |
comm |
a communicator number |
For mpi.gather and mpi.allgather, the message to be gathered
must be the same dim and the same type. The receive buffer can be prepared as
either integer(size * dim) or double(size * dim), where size is the total
number of members in a comm. For mpi.gatherv and mpi.allgatherv,
the message to be gathered can have different dims but must be the same type.
The argument rcounts records these different dims into an integer vector
in rank order. Then the receive buffer can be prepared as either
integer(sum(rcounts)) or double(sum(rcounts)).
For mpi.gather or mpi.gatherv, it returns the gathered
message for the root member. For other members, it returns what is in rdata,
i.e., rdata (or rcounts) is ignored. For mpi.allgather or
mpi.allgatherv, it returns the gathered message for all members.
Hao Yu
#Need 3 slaves to run properly #Or use mpi.spawn.Rslaves(nslaves=3) #mpi.bcast.cmd(id <-mpi.comm.rank(.comm), comm=1) #mpi.bcast.cmd(mpi.gather(letters[id],type=3,rdata=string(1))) #mpi.gather(letters[10],type=3,rdata=string(4)) # mpi.bcast.cmd(x<-rnorm(id)) # mpi.bcast.cmd(mpi.gatherv(x,type=2,rdata=double(1),rcounts=1)) # mpi.gatherv(double(1),type=2,rdata=double(sum(1:3)+1),rcounts=c(1,1:3)) #mpi.bcast.cmd(out1<-mpi.allgatherv(x,type=2,rdata=double(sum(1:3)+1), # rcounts=c(1,1:3))) #mpi.allgatherv(double(1),type=2,rdata=double(sum(1:3)+1),rcounts=c(1,1:3))#Need 3 slaves to run properly #Or use mpi.spawn.Rslaves(nslaves=3) #mpi.bcast.cmd(id <-mpi.comm.rank(.comm), comm=1) #mpi.bcast.cmd(mpi.gather(letters[id],type=3,rdata=string(1))) #mpi.gather(letters[10],type=3,rdata=string(4)) # mpi.bcast.cmd(x<-rnorm(id)) # mpi.bcast.cmd(mpi.gatherv(x,type=2,rdata=double(1),rcounts=1)) # mpi.gatherv(double(1),type=2,rdata=double(sum(1:3)+1),rcounts=c(1,1:3)) #mpi.bcast.cmd(out1<-mpi.allgatherv(x,type=2,rdata=double(sum(1:3)+1), # rcounts=c(1,1:3))) #mpi.allgatherv(double(1),type=2,rdata=double(sum(1:3)+1),rcounts=c(1,1:3))
mpi.gather.Robj gathers each member's object to the member
specified by the argument root.
The root member receives the objects as a list.
mpi.allgather.Robj is the same as mpi.gather.Robj
except that all members receive the result instead of just the root.
mpi.gather.Robj(obj=NULL, root = 0, comm = 1, ...) mpi.allgather.Robj(obj=NULL, comm = 1)mpi.gather.Robj(obj=NULL, root = 0, comm = 1, ...) mpi.allgather.Robj(obj=NULL, comm = 1)
obj |
data to be gathered. Could be different type. |
root |
rank of the gather |
comm |
a communicator number |
... |
optional arugments to |
Since sapply is used to gather all results, its default option "simplify=TRUE" is to simplify outputs. In some situations, this option is not desirable. Using "simplify=FALSE" as in the place of ... will tell sapply not to simplify and a list of outputs will be returned.
For mpi.gather.Robj, it returns a list, the gathered message
for the root member. For
mpi.allgatherv.Robj, it returns a list, the gathered message
for all members.
Hao Yu and Wei Xia
#Assume that there are some slaves running #mpi.bcast.cmd(id<-mpi.comm.rank()) #mpi.bcast.cmd(x<-rnorm(id)) #mpi.bcast.cmd(mpi.gather.Robj(x)) #x<-"test mpi.gather.Robj" #mpi.gather.Robj(x) #mpi.bcast.cmd(obj<-rnorm(id+10)) #mpi.bcast.cmd(nn<-mpi.allgather.Robj(obj)) #obj<-rnorm(5) #mpi.allgather.Robj(obj) #mpi.remote.exec(nn)#Assume that there are some slaves running #mpi.bcast.cmd(id<-mpi.comm.rank()) #mpi.bcast.cmd(x<-rnorm(id)) #mpi.bcast.cmd(mpi.gather.Robj(x)) #x<-"test mpi.gather.Robj" #mpi.gather.Robj(x) #mpi.bcast.cmd(obj<-rnorm(id+10)) #mpi.bcast.cmd(nn<-mpi.allgather.Robj(obj)) #obj<-rnorm(5) #mpi.allgather.Robj(obj) #mpi.remote.exec(nn)
mpi.get.count finds the length of a received message.
mpi.get.count(type, status = 0)mpi.get.count(type, status = 0)
type |
1 for integer, 2 for double, 3 for char. |
status |
a status number |
When mpi.recv is used to receive a message, the receiver
buffer can be set to be bigger than the incoming message. To find the
exact length of the received message, mpi.get.count is used to
find its exact length. mpi.get.count must be called
immediately after calling mpi.recv otherwise the status may be
changed.
the length of a received message.
Hao Yu
mpi.send, mpi.recv,
mpi.get.sourcetag, mpi.probe.
mpi.get.processor.name returns the host name (a string) where
it is executed.
mpi.get.processor.name(short = TRUE)mpi.get.processor.name(short = TRUE)
short |
a logical. |
a base host name if short = TRUE and a full host name otherwise.
Hao Yu
mpi.get.sourcetag finds the source and tag of a received message.
mpi.get.sourcetag(status = 0)mpi.get.sourcetag(status = 0)
status |
a status number |
When mpi.any.source and/or mpi.any.tag are
used by mpi.recv or mpi.probe, one can use
mpi.get.sourcetag
to find who sends the message or with what a tag number.
mpi.get.sourcetag must be called immediately after calling
mpi.recv or mpi.probe otherwise the obtained information may not
be right.
2 dim int vector. The first integer is the source and the second is the tag.
Hao Yu
mpi.send, mpi.recv, mpi.probe,
mpi.get.count
(Load balancing) parallellapply and related functions.
mpi.iapplyLB(X, FUN, ..., apply.seq=NULL, comm=1, sleep=0.01) mpi.iparApply(X, MARGIN, FUN, ..., job.num = mpi.comm.size(comm)-1, apply.seq=NULL, comm=1, sleep=0.01) mpi.iparLapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1,sleep=0.01) mpi.iparSapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, simplify=TRUE, USE.NAMES = TRUE, comm=1, sleep=0.01) mpi.iparRapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1, sleep=0.01) mpi.iparCapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1,sleep=0.01) mpi.iparReplicate(n, expr, job.num=mpi.comm.size(comm)-1, apply.seq=NULL, simplify = TRUE, comm=1,sleep=0.01) mpi.iparMM(A, B, comm=1, sleep=0.01)mpi.iapplyLB(X, FUN, ..., apply.seq=NULL, comm=1, sleep=0.01) mpi.iparApply(X, MARGIN, FUN, ..., job.num = mpi.comm.size(comm)-1, apply.seq=NULL, comm=1, sleep=0.01) mpi.iparLapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1,sleep=0.01) mpi.iparSapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, simplify=TRUE, USE.NAMES = TRUE, comm=1, sleep=0.01) mpi.iparRapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1, sleep=0.01) mpi.iparCapply(X, FUN, ..., job.num=mpi.comm.size(comm)-1, apply.seq=NULL, comm=1,sleep=0.01) mpi.iparReplicate(n, expr, job.num=mpi.comm.size(comm)-1, apply.seq=NULL, simplify = TRUE, comm=1,sleep=0.01) mpi.iparMM(A, B, comm=1, sleep=0.01)
X |
an array or matrix. |
MARGIN |
vector specifying the dimensions to use. |
FUN |
a function. |
simplify |
logical; should the result be simplified to a vector or matrix if possible? |
USE.NAMES |
logical; if |
n |
number of replications. |
A |
a matrix |
B |
a matrix |
expr |
expression to evaluate repeatedly. |
job.num |
Total job numbers. If job numbers is bigger than total slave numbers (default value), a load balancing approach is used. |
apply.seq |
if reproducing the same computation (simulation) is desirable, set it to the integer vector .mpi.applyLB generated in previous computation (simulation). |
... |
optional arguments to |
comm |
a communicator number |
sleep |
a sleep interval on master node (in sec) |
mpi.iparApply, mpi.iparLapply, mpi.iparSapply, mpi.iparRapply,
mpi.iparCapply, mpi.iparSapply, mi.iparReplicate, and mpi.iparMM
are nonblocking versions of mpi.parApply, mpi.parLapply, mpi.parSapply,
mpi.parRapply, mpi.parCapply, mpi.parSapply, mpi.parReplicate,
and mpi.parMM respectively. The main difference is that mpi.iprobe and
Sys.sleep are used so that master node consumes almost no CPU cycles while waiting for
slaves results. However, due to frequent wake/sleep cycles on master, those functions are not
suitable for running small jobs on slave nodes. If anticipated computing time for each job is
relatively long, e.g., minutes or hours, setting sleep to be 1 second or longer will further
reduce load on master (only slightly).
Many MPI APIs take an info argument for additional information passing. An info is an object which consists of many (key,value) pairs. Rmpi uses an internal memory to store an info object.
mpi.info.create creates a new info object.
mpi.info.free frees an info object and sets it to MPI_INFO_NULL.
mpi.info.get retrieves the value associated with key in an info.
mpi.info.set adds the key and value pair to info.
mpi.info.create(info = 0) mpi.info.free(info = 0) mpi.info.get(info = 0, key, valuelen) mpi.info.set(info = 0, key, value)mpi.info.create(info = 0) mpi.info.free(info = 0) mpi.info.get(info = 0, key, valuelen) mpi.info.set(info = 0, key, value)
info |
an info number. |
key |
a char (length 1). |
valuelen |
the length (nchar) of a key |
value |
a char (length 1). |
mpi.info.create, mpi.info.free, and mpi.info.set return
1 if success and 0 otherwise.
mpi.info.get returns the value (a char) for a given info and valuelen.
Hao Yu
Creates an intracommunicator from an intercommunicator
mpi.intercomm.merge(intercomm=2, high=0, comm=1)mpi.intercomm.merge(intercomm=2, high=0, comm=1)
intercomm |
an intercommunicator number |
high |
Used to order the groups of the two intracommunicators within comm when creating the new communicator |
comm |
a (intra)communicator number |
When master spawns slaves, an intercommunicator is created. To make
communications (point-to-point or groupwise) among master and slaves, an
intracommunicator must be created. mpi.intercomm.merge is used for
that purpose. This is a collective call so all master and slaves call
together. R slaves spawned by mpi.spawn.Rslaves should use
mpi.comm.get.parent to get (set) an intercomm to a number followed
by merging antercomm to an intracomm. One
can use mpi.comm.test.inter to test if a
communicator is an intercommunicator or not.
1 if success. Otherwise 0.
Hao Yu
Carry out parallel Monte Carlo simulation on R slaves spawned by using slavedaemon.R script and all executed results are returned back to master.
mpi.parSim(n=100, rand.gen=rnorm, rand.arg=NULL,statistic, nsim=100, run=1, slaveinfo=FALSE, sim.seq=NULL, simplify=TRUE, comm=1, ...)mpi.parSim(n=100, rand.gen=rnorm, rand.arg=NULL,statistic, nsim=100, run=1, slaveinfo=FALSE, sim.seq=NULL, simplify=TRUE, comm=1, ...)
n |
sample size. |
rand.gen |
the random data generating function. See the details section |
rand.arg |
additional argument list to |
statistic |
the statistic function to be simulated. See the details section |
nsim |
the number of simulation carried on a slave which is counted as one slave job. |
run |
the number of looping. See the details section. |
slaveinfo |
if TRUE, the numbers of jobs finished by slaves will be displayed. |
sim.seq |
if reproducing the same simulation is desirable, set it to the integer vector .mpi.parSim generated in previous simulation. |
simplify |
logical; should the result be simplified to a vector or matrix if possible? |
comm |
a communicator number |
... |
optional arguments to |
It is assumed that one simulation is carried out as
statistic(rand.gen(n)), where rand.gen(n) can return any
values as long as statistic can take them. Additional arguments can
be passed to rand.gen by rand.arg as a list. Optional
arguments can also be passed to statistic by the argument
....
Each slave job consists of replicate(nsim,statistic(rand.gen(n))),
i.e., each job runs nsim number of simulation. The returned values
are transported from slaves to master.
The total number of simulation (TNS) is calculated as follows. Let
slave.num be the total number of slaves in a comm and it is
mpi.comm.size(comm)-1. Then TNS=slave.num*nsim*run and the total
number of slave jobs is slave.num*run, where run is the number of
looping from master perspective. If run=1, each slave will run one slave
job. If run=2, each slave will run two slaves jobs on average, and so on.
The purpose of using run has two folds. It allows a tuneup
of slave job size and total number of slave jobs to deal with two
different cluster environments. On a cluster of slaves with equal CPU
power, run=1 is often enough. But if nsim is too big, one
can set run=2 and the slave jog size to be nsim/2 so that
TNS=slave.num*(nsim/2)*(2*run). This may improve R computation
efficiency slightly. On a cluster of slaves with different CPU power, one
can choose a big value of run and a small value of nsim
so that master can dispatch more jobs to slaves who run faster than
others. This will keep all slaves busy so that load balancing is
achieved.
The sequence of slaves who deliver results to master are saved into
.mpi.parSim. It keeps track which part of results done by which slaves.
.mpi.parSim can be used to reproduce the same simulation result if the same
seed is used and the argument sim.seq is equal to .mpi.parSim.
See the warning section before you use mpi.parSim.
The returned values depend on values returned by replicate
of statistic(rand.gen(n)) and the total number of simulation
(TNS). If statistic returns a single value, then the result is a
vector of length TNS. If statistic returns a vector (list) of
length nrow, then the result is a matrix of dimension
c(nrow, TNS).
It is assumed that a parallel RNG is used on all slaves. Run
mpi.setup.rngstream on the master to set up a parallel RNG. Though mpi.parSim
works without a parallel RNG, the quality of simulation is not guarantied.
mpi.parSim will automatically transfer rand.gen
and statistic to slaves. However, any functions that
rand.gen and statistic reply on but are not on slaves
must be transfered to slaves before using mpi.parSim. You
can use mpi.bcast.Robj2slave for that purpose. The same is
applied to required packages or C/Fortran codes. You can use either
mpi.bcast.cmd or put required(package) and/or
dyn.load(so.lib) into rand.gen and statistic.
If simplify is TRUE, sapply style simplication is applied. Otherwise a list of length
slave.num*run is returned.
Hao Yu
mpi.setup.rngstream
mpi.bcast.cmd
mpi.bcast.Robj2slave
mpi.probe uses the source and tag of incoming message to set a
status. mpi.iprobe does the same except it is a nonblocking call,
i.e., returns immediately.
mpi.probe(source, tag, comm = 1, status = 0) mpi.iprobe(source, tag, comm = 1, status = 0)mpi.probe(source, tag, comm = 1, status = 0) mpi.iprobe(source, tag, comm = 1, status = 0)
source |
the source of incoming message or mpi.any.source() for any source. |
tag |
a tag number or mpi.any.tag() for any tag. |
comm |
a communicator number |
status |
a status number |
When mpi.send or other nonblocking sends are used to send
a message, the receiver may not know the exact length before receiving
it. mpi.probe is used to probe the incoming message and put some
information into a status. Then the exact length can be found by using
mpi.get.count to such a status. If the wild card
mpi.any.source or mpi.any.tag are used, then one
can use mpi.get.sourcetag to find the exact source or tag of
a sender.
mpi.probe returns 1 only after a matching message has been found.
mpi.iproble returns TRUE if there is a message that can be
received; FALSE otherwise.
Hao Yu
mpi.send, mpi.recv,
mpi.get.count
mpi.comm.maxsize, mpi.request.maxsize, and
mpi.status.maxsize find the lengthes of comm, request, and status
arrayes respectively.
mpi.realloc.comm, mpi.realloc.request and
mpi.realloc.status increase the lengthes of comm, request and
status arrayes to newmaxsize respectively if newmaxsize is
bigger than the original maximum size.
mpi.realloc.comm(newmaxsize) mpi.realloc.request(newmaxsize) mpi.realloc.status(newmaxsize) mpi.comm.maxsize() mpi.request.maxsize() mpi.status.maxsize()mpi.realloc.comm(newmaxsize) mpi.realloc.request(newmaxsize) mpi.realloc.status(newmaxsize) mpi.comm.maxsize() mpi.request.maxsize() mpi.status.maxsize()
newmaxsize |
an integer. |
When Rmpi is loaded, Rmpi allocs comm array with size 10, request
array with 10,000 and status array with 5,000. They should be enough in
most cases. They use less than 150KB system memory. In rare case, one can
use mpi.realloc.comm, mpi.realloc.request and
mpi.realloc.status to increase them to bigger arrays.
Hao Yu
mpi.reduce and mpi.allreduce are global reduction operations.
mpi.reduce combines each member's result, using the operation
op, and returns the combined value(s) to the member specified by
the argument dest. mpi.allreduce is the same as
mpi.reduce except that all members receive the combined value(s).
mpi.reduce(x, type=2, op=c("sum","prod","max","min","maxloc","minloc"), dest = 0, comm = 1) mpi.allreduce(x, type=2, op=c("sum","prod","max","min","maxloc","minloc"), comm = 1)mpi.reduce(x, type=2, op=c("sum","prod","max","min","maxloc","minloc"), dest = 0, comm = 1) mpi.allreduce(x, type=2, op=c("sum","prod","max","min","maxloc","minloc"), comm = 1)
x |
data to be reduced. Must be the same dim and the same type for all members. |
type |
1 for integer and 2 for double. Others are not supported. |
op |
one of "sum", "prod", "max", "min", "maxloc", or "minloc". |
dest |
rank of destination |
comm |
a communicator number |
It is important that all members in a comm call either all mpi.reduce
or all mpi.allreduce even though the master may not be in
computation. They must provide exactly the same type and dim
vectors to be reduced. If the operation "maxloc" or "minloc" is used,
the combined vector is twice as long as the original one since the
maximum or minimum ranks are included.
mpi.reduce returns the combined value(s) to the member specified
by dest. mpi.allreduce returns the combined values(s) to
every member in a comm. The combined value(s) may be the summation,
production, maximum, or minimum specified by the argument op. If
the op is either "maxloc" or "minloc", then the maximum (minimum)
value(s) along the maximum (minimum) rank(s) will be returned.
Hao Yu
Remotely execute a command on R slaves spawned by using slavedaemon.R script and return all executed results back to master.
mpi.remote.exec(cmd, ..., simplify = TRUE, comm = 1, ret = TRUE)mpi.remote.exec(cmd, ..., simplify = TRUE, comm = 1, ret = TRUE)
cmd |
the command to be executed on R slaves |
... |
used as arguments to cmd (function command) for passing their (master) values to R slaves, i.e., if ‘myfun(x)’ will be executed on R slaves with ‘x’ as master variable, use mpi.remote.exec(cmd=myfun, x). |
simplify |
logical; should the result be simplified to a data.frame if possible? |
comm |
a communicator number. |
ret |
return executed results from R slaves if TRUE. |
Once R slaves are spawned by mpi.spawn.Rslaves with the
slavedaemon.R script, they are waiting for instructions from master. One can
use mpi.bcast.cmd to send a command to R slaves. However it
will not return executed results. Hence mpi.remote.exec can be
considered an extension to mpi.bcast.cmd.
return executed results from R slaves if the argument ret is
set to be TRUE. The value could be a data.frame if values
(integer or double) from each slave have the same dimension.
Otherwise a list is returned.
mpi.remote.exec may have difficult guessing invisible results
on R slaves. Use ret = FALSE instead.
Hao Yu
mpi.spawn.Rslaves,
mpi.bcast.cmd
#mpi.remote.exec(mpi.comm.rank()) # x=5 #mpi.remote.exec(rnorm,x)#mpi.remote.exec(mpi.comm.rank()) # x=5 #mpi.remote.exec(rnorm,x)
mpi.scatter and mpi.scatterv are the inverse operations of
mpi.gather and mpi.gatherv respectively.
mpi.scatter(x, type, rdata, root = 0, comm = 1) mpi.scatterv(x, scounts, type, rdata, root = 0, comm = 1)mpi.scatter(x, type, rdata, root = 0, comm = 1) mpi.scatterv(x, scounts, type, rdata, root = 0, comm = 1)
x |
data to be scattered. |
type |
1 for integer, 2 for double, and 3 for character. Others are not supported. |
rdata |
the receive buffer. Must be the same type as the sender |
scounts |
int vector specifying the block length inside a message to be scattered to other members. |
root |
rank of the receiver |
comm |
a communicator number |
mpi.scatter scatters the message x to all members. Each member receives
a portion of x with dim as length(x)/size in rank order, where size is the
total number of members in a comm. So the receive buffer can be prepared as
either integer(length(x)/size) or double(length(x)/size). For
mpi.scatterv, scounts counts the portions (different dims) of x sent to
each member. Each member needs to prepare the receive buffer as either
integer(scounts[i]) or double(scounts[i]).
For non-root members, mpi.scatter or scatterv returns the
scattered message and ignores whatever is in x (or scounts). For the root
member, it returns the portion belonging to itself.
Hao Yu
#Need 3 slaves to run properly #Or run mpi.spawn.Rslaves(nslaves=3) # num="123456789abcd" # scounts<-c(2,3,1,7) # mpi.bcast.cmd(strnum<-mpi.scatter(integer(1),type=1,rdata=integer(1),root=0)) # strnum<-mpi.scatter(scounts,type=1,rdata=integer(1),root=0) # mpi.bcast.cmd(ans <- mpi.scatterv(string(1),scounts=0,type=3,rdata=string(strnum), # root=0)) # mpi.scatterv(as.character(num),scounts=scounts,type=3,rdata=string(strnum),root=0) # mpi.remote.exec(ans)#Need 3 slaves to run properly #Or run mpi.spawn.Rslaves(nslaves=3) # num="123456789abcd" # scounts<-c(2,3,1,7) # mpi.bcast.cmd(strnum<-mpi.scatter(integer(1),type=1,rdata=integer(1),root=0)) # strnum<-mpi.scatter(scounts,type=1,rdata=integer(1),root=0) # mpi.bcast.cmd(ans <- mpi.scatterv(string(1),scounts=0,type=3,rdata=string(strnum), # root=0)) # mpi.scatterv(as.character(num),scounts=scounts,type=3,rdata=string(strnum),root=0) # mpi.remote.exec(ans)
mpi.scatter.Robj and mpi.scatter.Robj2slave are used to scatter a list
to all members. They are more efficient than using any parallel apply functions.
mpi.scatter.Robj(obj = NULL, root = 0, comm = 1) mpi.scatter.Robj2slave(obj, comm = 1)mpi.scatter.Robj(obj = NULL, root = 0, comm = 1) mpi.scatter.Robj2slave(obj, comm = 1)
obj |
a list object to be scattered from the root or master |
root |
rank of the scatter. |
comm |
a communicator number. |
mpi.scatter.Robj is an extension of mpi.scatter for
scattering a list object from a sender (root) to everyone. mpi.scatter.Robj2slave
scatters a list to all slaves.
mpi.scatter.Robj for non-root members, returns the
scattered R object. For the root member, it returns the
portion belonging to itself. mpi.scatter.Robj2slave returns no value
for the master and all slaves get their corresponding components in the list,
i.e., the first slave gets the first component in the list.
Hao Yu and Wei Xia
#assume that there are three slaves running #mpi.bcast.cmd(x<-mpi.scatter.Robj()) #xx <- list("master",rnorm(3),letters[2],1:10) #mpi.scatter.Robj(obj=xx) #mpi.remote.exec(x) #scatter a matrix to slaves #dat=matrix(1:24,ncol=3) #splitmatrix = function(x, ncl) lapply(.splitIndices(nrow(x), ncl), function(i) x[i,]) #dat2=splitmatrix(dat,3) #mpi.scatter.Robj2slave(dat2) #mpi.remote.exec(dat2)#assume that there are three slaves running #mpi.bcast.cmd(x<-mpi.scatter.Robj()) #xx <- list("master",rnorm(3),letters[2],1:10) #mpi.scatter.Robj(obj=xx) #mpi.remote.exec(x) #scatter a matrix to slaves #dat=matrix(1:24,ncol=3) #splitmatrix = function(x, ncl) lapply(.splitIndices(nrow(x), ncl), function(i) x[i,]) #dat2=splitmatrix(dat,3) #mpi.scatter.Robj2slave(dat2) #mpi.remote.exec(dat2)
The pair mpi.send and mpi.recv are two most used blocking
calls for point-to-point communications. An int, double or char vector
can be transmitted from any source to any destination.
The pair mpi.isend and mpi.irecv are the same except that
they are nonblocking calls.
Blocking and nonblocking calls are interchangeable, e.g., nonblocking sends can be matched with blocking receives, and vice-versa.
mpi.send(x, type, dest, tag, comm = 1) mpi.isend(x, type, dest, tag, comm = 1, request=0) mpi.recv(x, type, source, tag, comm = 1, status = 0) mpi.irecv(x, type, source, tag, comm = 1, request = 0)mpi.send(x, type, dest, tag, comm = 1) mpi.isend(x, type, dest, tag, comm = 1, request=0) mpi.recv(x, type, source, tag, comm = 1, status = 0) mpi.irecv(x, type, source, tag, comm = 1, request = 0)
x |
data to be sent or received. Must be the same type for source and destination. The receive buffer must be as large as the send buffer. |
type |
1 for integer, 2 for double, and 3 for character. Others are not supported. |
dest |
the destination rank. Use |
source |
the source rank. Use |
tag |
non-negative integer. Use |
comm |
a communicator number. |
request |
a request number. |
status |
a status number. |
The pair mpi.send (or mpi.isend) and mpi.recv
(or mpi.irecv) must be used together, i.e., if there is a sender,
then there must be a receiver. Any mismatch will result a deadlock
situation, i.e., programs stop responding. The receive buffer must be
large enough to contain an incoming message otherwise programs will be
crashed. One can use mpi.probe (or mpi.iprobe) and
mpi.get.count to find the length of an incoming message
before calling mpi.recv. If mpi.any.source or
mpi.any.tag is used in mpi.recv, one can use
mpi.get.sourcetag to find out the source or tag of the
received message. To send/receive an R object rather than an int, double
or char vector, please use the pair mpi.send.Robj and
mpi.recv.Robj.
Since mpi.irecv is a nonblocking call, x with enough buffer
must be created before using it. Then use nonblocking completion calls
such as mpi.wait or mpi.test to test if
x contains data from sender.
If multiple nonblocking sends or receives are used, please use request
number consecutively from 0. For example, to receive two messages from two
slaves, try
mpi.irecv(x,1,source=1,tag=0,comm=1,request=0)
mpi.irecv(y,1,source=2,tag=0,comm=1,request=1)
Then mpi.waitany, mpi.waitsome or mpi.waitall can be
used to complete the operations.
mpi.send and mpi.isend return no value. mpi.recv
returns the int, double or char vector sent from source. However,
mpi.irecv returns no value. See details for explanation.
Hao Yu
mpi.send.Robj,
mpi.recv.Robj,
mpi.probe,
mpi.wait,
mpi.get.count,
mpi.get.sourcetag.
#on a slave #mpi.send(1:10,1,0,0) #on master #x <- integer(10) #mpi.irecv(x,1,1,0) #x #mpi.wait() #x#on a slave #mpi.send(1:10,1,0,0) #on master #x <- integer(10) #mpi.irecv(x,1,1,0) #x #mpi.wait() #x
mpi.send.Robj and mpi.recv.Robj are two
extensions of mpi.send and mpi.recv. They are used to
transmit a general R object from any source to any destination.
mpi.isend.Robj is a nonblocking version of mpi.send.Robj.
mpi.send.Robj(obj, dest, tag, comm = 1) mpi.isend.Robj(obj, dest, tag, comm = 1, request=0) mpi.recv.Robj(source, tag, comm = 1, status = 0)mpi.send.Robj(obj, dest, tag, comm = 1) mpi.isend.Robj(obj, dest, tag, comm = 1, request=0) mpi.recv.Robj(source, tag, comm = 1, status = 0)
obj |
an R object. Can be any R object. |
dest |
the destination rank. |
source |
the source rank or mpi.any.source() for any source. |
tag |
non-negative integer or mpi.any.tag() for any tag. |
comm |
a communicator number. |
request |
a request number. |
status |
a status number. |
mpi.send.Robj and mpi.isend.Robj use
serialize to encode an R object into a binary
char vector. It sends the message to the destination. The receiver
decode the message back into an R object by using
unserialize.
If mpi.isend.Robj is used, mpi.wait or mpi.test must
be used to check the object has been sent.
mpi.send.Robj or mpi.isend.Robj return no value.
mpi.recv.Robj returns the the transmitted R object.
Hao Yu
mpi.send,
mpi.recv,
mpi.wait,
serialize,
unserialize,
mpi.sendrecv and mpi.sendrecv.replace execute blocking send
and receive operations. Both of them combine the sending of one message to
a destination and the receiving of another message from a source in one
call. The source and destination are possibly the same. The send buffer and
receive buffer are disjoint for mpi.sendrecv, while the buffers are
not disjoint for mpi.sendrecv.replace.
mpi.sendrecv(senddata, sendtype, dest, sendtag, recvdata, recvtype, source, recvtag, comm = 1, status = 0) mpi.sendrecv.replace(x, type, dest, sendtag, source, recvtag, comm = 1, status = 0)mpi.sendrecv(senddata, sendtype, dest, sendtag, recvdata, recvtype, source, recvtag, comm = 1, status = 0) mpi.sendrecv.replace(x, type, dest, sendtag, source, recvtag, comm = 1, status = 0)
x |
data to be sent or recieved. Must be the same type for source and destination. |
senddata |
data to be sent. May have different datatypes and lengths |
recvdata |
data to be recieved. May have different datatypes and lengths |
type |
type of the data to be sent or recieved. 1 for integer, 2 for double, and 3 for character. Others are not supported. |
sendtype |
type of the data to be sent. 1 for integer, 2 for double, and 3 for character. Others are not supported. |
recvtype |
type of the data to be recieved. 1 for integer, 2 for double, and 3 for character. Others are not supported. |
dest |
the destination rank. Use |
source |
the source rank. Use |
sendtag |
non-negative integer. Use |
recvtag |
non-negative integer. Use |
comm |
a communicator number. |
status |
a status number. |
The receive buffer must be large enough to contain an incoming message otherwise programs will be crashed. There is compatibility between send-receive and normal sends and receives. A message sent by a send-receive can be received by a regular receive and a send-receive can receive a message sent by a regular send.
Returns the int, double or char vector sent from the send buffers.
Kris Chen
mpi.send.Robj,
mpi.recv.Robj,
mpi.probe.
mpi.get.sourcetag.
#mpi.sendrecv(as.integer(11:20),1,0,33,integer(10),1,0,33,comm=0) #mpi.sendrecv.replace(seq(1,2,by=0.1),2,0,99,0,99,comm=0)#mpi.sendrecv(as.integer(11:20),1,0,33,integer(10),1,0,33,comm=0) #mpi.sendrecv.replace(seq(1,2,by=0.1),2,0,99,0,99,comm=0)
mpi.setup.rngstream setups RNGstream on all slaves.
mpi.setup.rngstream(iseed=NULL, comm = 1)mpi.setup.rngstream(iseed=NULL, comm = 1)
iseed |
An integer to be supplied to |
comm |
A comm number. |
mpi.setup.rngstream can be run only on master node. It can be run later on with the same or
different iseed.
No value returned.
Hao Yu
mpi.spawn.Rslaves spawns R slaves to those hosts automatically
chosen by MPI or specific hosts assigned by the argument hosts.
Those R slaves are running in R BATCH mode with a specific Rscript file.
The default Rscript file "slavedaemon.R" provides interactive R slave
environments.
mpi.close.Rslaves shuts down R slaves spawned by
mpi.spawn.Rslaves.
tailslave.log view (from tail) R slave log files (assuming they are all
in one working directory).
mpi.spawn.Rslaves(Rscript=system.file("slavedaemon.R", package="Rmpi"), nslaves=mpi.universe.size(), root = 0, intercomm = 2, comm = 1, hosts = NULL, needlog = TRUE, mapdrive=TRUE, quiet = FALSE, nonblock=TRUE, sleep=0.1) mpi.close.Rslaves(dellog = TRUE, comm = 1) tailslave.log(nlines = 3, comm = 1)mpi.spawn.Rslaves(Rscript=system.file("slavedaemon.R", package="Rmpi"), nslaves=mpi.universe.size(), root = 0, intercomm = 2, comm = 1, hosts = NULL, needlog = TRUE, mapdrive=TRUE, quiet = FALSE, nonblock=TRUE, sleep=0.1) mpi.close.Rslaves(dellog = TRUE, comm = 1) tailslave.log(nlines = 3, comm = 1)
Rscript |
an R script file used to run R in BATCH mode. |
nslaves |
number of slaves to be spawned. |
root |
the rank number of the member who spawns R slaves. |
intercomm |
an intercommunicator number |
comm |
a communicator number merged from an intercomm. |
hosts |
NULL or LAM node numbers to specify where R slaves to be spawned. |
needlog |
a logical. If TRUE, R BATCH outputs will be saved in log files. If FALSE, the outputs will send to /dev/null. |
mapdrive |
a logical. If TRUE and master's working dir is on a network, mapping network drive is attemped on remote nodes under windows platform. |
quiet |
a logical. If TRUE, do not print anything unless an error occurs. |
nonblock |
a logical. If TRUE, a nonblock procedure is used on all slaves so that they will consume none or little CPUs while waiting. |
sleep |
a sleep interval, used when nonblock=TRUE. Smaller sleep is, more response slaves are, more CPUs consume. |
dellog |
a logical specifying if R slave's log files are deleted or not. |
nlines |
number of lines to view from tail in R slave's log files. |
The R slaves that mpi.spawn.Rslaves spawns are really running a shell
program which can be found in system.file("Rslaves.sh",package="Rmpi")
which takes a Rscript file as one of its arguments. Other arguments are
used to see if a log file (R output) is needed and how to name it. The master
process id and the comm number, along with host names where R slaves are running are
used to name these log files.
Once R slaves are successfully spawned, the mergers from an intercomm (default ‘intercomm = 2’) to a comm (default ‘comm = 1’) are automatically done on master and slaves (should be done if the default Rscript is replaced). If additional sets of R slaves are needed, please use ‘comm = 3’, ‘comm = 4’, etc to spawn them. At most a comm number up to 10 can be used. Notice that the default comm number for R slaves (using slavedaemon.R) is always 1 which is saved as .comm.
To spawn R slaves to specific hosts, please use the argument hosts
with a list of those node numbers (an integer vector). Total node numbers
along their host names can be found by using lamhosts.
Notice that this is LAM-MPI specific.
Unless quiet = TRUE, mpi.spawn.Rslaves prints to stdio how many
slaves are successfully spawned and where they are running.
mpi.close.Rslaves return 1 if success and 0 otherwise.
tailslave.log returns last lines of R slave's log files.
Hao Yu
#mpi.spawn.Rslaves(nslaves=2) #tailslave.log() #mpi.remote.exec(rnorm(10)) #mpi.close.Rslaves()#mpi.spawn.Rslaves(nslaves=2) #tailslave.log() #mpi.remote.exec(rnorm(10)) #mpi.close.Rslaves()
mpi.universe.size returns the total number of CPUs available in a
cluster. Some MPI implements may not have this MPI call available.
mpi.universe.size()mpi.universe.size()
None.
Hao Yu
mpi.cancel cancels a nonblocking send or receive request.
mpi.test.cancelled tests if mpi.cancel cancels or not.
wait, waitall, waitany, and waitsome are used
to complete nonblocking send or receive requests. They are not local.
test, testall, testany, and testsome are used
to complete nonblocking send and receive requests. They are local.
mpi.cancel(request) mpi.test.cancelled(status=0) mpi.test(request, status=0) mpi.testall(count) mpi.testany(count, status=0) mpi.testsome(count) mpi.wait(request, status=0) mpi.waitall(count) mpi.waitany(count, status=0) mpi.waitsome(count)mpi.cancel(request) mpi.test.cancelled(status=0) mpi.test(request, status=0) mpi.testall(count) mpi.testany(count, status=0) mpi.testsome(count) mpi.wait(request, status=0) mpi.waitall(count) mpi.waitany(count, status=0) mpi.waitsome(count)
count |
total number of nonblocking operations. |
request |
a request number. |
status |
a status number. |
mpi.wait and mpi.test are used to complete a nonblocking
send and receive request: use the same request number by mpi.isend
or mpi.irecv. Once completed, the associated request is set to
MPI_REQUEST_NULL and status contains information such as source, tag,
and length of message.
If multiple nonblocking sends or receives are initiated, the following calls are more efficient. Make sure that request numbers are used consecutively as request=0, request=1, request=2, etc. In this way, the following calls can find request information in system memory.
mpi.waitany and mpi.testany are used to complete one out of
several requests.
mpi.waitall and mpi.testall are used to complete all
requests.
mpi.waitsome and mpi.testsome are used to complete all
enabled requests.
mpi.cancel returns no value.
mpi.test.cancelled returns TRUE if a nonblocking call is cancelled;
FALSE otherwise.
mpi.wait returns no value. Instead status contains information that
can be retrieved by mpi.get.count and mpi.get.sourcetag.
mpi.test returns TRUE if a request is complete; FALSE otherwise. If
TRUE, it is the same as mpi.wait.
mpi.waitany returns which request (index) has been completed. In
addition, status contains information that can be retrieved by
mpi.get.count and mpi.get.sourcetag.
mpi.testany returns a list: index— request index; flag—TRUE if
a request is complete; FALSE otherwise (index is no use in this case).
If flag is TRUE, it is the same as mpi.waitany.
mpi.waitall returns no value. Instead statuses 0, 1, ..., count-1
contain corresponding information that can be retrieved by
mpi.get.count and mpi.get.sourcetag.
mpi.testall returns TRUE if all requests are complete; FALSE
otherwise. If TRUE, it is the same as mpi.waitall.
mpi.waitsome returns a list: count— number of requests that have
been completed; indices—an integer vector of size count of those
completed request numbers (in 0, 1 ,..., count-1). In addition, statuses
0, 1, ..., count-1 contain corresponding information that can be
retrieved by mpi.get.count and mpi.get.sourcetag.
mpi.testsome is the same as mpi.waitsome except that count
may be 0 and in this case indices is no use.
Hao Yu
mpi.isend,
mpi.irecv,
mpi.get.count,
mpi.get.sourcetag.
Internal and hidden functions used by other MPI functions.
mpi.comm.is.null is used to test if a comm is MPI_COMM_NULL (empty
members).
string create a string (empty space character) buffer.
.docall a wrap to docall function.
.mpi.worker.apply apply like function used by workers.
.mpi.worker.applyLB apply like function used by workers (load balancing).
.mpi.worker.exec real execution by workers when using mpi.remote.exec.
.mpi.worker.sim real simulation by workers when using mpi.parSim.
.type.index identify input data type: integer, numeric, raw, or others.
.simplify simplify internal objects.
.splitIndices split parall apply jobs evenly.
.onUnload clean MPI when Rmpi is unloaded.
.mpi.undefined undefined mpi object.
.force.type force input data type object specified by type.
mpi.comm.is.null(comm) string(length) .docall(fun, args)mpi.comm.is.null(comm) string(length) .docall(fun, args)
comm |
a communicator number. |
length |
length of a string. |
fun |
a function object. |
args |
arguments to function. |
string returns an empty character string.
Hao Yu