标签: parallel-processing

我是C#Parallel.ForEach,和.NET的新手.我想并行化涉及数千个位置的搜索.对于每个位置,我计算大圆距离.这是我想要传播到不同核心的计算.我的问题是,如果我只有一个线程局部变量,我该怎么做呢,就像在这个MSDN TPL示例中一样？对于结果,我看了看Interlocked,看到它的选项Add,CompareExchange,Decrement,Exchange,Increment和Read,但我不只是增加,递增,递减,或测试是否相等.我希望通过并行运行的多个线程返回对象,这个线程总体上最短距离.我的直觉说这应该很容易,我应该能够创建一个包裹Location一个距离的小物体,但是如何从每个线程中捕获最佳答案然后选择它们之间的最短距离？这是非并行版本:

Location findClosestLocation(Location myLocation, List<Location> allLocations)
{
  double closest = double.MaxValue;
  Location closestLoc = null;
  foreach (Location aLoc in allLocations)
  {
    if (aLoc != myLocation)
    {
      double d = greatCircle(myLocation, aLoc);
      if (d < closest)
      {
        closest = d;
        closestLoc = aLoc;
      }
    }
  }
  return closestLoc;
}

我确实看到一个似乎提供了很好建议的DDJ博客文章,但我想知道这是否是最好的建议.我看到作者循环遍历数组,并想知道是否没有更多功能的方法来做到这一点.在我将使用的功能世界中map …

(编辑:澄清一下,我的主要目标是并发,但不一定是多核机器)

我对所有关于并发的概念都相当新,但我发现我需要有并行绘图例程,原因有很多:

• 我想分开绘制图形的不同部分(背景刷新的频率低于前景,保留在缓冲区中).
• 我希望控制优先级(UI响应性比绘制复杂图表更优先).
• 我希望每帧绘图计算多线程.
• 我想为复杂的缓冲区绘图程序提供取消.

然而,作为一个初学者,我的代码很快看起来像一团糟,重构或错误修复变得如此尴尬,我决定在做任何严肃的事情之前我需要更多地使用它.

所以,我想知道如何制作干净,易于保存的.NET多线程代码,这些代码在我第二天醒来后看到它时才有意义.我遇到的最大问题是构建应用程序,因此所有部分都以智能(而不是笨拙和hacky)方式相互交谈.

任何建议都是受欢迎的,但我喜欢我可以在空闲时间消化的来源(例如,不是500多页的并发论文)和C#/ VB.NET,直到最新版本(因为我看到那里)一直在进步).基本上我想要一些直截了当的东西,所以我可以开始玩我的玩具项目的概念.

我正在尝试在Java中并行化算法.我从合并排序开始,并在这个问题上发布了我的尝试.我修改过的尝试是在下面的代码中,我现在尝试并行快速排序.

在我的多线程实现或解决此问题的方法中是否存在任何新手错误？如果不是,我不应期望在双核上的顺序算法和并行算法之间的速度增加超过32%(参见底部的时间)？

这是多线程算法:

    public class ThreadedQuick extends Thread
    {
        final int MAX_THREADS = Runtime.getRuntime().availableProcessors();

        CountDownLatch doneSignal;
        static int num_threads = 1;

        int[] my_array;
        int start, end;

        public ThreadedQuick(CountDownLatch doneSignal, int[] array, int start, int end) {
            this.my_array = array;
            this.start = start;
            this.end = end;
            this.doneSignal = doneSignal;
        }

        public static void reset() {
            num_threads = 1;
        }

        public void run() {
            quicksort(my_array, start, end);
            doneSignal.countDown();
            num_threads--;
        }

        public void quicksort(int[] array, int start, int end) {
            int …

我在玩 tflite 并在我的多核 CPU 上观察到在推理期间它没有受到很大的压力。我通过预先用 numpy 创建随机输入数据（类似于图像的随机矩阵）消除了 IO 瓶颈，但随后 tflite 仍然没有充分利用 CPU 的潜力。

该文档提到了调整使用线程数的可能性。但是我无法在 Python API 中找到如何做到这一点。但是因为我看到人们为不同的模型使用多个解释器实例，所以我认为一个人可能会使用同一模型的多个实例并在不同的线程/进程上运行它们。我编写了以下简短脚本：

import numpy as np
import os, time
import tflite_runtime.interpreter as tflite
from multiprocessing import Pool


# global, but for each process the module is loaded, so only one global var per process
interpreter = None
input_details = None
output_details = None
def init_interpreter(model_path):
    global interpreter
    global input_details
    global output_details
    interpreter = tflite.Interpreter(model_path=model_path)
    input_details = interpreter.get_input_details()
    output_details = interpreter.get_output_details()
    interpreter.allocate_tensors()
    print('done init') …

让我们定义：

from multiprocessing import Pool
import numpy as np
def func(x):
    for i in range(1000):
        i**2
    return 1

请注意，func()它做了一些事情，它总是返回一个小数字1。

然后，我比较了一个 8 核并行Pool.map()v/sa 串行，python 内置，map()

n=10**3
a=np.random.random(n).tolist()

with Pool(8) as p:
    %timeit -r1 -n2  p.map(func,a)
%timeit -r1 -n2  list(map(func,a))

这给出了：

38.4 ms ± 0 ns per loop (mean ± std. dev. of 1 run, 2 loops each)
200 ms ± 0 ns per loop (mean ± std. dev. of 1 run, 2 loops each) …

假设我将caret在 R 中进行培训，但我想将此培训分为两个运行会话。

library(mlbench)
data(Sonar)
library(caret)
set.seed(998)
inTraining <- createDataPartition(Sonar$Class, p = .75, list = FALSE)
training <- Sonar[ inTraining,]
testing  <- Sonar[-inTraining,]

# First run session
nn.partial <- train(Class ~ ., data = training, 
                method = "nnet",
                 max.turns.of.iteration=5) # Non-existent parameter. But represents my goal

让我们假设nn整个对象我只有一个部分对象，它在第 5 回合（即nn.partial）之前具有训练信息。因此，将来我可以运行以下代码来完成训练工作：

library(mlbench)
data(Sonar)
library(caret)
set.seed(998)
inTraining <- createDataPartition(Sonar$Class, p = .75, list = FALSE)
training <- Sonar[ inTraining,]
testing  <- Sonar[-inTraining,]

nn <- train(Class ~ ., …

为什么在下面的 Julia 代码中，并行实现比串行运行慢？

using Distributed

@everywhere function ext(i::Int64)
   callmop = `awk '{ sum += $1 } END { print sum }' infile_$(i)`
   run(callmop)
end

function fpar()
   @sync @distributed for i = 1:10
      ext(i)
   end
end

function fnopar()
   for i = 1:10
      ext(i)
   end
end

val, t_par, bytes, gctime, memallocs = @timed fpar()
val, t_nopar, bytes, gctime, memallocs = @timed fnopar()

println("Parallel: $(t_par) s. Serial: $(t_nopar) s")  
# Parallel: 0.448290379 s. Serial: 0.028704802 s

这些文件infile_$(i)包含一列实数。经过一些研究，我遇到了处理类似问题的这篇文章和另一篇文章）。不过，如果考虑到 …

我试图确定一类可以从使用 Java 8 中引入的 parallelStream API 中受益的 Java 应用程序。

我知道其他 SO 帖子中描述的 API 的许多警告：

• 共享 fork/join 池，具有非平凡的启动时间，以及池中争用的一些潜在问题
• 以某种方式不受控制地使用系统资源，使得在服务器（已经具有多任务策略）上使用此类代码实际上可能是一个坏主意
• ...还有其他批评主要与性能有关

尽管如此，如果 Stream API 已经被使用，API 仍然可以使用现代多核机器，代码不是很具有侵入性，因此在低开发成本下没有麻烦的多线程。因此，我仍然认为它在某些情况下很有用。

我认为应用程序上下文因此必须是这样的：

1. 我的申请目前是连续的
2. 存在响应时间问题，就挂钟时间而言，例如，用户单击了 GUI 按钮并正在等待回复
3. 应用程序在客户端机器上运行，大多数时候我们可以期望有一些可用的 CPU 内核，而不是在资源已经竞争的服务器上
4. 我的开发团队没有人力/技能来开发他们自己的任务分配/线程机制，所以他们不会去并行化，除非他们可以使用这个 API 轻松做到

我在github上搜索过，但很难找到不是练习或教科书示例的parallelStream用法的相关示例（我欢迎链接到API的中型+项目中的一些用法）。

那么，Java 语言开发人员使用此 API 的目标是哪种应用程序？

您是否同意上述对 API 有用的应用程序上下文的要求？

假设我有 N 个生成器gen_1, ..., gen_N，其中每个生成器都会产生相同数量的值。我想要一个生成器gen，使其在 N 个并行进程中运行 gen_1, ..., gen_N 并产生(next(gen_1), next(gen_2), ... next(gen_N))

那是我想要的：

def gen():
   yield (next(gen_1), next(gen_2), ... next(gen_N))

这样每个 gen_i 都在自己的进程上运行。是否有可能做到这一点？我尝试在以下虚拟示例中执行此操作但没有成功：

A = range(4)

def gen(a):
    B = ['a', 'b', 'c']
    for b in B:
        yield b + str(a)

def target(g):
    return next(g)

processes = [Process(target=target, args=(gen(a),)) for a in A]

for p in processes:
    p.start()

for p in processes:
    p.join()

但是我得到了错误TypeError: cannot pickle 'generator' object。

编辑：

我修改了@darkonaut 的答案以满足我的需要。我发布它以防你们中的一些人觉得它有用。我们首先定义几个效用函数： …

首次启动 MATLAB 并行池时，通常需要几秒钟的时间。因此，在用户交互式应用程序中，有动力确保在第一个计算任务需求到达之前运行并行池，因此启动并行池的过程不会添加到响应请求的总时间中.

然而parpool，我见过的每一个程序操作都会启动并行池，直到池完成启动。这意味着即使用户在一段时间内不需要调用并行池，在并行池启动完成之前，他们也不能做任何其他事情，比如开始设置他们的计算成本高的请求——例如填写用户界面。

这非常令人沮丧！如果是任何其他耗时的准备操作，一旦并行池就位，它就可以在后台使用parfeval并且不会阻碍用户的工作流程，直到任何实际调用该准备完成的请求。但是因为这个任务实际上解决了缺少运行并行池的问题，所以用户似乎必须等待他们可能实际上不需要使用的东西，直到任务完成很久之后。

有什么办法可以绕过这种明显的可用性限制吗？