问题标签 [python-multiprocessing]

我有一个作为守护进程运行的 Python 脚本。启动时，它产生 5 个进程，每个进程都连接到 Postgres 数据库。现在，为了减少数据库连接的数量（最终会变得非常大），我正在尝试找到一种跨多个进程共享单个连接的方法。为此，我正在研究multiprocessing.sharedctypes.ValueAPI。但是，我不确定如何psycopg2.connection跨进程使用此 API 传递对象。谁能告诉我怎么做？

为了解决这个问题，我也愿意接受其他想法。

我没有考虑将连接作为构造函数的一部分传递给 5 个进程的原因是互斥处理。如果我遵循这种方法，我不确定如何防止多个进程访问连接。有人能告诉我这是否是正确的做法吗？

I need to run a long foobar.py process with Popen and parse its output with a multiprocessing process.

My problem is that sometimes I cannot wait for the parser to finish, so I need to daemonize the parser using the multiprocessing daemon property. I need the parser to be usable both deamonic non daemonic ways. The doc also says that a daemonic process is not allowed to create child processes. So in that case I run the Popen process before the parser is forked (see the start method below).

Let's say here that foobar.py just waits some seconds and print something, and parseFoobar method just prints output in a file. In my case both functions are a lot more complex than this.

Running Parser(daemon=False, "sync.txt").start() works fine and there is some output in sync.txt. But running Parser(daemon=True, "async.txt") does not produce anything in async.txt and seem to be blocked at line for line in iter(self.process.stdout): because the file is created, but it is empty.

Why doesn't it work? How can I fix it?

You can find gists for parser.py and foobar.py for testing. Just run python parser.py and look at output files.

Edit: There are some tips in django daemonize methods

TL;博士

根据 Valgrind 的 memcheck 工具，如果我在函数中分配一个大的局部变量并使用 启动该函数multiprocessing.Pool().apply_async()，则子进程和主进程的堆大小都会增加。为什么 main 的堆大小会增加？

背景

我正在与一个多处理工作人员池一起工作，每个工作人员都将处理来自输入文件的大量数据。我想看看我的内存占用如何根据输入文件的大小进行缩放。为此，我在 Valgrind 下使用 memcheck 和此 SO answer中描述的技术运行我的脚本。（从那以后我了解到 Valgrind 的 Massif 工具更适合这个，所以我将使用它来代替。）

memcheck 输出中有一些看起来很奇怪的东西，我想帮助理解。

我在 Red Hat Linux 上使用 CPython 2.7.6，并像这样运行 memcheck：

valgrind --tool=memcheck --suppressions=./valgrind-python.supp python test.py

代码和输出

堆摘要（每个进程一个）：

总堆使用量：45,193 分配，32,392 释放，7,221,910 字节分配
总堆使用：44,832 分配，22,006 释放，7,181,635 字节分配

如果我将tmp = 'a'*1行更改为tmp = 'a'*10000000我得到这些摘要：

总堆使用量：44,835 次分配，22,009 次释放，分配 27,181,763 字节
总堆使用量：45,195 次分配，32,394 次释放，分配 17,221,998 字节

问题

为什么两个进程的堆大小都会增加？我知道对象的空间是在堆上分配的，因此较大的堆对于其中一个进程当然是有意义的。但是我希望子进程拥有自己的堆、堆栈和解释器实例，所以我不明白为什么在子进程中分配的局部变量也会增加主进程的堆大小。如果它们共享同一个堆，那么 CPython 是否实现了自己的 fork() 版本，它不会为子进程分配唯一的堆空间？

将此标记为已回答并围绕速度问题真正出现的地方开始一个更简单的主题

Python慢​​读性能问题

感谢您迄今为止的所有评论，非常有用

我有大约 40M 的 XML 文件分布（不均匀）。60K 子目录，结构基于 10 位数字拆分，因此：

12/34/56/78/90/files.xml

我有一个 perl 脚本，它针对文件运行，将单个字段的值拉出并打印值和文件名。Perl 脚本包含在一个 bash 脚本中，该脚本在深度 2 的所有目录列表中运行最多 12 个并行实例，然后遍历每个实例并在找到它们时在底层处理文件。

从多个运行中取出磁盘缓存，进程的 unix 时间返回大约：

我想将其迁移到 python 脚本（作为学习练习和测试），因此创建了以下内容（在大量阅读各种 python 方法之后）：

但是无论我在运行它时如何对内容进行切片，unix time 都会给我这样的结果：

如果我在一个线程中针对一个小子集（最终被完全缓存）同时运行 python 和 perl 脚本，那么就没有磁盘 io（根据 iotop），那么脚本运行的时间几乎相同。

到目前为止，我能想到的唯一结论是，文件 io 在 python 脚本中的效率远低于在 perl 脚本中的效率，因为似乎是 io 导致了问题。

所以希望这是足够的背景，我的问题是我是在做一些愚蠢的事情还是错过了一个技巧，因为我的想法已经用完了，但不能相信 io 在处理时间上造成了如此大的差异。

感谢任何指针，并将根据需要提供更多信息。

谢谢

硅

参考 Perl 脚本如下：

I have few basic questions when it comes to using Python's multiprocessing module :

Is it necessary to pass work_queue and result_queue as args to Process? Does the answer depends on the OS? The more fundamental question is: does the child process get a copied (COW) address space from the parent process, and hence knows the definition of the class/class method? If yes, how does it know that the queues are to be shared for IPC, and that it shouldn't make duplicates of the work_queue and result_queue in the child process? I tried searching this online but most of the documentation I found was vague, and didn't go into enough details as what exactly is happening underneath.

我已经问过了，很少有人给出好的建议，但我还是初学者，对我来说有很多未知数。因此我决定再次寻求帮助而不给出错误的代码。

我需要一个脚本，它会在另一个仍在运行时执行将文件复制到目录。基本上我运行第一个命令，它生成文件（直到用户按回车键）然后这些文件消失了（自动删除）。
我想要的是复制这些文件（也不必按“Enter”）。我用 bash 制作，但是我想在 python 上实现这一点。请看下面：

我想multiprocessing在我的代码中使用以获得更好的性能。

但是，我收到如下错误：

我已经尝试了另一种方式并得到了这个错误：

我的代码如下所示：

我认为由于未Pool在 main 函数中使用而引发错误。

我的猜想对吗？以及如何修改我的代码来修复错误？

我在 python 中有一个进程池，使用from multiprocessing import Pool. 我将要运行的不同函数传递给这些进程，调用该Pool.apply_async函数。如果我在其中一个函数中引入错误（即：一行代码，例如5/0），则预期的命令行报告ZeroDivisionError: integer division or modulo by zero永远不会出现，并且程序永远不会终止。即使我在对 的调用中引入了回调函数，Pool.apply_async如果进程必须执行的函数有错误，回调函数也不会被调用。

如果出现问题，如何让池中的这些进程报告错误并终止？

今天玩锁，注意到以下情况。假设我有以下需要锁定的代码。使用获取和释放的旧方式：

with使用以下构造无法实现这段代码：

或者可以吗？如果我错了，请纠正我。

假设您要并行运行多个进程（使用多处理，可能在多个单独的机器上，如在集群中），其中每个进程创建特定类的新实例列表。然后，您将所有这些列表发送回父进程，并希望将它们组合起来。现在，我们可以通过它们的对象 id 来索引这些实例吗？鉴于每个对象都是在单独的进程（可能是单独的机器）上生成的，我可以期望 id 唯一地标识对象吗？

换句话说，对象的 id 是否能够在进程之间发送数据所需的酸洗中存活下来，或者解释器在解开对象时是否为对象分配了一个新鲜且唯一的 id？