标签: parallel-processing

感觉需要学习如何使用asyncio，但想不出可以帮助我学习这项新技术的适用问题（或问题集）。

你能提出一个可以帮助我在实践中理解和学习 asyncio 用法的问题吗？

换句话说：你能给我推荐一些抽象问题或应用程序的例子，在编码时，这将帮助我学习如何在实践中使用 asyncio。

谢谢

我想加速一个算法（完整的本地二进制模式与圆形邻居），我遍历所有像素并用它的邻居计算一些东西（所以我需要邻居像素访问）。

目前我通过用一个线程/进程迭代所有像素来做到这一点。我想通过将输入图像分成多个 ROI 并单独计算每个 ROI（使用多个线程）来并行化此任务。

这里的问题是，ROI 是重叠的（因为要计算一个像素，有时我需要看远处的邻居），并且多个线程可能同时访问像素数据（READING）。如果两个或多个线程同时在相同的索引上读取相同的 Mat，这是一个问题吗？

如果我写入相同的 Mat 并行但使用不同的索引，这也是一个问题吗？

我有一个依赖于 MPI 的 mpi4py 实现的 python 脚本。这需要通过 mpiexec（或 mpirun）运行。

我的问题是：如何直接从 PyCharm 使用 mpiexec（或 mpirun）运行（并希望调试）python 脚本？

将 mpiexec 作为外部工具运行不是一个好主意，因为我将无法调试它。

我可以做什么？

我正在尝试编写一个具有多线程功能的 CRAN 包。我使用 实现了一个完美的解决方案doSNOW，但该软件包已被 CRAN 团队标记为“已取代”，他们要求我切换到一个doParallel解决方案。这很好，但是我找不到一种方法来监视已完成的作业数量，doParallel就像我使用doSNOW. 这是我的doSNOW解决方案：

# Set up parameters
nthreads<-2
nreps<-100
funrep<-function(i){
    Sys.sleep(0.1)
    res<-c(log2(i),log10(i))
    return(res)
}
# doSNOW solution
library(doSNOW)
cl<-makeCluster(nthreads)
registerDoSNOW(cl)
pb<-txtProgressBar(0,nreps,style=3)
progress<-function(n){
    setTxtProgressBar(pb,n)
}
opts<-list(progress=progress)
i<-0
output<-foreach(i=icount(nreps),.combine=c,.options.snow=opts) %dopar% {
    s<-funrep(i)
    return(s)
}
close(pb)
stopCluster(cl)

这是之前 Stack Overflow 帖子中建议的 doParallel 解决方案。但是，如您所见，它不会在作业完成时打印进度，它只会在最后合并结果时打印。

# doParallel solution
library(doParallel)
progcombine<-function(){
  count<-0
  function(...) {
    count<<-count+length(list(...))
    setTxtProgressBar(pb,count)
    utils::flush.console()
    c(...)
  }
}
cl <- makeCluster(nthreads)
registerDoParallel(cl)
output<-foreach(i = icount(nreps),.combine=progcombine()) %dopar% {
    funrep(i) …

我在玩 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 ~ ., …

我试图确定一类可以从使用 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并且不会阻碍用户的工作流程，直到任何实际调用该准备完成的请求。但是因为这个任务实际上解决了缺少运行并行池的问题，所以用户似乎必须等待他们可能实际上不需要使用的东西，直到任务完成很久之后。

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