如何告诉 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
因此需要一个更好的（即有效的）解决方案！
python serialization pickle multiprocessing python-multiprocessing
我正在使用aiohttp创建一个异步/IO 网络服务器。但是，据我了解，Async/IO 意味着服务器只能在一个处理核心上运行。uwsgi另一方面，像 一样的常规同步服务器可以通过真正并行的线程和进程充分利用计算机的计算资源。那么，为什么是异步/ IO新时尚，如果它少比多处理器并行？异步服务器aiohttp可以多处理吗？
python python-asyncio aiohttp python-multiprocessing python-3.5
假设我有以下内容：
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 …与Python Multiprocessing相关的错误：AttributeError: module '__main__' has no attribute '__spec__'，但由不同情况引起。
当我尝试使用 pdb 运行多处理代码时，我在 Python 3.7.4 中遇到了一个问题。该问题与https://docs.python.org/3.6/library/multiprocessing.html 中的基本多处理示例相同：
from multiprocessing import Pool
def f(x):
return x*x
if __name__ == '__main__':
with Pool(5) as p:
print(p.map(f, [1, 2, 3]))
这个运行正常（输出[1, 4, 9]时在Python经由直接运行）python.exe testcase.py。但是，它在 pdb 下不起作用；python.exe -m pdb testcase.py失败并出现错误：
Traceback (most recent call last):
File "c:\python37\lib\pdb.py", line 1697, in main
pdb._runscript(mainpyfile)
File "c:\python37\lib\pdb.py", line 1566, in _runscript
self.run(statement)
File "c:\python37\lib\bdb.py", line 585, in …我有一个不确定的生产问题，其中 amultiprocessing.Pool卡住并且永远不会从join.
我已经设法将问题减少到这个小例子，并让它在某种程度上可靠地挂起。
#!/usr/bin/env python3
import os
import time
import multiprocessing.pool
def run_task(i):
print(f'[{os.getpid()}] task({i}) complete')
if __name__ == '__main__':
tasks = iter(range(10))
processes = 4
pool = multiprocessing.pool.Pool(processes=processes, maxtasksperchild=1)
running = []
while True:
try:
running = [ f for f in running if not f.ready() ]
avail = processes - len(running)
if avail:
for _ in range(avail):
i = next(tasks)
print(f'[{os.getpid()}] add task({i})')
future = pool.apply_async(run_task, ( i, ))
running.append(future)
else:
time.sleep(0.1)
except …我有一个直到今天都可以进行多处理的脚本。为了重现这个问题，我简化了与下面显示的函数并行化的函数：
from multiprocessing import Process, Queue
import random
def rand_num():
num = random.random()
print(num)
if __name__ == "__main__":
queue = Queue()
processes = [Process(target=rand_num, args=()) for x in range(4)]
for p in processes:
p.start()
for p in processes:
p.join()
呈现完全相同的错误消息（重复 4 次，为了可读性我省略了重复）：
Traceback (most recent call last):
File "<string>", line 1, in <module>
File "/usr/local/Cellar/python@3.8/3.8.1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/multiprocessing/spawn.py", line 116, in spawn_main
exitcode = _main(fd, parent_sentinel)
File "/usr/local/Cellar/python@3.8/3.8.1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/multiprocessing/spawn.py", line 125, in _main
prepare(preparation_data)
File "/usr/local/Cellar/python@3.8/3.8.1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/multiprocessing/spawn.py", line 236, in prepare
_fixup_main_from_path(data['init_main_from_path'])
File "/usr/local/Cellar/python@3.8/3.8.1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/multiprocessing/spawn.py", line …下面是我的代码，为此我面临着多处理问题。我看到以前有人问过这个问题，我已经尝试过这些解决方案，但似乎不起作用。有人可以帮我吗？
from multiprocessing import Pool, Manager
Class X:
def _init_():
def method1(number1,var_a, var_b, var_c, var_d):
return values
if __name__ == 'main':
for value in ["X", "Y"]:
dict_values = Manager().dict()
with Pool(1) as p:
p.starmap(method1,
[
(1, dict_values, var_a, var_b, var_c, var_d),
(2, dict_values, var_a, var_b, var_c, var_d),
(3, var_a, var_b, var_c, var_d)
])
我面临的错误
multiprocessing.pool.RemoteTraceback:
Traceback (most recent call last):
File "/usr/lib64/python3.6/multiprocessing/managers.py", line 749, in _callmethod
conn = self._tls.connection
AttributeError: 'ForkAwareLocal' object has no attribute 'connection'
During handling of the …假设我有 N 个生成器gen_1, ..., gen_N，其中每个生成器都会产生相同数量的值。我想要一个生成器gen，使其在 N 个并行进程中运行 gen_1, ..., gen_N 并产生(next(gen_1), next(gen_2), ... next(gen_N))
那是我想要的：
def gen():
yield (next(gen_1), next(gen_2), ... next(gen_N))
这样每个 gen_i 都在自己的进程上运行。是否有可能做到这一点？我尝试在以下虚拟示例中执行此操作但没有成功：
A = range(4)
def gen(a):
B = ['a', 'b', 'c']
for b in B:
yield b + str(a)
def target(g):
return next(g)
processes = [Process(target=target, args=(gen(a),)) for a in A]
for p in processes:
p.start()
for p in processes:
p.join()
但是我得到了错误TypeError: cannot pickle 'generator' object。
编辑：
我修改了@darkonaut 的答案以满足我的需要。我发布它以防你们中的一些人觉得它有用。我们首先定义几个效用函数： …
python parallel-processing generator multiprocessing python-multiprocessing
基本上，我有 2 个线程，接收和发送。我希望能够输入一条消息，每当我收到一条新消息时，它就会“打印在我正在输入的行上方”。首先我认为会起作用，您可以粘贴它，它将运行：
import multiprocessing
import time
from reprint import output
import time
import random
import sys
def receiveThread(queue):
i = 0
while True:
queue.put(i)
i+=1
time.sleep(0.5)
def sendThread(queue):
while True:
a = sys.stdin.read(1)
if (a != ""):
queue.put(a)
if __name__ == "__main__":
send_queue = multiprocessing.Queue()
receive_queue = multiprocessing.Queue()
send_thread = multiprocessing.Process(target=sendThread, args=[send_queue],)
receive_thread = multiprocessing.Process(target=receiveThread, args=[receive_queue],)
receive_thread.start()
send_thread.start()
with output(initial_len=2, interval=0) as output_lines:
while True:
output_lines[0] = "Received: {}".format(str(receive_queue.get()))
output_lines[1] = "Last Sent: {}".format(str(send_queue.get()))
但是这里发生的是我无法发送数据。与我 put 时不同，输入不会给我一个 EOF a …
python ×10
python-3.x ×3
aiohttp ×1
generator ×1
input ×1
pdb ×1
pickle ×1
python-3.5 ×1
queue ×1