Ejecutar bucle foreach en paralelo o secuencialmente dado una condición

Question

A menudo termino con varios bucles anidados foreach y, a veces al escribir funciones generales (por ejemplo, para un paquete) no hay ningún nivel que sea obvio para paralelizar en. ¿Hay alguna manera de lograr lo que describe la maqueta a continuación?

foreach(i = 1:I) %if(I < J) `do` else `dopar`% { 
    foreach(j = 1:J) %if(I >= J) `do` else `dopar`% { 
     # Do stuff 
    } 
} 

Por otra parte, ¿hay alguna manera de detectar si un motor en paralelo se ha registrado para que pueda evitar recibir mensajes de advertencia innecesarias? Esto sería útil tanto para verificar los paquetes antes de la presentación de CRAN como para no molestar a los usuarios que ejecutan R en las computadoras de núcleo único.

foreach(i=1:I) %if(is.parallel.backend.registered()) `dopar` else `do`% { 
    # Do stuff 
} 

Gracias por su tiempo.

Edit: Muchas gracias por todos los comentarios sobre los núcleos y los trabajadores, y tiene razón en que la mejor manera de tratar con el ejemplo anterior sería replantear toda la configuración. Prefiero algo así como debajo de la idea triu, pero es esencialmente el mismo punto. Y, por supuesto, también podría hacerse con un tapply paralelo, como sugirió Joris.

ij <- expand.grid(i=1:I, j=1:J) 
foreach(i=ij$I, j=ij$J) %dopar% { 
    myFuction(i, j) 
} 

Sin embargo, en mi intento de simplificar la situación que dio lugar a este hilo, omití algunos detalles cruciales. Imagine que tengo dos funciones analyse y batch.analyse y el mejor nivel para paralelizar en podría ser diferente dependiendo de los valores de n.replicates y n.time.points.

analyse <- function(x, y, n.replicates=1000){ 
    foreach(r = 1:n.replicates) %do% { 
     # Do stuff with x and y 
    } 
} 
batch.analyse <- function(x, y, n.replicates=10, n.time.points=1000){ 
    foreach(tp = 1:time.points) %do% { 
     my.y <- my.func(y, tp) 
     analyse(x, my.y, n.replicates) 
    } 
} 

Si n.time.points > n.replicates tiene sentido para paralelizar en batch.analyse pero por lo demás tiene más sentido para paralelizar en analyse. ¿Alguna idea sobre cómo abordarlo? ¿Sería de alguna manera posible detectar en analyse si la paralelización ya ha tenido lugar?

Answer 1

La cuestión que usted plantea fue la motivación para el operador foreach anidación, '%:%'. Si el cuerpo del bucle interno necesita una cantidad considerable de tiempo de cálculo, que está bastante segura usando:

foreach(i = 1:I) %:% 
    foreach(j = 1:J) %dopar% { 
     # Do stuff 
    } 

Este "desenrolla" los bucles anidados, lo que resulta en (I * J) tareas que pueden todos ser ejecutado en paralelo.

Si el cuerpo del lazo interno no toma mucho tiempo, la solución es más difícil. La solución estándar es paralelizar el bucle externo, pero eso podría resultar en muchas tareas pequeñas (cuando I es grande y J es pequeño) o en algunas tareas grandes (cuando soy pequeño y J es grande).

Mi solución favorita es utilizar el operador de anidamiento con división de tareas. Aquí está un ejemplo completo utilizando el backend doMPI:

library(doMPI) 
cl <- startMPIcluster() 
registerDoMPI(cl) 
I <- 100; J <- 2 
opt <- list(chunkSize=10) 
foreach(i = 1:I, .combine='cbind', .options.mpi=opt) %:% 
    foreach(j = 1:J, .combine='c') %dopar% { 
     (i * j) 
    } 
closeCluster(cl) 

Esto se traduce en 20 "trozos de tareas", cada uno que consta de 10 cálculos del cuerpo del bucle.Si usted quiere tener un solo trozo de tareas para cada trabajador, se puede calcular el tamaño del fragmento como:

cs <- ceiling((I * J)/getDoParWorkers()) 
opt <- list(chunkSize=cs) 

Por desgracia, no todos los backends CHUNKING tarea de apoyo paralelo. Además, doMPI no es compatible con Windows.

Para obtener más información sobre este tema, véase mi viñeta de "anidamiento Foreach Loops" en el paquete foreach:

library(foreach) 
vignette('nesting') 

Answer 2

Si termina con varios bucles foreach anidados, reconsideraría mi enfoque. Usar versiones paralelas de tapply puede resolver una gran cantidad de esa molestia. En general, no debe usar la paralelización anidada, ya que eso no le aporta nada. Paraleliza el bucle externo y olvídate del bucle interno.

La razón es simple: si tiene 3 conexiones en su clúster, el bucle dopar externo utilizará las tres. El ciclo de dopar interno no podrá usar ninguna conexión adicional, ya que no hay ninguno disponible. Entonces no ganas nada. Por lo tanto, la maqueta que proporcione no tiene ningún sentido desde un punto de vista de programación.

Su segunda pregunta es respondida con bastante facilidad por la función getDoParRegistered() que devuelve TRUE cuando se registra un backend, y FALSE de lo contrario. Sin embargo, preste atención:

• también devuelve TRUE si se registra un back-end secuencial (es decir, después de llamar a registerDoSEQ).
• Devolverá VERDADERO también después de que se haya detenido un clúster, pero en ese caso% dopar% devolverá un error.

por ejemplo:

require(foreach) 
require(doSNOW) 
cl <- makeCluster(rep("localhost",2),type="SOCK") 
getDoParRegistered() 
[1] FALSE 
registerDoSNOW(cl) 
getDoParRegistered() 
[1] TRUE 
stopCluster(cl) 
getDoParRegistered() 
[1] TRUE 

Pero ahora la ejecución de este código:

a <- matrix(1:16, 4, 4) 
b <- t(a) 
foreach(b=iter(b, by='col'), .combine=cbind) %dopar% 
    (a %*% b) 

volverá en un error:

Error in summary.connection(connection) : invalid connection 

Se puede construir una comprobación adicional. A (horriblemente feo) truco que puede utilizar para comprobar que la conexión registrada por doSNOW es válida, puede ser:

isvalid <- function(){ 
    if (getDoParRegistered()){ 
     X <- foreach:::.foreachGlobals$objs[[1]]$data 
     x <- try(capture.output(print(X)),silent=TRUE) 
     if(is(x,"try-error")) FALSE else TRUE 
    } else { 
     FALSE 
    } 
} 

que se podría utilizar como

if(!isvalid()) registerDoSEQ() 

Esto registrará el backend secuencial si getDoParRegistered() devuelve TRUE pero ya no hay una conexión de clúster válida. Pero, nuevamente, este es un truco, y no tengo idea de si funciona con otros backends o incluso con otros tipos de clúster (generalmente uso sockets).

Answer 3

En orden inverso de las preguntas que le preguntó:

1. @Joris es correcta con respecto a la comprobación de un motor paralelo registrado. Sin embargo, tenga en cuenta que existe una diferencia entre una máquina que es un solo núcleo y si se registra o no un back-end paralelo. Comprobar el número de núcleos es una tarea específica de la plataforma (sistema operativo). En Linux, esto puede funcionar para usted:

   CountUnixCPUs <- function(cpuinfo = "/proc/cpuinfo"){ 
   tmpCmd <- paste("grep processor ", cpuinfo, " | wc -l", sep = "") 
   numCPU <- as.numeric(system(tmpCmd, intern = TRUE)) 
   return(numCPU) 
   } 
   

   Editar: @ Ver enlace de Joris a otra página, a continuación, que da consejos para Windows y Linux. Es probable que reescriba mi propio código, al menos para incluir más opciones para contar núcleos.

2. En cuanto a los bucles anidados, tomo una táctica diferente: preparo una tabla de parámetros y luego repito sobre las filas. Una forma muy simple es, por ejemplo .:

   library(Matrix) 
   ptable <- which(triu(matrix(1, ncol = 20, nrow = 20))==1, arr.ind = TRUE) 
   foreach(ix_row = 1:nrow(ptable)) %dopar% { myFunction(ptable[ix_row,])}