标签: parallel-processing

我有一个结构：

public class DataItem {
    public int wordID, categoryID, documentID, count;
}

我有一个如下所示的列表：

final public ArrayList<DataItem> data = new ArrayList<>();

我写了一个在其中搜索的方法：

public DataItem FindDataItem(final int wordID, final int categoryID, final int documentID)
{
    for(DataItem dataItem : data)
        if(dataItem.wordID == wordID && dataItem.documentID == documentID && dataItem.categoryID == categoryID)
            return dataItem;
    return null;
}

但它太慢了。我怎样才能加快速度？

我正在考虑四个HashMap彼此内部，但我想像数据库表一样使用这些数据，因此很难在HashMap中按计数进行分组

我也在考虑ParalellStream，但我不知道如何使用它。看起来很复杂。但仍然是 O(n)。

我也在考虑使用数据库。但我不想有IO。我想把它全部放在内存中。

请引导我完成这个过程。

我有 ~100 个子目录，每个子目录有 ~1000 个文件，我想转换JPG为在ie下PNG使用。我的脚本很慢，我可以加快速度吗？Image MagickBASH for Win10LINUX script

find . -type f -name '*.jpg' -exec sh -c '
    orgfile="$0"
    newfile="$(echo "$0" | sed 's/.jpg/.png/')"
    echo $orgfile $newfile
    convert $orgfile -unsharp 0x5 $newfile
    rm $orgfile
' {} \;

我喜欢循环过程，因为转换是许多过程中的第一个，因此输入和输出名称可以重复使用。然而，它的速度很慢，并且有回声可供反馈（更改为每个目录？）

在相关帖子中给出了以下解决方案

# Runs these conversions serially
ls *.NEF | sed 's#.NEF##' | xargs -I^ convert ^.NEF ^.jpg
# Runs these conversions with 8 different processes
ls *.NEF | sed 's#.NEF##' | xargs …

我正在尝试使用 --parallel 选项提高 Linux 上的 perforce 同步性能，但对于我指定的每个线程，错误都会重复一次：

强制密码 (P4PASSWD) 无效或未设置。

如果我设置P4PORT、P4USER和环境变量，然后执行简单的 p4 同步，它就可以正常工作P4PASSWD。P4CLIENT如果我使用以下命令，我会收到 4 次错误消息（每个线程一次）：

p4 sync --parallel "threads=4,min=1,minsize=1"

如果我在命令行上将密码指定为全局密码，我会得到相同的结果：

p4 -P passwd sync --parallel "threads=4,min=1,minsize=1"

如果我首先使用票证p4 login -p并将其替换为同步命令中的密码，我会得到相同的结果。

我创建了库uwork，但是启动每个进程都需要一些时间，因此它不适合实时应用程序。通过进行一些性能基准测试，我得到了这个片段来说明问题：

let external = (new Date()).getTime();
let blob = new Blob([`
  let internal = new Date();
  console.log("Creation:", internal - ${external} + 'ms');
`], {"type": "text\/plain"});
let file = URL.createObjectURL(blob);
let ww = new Worker(file);

创建文件和 Web Worker 需要 50-400 毫秒之间的任何时间。有什么办法可以优化这个吗？为什么需要这么长时间？

这对于“长”处理时间来说是完全可以的，因为启动时间可以忽略不计，但对于视频分析等实时应用程序来说就不行了，我想以至少 10fps（100ms/运行）的速度应用它。

编辑

经过一些测试后，这也需要 50-400 毫秒，所以我强烈认为是URL.createObjectURL()（或new Blob()）花费了这一时间，而不是 Web Worker 本身的创建：

let external = (new Date()).getTime();
let blob = new Blob([external], {"type": "text\/plain"});
let file = URL.createObjectURL(blob);
fetch(file).then(res => res.text()).then(file => { …

我正在尝试对 fit_generator 进行多处理。

这些是我面临的问题。

trainable_model.fit_generator(load_random_cached_bottlenecks(BATCH_SIZE, label_map, training_addr_label_map, train_npy_dir, 'h5py', h5py_file_train),epochs = EPOCHS, steps_per_epoch=iterations_per_epoch_t, validation_data = load_random_cached_bottlenecks(BATCH_SIZE, label_map, validation_addr_label_map, val_npy_dir, 'h5py', h5py_file_val), validation_steps=iterations_per_epoch_v, workers = 1, callbacks = callback_list, use_multiprocessing = True, max_queue_size = 32)

造成问题的主要论点：workers和use_multiprocessing。

当 时worker=1，use_multiprocessing=True/False运行没有问题。

如果workers=5，use_multiprocessing=True则抛出错误。奇怪的是它正在运行，但是在一些随机迭代中我遇到了类似的错误

KeyError: 'Unable to open object (bad local heap signature)'

或者

KeyError: 'Unable to open object (wrong B-tree signature)'

我使用 h5py 来读取文件。我为此目的编写了自定义生成器。

def load_random_cached_bottlenecks(batch_size, label_map,
 addr_label_map, dirs, comp_type = 'h5py', hdf5_file …

我需要并行执行 Python 脚本，因此我使用以下批处理文件：

start python C:\myfolder\1.py
start python C:\myfolder\2.py
start python C:\myfolder\3.py

它工作正常，但现在我需要在上述前三个脚本完成后并行运行三个脚本。如何在同一个批处理文件中指定它？

假设我有N 个生成器，可以生成项目流gs = [..] # list of generators。

我可以轻松地将它们组合在一起，从:zip中的每个相应生成器中获取元组生成器。gstuple_gen = zip(*gs)

这会按顺序调用next(g)每个并将结果收集到一个元组中。但如果每个项目的生产成本都很高，我们可能希望在多个线程上并行化工作。ggsnext(g)

我怎样才能实现pzip(..)这个功能呢？

我正在使用 Pythonmultiprocessing创建并行应用程序。进程需要共享一些数据，为此我使用Manager. 但是，我有一些进程需要调用的常用函数以及需要访问对象存储的数据的函数Manager。我的问题是我是否可以避免需要将Manager实例作为参数传递给这些通用函数，而是像全局函数一样使用它。换句话说，请考虑以下代码：

import multiprocessing as mp

manager = mp.Manager()
global_dict = manager.dict(a=[0])

def add():
    global_dict['a'] += [global_dict['a'][-1]+1]

def foo_parallel(var):
    add()
    print var

num_processes = 5
p = []
for i in range(num_processes):
    p.append(mp.Process(target=foo_parallel,args=(global_dict,)))

[pi.start() for pi in p]
[pi.join() for pi in p]

这运行良好并返回p=[0,1,2,3,4,5]到我的机器上。然而，这是“好形式”吗？这是一个好方法吗，就像定义add(var)和调用add(var)一样好？

我几天来一直在寻找这个问题的答案，但没有结果。我可能只是不理解那些漂浮在外面的部分，并且该multiprocessing模块的 Python 文档相当大，对我来说不清楚。

假设您有以下 for 循环：

import timeit


numbers = []

start = timeit.default_timer()

for num in range(100000000):
    numbers.append(num)

end = timeit.default_timer()

print('TIME: {} seconds'.format(end - start))
print('SUM:', sum(numbers))

输出：

TIME: 23.965870224497916 seconds
SUM: 4999999950000000

对于此示例，假设您有一个 4 核处理器。有没有办法总共创建 4 个进程，其中每个进程都在单独的 CPU 核心上运行，并且完成速度大约快 4 倍，因此 24 秒/4 个进程 = 约 6 秒？

以某种方式将 for 循环分成 4 个相等的块，然后将这 4 个块添加到数字列表中以等于相同的总和？有一个 stackoverflow 线程：Parallel Simple For Loop但我不明白。谢谢大家。

考虑以下张量流代码片段：

import time
import numpy as np
import tensorflow as tf

def fn(i):
    # do some junk work
    for _ in range(100):
        i ** 2
    return i

n = 1000
n_jobs = 8
stuff = np.arange(1, n + 1)
eager = False
t0 = time.time()
if eager:
    tf.enable_eager_execution()
res = tf.map_fn(fn, stuff, parallel_iterations=n_jobs)
if not eager:
    with tf.Session() as sess:
        res = sess.run(res)
        print(sum(res))
else:
    print(sum(res))
dt = time.time() - t0
print("(eager=%s) Took %ims" % (eager, dt * 1000))

如果使用 …