代码之家 › 专栏 › 技术社区 › akraf

为什么我的二级期货会连续执行?

r-future parallel-processing r

akraf · 技术社区 · 6 年前

我试图复制 topology vignette of the future package . 引述:

期货可以嵌套在R中,这样一个期货创造另一套期货。等等。例如,这可能发生在嵌套for循环内[…]

有一部分作者使用 plan(list(multicore, multicore)) (进一步的论证和 tweak 省略)同时处理两个期货,每个期货同时处理四个期货。这应该等于同步处理的八个期货。

然而,当我试图用下面的代码重现这一点时,我看到第二层期货是按顺序处理的。我做错什么了?

麦克维尔

library(future)
library(ggplot2)
plan(list(multiprocess, multiprocess))


# Run for a random amount of time and return start and stop time
startStop <- function(){
  start <- Sys.time()
  x <- runif(1, 1, 3)
  Sys.sleep(x)
  stop <- Sys.time()
  return(data.frame(start = start, stop = stop))
}

nGrp <- 3
nCV <- 4

l <- rep(list(NULL), nGrp)


for(i in seq_along(l)){
  l[[i]] <- future({
    m <- rep(list(NULL), nCV)
    for(j in seq_along(m)){
      m[[j]] <- future(startStop())
    }
    m <- lapply(m, value)
    m <- do.call(rbind, m)
    m
  })
}
l <- lapply(l, value)
d <- do.call(rbind, l)
d$iGrp <- rep(seq_len(nGrp), each = nCV)
d$iCV <- rep(seq_len(nCV), times = nGrp)

d$x <- paste(d$iGrp, d$iCV, sep = "_")
d$iGrp <- as.character(d$iGrp)
ggplot(d, aes(x = x, ymin = start, ymax = stop, color = iGrp)) + geom_linerange() + coord_flip()

期望值

会话信息

R version 3.4.3 (2017-11-30)
Platform: x86_64-pc-linux-gnu (64-bit)
Running under: CentOS Linux 7 (Core)

Matrix products: default
BLAS: /opt/Bio/R/3.4.3/lib64/R/lib/libRblas.so
LAPACK: /opt/Bio/R/3.4.3/lib64/R/lib/libRlapack.so

locale:
 [1] LC_CTYPE=en_US.UTF-8       LC_NUMERIC=C              
 [3] LC_TIME=en_US.UTF-8        LC_COLLATE=en_US.UTF-8    
 [5] LC_MONETARY=en_US.UTF-8    LC_MESSAGES=en_US.UTF-8   
 [7] LC_PAPER=en_US.UTF-8       LC_NAME=C                 
 [9] LC_ADDRESS=C               LC_TELEPHONE=C            
[11] LC_MEASUREMENT=en_US.UTF-8 LC_IDENTIFICATION=C       

attached base packages:
[1] stats     graphics  grDevices utils     datasets  methods   base     

other attached packages:
[1] ggplot2_2.2.1 future_1.8.1 

loaded via a namespace (and not attached):
 [1] Rcpp_0.12.17     devtools_1.13.4  munsell_0.4.3    colorspace_1.3-2
 [5] R6_2.2.2         rlang_0.1.6      httr_1.3.1       plyr_1.8.4      
 [9] globals_0.11.0   tools_3.4.3      parallel_3.4.3   grid_3.4.3      
[13] gtable_0.2.0     git2r_0.21.0     withr_2.1.1      yaml_2.1.16     
[17] lazyeval_0.2.1   digest_0.6.15    tibble_1.4.2     codetools_0.2-15
[21] curl_3.1         memoise_1.1.0    compiler_3.4.3   pillar_1.1.0    
[25] scales_0.5.0     listenv_0.7.0

2 回复 | 直到 6 年前

HenrikB 6 年前

作者 future 这里:这是因为有一个针对嵌套并行性的内置保护。如果没有它,你会用太多的并行进程使你的计算机超载,这不仅会使它过热,还会降低整体性能。

我已经为下一个版本更新了“未来拓扑”小插曲,内容如下:

内置的递归并行保护

上面我们已经平行地处理了未来的外部或内部集合。如果我们想同时处理这两层呢?很容易使用:
plan(list(multiprocess, multiprocess))
虽然这并没有给出错误,但是我们会发现,期货的内层将按顺序处理,就像我们使用 plan(list(multiprocess, sequential)) . 此行为是由于内置的对嵌套并行性的保护。如果两个层都并行运行,每个层都使用机器上可用的8个内核,那么我们将运行8*8=64个并行进程——这肯定会使我们的计算机过载。内部发生的是外层, availableCores() 等于八(8),而内层等于一(1)。

现在,我们可以想象,我们处理外层的两个平行期货,然后处理内层的四个平行期货。在这种情况下,我们最多只能运行8个内核(=2*4)。这可以通过强迫每个层的固定数量的工人来实现(不推荐):
plan(list(tweak(multiprocess, workers = 2), tweak(multiprocess, workers = 4)))

cywhale Weng 6 年前

如果你想像你期望的那样实现并行处理, future.callr 是个选择。只需使用: library(future.callr) plan(list(callr, callr))