标签: python-multiprocessing

背景：我正在尝试在 python 3.4 应用程序中实现多核处理PyQT5。

在我的框架应用程序中numpy.ndarrays，将其想象为一个[n,m,t]数组。我需要处理每个[n,m,:]数组，使用多核会线性加速我的过程。

我尝试过多重处理，并使用部分示例脚本编写了一个简单的脚本，并给了我以下想法：

简单无 GUI 代码：

import multiprocessing as mp
import numpy

aa = numpy.random.rand(4,2,3)

def random_function(x):
    return x,x**3

if __name__ == '__main__':
    pool = mp.Pool(processes=4)

    #with apply_asynch

    #results = [pool.apply_async(cube, args=(aa[:,:,x],)) for x in range(0,aa.shape[2])]
    #output = [p.get() for p in results]
    #test_va = numpy.asarray( output)


    #with apply

    results = [pool.apply(random_function, args=(aa[:,:,x],)) for x in range(0,aa.shape[2])]
    test_va = numpy.asarray( results)

这有效并且可以满足我需要它做的事情。

问题：现在，当我实现这个时，PyQT5 …

我正在运行两个处理来访问导入的模块，如下所示：

import foo

def bar():
  while True:
    foo.a = True

def baz():
  while True:
    print foo.a

p1 = Process(target=bar)
p2 = Process(target=baz)
p1.start()
p2.start()

似乎每个进程都有自己的模块 foo 实例，bar()将值更改为 True，但实际上baz()是 False。有什么解决办法吗？

最终目标是在后台执行一个方法，但不是并行：当多个对象调用此方法时，每个对象都应该等待轮到它们继续执行。为了实现在后台运行，我必须在子进程（而不是线程）中运行该方法，并且我需要使用spawn（而不是fork）来启动它。为了防止并行执行，显而易见的解决方案是在进程之间共享全局锁。
当进程被分叉时（这是 Unix 上的默认设置），很容易实现，如以下两个代码中所强调的那样。
我们可以将它作为类变量共享：

import multiprocessing as mp
from time import sleep

class OneAtATime:

    l = mp.Lock()

    def f(self):
        with self.l:
            sleep(1)
        print("Hello")

if __name__ == "__main__":
    a = OneAtATime()
    b = OneAtATime()
    p1 = mp.Process(target = a.f)
    p2 = mp.Process(target = b.f)
    p1.start()
    p2.start()

或者我们可以将它传递给方法：

import multiprocessing as mp
from time import sleep

class OneAtATime:
    def f(self, l):
        with l:
            sleep(1)
        print("Hello")

if __name__ == "__main__":
    a = OneAtATime()
    b = OneAtATime()
    m = mp.Manager()
    l = mp.Lock() …

我一直在阅读并发性以及 SysV 和 POSIX 风格的共享内存，特别是 Python 的多处理包。现在我有几个问题，到目前为止我的 Google-fu 还没有找到答案：

Q1据我从Python bugtracker 中的这个问题了解到，最近引入的multiprocessing.shared_memory.SharedMemory使用 POSIX 命名的共享内存。然而，当涉及到RawArray和RawValue中定义的类型时multiprocessing.sharedctypes，Python 文档还有很多不足之处。它们在内部是如何实施的？

Q2例如， aRawArray(\'B\')和 a 的实例之间唯一的实际区别是multiprocessing.shared_memory.SharedMemory后者有名称并且也可以从不相关（可能是非 Python）的进程访问吗？他们的表现有可比性吗？

Q3 Python 中多处理锁到底是如何实现的？难道他们不需要使用某种共享内存吗？文档说：

\n

笔记

\n

此包\xe2\x80\x99s 的某些功能需要主机操作系统上有效的共享\nsemaphore 实现。如果没有，multiprocessing.synchronize 模块将被禁用，并且尝试导入它会导致导入错误。请参阅bpo-3770了解\n更多信息。

\n

（请注意，multiprocessing.Lock实际上源自multiprocessing.synchronize.Lock。）现在根据链接的问题bpo-3770，Lock使用 libc\'sshm_open()创建一个命名信号量。这与工作原理相似吗SharedMemory？

更新：根据源代码 \xe2\x80\x93 特别是，这些行：Link 1，link 2（谢谢你，bnaecker！），非 …

我寻找了其他问题，这个未接受的答案问题是我能找到的唯一一个以某种方式涵盖了这个问题并且没有真正帮助的问题。另外，我需要它来处理进程，而不是线程。

因此，我从头开始编写了一个示例程序来展示我的问题，您应该能够粘贴它并且它将运行：

import multiprocessing
import time 

class Apple:
   def __init__(self, color):
      self.color = color

def thinkAboutApple(apple):
   while True:
      print(apple.color)
      time.sleep(2)

my_apple = Apple("red")
new_process = multiprocessing.Process(target=thinkAboutApple, args=(my_apple,))
new_process.start()
time.sleep(4)
print("new: brown")
my_apple.color = "brown"

#so that the program doesn't exit after time.sleep(4)
while True:
    pass

# actual output | # wanted output
red             | red
red             | red
new: brown      | new: brown
red             | brown
red             | brown

这告诉我，要么苹果处于一个奇怪的假设，它同时有两种颜色，要么 new_process' 苹果位于 ram 中的另一个位置，并与主进程中的苹果分开。

所以问题是：有没有办法让进程中的苹果指针指向同一个苹果，或者什么是Pythonic方法来保持所有进程中苹果的所有实例相同？如果我在许多进程中有相同的苹果，甚至更多进程没有苹果，我如何确保它们始终相同？

我开始将多重处理纳入我的代码中，因为我试图自动化的任务从计算角度来说相当昂贵。根据我收集的 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 * …