使用顺序初始值设定项参数的多进程池初始化

Question

我有一些代码如下：

import multiprocessing as mp
connection: module.Connection
def client_id():
for i in range(mp.cpu_count*2):
yield i
def initproc(host: str, port: int, client_id: int):
global connection
connection.connect(host, port, client_id)
def main():
host = "something"
port = 12345
mp.get_context("spawn").Pool(processes=mp.cpu_count()*2,
initializer=initproc,
initargs=(host, port, client_id())) as p:
res = p.starmap(processing_function, arg_list)

就本问题而言，processing_function和arg_list不相关。

问题是我收到了一个错误：

ForkingPickler(file, protocol).dump(obj)
TypeError: cannot pickle 'generator' object

有没有任何方法可以在池中创建一个初始化进程，使初始化的参数中的一个成为序列中的下一个数字？

附言：在编写的代码中，可能可以初始化初始化器函数之外的所有连接对象，但在我的特定实例中则不然。我需要将连接的参数传递到初始值设定项中。

Answer 1

对于您的案例，一个简单的解决方案是使用Process.name中包含的子进程的序列号。你可以用…提取它。。。

mp.current_process().name.split('-')[1]

如果您需要更多地控制序列的起始位置，可以使用multiprocessing.Value作为计数器，工人可以从中获得他们的唯一编号。

import multiprocessing as mp
import time

def init_p(client_id):
with client_id.get_lock():
globals()['client_id'] = client_id.value
print(f"{mp.current_process().name},"
f" {mp.current_process().name.split('-')[1]},"  # alternative
f" client_id:{globals()['client_id']}")
client_id.value += 1

if __name__ == "__main__":
ctx = mp.get_context("spawn")
client_ids = ctx.Value('i', 0)
with ctx.Pool(
processes=4,
initializer=init_p,
initargs=(client_ids,)
) as pool:
time.sleep(3)

输出：

SpawnPoolWorker-2, 2, client_id:0
SpawnPoolWorker-3, 3, client_id:1
SpawnPoolWorker-1, 1, client_id:2
SpawnPoolWorker-4, 4, client_id:3
Process finished with exit code 0