标签: parallel-processing

我想并行化以下代码段，但我是 openmp 和创建并行代码的新手。

std::vector<DMatch> good_matches;
for (int i = 0; i < descriptors_A.rows; i++) {
   if (matches_RM[i].distance < 3 * min_dist) {
      good_matches.push_back(matches_RM[i]);
   }
}

我试过了

std::vector<DMatch> good_matches;
#pragma omp parallel for
for (int i = 0; i < descriptors_A.rows; i++) {
   if (matches_RM[i].distance < 3 * min_dist) {
      good_matches[i] = matches_RM[i];
   }
}

和

std::vector<DMatch> good_matches;
cv::DMatch temp;
#pragma omp parallel for
for (int i = 0; i < descriptors_A.rows; i++) {
   if (matches_RM[i].distance < 3 * min_dist) { …

我正在尝试将“旧方式”循环转换为基于流的方法。该循环采用一大组元素并返回一个落在给定半径内的子集。结果按距离排序，并且结果本身具有方便的计算距离（用于演示）。它以旧方式工作正常，我不需要对它进行 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）？

from joblib import Parallel, delayed

def func(v):
    temp.append(v)
    return

temp = []
Parallel(n_jobs=4)(delayed(func)(v) for v in range(10))
print temp

我想让共享内存变量。但是temp的值为空[]。我该怎么做？

对于其他方法，我尝试了 pickle.dump 和 load。但是有一个锁定问题。请给我建议！

我正在尝试通过hargreaves方法计算蒸发量package SPEI。这涉及使用最低温度 ( TMIN) 和最高温度 ( TMAX)。鉴于此Tmin，并行计算是我最好的选择，并且Tmax rasterstacks拥有500,000 cells and 100 layers each. Hargreaves function取Tmin，Tmax并latitude在each grid作为输入。以下是我的第一个猜测如何解决这个问题：

library(SPEI)
# go parallel 
library(parallel)
clust <- makeCluster(detectCores())

#har <- hargreaves(TMIN,TMAX,lat=37.6475) # get evaporation for a station. 

但是，我的数据是网格化的。

Tmin并且Tmax是名单中，每个数据帧Tmin，并Tmax有一个$latitude连接到它。在 中pet，k$d是 Tmin，k$d是 Tmax（也许我应该在petegfunction(k,y)而不是仅仅提供两个参数k？） …

该包的摘要的java.util.stream规定如下：

有状态 lambda 的一个例子是map()in的参数：

Set<Integer> seen = Collections.synchronizedSet(new HashSet<>());
stream.parallel().map(e -> { if (seen.add(e)) return 0; else return e; })...

Run Code Online (Sandbox Code Playgroud)

在这里，如果映射操作是并行执行的，由于线程调度差异，相同输入的结果可能会因运行而异，而对于无状态 lambda 表达式，结果将始终相同。

我不明白为什么这不会产生一致的结果，因为该集合是同步的并且一次只能处理一个元素。您能否以一种演示结果如何因并行化而变化的方式完成上述示例？

并行 STL 算法是否符合std::back_insert_iterator??

我可能误解了std::par和之间的区别std::par_vec，是否std::par_vec意味着需要预先分配输出范围？

代码示例：

auto numbers = {1,2,3,4,5,6};
auto squared = std::vector<int>{};
std::transform(
  **std::par/std::par_vec,**
  numbers.begin(),
  numbers.end(),
  std::back_inserter(squared),
  [](auto val) { 
    return val*val; 
  }
);

更新

简化问题作为我的第一个问题是误读文章的结果。