标签: parallel-processing

我正在考虑为我想解决的一个问题利用并行性.问题大致如下:给定输入(点序列)找到最佳输出(由这些点组成的最大三角形,最长线等).在点序列中有3种不同的"形状",但我只对"最佳得分"(通常是某种形式的"长度"倍数系数)感兴趣.我们称之为形状S1,S2,S3.

我有2种不同的算法来解决S1 - 'S1a'在O(n ²)中,'S1b'大多表现得更好,但最坏的情况是O(n ⁴).

第一个问题:是否有一些简单的方法可以并行运行S1a和S1b,使用先完成并停止另一个的方法？至于我正在阅读文档,这可以使用一些forkIO编程并在获得结果时杀死线程 - 只是询问是否有更简单的东西？

第二个问题 - 更加困难:我以这种方式调用优化函数:

optimize valueOfSx input

valueOfSx特定于每个形状,并返回"得分"(或得分的猜测)可能的解决方案.优化调用此函数以找出最佳解决方案.我感兴趣的是:

s1 = optimize valueOfS1 input
s2 = optimize valueOfS2 input
s3 = optimize valueOfS3 input
<- maximum [s1,s2,s3]

但是,如果我知道S1的结果,我可以丢弃所有较小的解决方案,从而使s2和s3收敛得更快,如果不存在更好的解决方案(或者至少丢弃最差的解决方案,从而提高空间效率).我现在在做的是:

zeroOn threshold f = decide .f
    where decide x = if (x < threshold) then 0 else x
s1 = optimize valueOfS1 input
s2 = optimize (zeroOn s1 valueOfS2) input
s3 = optimize (zeroOn (max s1 s2) valueOfS3) input

现在的问题是:我能以这样的方式如运行S2和S3并行,无论哪完成第一次将更新其他线程运行的得分功能的"门槛"参数？在某种意义上的东西:

threshold = 0 …

我在 R 中进行了一些并行模拟，我注意到使用“L'Ecuyer-CMRG”rng 时种子没有改变。我正在阅读“Parallel R”一书，每次调用 mclapply() 时，选项 mc.set.seed = TRUE 应该给每个工人一个新种子。

这是我的代码：

library(parallel)
RNGkind("L'Ecuyer-CMRG")

mclapply(1:2, function(n) rnorm(n), mc.set.seed = TRUE)
[[1]]
[1] -0.7125037

[[2]]
[1] -0.9013552  0.3445190

mclapply(1:2, function(n) rnorm(n), mc.set.seed = TRUE)
[[1]]
[1] -0.7125037

[[2]]
[1] -0.9013552  0.3445190

编辑：同样的事情发生在我的台式机和我的笔记本电脑上（都是 Ubuntu 12.04 LTS）。

我正在尝试使用 concurrent.futures 模块将一个需要很长时间的进程拆分为多个进程。附上下面的代码

主功能：

with concurrent.futures.ProcessPoolExecutor() as executor:
for idx, score in zip([idx for idx in range(dataframe.shape[0])],executor.map(get_max_fuzzy_score,[dataframe[idx:idx+1] for idx in range(dataframe.shape[0])])):
    print('processing '+str(idx+1)+' of '+str(dataframe.shape[0]+1))
    dataframe['max_row_score'].iloc[idx] = score

get_max_fuzzy_score 函数：

def get_max_fuzzy_score(picklepath_or_list, df):
import numpy as np
extracted_text_columns = list(df.filter(regex='extracted_text').columns)
data_list = [df[data].iloc[0] for data in extracted_text_columns if not df[data].isnull().values.any()]
try:
    size = len(picklepath_or_list)
    section_snippet_list = picklepath_or_list
except:
    section_snippet_list = pickle.load(open(picklepath_or_list,'rb'))
scores = []
for section_snippet in section_snippet_list:
    for data in data_list:
        scores.append(fuzz.partial_ratio(data,section_snippet))
score = max(scores)

return score

该函数采用几列的值，并从先前构建的列表中返回最大模糊分数。

这是我得到的错误： …

介绍

我正在使用dask.distributed（令人尴尬的并行任务）并行化一些代码。

• 我有一个指向不同图像的路径列表，这些图像散布给工作人员。
• 每个工作人员加载和过滤图像（3D 堆栈）并运行一些过滤。使用 scipy 进行 3D 过滤可节省中间体输出。
• 每个过滤后的图像在磁盘上保存为 npy 和/或 png。
• 我在集群上运行之前在本地进行测试，我的设置是：

.

from dask.distributed import Client, LocalCluster
cluster = LocalCluster(n_workers=2, threads_per_worker=1,memory_limit =8e9)
client = Client(cluster)

问题：

• 当我只处理两个图像（1 个图像/工人）时，一切都很好
• 当我为每个工作人员散布多个图像时，我会收到此警告，其中进程内存值增加。

.

distributed.worker - WARNING - Memory use is high but worker has no data to store to disk.
Perhaps some other process is leaking memory?  Process memory: 6.21 GB -- Worker memory limit: 8.00 GB

暗示工作人员使用的部分 RAM 不在freed不同文件之间（我猜是剩余的过滤中间体......）

题

有没有办法在开始处理下一个图像之前释放工人的内存？我必须garbage collector在运行任务之间运行一个循环吗？

编辑 …

我有一个很大的句子列表（约 7 百万个），我想从中提取名词。

我使用joblib库来并行化提取过程，如下所示：

import spacy
from tqdm import tqdm
from joblib import Parallel, delayed
nlp = spacy.load('en_core_web_sm')

class nouns:

    def get_nouns(self, text):
        doc = nlp(u"{}".format(text))
        return [token.text for token in doc if token.tag_ in ['NN', 'NNP', 'NNS', 'NNPS']]

    def parallelize(self, sentences):
        results = Parallel(n_jobs=1)(delayed(self.get_nouns)(sent) for sent in tqdm(sentences))
        return results

if __name__ == '__main__':
    sentences = ['we went to the school yesterday',
                 'The weather is really cold',
                 'Can we catch the dog?',
                 'How old are you John?', …

我的任务是将依赖节点列表转换为 AWS Step Functions。AWS Step Function 定义允许并行分支甚至分支嵌套到多级深度。不幸的是，它不支持分支中任务之间的依赖关系，因此强制您在两个结果可用于步骤函数中的后续任务之前完成并行步骤。

在我的图表中，Step Functions 可以轻松支持如图 1 所示的简单并行分支。

当涉及到图 2，尤其是图 3 时，它就成了一个问题。

作为一种简单的方法，我们可以引入额外的节点来为其依赖节点收集结果，如图 2b 和 3b 所示，但这现在引入了以前不存在的依赖关系：

• 在图 2b 中，引入了这些新的依赖项：
  • E -> A，F -> A
• 在图 3b 中，引入了这些新的依赖项：
  • E -> A、F -> A、F-> B、C -> E

这是一个问题，因为在手动审批任务的情况下，这些任务的时间可能是几小时到几天。这将导致后面的步骤被它们不依赖的任务不必要地延迟。

有关如何解决此问题的任何建议？也许我可以采取不同的方法？也许我可以应用一些花哨的图论算法？我什至不知道用什么词来解释图论中的这个问题。

如果需要，这里有一个在 draw.io 上使用这些图表的网址。

我已经使用 FastAI（PyTorch 后端）在 GPU 上训练了一个 CNN 模型。我现在尝试在同一台机器上使用该模型进行推理，但使用 CPU 而不是 GPU。除此之外，我还尝试使用多处理模块来利用多个 CPU 内核。现在问题来了，

在单 CPU 上运行代码（无多处理）只需 40 秒即可处理近 50 张图像

使用 Torch 多处理在多个 CPU 上运行代码需要 6 多分钟才能处理相同的 50 个图像

from torch.multiprocessing import Pool, set_start_method
os.environ['CUDA_VISIBLE_DEVICES']=""
from fastai.vision import *
from fastai.text import *
defaults.device = torch.device('cpu')

def process_image_batch(batch):

    learn_cnn  = load_learner(scripts_folder, 'cnn_model.pkl')
    learn_cnn.model.training = False    
    learn_cnn.model = learn_cnn.model.eval()
    # for image in batch: 
    #     prediction = ... # predicting the image here
    #     return prediction

if __name__ == '__main__':
    #
    # …

std::accumulate并std::reduce做几乎相同的事情。

的总结std::reduce说明了一切：

similar to `std::accumulate`, except out of order 

在许多情况下，这些函数应该产生相同的最终结果并展示相同的整体功能。很明显，如果您有一些非常重的负载计算等，您可以尝试std::reduce进行parrelization。IE。从鸟类的角度来看，这里的传统智慧是什么 - 除非明确优化，否则您是否应该始终坚持直率的 std::accumulate ？还是应该默认使用std::reduce？

如果std::reduce（选择默认/未选择执行策略）总是至少与std::accumulate（保存一些指令）一样快，那么我认为只有在订单严格时才应使用累积。

这是查找 LCM 和 HCF 之和等于该数字的第一对数字（1 除外）的代码。

import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

class PerfectPartition {
    static long gcd(long a, long b) {
        if (a == 0)
            return b;
        return gcd(b % a, a);
    }

    // method to return LCM of two numbers
    static long lcm(long a, long b) {
        return (a / gcd(a, b)) * b;
    }

    long[] getPartition(long n) {
        var ref = new Object() {
            long x;
            long y;
            long[] ret = null;
        };

        Thread mainThread = Thread.currentThread();
        ThreadGroup t …

我过去使用过 Joblib 和 Airflow 并且没有遇到过这个问题。我正在尝试通过 Airflow 运行一个使用 Joblib 运行并行计算的作业。当 Airflow 作业启动时，我看到以下警告

UserWarning: Loky-backed parallel loops cannot be called in multiprocessing, setting n_jobs=1

将警告追溯到源头我看到 LokyBackend 类的 joblib 包中触发了以下函数（MultiprocessingBackend 类中也有类似的逻辑）

def effective_n_jobs(self, n_jobs):
    """Determine the number of jobs which are going to run in parallel"""
    if n_jobs == 0:
        raise ValueError('n_jobs == 0 in Parallel has no meaning')
    elif mp is None or n_jobs is None:
        # multiprocessing is not available or disabled, fallback
        # to sequential mode
        return 1
    elif mp.current_process().daemon: …