并行执行foreach循环或按顺序执行

Question

我经常最终得到几个嵌套foreach循环,有时候在编写一般函数时(例如对于一个包),没有明显的并行化级别.有没有办法完成下面的模型描述？

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

此外,有没有办法检测并行后端是否已注册,以便我可以避免收到不必要的警告消息？在CRAN提交之前检查包并且不打扰在单核计算机上运行R的用户时,这将非常有用.

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

谢谢你的时间.

编辑:非常感谢您对核心和工作人员的所有反馈,并且你是对的,处理上述示例的最佳方法是重新考虑整个设置.我更喜欢下面这个triu想法,但它基本上是相同的.它当然也可以tapply像Joris建议的那样平行完成.

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

然而,在我试图简化引起这个线程的情况时,我遗漏了一些关键的细节.想象一下,我有两个功能analyse,并batch.analyse和最好的水平,在并行依据的值可能是不同的n.replicates和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)
    }
}

如果n.time.points > n.replicates并行化是有意义的,那么并行化batch.analyse更有意义analyse.有关如何解决它的任何想法？是否有可能检测到analyse并行化是否已经发生？

Answer 1

您提出的问题是foreach嵌套运算符'%:%'的动机.如果内循环的主体需要大量的计算时间,那么使用它是非常安全的:

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

这会"展开"嵌套循环,从而导致(I*J)任务可以并行执行.

如果内环的主体不需要花费太多时间,则解决方案更加困难.标准解决方案是并行化外部循环,但这仍然可能导致许多小任务(当我很大而J很小时)或一些大任务(当我很小而J很大时).

我最喜欢的解决方案是使用嵌套操作符和任务分块.这是使用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)

这导致20个"任务块",每个包含10个循环体的计算.如果要为每个worker创建一个任务块,可以将块大小计算为:

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

不幸的是,并非所有并行后端都支持任务分块.此外,doMPI不支持Windows.

有关此主题的更多信息,请参阅foreach包中的插图"嵌套Foreach循环":

library(foreach)
vignette('nesting')

Answer 2

如果你最终得到几个嵌套的foreach循环,我会重新考虑我的方法.使用并行版本tapply可以解决很多麻烦.通常,您不应该使用嵌套并行化,因为这不会带来任何东西.并行化外部循环,忘记内部循环.

原因很简单:如果群集中有3个连接,则外部多普勒循环将使用所有三个连接.内部多巴环路将无法使用任何额外的连接,因为没有可用的连接.所以你没有得到任何东西.因此,从编程的角度来看,你提供的模型根本没有意义.

您的第二个问题很容易通过函数回答,该函数getDoParRegistered()在注册后端时返回TRUE,否则返回FALSE.注意:

• 如果注册了顺序后端(即在调用registerDoSEQ之后),它也返回TRUE.
• 在群集停止后它也将返回TRUE,但在这种情况下%dopar%将返回错误.

例如:

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

但现在运行此代码:

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

将返回错误:

Error in summary.connection(connection) : invalid connection

你可以建立一个额外的支票.你可以用来检查注册的连接doSNOW是否有效的(可怕的丑陋)hack ,可以是:

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
    }
}

您可以使用哪个

if(!isvalid()) registerDoSEQ()

如果getDoParRegistered()返回TRUE但是不再有有效的集群连接,这将注册顺序后端.但同样,这是一个黑客,我不知道它是否适用于其他后端甚至其他类型的集群类型(我主要使用套接字).