标签: parallel-processing

考虑以下简单代码：

Stream.of(1)
  .flatMap(x -> IntStream.range(0, 1024).boxed())
  .parallel() // Moving this before flatMap has the same effect because it's just a property of the entire stream
  .forEach(x -> {
     System.out.println("Thread: " + Thread.currentThread().getName());
  });

很长一段时间，我认为即使在flatMap. 但是上面的代码打印了所有的“Thread:main”，证明我的想法是错误的。

一种使其并行的简单方法flatMap是收集然后再次流式传输：

Stream.of(1)
  .flatMap(x -> IntStream.range(0, 1024).boxed())
  .parallel() // Moving this before flatMap has the same effect because it's just a property of the entire stream
  .collect(Collectors.toList())
  .parallelStream()
  .forEach(x -> {
     System.out.println("Thread: " + Thread.currentThread().getName());
  });

我想知道是否有更好的方法，以及flatMap仅在调用之前并行化流的设计选择，而不是在调用之后并行化。

========关于问题的更多说明========

从一些答案来看，我的问题似乎没有完全表达出来。正如@Andreas 所说，如果我从 …

我在矢量化循环时遇到了麻烦。我正在寻找重写下面的代码，使其矢量化。我已经运行了 Complete Banerjee 的测试，我发现所有依赖项都被破坏了，但我不知道从哪里开始。编译器是gcc。体系结构是 x86，数组是整数数组。

for (int i = 0; i < 100; i++) { 
     x[20 + i] = y[i] * z[i];
     p[i] = x[21 + i] + q[i];
}

我一直在寻找多线程教程和那些特定于同步的教程，但我一直无法实现我需要的东西。

它展示了我的程序中发生的同步。

基本上，我有一个类从其他类继承了一些函数，这些函数需要同步以便两个线程不会同时修改对象（没有数据损坏）。

我以前在没有同步关键字的情况下实现了代码，所以我可以设法看到数据损坏的发生。

 EditOptionsMethods e1;
    int threadNo;

    public EditOptions() {
        e1 = new BuildAuto();
        run();
    }

    public void run() {
        System.out.println("Thread running.");

        switch (threadNo) {
            case 0:

                break;

            case 1:
                break;
        }

    }

    public void setOptions(String optSetName1, String desiredOption1, String optSetName2, String desiredOption2) {
        e1.s_Option(optSetName1, desiredOption1); // 
        e1.s_Option(optSetName2, desiredOption2);
    }

s_Option 必须同步，因此两个线程都不会发生。我首先将在没有同步的情况下使用它，然后我可以初始化一个循环（具有高索引量，假设为 1000，然后我用第一个线程添加，并用第二个线程减去）以查看发生的损坏作为示例。

但我没有找到一种方式来展示这一点。

如果有人知道我如何实现这一点，那就太棒了。

public class Test {
    private static volatile boolean flag = false;
    private static int i = 1;
    public static void main(String[] args) {
        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            flag = true;
            i += 1;
        }).start();
        new Thread(() -> {
            while (!flag) {
                if (i != 1) {
                    System.out.println(i);
                }
            }
            System.out.println(flag);
            System.out.println(i);
        }).start();
    }
}

变量i写在 volatile 变量标志之后，但代码输出 true 2。看起来i第一个线程的修改对第二个线程是可见的。

按照我的理解，变量I应该写在flag之前，这样第二个线程就可以知道变化了。

所以我有一个函数，它可以在如下所示的大型未排序数字数组中找到一个大于 N 的数字。

import java.util.*;

public class program {

    // Linear-search function to find the index of an element
    public static int findIndex(int arr[], int t)
    {
        // if array is Null
        if (arr == null) {
            return -1;
        }

        // find length of array
        int len = arr.length;
        int i = 0;

        // traverse in the array
        while (i < len) {

            // if the i-th element is t
            // then return the index
            if (arr[i] > t) { …

从这个来源可以阅读：

值得一提的是，同步和并发集合只会使集合本身线程安全，而不是内容。

我认为如果Collection是线程安全的，那么它的内容将隐式是线程安全的。

我的意思是，如果两个线程无法访问我的Collection对象，那么我的Collection对象所持有的对象将隐式成为线程安全的。

我错过了这一点，有人可以用一个例子来解释我吗？

我编写了一个代码，它使用 OpenMP 执行 LU 分解。我觉得代码执行得太慢了。我正在使用的计算机有 16 个内核，16 个线程的时间为 12.5 秒，4 个线程的时间为 14.3 秒。这是发生并行化的代码部分。我是 C 编程的新手，觉得我错过了一些会减慢多线程速度的东西。

    /*__Initialize lock__*/
omp_lock_t lock[n];

for(i=0;i<n;i++)
   omp_init_lock(&lock[i]);

/*___Create theads____*/
#pragma omp parallel private(i,j,k,thread) num_threads(nThreads)
{
  thread=omp_get_thread_num();
  #pragma omp for schedule(static)
  for(i=0;i<n;i++) omp_set_lock(&lock[i]);
  if(thread==0)
    omp_unset_lock(&lock[0]);

  for(i=0;i<n;i++){
    div=1/U[i][i];
    #pragma omp for schedule(static)
      for(j=i+1;j<n;j++)
         L[j][i]=U[j][i]*div;

    #pragma omp for schedule(static)
      for(j=i+1;j<n;j++){
        for(k=i;k<n;k++)
           U[j][k]=U[j][k]-L[j][i]*U[i][k];
        if(j==i+1)
           omp_unset_lock(&lock[i+1]);
      }
}
}

在某些机器上，在所有内核上加载软件包会占用所有可用 RAM，从而导致错误 137 并且我的 R 会话被终止。在我的笔记本电脑 (Mac) 和 Linux 计算机上，它运行良好。在我想要运行它的 Linux 计算机上，它没有 32 核和 32 * 6GB RAM。系统管理员告诉我计算节点上的内存是有限的。然而，根据我在下面的编辑，我的记忆需求并没有超出任何想象。

我如何调试它并找出不同之处？我是这个parallel包的新手。

这是一个示例（假设该命令install.packages(c(“tidyverse”,”OpenMx”))已在 4.0.3 版的 R 中运行）：

我还注意到，这似乎只适用于OpenMx和mixtools包。我mixtools从 MWE 中排除，因为OpenMx足以产生问题。tidyverse单独工作正常。

我尝试过的一种解决方法是不在集群上加载包，而只是.libPaths("~/R/x86_64-pc-linux-gnu-library/4.0/")在 of 的主体中进行评估 expr，clusterEvalQ并像OpenMx::vec在我的函数中一样使用命名空间命令，但这会产生相同的错误。所以我被卡住了，因为在三台机器中的两台上它运行良好，只是不在我应该使用的一台（计算节点）上。

.libPaths("~/R/x86_64-pc-linux-gnu-library/4.0/")
library(parallel)
num_cores <- detectCores()
cat("Number of cores found:") 
print(num_cores)
working_mice <- makeCluster(num_cores) 
clusterExport(working_mice, ls())
clusterEvalQ(working_mice, expr = {
  library("OpenMx")
  library("tidyverse")
  })

通过简单地加载包，它似乎消耗了所有可用的 RAM，从而导致错误 137。这是一个问题，因为我需要在每个可用内核中加载库，它们的功能正在执行任务。

随后，我正在使用DEoptim但加载包足以产生错误。 …

我在一个目录中有 10 张 jpeg 图像。我想使用 pyspark 同时阅读所有这些内容。我尝试如下。

from PIL import Image


from pyspark import SparkContext, SparkConf    

conf = SparkConf()
spark = SparkContext(conf=conf)       

files = glob.glob("E:\\tests\\*.jpg")

files_ = spark.parallelize(files)    

arrs = []

for fi in files_.toLocalIterator():      

    im = Image.open(fi)
    data = np.asarray(im)
    arrs.append(data)

img = np.array(arrs)    
print (img.shape)

代码无误地结束并打印出来img.shape；然而，它并没有并行运行。你可以帮帮我吗？

问题

Julia 是否有一种有效的方法从给定的条目列表 (u,v,w)构建一个巨大的稀疏矩阵，其中一些可以具有相同的位置(u,v)，在这种情况下，它们的权重 w 必须是总和。因此u,v,w是输入向量，我希望创建一个w[i]在 position 处具有值的稀疏矩阵u[i],v[i]。例如，Mathematica 代码

n=10^6;   m=500*n;   
u=RandomInteger[{1,n},m];   
v=RandomInteger[{1,n},m];   
w=RandomInteger[{-9, 9}, m];   AbsoluteTiming[
SetSystemOptions["SparseArrayOptions"->{"TreatRepeatedEntries"->1}]; 
a= SparseArray[{u,v}\[Transpose] -> w, {n,n}]; ]

需要 135 秒和 60GB 的 RAM。等效的 Python 代码

import scipy.sparse as sp   
import numpy as np
import time
def ti(): return time.perf_counter() 
n=10**6; m=500*n;
u=np.random.randint(0,n,size=m);            
v=np.random.randint(0,n,size=m);            
w=np.random.randint(-9,10,size=m); t0=ti(); 
a=sp.csr_matrix((w,(u,v)),shape=(n,n),dtype=int); t1=ti(); print(t1-t0)

需要 36 秒和 20GB，但不支持 (2)。等效的 Julia 代码

using SparseArrays;
m=n=10^6; r=500*n; …