问题标签 [pathos]

我有一个脚本，它使用 pathos ProcessPooldata()在多个数据点上并行执行一个函数。

调用的函数data()计算并将结果发送到调用的单个对象Database，由于大量推送被发送到数据库，我添加了一个简单的队列，当队列大小大于 500 时，它被推送到数据库。

当所有数据点都已映射但类队列中仍有一定量的数据Database（在每个进程中，进程 1 可能有 100 个进程 2 可能有 200 个）时，就会出现问题。

我写了一个函数来在退出时刷新，但是如何在所有进程上调用该函数？

下面是一些玩具示例，用于重现我在 Python 3.5 上使用 pathos.multiprocessing 时遇到的问题。一个问题是并行过程无法识别是什么Test，即使Test没有在test(). 我已经看到一些帖子解决了第二个问题，说我需要一个import numpy as np内部test函数，但这对我不起作用。

我得到的错误是：_pickle.PicklingError: Can't pickle <enum 'TestEnum'>: it's not found as builtins.TestEnum

如果我要摆脱所有出现的TestEnum，那么下一个错误将np是无法识别。我在该站点上看到其他帖子表明import numpy as np在顶部需要an main()，但这对我不起作用。当我尝试在其中导入模块时遇到的错误main()是：ImportError: __import__ not found

我试图了解如何使用 Pathos 包来运行调用函数的函数。我的理解是 Pathos 相对于主要的多处理包的优势在于它允许函数内部的函数。但是，我似乎无法让它工作。这是我能想到的最简单的例子：

输出：

编辑：似乎如果我将此代码粘贴到 shell 控制台中，它工作正常。如果我从我选择的 IDE PyCharm 中运行它，那就没有任何问题了。所以现在我的问题是为什么相同的代码在相同版本的python解释器（3.6.1）中的工作方式不同，具体取决于它是从shell还是应用内控制台运行/

我有一堆长时间运行的进程，我想将它们分成多个进程。那部分我可以做没有问题。我遇到的问题是有时这些进程会进入挂起状态。为了解决这个问题，我希望能够为进程正在处理的每个任务设置时间阈值。当超过该时间阈值时，我想重新启动或终止任务。

最初我的代码使用进程池非常简单，但是使用池我无法弄清楚如何检索池内的进程，更不用说如何重新启动/终止池中的进程。

我已使用队列和进程对象，如本示例所示（https://pymotw.com/2/multiprocessing/communication.html#passing-messages-to-processes进行了一些更改。

我试图弄清楚这一点在下面的代码中。在其当前状态下，进程实际上并未终止。除此之外，我无法弄清楚在当前任务终止后如何让进程转移到下一个任务。任何建议/帮助表示赞赏，也许我会以错误的方式解决这个问题。

谢谢

我正在尝试将 pathos 多处理用于一个简单的函数，我为其传递一个数组，它拆分为并行进程，然后在最后加入。

这是我使用多处理功能移植过来的代码，然后我没有导入多处理，而是导入了 pathos。

但是，我收到此错误：

我不确定如何提取输出结果。pathos 的文档很薄。

有没有人有任何想法或经验？

仅供参考，我在 jupyter 笔记本中运行 python。

我正在为我们在团队中进行的一些测试开发一个自动化框架。我们遇到了这个 Pickling 错误，在做了一些研究之后，我遇到了一些解决方案并查看了随附的 github 页面：

• Python多处理酸洗错误
• 多处理：如何在类中定义的函数上使用 Pool.map？
• python pathos 的酸洗问题<---从这个链接，我确认代码引用了正确的库。
• https://github.com/uqfoundation/pathos

我安装了 pathos 并相信我已经正确实现了它，但我仍然收到 Pickling 错误。这是示例代码和回溯：

test_start_callback 是一个工厂函数，用于生成不带参数的函数。我编辑了未使用的代码段。

追溯

我错过了其他 SO 链接中未提及的内容，或者我传递此函数的方式也不被 pathos 支持。

我有以下问题：我想使用 pathos.multiprocessing 中的 amap。

尽管我可以在 f 中做任何事情，但必须有这两个函数 f 和 g。

如果我运行此代码，我会得到 g 需要 3 个参数，但给出了一个。

我怎样才能解决这个问题？

我目前正在创建一个类，该类应该使用该multiprocessing模块以多线程方式执行某些方法。Pool我使用一个n工人执行真正的计算。现在我想为每个当前n活跃的工作人员分配一个索引0，n用于其他计算。为此，我想使用 sharedQueue以某种方式分配索引，即每次都没有两个工人具有相同的 id。为了在不同线程之间共享Queue类内部相同的内容，我想将其存储在Manager.Namespace(). 但是这样做，我遇到了一些问题Queue。因此，我创建了我的问题的最小版本，并最终得到了这样的结果：

在这段代码中，执行在第二个 print 语句之前停止 - 因此，我认为实际上没有数据写入Queue. 当我删除namespace相关的东西时，代码可以完美地工作。这是multiprocessings 对象的预期行为吗？我做错了什么吗？或者这是某种错误？

我有一个代码，其中共享资源由函数调用更改。到目前为止，对于每个输入向量（维度 rxc 的输入矩阵），我都在串行运行它。

我想在每个函数调用中更改一个共享资源（比如 R）。到目前为止，我已经尝试使用 Python Multiprocessing 和 Pathos Multiprocessing 来加快速度。我什至尝试过 np.apply_along_axis 来加快速度。

到目前为止，我注意到串行处理是最快的方式。我不知道为什么会这样。

我尝试了以下方法

1. np.apply_along_axis ：只有轻微的延迟（恒定时移）
2. Pathos.multiprocessing.ProcessingPool.Map ：最多延迟 10 倍
3. multiprocessing.Process（手动拆分）：最多延迟 5 倍
4. multiprocessing.Pool ：符合 Pathos 结果。

我是 python 并行编程的新手，可能做错了什么。有什么好的方法吗？

更新：附上自组织地图代码

我在引用我正在做的事情。

我在跑

并在运行时打印以下内容：

这是什么意思？这是否意味着我可以采取任何措施来防止错误提高性能？