标签: python-multiprocessing

我正在使用多处理做一些非常简单的事情:

data = {'a': 1}
queue.put(data, True)
data.clear()

当我在另一个进程(使用get()方法)上使用队列时,我得到一个空字典.如果我删除data.clear()我得到按预期的键.有没有办法等待put()完成序列化？

我正在从大型CSV文件中读取数据,对其进行处理并将其加载到SQLite数据库中.分析表明80%的时间花在I/O上,20%是处理输入以准备数据库插入.我加快了处理步骤,multiprocessing.Pool以便I/O代码永远不会等待下一条记录.但是,这导致了严重的内存问题,因为I/O步骤无法跟上工作人员的步伐.

以下玩具示例说明了我的问题:

#!/usr/bin/env python  # 3.4.3
import time
from multiprocessing import Pool

def records(num=100):
    """Simulate generator getting data from large CSV files."""
    for i in range(num):
        print('Reading record {0}'.format(i))
        time.sleep(0.05)  # getting raw data is fast
        yield i

def process(rec):
    """Simulate processing of raw text into dicts."""
    print('Processing {0}'.format(rec))
    time.sleep(0.1)  # processing takes a little time
    return rec

def writer(records):
    """Simulate saving data to SQLite database."""
    for r in records:
        time.sleep(0.3)  # writing takes the longest
        print('Wrote {0}'.format(r)) …

There as several similar questions that I have already read on Stack Overflow. Unfortunately, I lost links of all of them, because my browsing history got deleted unexpectedly.

All of the above questions, couldn't help me. Either, some of them have used CELERY or some of them SCRAPYD, and I want to use the MULTIPROCESSISNG Library. Also, the Scrapy Official Documentation shows how to run multiple spiders on a SINGLE PROCESS, not on MULTIPLE PROCESSES.

None of them couldn't help …

我有一个相当简单的嵌套for循环,迭代四个数组:

for a in a_grid:
    for b in b_grid:
        for c in c_grid:
            for d in d_grid:
                do_some_stuff(a,b,c,d)  # perform calculations and write to file

也许这不是开始在4D网格上执行计算的最有效方法.我知道joblib能够像这样并行化两个嵌套的for循环,但是我很难将它概括为四个嵌套循环.有任何想法吗？

根据 https://github.com/joblib/joblib/issues/180,有没有一种安全的方法从python中的线程创建子进程？ Python多处理模块不允许在线程内使用.这是真的？

我的理解是,从线程中分叉是好的,只要你没有持有线程.当你这样做时(在当前线程中？在进程中的任何地方？).但是,Python的文档没有说明在fork之后是否安全地共享threading.Lock对象.

还有这个:从日志记录模块共享的锁导致fork问题.https://bugs.python.org/issue6721

我不确定这个问题是怎么产生的.听起来当进程中的任何锁的状态都被复制到当前线程分叉的子进程中(这看起来像设计错误并且肯定会死锁).如果是这样,那么使用多真正提供对这种任何保护(因为我可以自由地创建我multiprocessing.Pool被创建和被其它线程进入后threading.Lock,经过线程已经开始,使用不是叉安全日志模块) - 多处理模块docs也没有说明是否应该在Locks之前分配multiprocessing.Pools.

用多处理替换threading.Lock.Lock到处避免这个问题并允许我们安全地组合线程和分支？

我正在改进一个解析XML并对其子树进行分类和索引的程序.实际程序太大而无法显示在这里,所以我把它归结为一个最小的测试用例,显示我遇到的问题.

这个想法是:

1. 逐个处理目录中的XML文件
2. alpino_ds并行处理文件中的所有节点
3. 在此过程中,该进程需要对共享变量进行读/写访问,以便我们可以检查属性总共发生了多少次,或者跟踪文件句柄

请注意,在实际代码中还有一些注意事项:

• simply returning new values per process and then merging them in the main thread seems not advisable and presumably quite slow because the actual data structure are dicts of four levels deep consisting of dicts, sets, ints, and strings, as well as dict-to-filehandle, and Counter() objects;
• I tried using threads (with ThreadPoolExecutor) and even though there was some gain (I calculated around 5% improvement in speed), this …

多处理模块对于python初学者来说非常困惑,特别是那些刚刚从MATLAB迁移并且使用并行计算工具箱变得懒惰的人.我有以下功能需要大约80秒运行,我想通过使用Python的多处理模块来缩短这个时间.

from time import time

xmax   = 100000000

start = time()
for x in range(xmax):
    y = ((x+5)**2+x-40)
    if y <= 0xf+1:
        print('Condition met at: ', y, x)
end  = time()
tt   = end-start #total time
print('Each iteration took: ', tt/xmax)
print('Total time:          ', tt)

这按预期输出:

Condition met at:  -15 0
Condition met at:  -3 1
Condition met at:  11 2
Each iteration took:  8.667453265190124e-07
Total time:           86.67453265190125

由于循环的任何迭代都不依赖于其他循环,我尝试从官方文档中采用此服务器进程来在单独的进程中扫描范围的块.最后我想出了vartec对这个问题的回答,可以准备以下代码.我还根据Darkonaut对当前问题的回答更新了代码.

from time import time 
import multiprocessing as …

multiprocessing我正在使用 SLURM 在集群上测试 Python 的模块。我想绝对确保我的每个任务实际上都按照我的意图在单独的 cpu 核心上运行。由于配置 SLURM 的可能性有很多，因此这一点并不明显。

因此，我想知道是否有一种方法可以从正在运行的Python任务中获取核心特定信息。我需要我的 Python 脚本来获取有关其运行的核心的信息，以便区分各个核心。

考虑以下脚本。已启动 10 个任务。有没有办法让它们每个人打印有关它们所分配到的核心的核心特定信息？

import multiprocessing

def hello():
    print "Hello World"

pool = multiprocessing.Pool()
jobs = []
for j in range(10):
    p = multiprocessing.Process(target = hello)
    jobs.append(p)
    p.start()

我在这里和这里找到的模块提供了 CPU 特定信息，但这些是关于可用 CPU 的信息，而不是运行时的核心信息。

我正在使用 python 3.6.7 和 Ubuntu 18.04

运行以下脚本后，每个进程都有自己的共享锁：

from multiprocessing import Process, Manager


def foo(l1):
    with l1:
        print('lol')


if __name__ == '__main__':
    processes = []
    with Manager() as manager:
        for cluster in range(10):
            lock1 = manager.Lock()
            calc_args = (lock1, )
            processes.append(Process(target=foo,
                                     args=calc_args))
        for p in processes:
            p.start()

        for p in processes:
            p.join()

我有一个奇怪的例外：

Process Process-2:
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "temp.py", line 5, in foo …

在我的代码中，我使用 multiprocessing.Pool 来同时运行一些代码。简化的代码看起来有点像这样：

    class Wrapper():
        session: Session       


        def __init__(self):
            self.session = requests.Session()
            # Session initialization


        def upload_documents(docs):
            with Pool(4) as pool:
                upload_file = partial(self.upload_document)
                pool.starmap(upload_file, documents)

                summary = create_summary(documents)

            self.upload_document(summary)


        def upload_document(doc):
            self.post(doc)


        def post(data):
            self.session.post(self.url, data, other_params)

所以基本上通过 HTTP 发送文档是并行的。现在我想测试这段代码，但无法做到。这是我的测试：

    @patch.object(Session, 'post')
    def test_study_upload(self, post_mock):
        response_mock = Mock()
        post_mock.return_value = response_mock
        response_mock.ok = True

        with Wrapper() as wrapper:
            wrapper.upload_documents(documents)

        mc = post_mock.mock_calls

在调试中我可以检查模拟调用。有一个看起来有效，它是上传摘要的那个，以及一堆诸如call.json()、call.__len__()等call.__str__()调用。

没有电话上传文件。当我在方法中设置断点时upload_document，我可以看到每个文档都调用一次它，它按预期工作。但是，我无法测试它，因为我无法通过模拟验证此行为。我认为这是因为有许多进程调用同一个模拟，但仍然 - 我该如何解决这个问题？

我使用Python 3.6