MAT8186 -- Cours 7

basic.R

library(Rmpi)

## Crée les workers.
nworkers <- 5
mpi.spawn.Rslaves(nslaves = nworkers)

## Demande à chaque worker de retourner son rang.
mpi.remote.exec(mpi.comm.rank())

## On termine!
mpi.close.Rslaves()
mpi.quit()

dist_scale.R

library(Rmpi)

## Données
n <- 100000
set.seed(666)

dat <- data.frame(rnorm(n, 1234, 10),
                  rchisq(n, 20),
                  rt(n, 42),
                  rbeta(n, 0.1, 2))

## On doit avoir autant de worker que la longueur de la liste.
mpi.spawn.Rslaves(nslaves = length(dat))

## Extension de `mpi.scatter` permettant de travailler facilement avec
## une liste.
mpi.scatter.Robj2slave(dat)

mpi.bcast.cmd({
    ## Réduit-centre le vecteur
    dat <- scale(dat)

    ## Retourne le résultat au manager.
    mpi.gather.Robj(dat)
})

## Récupère les vecteurs
dat <- as.data.frame(mpi.gather.Robj(dat))
colnames(dat) <- c(paste0("Orig", seq_len(ncol(dat) / 2)),
                   paste0("Scaled", seq_len(ncol(dat) / 2)))

print(head(dat))
cat("--------\nMoyennes\n--------\n")
print(colMeans(dat))
cat("--------\nÉcarts-type\n--------\n")
print(sapply(dat, sd))

mpi.close.Rslaves()
mpi.quit()

ring_add.R

library(Rmpi)

## Crée les workers.
nworkers <- 5
mpi.spawn.Rslaves(nslaves = nworkers)

## Valeur initiale de `x`.
x <- 0L

## Définit la fonction à faire exécuter à chaque worker.
task <- function(nworkers) {
    rank <- mpi.comm.rank()

    ## Reçoit la variable `x` du worker précédent ou du manager si rank = 1.
    mpi.recv(x, 1, rank - 1, 0)

    ## Ajoute 1.
    x <- x + 1

    ## Envoie `x` au prochain worker ou au manager si rank = nworkers.
    dst <- ifelse(rank < nworkers, rank + 1, 0)
    mpi.send(x, 1, dst, 0)
}

## Envoie la fonction à tous les workers.
mpi.bcast.Robj2slave(task)

## Définit la variable `x` pour tous les workers.
mpi.bcast.cmd(x <- integer(1L))

## Envoie `x` au premier worker.
mpi.send(x, 1, 1, 0)

## Exécute la tâche.
mpi.bcast.cmd(cmd = task, nworkers)

x <- mpi.recv(x, 1, nworkers, 0)

cat("x =", x, "\n")

mpi.close.Rslaves()
mpi.quit()

snow_basic.R

library(snow)

cl <- makeCluster(4)

parLapply(cl, list(a = 1:10, b = 11:20, c = 21:30, d =  31:40), mean)

stopCluster(cl)