问题标签 [multiprocessing-manager]

我需要在 python 中的多个进程之间共享一个对象及其方法。我正在尝试使用managers（在模块中multiprocessing），但它崩溃了。这是一个生产者-消费者的愚蠢示例，其中两个进程之间的共享对象只是具有四种方法的数字列表。

无论如何，它总是这样崩溃，告诉我：

那是怎么回事？我应该如何使用这些managers来共享对象及其方法？

我有一个存储对象的字典：

现在我想将它转换为使用管理器字典，因为我想使用多处理并且需要共享这个字典 amonst 进程

只是通过转换为使用 manager.dict() 事情变得非常缓慢。

例如，如果使用本机 dict，则只需 0.65 秒即可创建 625 个对象并将其存储到 dict 中。

同样的任务现在需要 126 秒！

我可以做任何优化以使 manager.dict() 与 python {} 保持一致？

我正在尝试更新多处理模块 manager.dict() 的嵌套字典中的键，但无法这样做。它不会更新值，也不会抛出任何错误。

代码：

输出：

进行中

{'abc': {'status': 'ongoing', 'fail': 0, 'repeat': 5, 'success': 5}}

我的期望是，一旦 while 循环结束，它应该使 d['abc']['status'] = 完成。但在最终打印时，它仅将其状态打印为“进行中”。

我有一个给定两个参数的代码， (k, m) 将返回一个 4d numpy 数组，我的要求是我需要计算这个数组以获得 (k,m) 的可能值，其中 k,m < N 并将它们相加. 这在串行上很慢，所以我正在尝试学习 python 中的多处理模块来做到这一点。 https://docs.python.org/2/library/multiprocessing.html

本质上，我想使用我的 8 个内核来并行计算这些 4d 数组并将它们全部加起来。现在的问题是如何设计这个。每个数组大约为 100 MB，N 大约为 20。因此，无法在队列中存储 20**2 * 100 MB。解决方案是拥有一个共享内存对象，一个结果数组，每个进程将不断将结果添加到该数组中。

multiprocessing 有两种方法可以做到这一点，使用共享内存或服务器进程。它们似乎都不支持 mutlidim 数组。谁能建议一种方法来实现我的程序？提前谢谢。

我有一个我正在多处理中处理的元素列表apply_async任务中处理这些元素，并使用 manager dict 中的一个键逐个更新处理的元素，我想在其上映射整个列表。

我尝试了以下代码：

最后，它只是在 dict 中发布空列表。输出：

{'task': {'processed_list': []}}

现在经过一番研究，我知道 manager dict 中的元素变得不可变，所以你必须用新数据重新初始化整个 dict 才能更新它。所以我尝试了以下代码，它给出了一个奇怪的错误。

输出：

回溯（最后一次调用）：文件“./a.py”，第 29 行，在 p.apply_async(spammer_task (d, my_list)) 文件“./a.py”，第 14 行，在 spammer_task new_list = old_list。 append(ele) AttributeError: 'NoneType' 对象没有属性 'append'

不知何故，它似​​乎附加None到我无法弄清楚原因的列表中。

我正在运行一个应用程序，它不能坐下来等待与 Python 管理器的成功/不成功连接。客户端应用程序应该尝试将一些信息发送到假定正在运行的服务器，如果它失败，则采取另一种措施。问题是每当服务器关闭时，连接需要很长时间才能将控制权返回给客户端应用程序，并且它不能浪费时间等待它，因为还有其他事情要做。

我想出了一个方案，其中一个中间对象负责连接，但它只工作一次。假设第一次，当仍然没有连接到服务器时，这个中间对象在不阻塞客户端应用程序的情况下处理连接部分。如果由于某种原因，服务器宕机又回来了，我就不能让它工作了。

假设我有以下服务器：

这是处理连接的中间对象：

最后是客户端应用程序：

我肯定在线程上做错了什么，但我无法弄清楚。任何想法？

我的问题与这里的问题非常相似，除了捕获的解决方案对我来说不太有效。

问题：我正在使用多处理来并行处理文件。大约 97%，它有效。但是，有时，父进程会永远空闲，CPU 使用率显示为 0。

这是我的代码的简化版本

问题是我可以看到进程正在启动，并且我可以看到所有项目都在处理中。但是，有时，进程不会重新加入。当我回来查看时，只有父进程在那里，但它处于空闲状态，就好像它在等待什么一样。没有任何子进程，但我不认为 join 被调用，因为我的 print 语句没有出现。

我的假设是，这发生在有很多断帧的视频上。但是，重现此错误有点困难，因为它很少发生。

编辑：代码现在应该有效。试图找到可以重现此错误的文件。

这是主要代码：

这是 func() 定义：

如代码中所述，它正在打印空列表而不是预期值。如何使这个共享的字典和列表正常工作？

第一次打印显示预期值。第二次打印显示预期的键，但值部分是空列表。由于它正确显示了密钥，因此共享工作正常。附加部分存在一些问题。我无法弄清楚。

否决选民，在您认为我的问题与那个问题相同之前，请仔细阅读链接的问题。如果您仍然坚持认为它们是相同的，请发表评论说明您为什么这么认为。在我看来，这两个问题是不同的。链接的问题询问如何与建议使用 Manager.dict() 的答案共享 dict。我的问题已经使用 Manager.dict() 并且由于异常而失败。谢谢你。

我有一个简单的程序，可以搜索本地网络中的开放端口，并将连接的套接字连同它们的本地地址一起存储在字典中。现在，我使用 Manager 共享字典来存储这些条目，但它只接受简单对象而不接受套接字实例。这是代码：

我得到的错误是：

我对代理进行了简短的研究，但找不到将它们实现到我的代码中的方法，所以我寻求帮助。如何使套接字实例与 Manager 字典兼容？

问题：

我正在从服务器下载财务数据，然后处理这些数据。我确实会同时收集多只股票的数据。

我需要并行处理数据下载器和数据处理器（因为数据处理器本身将由多个进程组成）。

我绝对需要以序列化方式处理每只股票的数据，但如果我有超过 1 只股票，我必须以并行方式处理这些股票。

我对问题的理解：

从我收集到的信息中，我需要一种方法从单一来源获得一种将这些数据传输到并行进程的方法，这些进程预先确定哪些数据（根据股票 ID）进入每个进程（每个股票都有自己的进程）。

到目前为止，我已经尝试了几种不同的方法，但都没有成功，我只需要解决这个错误：

可能的解决方案

我将尝试实现的下一件事是使用multiprocessing.Manager().dict()withcollections.deque或multiprocessing.Queue()orlist()作为元素并为mp.Process()实例（为每只股票）创建一个字典。

重要的是可以动态分配这些数据结构，因为我可能会在运行时更改库存。

问题

解决此问题的有效方法是什么？

直觉上似乎有比使用更好的方法multiprocessing.Manager().dict()来完成这项任务，但我还没有找到。有这种事吗？