标签: parallel-processing

假设我有 5 个线程，它们必须1,000,000对并行蒙特卡罗方法程序进行总计函数调用。我1,000,000 / 5为 5 个线程中的每一个分配了函数调用。然而，经过多次测试（一些测试的迭代次数高达 1 万亿次），我意识到某些线程的完成速度比其他线程快得多。因此，我想动态地将工作负载分配给每个线程。我的第一个方法涉及一个AtomicLong初始值设置为 10 亿的变量。在每次函数调用之后，我会将 减AtomicLong1。在每次函数调用之前，程序都会检查 是否AtomicLong大于0，如下所示：

AtomicLong remainingIterations = new AtomicLong(1000000000);
ExecutorService threadPool = Executors.newFixedThreadPool(5);

for (int i = 0; i < 5; i++) {//create 5 threads
   threadPool.submit(new Runnable() {
       public void run() {
          while (remainingIterations.get() > 0) {//do a function call if necessary 
             remainingIterations.decrementAndGet();//decrement # of remaining calls needed
             doOneFunctionCall();//perform a function call
          }
       }
   });
}//more unrelated code is …

我正在学习如何使我的脚本运行得更快。我认为并行是一个好方法。所以我尝试了 gevent 和多处理。但我对它不同的结果感到困惑。举两个我遇到的例子

例1：

a=np.zeros([3])
def f(i):
    a[i]=1
    print a
def f_para():
    p=multiprocessing.Pool()
    p.map(f, range(3))

def f_asy():
    threads = [gevent.spawn(f, i) for i in xrange(3)]
    gevent.joinall(threads)

f_para()
[ 0.  1.  0.]
[ 0.  0.  1.]
[ 1.  0.  0.]

f_asy()
[ 1.  0.  0.]
[ 1.  1.  0.]
[ 1.  1.  1.]

我发现使用多处理，全局对象a在 fat 中永远不会改变，并且运行后f_para()仍然a是原始数组。跑步的时候f_asy()，就不一样了，a变了。

例2：

def f2(i):
    subprocess.call(['./a.out', str(i)])
    time.sleep(0.2)

def f2_loop():
    for i in xrange(20):
        f2(i)

def f2_para(): …

我在这里很困惑。我有一个受 CPU 限制的应用程序，因此我开始使用多进程实现并行化来克服 GIL 问题。

我第一次尝试使用multiprocessing，futures但遇到了酸洗问题，所以我去pathos使用哪个dill作为pickle替代品。

一切正常，但我想知道我是否使用的是最“面向未来”的解决方案。我也见过dask，但我不知道它是否会在酸洗类问题的情况下起作用（请参阅Python：（Pathos）多处理与类方法）。从文档来看，它使用了futures所以我假设它不会完成这项工作。

其次，我希望能够一次使用两台服务器，并且我已经看到这可以通过pathos(also dask) 实现，但我不明白它到底是如何工作的。这个答案/sf/answers/1886378091/仅显示如何使用一台服务器。使用 2 个或更多怎么样？我找不到任何关于此的示例，尽管包信息中描述的似乎是可能的。

感谢您的帮助！

我有一个结构：

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 => { …

我目前正在使用一个工作人员运行 celery 4.0.2，如下所示：

芹菜.py：

app = Celery('project',
         broker='amqp://jimmy:jimmy123@localhost/jimmy_vhost',
         backend='rpc://',
         include=['project.tasks'])

if __name__ == '__main__':
    app.start()
    app.name

任务.py：

from .celery import app
from celery.schedules import schedule
from time import sleep, strftime

app.conf.beat_schedule = {
    'planner_1': {
        'task': 'project.tasks.call_orders',
        'schedule': 1800,
    },
    'planner_2': {
        'task': 'project.tasks.call_inventory',
        'schedule': 900,
    },
}

我使用以下命令来运行beat：

 celery -A project worker -l info --concurrency=3 --beat -E

现在它只是一个队列，只有一个工作人员在运行。

我的问题是如何使用多个工作人员和单个队列运行 celery，以便使用多处理并行执行任务而不重复？

我在互联网上查找了如何使用多处理运行芹菜。根据这篇文章：

celery worker -l info -P processes -c 16 将导致单个消息使用者将工作委托给 16 个操作系统级池进程。在多核环境中，每个操作系统级进程可以分配给不同的CPU，因此它将并行处理任务，但不会并行消费消息。

使用-p processes论证可以解决我的问题吗？另外，“它将并行处理任务，但不会并行消费消息”是什么意思？

I'm trying to use these all things together to run parallel tests in a headless chrome:

Docker, Selenium, Pytest

However, I'm wondering where it makes sense to run the parallel part of the system?

Docker can do this (using selenium grid). Both these can be used to run parallel (and distributed) selenium tests. e.g.

https://github.com/elgalu/docker-selenium

https://github.com/zalando/zalenium

Also Pytest has its own way of running parallel tests (using pytest-xdist) e.g.

http://pytest.org/dev/xdist.html

Would it be easier to run 10 parallel pytest-xdist than …

我有这样的代码：

def generator():
    while True:
        # do slow calculation
        yield x

我想将缓慢的计算转移到单独的进程中。

我正在 python 3.6 中工作，所以我有concurrent.futures.ProcessPoolExecutor. 如何使用它来并发生成器并不明显。

与常规并发场景使用的区别map在于，这里没有什么可映射的（生成器永远运行），并且我们不希望一次获得所有结果，我们希望将它们排队并等到队列未满之前计算更多结果。

我不用用concurrent，multiprocessing也可以。这是一个类似的问题，如何在生成器内部使用它并不明显。

略有不同：生成器返回的每个值都是一个大的 numpy 数组（10 MB 左右）。如何在不酸洗和不酸洗的情况下转移它？我已经看过文档，multiprocessing.Array但如何使用它来传输 numpy 数组并不完全明显。