在工作进程之间共享对象

Question

我想在许多不同的工作进程上运行f(x)，这些工作进程运行一台（多于一台的奖励积分）远程机器，其中x 是一个大对象。

我的交互式 R 会话在 node0 上运行，并且我使用 parallel 库，所以我执行以下操作：

library(parallel)

cl <- makeCluster(rep("node1", times = 64))
clusterExport(cl, "x")
clusterExport(cl, "f")

clusterEvalQ(cl, f(x))

问题是发送x需要相当长的时间，因为它是通过网络连接单独传输到每个工作进程的，而主进程运行的机器是分开的。

问题： 是否可以只向每个节点发送一次x 并让工作进程在本地复制它？

Answer 1

假设主服务器和远程主机之间的连接是瓶颈，您可以将一个副本传输到第一个工作人员，然后将其缓存到文件中，并让其他工作人员从该缓存文件中读取数据。比如：

library("parallel")

## Large data object
x <- 1:1e6
f <- function(x) mean(x)

## All N=64 workers are on the same host
cl <- makeCluster(rep("node1", times = 64))

## Send function
clusterExport(cl, "f")

## Send data to first worker (over slow connection)
clusterExport(cl[1], "x")

## Save to cache file (on remote machine)
cachefile <- clusterEvalQ(cl[1], {
  saveRDS(x, file = (f <- tempfile())); f
})[[1]]

## Load cache file into remaining workers
clusterExport(cl[-1], "cachefile")
clusterEvalQ(cl[-1], { x <- readRDS(file = cachefile); TRUE })

# Resolve function on all workers
y <- clusterEvalQ(cl, f(x))

Answer 2

这是一个使用 fifos 的版本，我不确定它的可移植性如何，在 Linux 下工作，我不确定这与 @HenrikB 的答案在性能方面如何比较：

library(parallel)

# create a very large cluster on a single (remote) node:
cl <- makePSOCKcluster(3)

# create a very large object
o <- 1:10

# create a fifo on the node and retrieve the name
fifo_name <- clusterEvalQ(cl[1], {
                        fifo_name <- tempfile()
                        system2("mkfifo", fifo_name)
                        fifo_name
})[[1]]

# send the very large object to one process on the node and the name of the fifo to all nodes
clusterExport(cl[1], "o")
clusterExport(cl, "fifo_name")

# does the actual sharing through the fifo
# note that a fifo has to be opened for reading 
# before writing on it
for(i in 2:length(cl)) {
  clusterEvalQ(cl[i], { ff <- fifo(fifo_name, "rb")  })
  clusterEvalQ(cl[1], { ff <- fifo(fifo_name, "wb")
                        saveRDS(o, ff)
                        close(ff)                    })
  clusterEvalQ(cl[i], { o <- readRDS(ff)
                        close(ff)                    })
}

# cleanup
clusterEvalQ(cl[1], {   unlink(fifo_name)            })

# check if everything is there
clusterEvalQ(cl, exists("o"))

# now you can do the actual work
...