标签: parallel-processing

我创建了各种参考类来适应一些 arima、garch 过程，并希望在并行计算中使用它们 parSapply

我先做了一些导出

cl <- makeCluster(mc <- getOption("cl.cores", 20))
        clusterExport(cl, c("merge.xts", "index", "coredata", "xts", "lag.xts", "zoo", "LearnerPredict", "arima", "generic_learner", "arma_simple", "logwarn"))
        clusterEvalQ(cl, "arma_simple")
        clusterEvalQ(cl, "generic_learner")


generic_learner <- setRefClass(

        Class = "generic_learner",

        fields = list(

            params = "list"

        ),

        methods = list(

            fitModel = function() {cat("overload function with fitting function \n")},

            fcastModel = function() {cat("overload function with forecast function \n")},

            fmt_params = function() {cat("overload function with formatted parameters \n")},

            fmt_class = function() {cat("overload class\n")},

            fmt_ref = function() {paste(.self$fmt_class(), .self$fmt_params(), …

我熟悉foreach,%dopar%之类的。我也是熟悉parallel的选项cv.glmnet。但是你如何设置嵌套的并行性如下？

library(glmnet)
library(foreach)
library(parallel)
library(doSNOW)
Npar <- 1000
Nobs <- 200
Xdat <- matrix(rnorm(Nobs * Npar), ncol = Npar)
Xclass <- rep(1:2, each = Nobs/2)
Ydat <- rnorm(Nobs)

并行交叉验证：

cl <- makeCluster(8, type = "SOCK")
registerDoSNOW(cl)
system.time(mods <- foreach(x = 1:2, .packages = "glmnet") %dopar% {
    idx <- Xclass == x
    cv.glmnet(Xdat[idx,], Ydat[idx], nfolds = 4, parallel = TRUE)
})
stopCluster(cl)

非并行交叉验证：

cl <- makeCluster(8, type = "SOCK")
registerDoSNOW(cl)
system.time(mods <- foreach(x …

很长一段时间以来,我一直在使用sfLapply来处理很多并行r脚本.然而,最近我已经深入研究并行计算,我一直在使用sfClusterApplyLB,如果单个实例不需要花费相同的时间来运行,那么可以节省大量时间.如果sfLapply将在加载新批处理之前等待批处理的每个实例完成(这可能导致空闲实例),完成任务的sfClusterApplyLB实例将立即分配给列表中的其余元素,因此可能会节省相当多的时间当实例没有花费相同的时间时.这让我质疑为什么我们在使用降雪时不想平衡我们的跑步？到目前为止我唯一发现的是,当并行脚本出现错误时,sfClusterApplyLB仍会在发出错误之前循环遍历整个列表,而sfLapply将在尝试第一批后停止.我还缺少什么？是否存在负载平衡的任何其他成本/缺点？下面是一个示例代码,显示了两者之间的差异

rm(list = ls()) #remove all past worksheet variables
working_dir="D:/temp/"
setwd(working_dir)
n_spp=16
spp_nmS=paste0("sp_",c(1:n_spp))
spp_nm=spp_nmS[1]
sp_parallel_run=function(sp_nm){
  sink(file(paste0(working_dir,sp_nm,"_log.txt"), open="wt"))#######NEW
  cat('\n', 'Started on ', date(), '\n') 
  ptm0 <- proc.time()
  jnk=round(runif(1)*8000000) #this is just a redundant script that takes an arbitrary amount of time to run
  jnk1=runif(jnk)
  for (i in 1:length(jnk1)){
    jnk1[i]=jnk[i]*runif(1)
  }
  ptm1=proc.time() - ptm0
  jnk=as.numeric(ptm1[3])
  cat('\n','It took ', jnk, "seconds to model", sp_nm)

  #stop sinks
  sink.reset <- function(){
    for(i in seq_len(sink.number())){
      sink(NULL)
    }
  }
  sink.reset()
}
require(snowfall)
cpucores=as.integer(Sys.getenv('NUMBER_OF_PROCESSORS'))

sfInit( parallel=T, cpus=cpucores) # …

我的理解是 concurrent.futures 依靠酸洗参数来让它们在不同的进程（或线程）中运行。酸洗不应该创建参数的副本吗？在 Linux 上它似乎没有这样做，即，我必须明确地传递一个副本。

我试图理解以下结果：

<0> rands before submission: [17, 72, 97, 8, 32, 15, 63, 97, 57, 60]
<1> rands before submission: [97, 15, 97, 32, 60, 17, 57, 72, 8, 63]
<2> rands before submission: [15, 57, 63, 17, 97, 97, 8, 32, 60, 72]
<3> rands before submission: [32, 97, 63, 72, 17, 57, 97, 8, 15, 60]
in function 0 [97, 15, 97, 32, 60, 17, 57, 72, 8, 63]
in function 1 [97, 32, …

我正在使用 OpenMP 并且我想生成线程，以便一个线程执行一段代码并完成，与运行并行 for 循环迭代的 N 个线程并行。

执行应该是这样的：

Section A (one thread)       ||      Section B (parallel-for, multiple threads)
         |                   ||        | | | | | | | | | |
         |                   ||        | | | | | | | | | |
         |                   ||        | | | | | | | | | |
         |                   ||        | | | | | | | | | |
         |                   ||        | | | | | | | | | |
         V                   || …

我正在尝试将“旧方式”循环转换为基于流的方法。该循环采用一大组元素并返回一个落在给定半径内的子集。结果按距离排序，并且结果本身具有方便的计算距离（用于演示）。它以旧方式工作正常，我不需要对它进行 Java8 化。但我真的很想。:-) 如果只是为了能够在这个傻瓜上使用 .parallel() 就好了。

问题是……我的 filter() 使用了一个计算值（距离），然后我需要在后续的 map() 步骤中使用它（以构建“with distance”实例）。假设距离计算很昂贵。这是 Java 7 的方式……向下滚动以查看 getNearestStations() 方法：

public interface Coordinate {
    double distanceTo(Coordinate other);
}

public class Station {
    private final String name;
    private final Coordinate coordinate;

    public Station(String name, Coordinate coordinate) {
        this.name = name;
        this.coordinate = coordinate;
    }

    public String getName() {
        return name;
    }

    public Coordinate getCoordinate() {
        return coordinate;
    }
}

public class StationWithDistance extends Station implements Comparable<StationWithDistance> {
    private final double distance;

    public …

我在 C++ AMP 中处理大型数组（超过 65536 个元素）时遇到问题。我正在使用 C++ amp 来计算多边形列表的法线、切线和双切线向量。输入由位置数组（每个位置 3 个浮点数）和 uv 坐标数组（每个顶点 2 个浮点数）组成。在我的parallel_for_each函数中，我计算法线、切线和双切线（每组3个顶点各1个）。我将它们写回数组（封装在 array_view 中）。该算法如下所示：

concurrency::extent<2> ePositions(positionsVector.size() / 3, 3);
concurrency::array_view<const float, 2> positions(ePositions, positionsVector);
concurrency::extent<2> eUVs(uvsVector.size() / 2, 2);
concurrency::array_view<const float, 2> UVs(eUVs, uvsVector);
concurrency::extent<2> eNormalDirections(normalDirectionsVector.size() / 3, 3);
concurrency::array_view<float, 2> normalDirections(eNormalDirections, normalDirectionsVector);
normalDirections.discard_data();
concurrency::extent<2> eTangentDirections(tangentDirectionsVector.size() / 3, 3);
concurrency::array_view<float, 2> tangentDirections(eTangentDirections, tangentDirectionsVector);
tangentDirections.discard_data();
concurrency::extent<2> eBitangentDirections(bitangentDirectionsVector.size() / 3, 3);
concurrency::array_view<float, 2> bitangentDirections(eBitangentDirections, bitangentDirectionsVector);
bitangentDirections.discard_data();

concurrency::parallel_for_each(eNormalDirections.tile<1, 3>(), [=](concurrency::tiled_index<1, 3> t_idx) restrict(amp)
{
    < ... calculate the normals, tangents …

我已经使用pmap. 使用该-p选项在一台机器上的性能改进非常好。现在我想在多台机器上运行。

我--machinefile在 julia start 上使用了这个选项。它可以工作，但它只在远程机器上启动一个进程。我想在每台机器上运行多个进程。选项-p仅在本地机器上启用多个进程。有没有办法指定远程机器上的进程数？

我有一个关于我应该如何使用ThreadLocal.

背景和情况

有几个单例对象用于ThreadLocal为每个线程创建一个副本。这个单例对象有一个函数foo()。

public class SingletonA {
    protected static ThreadLocal<SingletonA> singleton = new ThreadLocal<SingletonA>() {
        @Override
        protected SingletonA initialValue() {
            return new SingletonA();
        }
    };

    private SingletonA() { ... }
    public static SingletonA getInstance() { return singleton.get(); }
    public static void remove() { singleton.remove(); }
    public static void foo() { ... }
}

... 有 SingletonB、SingletonC 等等。

有一个单例存储库可以缓存ThreadLocal上面的单例。这个类也是一个ThreadLocal单例——

public class SingletonRepo {
        protected static ThreadLocal<SingletonRepo> singleton = new ThreadLocal<SingletonRepo>() { …

我有一系列文档（约 50,000 个），我已经将这些文档转换为语料库并使用 R 中的 topicmodels 包构建 LDA 对象。不幸的是，为了测试 150 多个主题，需要几个小时。

到目前为止，我发现我可以使用以下方法同时测试几个不同的集群大小：

library(topicmodels)
library(plyr)
library(foreach)
library(doMC)
registerDoMC(5) # use 5 cores

dtm # my documenttermmatrix

seq <- seq(200,500, by=50)

models <- llply(seq, function(d){LDA(dtm, d)}, .parallel=T)

有没有办法并行化 LDA 函数，使其运行得更快（而不是一次运行多个 LDA）？