问题标签 [process-pool]

这是我的代码：

如果我在上层范围内运行（不包含在test函数中），它会按预期工作，但是在将它包含在另一个函数中之后 - 它们只是没有被调用。

这是我收到的test功能输出：

没有：

所以问题是 -我怎样才能让它在另一个函数中工作？

我正在尝试使用 concurrent.futures.ProcessPoolExecutor() 运行一些非常大的时间序列数据。数据集包含多个时间序列（它们是独立的）。整个数据集在data我通过辅助函数传递的元组列表中可用，如下所示：

现在，虽然数据中包含的不同时间序列在列之间是独立的，但是由于时间序列数据的性质，一个时间序列内的值需要一个接一个地处理。通过data根据不同的时间序列对变量进行排序，我确信 map 总是会随着时间的推移按顺序进行调用。

由于executor.map我无法找到始终将特定时间序列映射到同一个核心的方法，或者以某种方式将先前尝试的状态共享到在新核心上运行的进程。

使用当前设置，每当在新内核上调用特定时间戳的处理时，它都会从初始化步骤开始。

这个问题有什么优雅的解决方案吗？

我们有一个异步任务调度系统，它使用 golang 的 exec.Command 来执行 php 脚本。每次，调度程序从消息队列中获取一个任务，它会创建一个新的 exec.Command 来执行该任务。

有时，几乎有数千个任务需要同时执行。在这种情况下，调度程序将创建数千个 exec.Command，然后在一段时间后将它们销毁。

我想知道是否有办法创建一个进程池（或类似的东西），以便我们可以重用 exec.Command 来降低创建新子进程的成本。

ps：我注意到exec.Command在调用运行后不能重用。

UPATE 当前的逻辑如下：

所以我有以下代码可以正常工作：

然后我尝试 zip() 打印 ProcessPoolExecuter 正在处理的迭代，但是当我运行以下代码时没有打印任何内容：

我不知道为什么。有人可以帮忙吗？

我有一个类A，它在启动时会更改一个可变类属性nums。

当通过带有 的进程池启动类时，我注意到它具有几个不同进程的值。这对我来说是一种不良行为。maxtasksperchild= 1nums

我的问题是：

• 进程是否共享内存？
• 我不maxtasksperchild正确理解进程池的工作原理吗？

编辑：我猜这个池会腌制它开始的前一个进程（而不是原来的进程），从而保存 的值nums，对吗？如果是这样，我该如何强制它使用原始流程？

这是一个示例代码：

编辑因为 k.wahome 的回答：使用实例属性不能回答我的问题我需要使用类属性，因为在我的原始代码（此处未显示）中，每个进程有几个实例。我的问题专门关于多处理池的工作原理。

顺便说一句，执行以下操作确实有效

I have a method inside a class that needs to do a lot of work in a loop, and I would like to spread the work over all of my cores.

I wrote the following code, which works if I use normal map(), but with pool.map() returns an error.

Error 1:

AttributeError: Can't pickle local object 'SomeClass.some_method..single'

Why can't it pickle single()? I even tried to move single() to the global module scope (not inside the class - makes it independent of the context):

and I get the following ...

Error 2:

AttributeError: Can't get attribute 'single' on module 'main' from '.../test.py'

当其中一个孩子中止和/或抛出异常时，中止多处理的正确方法是什么？

我发现了各种各样的问题（通用多处理错误处理，如何在异常但没有答案时关闭多处理池，...），但没有关于如何停止对子异常进行多处理的明确答案。

例如，我期望以下代码：

输出：

相反，它f在检测/处理异常之前对所有项目应用函数：

没有办法直接捕获异常吗？

我正在使用多处理池来管理 tesseract 进程（缩微胶片的 OCRing 页面）。很多时候，在一个包含 20 个 tesseract 进程的池中，有几页会更难 OCR，因此这些进程比其他进程花费的时间要长得多。同时，池只是挂起，大多数 CPU 都没有被利用。我希望让这些落后者继续下去，但我也想启动更多进程来填满现在闲置的许多其他 CPU，而这几个粘性页面正在完成。我的问题：有没有办法加载新进程来利用那些空闲的 CPU。换句话说，可以在等待整个池完成之前将池中的空位填满吗？

我可以使用星图的异步版本，然后在当前池下降到一定数量的活动进程时加载一个新池。但这似乎不优雅。根据需要自动保持在进程中的插槽会更优雅。

这是我的代码现在的样子：

类似的问题（但答案对我不起作用）：如何取消使用 concurrent.futures.ProcessPoolExecutor 运行的长时间运行的子进程？

与上面链接的问题和提供的解决方案不同，在我的情况下，计算本身相当长（受 CPU 限制）并且不能循环运行以检查是否发生了某些事件。

以下代码的简化版本：

这个想法是有一个主模拟循环运行和监控三个机器人线程。然后，这些机器人线程中的每一个都会执行一些推理，但也会使用 启动一个非常长的后台进程ProcessPoolExecutor，这可能最终会运行更长的阈值/最大执行时间来对事物进行推理。

.cancel()正如您在上面的代码中看到的，我在发生超时时尝试执行这些任务。虽然这并没有真正取消实际的计算，但它一直在后台发生，并且asyncio循环不会终止，直到所有长时间运行的计算完成之后。

如何在方法中终止如此长时间运行的 CPU 密集型计算？

其他类似的 SO 问题，但不一定相关或有帮助：

1. asyncio：是否可以取消由 Executor 运行的未来？
2. 如果单个异步任务超过 Python 中的阈值时间，如何在多处理中终止单个异步任务
3. Python中带有工作池的异步多处理：超时后如何继续？

我有以下代码片段，它尝试在多个子流程之间拆分处理。

主进程在 while 循环中调用搜索函数，如果队列达到阈值计数，则处理池将映射到进程函数，其中作业来自队列。我的问题是，python多处理池在执行期间会阻塞主进程还是立即继续执行？我不想遇到“has_jobs_to_process()”评估为 true 并且在处理作业期间，它为另一组作业评估为 true 并且调用“self._process_pool.map(process,populated_sub_lists)”的情况再次因为我不知道在进程运行时再次调用 map 的后果。