标签: python-multiprocessing

在我的项目中，我有一长串要定价的交易，因此我将其分成多个块并将它们委托给 multiprocessing#Pool 进行处理。假设包装类称为Valuation，方法是pool_calculate () ，它看起来像：

    multiprocessing.freeze_support() 
    p = multiprocessing.Pool()  # default is a number of processes equal to the number of CPU cores
    results = p.map(self.pooled_valuation_tasks, chunks)
    p.close()
    p.join()

问题是我们必须为所有定价导入一个内部模块，比如magic。

import magic
class Valuation(object):
    ....

该模块似乎没有传播到池工作人员中，错误消息如下所示：

Process SpawnPoolWorker-3:
Traceback (most recent call last):
  File "c:\opt\magic0.0.46\PythonVenv\lib\multiprocessing\process.py", line 258, in _bootstrap
    self.run()
  File "c:\opt\magic0.0.46\PythonVenv\lib\multiprocessing\process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "c:\opt\magic0.0.46\PythonVenv\lib\multiprocessing\pool.py", line 108, in worker
    task = get()
  File "c:\opt\magic0.0.46\PythonVenv\lib\multiprocessing\queues.py", line 337, in get
    return _ForkingPickler.loads(res) …

Python 3.8 引入了新的共享内存功能。我们正在尝试使用SharedMemoryManager并且 aNameError被抛出。

我认为我们可能在复杂的场景中做错了什么，所以我使用 python 文档片段对其进行了分解。

try:
    # python >= 3.8
    from multiprocessing.managers import SharedMemoryManager as Manager
except:
    # python < 3.8
    from multiprocessing.managers import BaseManager as Manager

class MathsClass:
    def add(self, x, y):
        return x + y
    def mul(self, x, y):
        return x * y

class MyManager(Manager):
    pass

MyManager.register('Maths', MathsClass)

if __name__ == '__main__':
    with MyManager() as manager:
        maths = manager.Maths()
        print(maths.add(4, 3))         # prints 7
        print(maths.mul(7, 8))         # prints 56

这几乎取自多处理文档（回退导入除外），在 python 3.7 …

我创建了一个简单的脚本，它遍历我需要 ping 和 nslookup 的服务器列表。问题是，ping 可能需要一些时间，尤其是 ping 比一天中的秒数更多的服务器。

我对编程相当陌生，我知道多处理或多线程可能是使我的工作运行得更快的解决方案。

我的计划是获取我的服务器列表和 1. 将它分成大小均匀的列表，列表的数量与线程/进程匹配，或者 2. 如果这些选项之一支持它，则遍历单个列表，每个列表都传递一个新的在完成之前的 ping 和 nslookup 后，线程或进程的服务器名称。这是可取的，因为它确保我花费最少的时间，就好像列表 1 有 200 个离线服务器而列表 6 有 2000 个，它需要等待使用列表 6 的过程完成，即使所有其他人在那个时候都是免费的观点。

1. 哪一个更适合这项任务，为什么？

2. 如果可能，我将如何确保每个线程或进程具有基本相同的运行时

代码片段即使现在相当简单

import subprocess
import time
server_file = open(r"myfilepath", "r")
initial_time = time.time()
for i in range(1000):
    print(server_file.readline()[0:-1]+ ' '+str(subprocess.run('ping '+server_file.readline()[0:-1]).returncode)) #This returns a string with the server name, and return code,
print(time.time()-initial_time)

出现此问题是因为每次失败的 ping 平均需要 3 秒以上。此外，我知道不放置打印语句会使它更快，但我想监视它的一个小案例。我正在对 100,000 台服务器的影响进行 ping 操作，这将需要定期执行，并且该列表将继续增长

我正在尝试使用 python 多处理模块在多个设备上并行运行一堆 linux 命令。脚本能够串行连接到多个设备并开始并行执行 linux 命令。脚本在第一次尝试中失败，但在第二次尝试中成功。

请查找错误日志：

    -E- !!! *** Exception occurred during execution *** !!!
    -E- Exception Name: ExceptionPexpect
    -E- Exception trace back: 
        Traceback (most recent call last):
      File "/home/venkat/Desktop/setup_cleanup_with_parallel/test/lib/python3.5/site-packages/ptyprocess/ptyprocess.py", line 705, in isalive
        pid, status = os.waitpid(self.pid, waitpid_options)
    ChildProcessError: [Errno 10] No child processes

    During handling of the above exception, another exception occurred:

    Traceback (most recent call last):
      File "/home/venkat/Desktop/setup_cleanup_with_parallel/test/lib/python3.5/site-packages/pexpect/pty_spawn.py", line 23, in _wrap_ptyprocess_err
        yield
      File "/home/venkat/Desktop/setup_cleanup_with_parallel/test/lib/python3.5/site-packages/pexpect/pty_spawn.py", line 705, in isalive
        alive = ptyproc.isalive()
      File "/home/venkat/Desktop/setup_cleanup_with_parallel/test/lib/python3.5/site-packages/ptyprocess/ptyprocess.py", line …

如果我有一些 Python 代码的设置阶段很长，最终会生成进程，生成的进程是基于父进程开始时的 Python 文件还是生成时的文件？

也就是说，我启动了父 Python 进程。然后我去编辑 Python 文件并在父进程产生其他进程之前完成它们的编辑。最后，父进程产生使用这些文件中的代码的子进程。子进程会使用父进程启动时的代码吗？还是进程产生时的代码？

我正在使用"spawn"多处理上下文在 Python 中创建一个新进程

import multiprocessing
ctx = multiprocessing.get_context("spawn")

proc = ctx.Process(target=my_func)
proc.start()

我希望这个过程有一组环境变量。理想情况下，我会在创建流程时指定这一点，如下所示：

proc = ctx.Process(target=my_func, environment={"MY_NAME": "MY_VALUE"})

有没有办法做到这一点？我希望 Python 运行时执行此操作，以便我可以确保在我的任何 Python 代码运行之前设置这些环境变量。

我正在尝试配置 Python 日志记录框架来处理来自multiprocessing.Pool. 在大多数情况下，脚本将无限期挂起，尽管已经观察到一些其他行为，例如退出而不打印所有日志消息。

我的实际代码更复杂，但我已将其简化为以下脚本，该脚本在我测试过的计算机上相当可靠地中断。

#!/usr/bin/env python3

import logging
import logging.handlers
import multiprocessing
import multiprocessing.util

L = logging.getLogger(__name__)

_globalQueue = None
_globalListener = None

def basicsetup():
    global _globalQueue
    global _globalListener

    cf = logging.Formatter("[{levelname}] {created:.7f} {name} ({process}~{processName}): {message}", style="{")

    handler = logging.StreamHandler()
    handler.setLevel(logging.DEBUG)
    handler.setFormatter(cf)

    # Subprocesses should use the queue to send log messages back to a thread in the main process
    _globalQueue = multiprocessing.Queue()

    _globalListener = logging.handlers.QueueListener(_globalQueue, handler, respect_handler_level=True)
    _globalListener.start()

    # Configure logging for main thread
    process_setup(get_queue())

def …

问题

我正在尝试优化一些适合所谓的令人尴尬的并行计算的计算，但我发现使用 python 的multiprocessing包实际上会减慢速度。

我的问题是：我做错了什么，或者并行化实际上减慢了速度是否有内在原因？是因为我在使用 numba 吗？像 joblib 或 dak 这样的其他软件包会有很大的不同吗？

有很多类似的问题，其中的答案总是开销比节省的时间要多，但所有这些问题往往都围绕着非常简单的函数展开，而我原以为带有嵌套循环的东西更适合并行化. 我也没有找到 joblib、multiprocessing 和 dask 之间的比较。

我的功能

我有一个函数，它将一维 numpy 数组作为形状 n 的参数，并输出一个形状为 (nxt) 的 numpy 数组，其中每一行都是独立的，即输出的第 0 行仅取决于输入的第 0 项，项目 1 上的第 1 行，等等。像这样：

底层计算使用numba进行了优化，可将速度提高多个数量级。

玩具示例 - 结果

我无法分享确切的代码，所以我想出了一个玩具示例。中定义的计算my_fun_numba实际上无关紧要，它只是一些非常平庸的数字运算以保持 CPU 忙碌。

对于玩具示例，我的 PC 上的结果是这些，它们与我使用实际代码得到的结果非常相似。

如您所见，将输入数组拆分为不同的块并将它们中的每一个发送到 multiprocessing.pool 实际上比仅在单个核心上使用 numba 会减慢速度。

我试过的

我在装饰器中尝试了cache和nogil选项的各种组合numba.jit，但差异很小。

我已经用 PyCharm 分析了代码（不是 timeit.Timer 部分，只是一次运行），如果我正确理解输出，似乎大部分时间都花在等待 pool 上。

按时间排序：

按自己时间排序：

玩具示例 - 代码

import …

我们注意到在我们的一个部署中遗留了一堆已失效的（僵尸）进程，并设法生成了一个显示问题的非常小的程序：

多.py：

from multiprocessing import Pool, set_start_method

def f(x):
    return x*x

if __name__ == '__main__':
    set_start_method('spawn')
    with Pool(5) as p:
        print(p.map(f, [1, 2, 3]))
        p.close()
        p.join()

该程序似乎正在离开僵尸进程，但很难捕获，因为从常规 shell 运行它会导致 shell 收割僵尸进程。

在我们的部署中，我们从另一个 python 程序运行它，所以为了模拟它，我们有这个：

主要.py：

from subprocess import run
from time import sleep

while True:
    result = run(["python", "multi.py"], capture_output=True)
    print(result.stdout.decode('utf-8'))
    result = run(["ps", "-ef", "--forest"], capture_output=True)
    print(result.stdout.decode('utf-8'), flush=True)
    sleep(1)

运行main.py会产生以下输出：

[1, 4, 9]

UID        PID  PPID  C STIME TTY          TIME CMD
root         1     0 11 …

进程正在改变他们不应该改变的东西。

AWorker有一个状态变量 (an mp.Value)。此值设置为-1，并且它 (the Worker)1在循环中将其更改为。

然而，这似乎是可以重置该值回至-1由产卵第二Worker，即使这股没有与原来的对。这似乎应该是不可能的。

行为：

当第二个工作器Worker启动时，第一个工作器 ( self.state.value)的状态被重置为-1。这被捕获，我们打印出发现错误。

代码：

import multiprocessing as mp
import time


class Worker:
    def __init__(self, tag, service_state) -> None:
        self.tag = tag
        self.local_state = int(service_state.value)
        self.state = service_state
        self.run_work_loop()

    def run_work_loop(self) -> None:
        print(f"[{self.tag}] Running... {self.state.value} {self.local_state}")
        while True:
            if self.state.value != self.local_state:
                print(f"[{self.tag}] Illegal change. Shared state: {self.state.value} Local State: {self.local_state}")
                break

            elif …