代码之家  ›  专栏  ›  技术社区  ›  gdkrmr Pittoro

在工作进程之间共享对象

  •  0
  • gdkrmr Pittoro  · 技术社区  · 8 年前

    我想跑步 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个 仅向每个节点复制一次,并让工作进程在本地复制它?

    2 回复  |  直到 8 年前
        1
  •  2
  •   HenrikB    8 年前

    假设主服务器和远程主机之间的连接是瓶颈所在,您可以将一个副本传输给第一个辅助服务器,然后将其缓存到文件中,并让其他辅助服务器从该缓存文件中读取数据。类似于:

    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))
    
        2
  •  0
  •   gdkrmr Pittoro    8 年前

    这是一个使用 fifos ,我不确定这在Linux下的可移植性如何,我不确定这在性能方面与 @HenrikB 的anwser:

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