标签: python-multiprocessing

我正在编写一个机器学习工具包来并行运行具有不同设置的算法（每个进程运行一个设置的算法）。我正在考虑使用 mpi4py 或 python 的内置多处理？

我正在考虑一些优点和缺点。

1. 便于使用：

   • mpi4py：似乎有更多的概念需要学习，还有更多的技巧可以让它运行良好
   • 多处理：非常简单和干净的 API

2. 速度：

   • mpi4py：人们说它的级别更低，所以我希望它比 python 多处理更快？
   • 多处理：与mpi4py相比，慢得多？

3. 干净和简短的代码：

   • mpi4py：似乎要编写更多代码
   • 多处理：首选，易于使用的 API

工作环境是我的目标是基本上在一台计算机或 GPU 服务器上运行代码。并不是真正针对在网络中的不同机器上运行（只有 MPI 可以做到）。

由于主要目标是进行机器学习，因此并行化并不是真的需要非常优化，我想要的关键目标是平衡简单、干净和快速维护代码库，但同时喜欢利用并行化的好处。

有了上述背景，是否建议使用多处理就足够了？或者是否有充分的理由使用 mpi4py ？

我正在测试 python 2.7 中的多处理管理器包，并遇到了一个问题，该问题已经记录在 Unix 上运行的 NFS ( https://bugs.python.org/issue13978 )。问题是虽然这突出了问题，但我不清楚你是如何解决这个问题的，或者替代方案是什么？

from multiprocessing import Manager

if __name__ == '__main__':
    Manager()

然后我调用我的脚本并得到：

Traceback (most recent call last):
File "/usr/lib/python2.7/multiprocessing/util.py", line 274, in 
_run_finalizers
finalizer()
File "/usr/lib/python2.7/multiprocessing/util.py", line 207, in __call__
res = self._callback(*self._args, **self._kwargs)
File "/usr/lib/python2.7/shutil.py", line 252, in rmtree
onerror(os.remove, fullname, sys.exc_info())
File "/usr/lib/python2.7/shutil.py", line 250, in rmtree
os.remove(fullname)
OSError: [Errno 16] Device or resource busy: '/home/user/.tmp/pymp-xrl_Vc/.nfs0000000196'

如果我不能使用 Manager，是否有另一种方法可以在多个进程之间共享一个对象，或者是否有解决方法（manager.shutdown 和 del manager 的建议方法没有任何效果）？

今天有更多的时间来研究这个，经过一番挖掘，我找到了一个导致干净退出的解决方案（即没有错误）。

from multiprocessing import Manager
mp = Manager() …

如何告诉 python multiprocessing 应该使用哪个 pickle 协议进行序列化？

问题是我想在并行进程中使用一个大的 Pandas DataFrame 处理multiprocessing.Pool.apply_async，结果我得到以下错误：

溢出错误：无法序列化大于 4 GiB 的字节对象

根据此链接https://github.com/stan-dev/pystan/issues/197，问题可能是由于使用默认版本的 pickle 协议（pickle.DEFAULT_PROTOCOL=3）进行多处理，而 protocol=pickle.HIGHEST_PROTOCOL =4 可以用来避免这个问题。

感谢您的帮助！

更新：

以下候选解决方案（基于@juanpa.arrivillaga 引用的链接）

from multiprocessing.reduction import ForkingPickler

class ForkingPicklerHighest(ForkingPickler):
    @classmethod
    def dumps(cls, obj, protocol=None):
        return ForkingPickler.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)

multiprocessing.connection._ForkingPickler = ForkingPicklerHighest

产生以下错误：

struct.error: 'i' 格式需要 -2147483648 <= number <= 2147483647

因此需要一个更好的（即有效的）解决方案！

我正在使用aiohttp创建一个异步/IO 网络服务器。但是，据我了解，Async/IO 意味着服务器只能在一个处理核心上运行。uwsgi另一方面，像 一样的常规同步服务器可以通过真正并行的线程和进程充分利用计算机的计算资源。那么，为什么是异步/ IO新时尚，如果它少比多处理器并行？异步服务器aiohttp可以多处理吗？

假设我有以下内容：

• 具有 4 个 GPU 的系统。
• 一个函数，foo在每个 GPU 上最多可以同时运行 2 次。
• files需要foo按任何顺序处理的列表。但是，每个文件都需要不可预测的时间来处理。

我想处理所有文件，通过确保foo在任何给定时间始终有 8 个运行实例（每个 GPU 上 2 个实例），直到剩下的文件少于 8 个，从而使所有 GPU 尽可能忙碌。

调用 GPU 的实际细节不是我的问题。我想弄清楚的是如何编写并行化，以便我可以保持 8 个foo运行实例，但以某种方式确保始终使用每个 GPU ID 中的 2 个。

我想出了一种使用 解决这个问题的方法multiprocessing.Pool，但该解决方案非常脆弱，并且依赖于（AFAIK）未记录的功能。它依赖于一个事实，即内部流程Pool的格式命名FormPoolWorker-%d，其中%d是一个与池中的进程数之间的数字。我取这个值并用 GPU 的数量修改它，这给了我一个有效的 GPU id。但是，如果我能以某种方式将 GPU id 直接提供给每个进程（可能是在初始化时），而不是依赖于进程名称的字符串格式，那就更好了。

我考虑过的一件事是，如果 允许一个列表的initializer和initargs参数，以便每个进程都可以使用不同的参数集进行初始化，那么问题就没有实际意义了。不幸的是，这似乎不起作用。Pool.__init__initargs

任何人都可以推荐一个更健壮或pythonic的解决方案来解决这个问题吗？

黑客解决方案（Python 3.7）：

from multiprocessing import Pool, current_process

def foo(filename):
    # Hacky …

我开始将多重处理纳入我的代码中，因为我试图自动化的任务从计算角度来说相当昂贵。根据我收集的 stackoverflow 信息，我的代码中的模块结构如下。我正在 Win10 中开发 python 3.7。

main：除了加载输入等之外，还调用多处理函数的地方。

import pandas as pd\nimport run\n\ndef do():\n    df=pd.DataFrame({'Identifier': ['id_1', 'id_1', 'id_1', 'id_1', 'id_1', 'id_2', 'id_2', 'id_2', 'id_2', 'id_2', 'id_3', 'id_3', 'id_3', 'id_3', 'id_3'],\n                 'float_id': [1, 2, 3, 4, 5, 10, 25, 33, 45, 50, .1, .2, .3, .4, .5],\n                 'a': np.random.rand(15),\n                 'b': np.random.rand(15),\n                 'c': np.random.rand(15)})\n    \n    v_column=['a', 'b', 'c']\n    \n    df_out=run.function_multiprocessing(df, v_column)\n\n    return df_out\n\nif __name__=='__main__':\n    df_out=do()\n

\n
• run：多处理函数所在的位置
\n

import defs\nimport pandas as pd\n\nimport multiprocessing\n\n\ndef iterator(data, id_col, …

在我的 Macbook Pro（英特尔，2020）上，我可以成功使用multiprocessing.Pool：

from multiprocessing import Pool

p = Pool(8)
results = p.map(worker_function, list_of_inputs)
p.close()

但是，如果我在 Macbook Air（M1，2020）上运行相同的代码，我会一次又一次地重复出现奇怪的错误（片段如下）：

from multiprocessing import Pool

p = Pool(8)
results = p.map(worker_function, list_of_inputs)
p.close()

我必须在 Linux 中使用带有“spawn”上下文的 Process 方法。然后我写了一个示例代码如下：

from multiprocessing import Value
import multiprocessing

class Test(object):
    def __init__(self, m_val):
        print("step1")
        self.m_val = m_val
        print("step2")
        self.m_val_val = m_val.value
        self.prints()
    def prints(self):
        print("self.m_val_val:%d"%self.m_val_val)


def main(m_val):
    t = Test(m_val)

if __name__   == "__main__":
    N = 2
    procs = []
    v = Value("i",10)
    for i in range(0,N):
        proc_i = multiprocessing.get_context("spawn").Process(target=main,args=(v,))
        proc_i.daemon=True
        procs.append(proc_i)
    for i in range(0,N):
        procs[i].start()
    
    for i in range(0,N):
        procs[i].join()

当我在 Linux 中运行这段代码时，它将打印：

step1
step2
step1
step2

而在Windows中，打印内容将是：

step1
step2
self.m_val_val:10
step1
step2
self.m_val_val:10

此外，屏幕上也没有打印任何错误信息。那么，我该如何解决这个问题，即如何在Linux中使用“spawn”上下文时在进程之间使用多处理值？

操作系统和Python信息：

uname -a
Linux debian 5.10.0-8-amd64 #1 SMP Debian 5.10.46-4 (2021-08-03) x86_64 GNU/Linux
python3 --version
Python 3.9.2

这是一个可以启动多重处理的简单类。

from multiprocessing.pool import Pool    

class my_mp(object):
    def __init__(self):
        self.process_num = 3
        fh = open('test.txt', 'w')
    def run_task(self,i):
        print('process {} start'.format(str(i)))
        time.sleep(2)
        print('process {} end'.format(str(i)))
    def run(self):
        pool = Pool(processes = self.process_num)
        for i in range(self.process_num):
            pool.apply_async(self.run_task,args = (i,))
        pool.close()
        pool.join()

初始化my_mp类，然后启动多进程。

ins = my_mp()
ins.run()
process 0 start
process 1 start
process 2 start
process 0 end
process 2 end …

我正在尝试为我的 Python 应用程序创建一个共享内存，该内存应该在父进程以及从该父进程生成的另一个进程中使用。在大多数情况下，工作正常，但是，有时我会得到以下堆栈跟踪：

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "/usr/lib/python3.8/multiprocessing/spawn.py", line 116, in spawn_main
    exitcode = _main(fd, parent_sentinel)
  File "/usr/lib/python3.8/multiprocessing/spawn.py", line 126, in _main
    self = reduction.pickle.load(from_parent)
  File "/usr/lib/python3.8/multiprocessing/synchronize.py", line 110, in __setstate__
    self._semlock = _multiprocessing.SemLock._rebuild(*state)
FileNotFoundError: [Errno 2] No such file or directory: '/psm_47f7f5d7'

我想强调的是，我们的代码/应用程序在 99% 的情况下都可以正常工作。我们在应用程序中定期为每个此类进程生成新的共享内存（这是一个服务器进程，因此它每天 24/7 运行）。几乎所有时候这都工作正常，只是偶尔会抛出上面的错误，然后杀死整个应用程序。

更新：我注意到这个问题主要发生在应用程序已经运行了一段时间的时候。当我启动它时，共享内存的创建和生成新进程工作正常，没有此错误。

共享内存是这样创建的：

# Spawn context for multiprocessing
_mp_spawn_ctxt = multiprocessing.get_context("spawn")
_mp_spawn_ctxt_pipe = _mp_spawn_ctxt.Pipe

# Create shared memory
mem_size = width * height * …