Python在进程之间共享锁

2024-04-25 19:47:53 发布

您现在位置:Python中文网/ 问答频道 /正文

我正在尝试使用一个部分函数,以便pool.map()可以针对具有多个参数的函数(在本例中是Lock()对象)。

下面是示例代码(取自对我前面一个问题的回答):

from functools import partial

def target(lock, iterable_item):
    for item in items:
        # Do cool stuff
        if (... some condition here ...):
            lock.acquire()
            # Write to stdout or logfile, etc.
            lock.release()

def main():
    iterable = [1, 2, 3, 4, 5]
    pool = multiprocessing.Pool()
    l = multiprocessing.Lock()
    func = partial(target, l)
    pool.map(func, iterable)
    pool.close()
    pool.join()

但是,当我运行此代码时,会出现错误:

Runtime Error: Lock objects should only be shared between processes through inheritance.

我错过了什么?如何在子进程之间共享锁?


Tags: 函数代码lockmaptarget参数defiterable
2条回答

对不起,我应该在回答你的另一个问题时注意到这一点。不能将普通的multiprocessing.Lock对象传递给Pool方法,因为它们不能被pickle。有两种方法可以解决这个问题。一个是创建^{}并传递^{}

def main():
    iterable = [1, 2, 3, 4, 5]
    pool = multiprocessing.Pool()
    m = multiprocessing.Manager()
    l = m.Lock()
    func = partial(target, l)
    pool.map(func, iterable)
    pool.close()
    pool.join()

不过,这有点沉重;使用Manager需要生成另一个进程来托管Manager服务器。所有对acquire/release锁的调用都必须通过IPC发送到该服务器。

另一个选项是在创建池时使用initializerkwarg传递常规的multiprocessing.Lock()。这将使您的锁实例在所有子工作进程中都是全局的:

def target(iterable_item):
    for item in items:
        # Do cool stuff
        if (... some condition here ...):
            lock.acquire()
            # Write to stdout or logfile, etc.
            lock.release()
def init(l):
    global lock
    lock = l

def main():
    iterable = [1, 2, 3, 4, 5]
    l = multiprocessing.Lock()
    pool = multiprocessing.Pool(initializer=init, initargs=(l,))
    pool.map(target, iterable)
    pool.close()
    pool.join()

第二种解决方案的副作用是不再需要partial

这里有一个版本(使用Barrier而不是Lock,但您可以理解),它也可以在Windows上工作(其中缺少的fork会导致其他问题):

import multiprocessing as mp

def procs(uid_barrier):
    uid, barrier = uid_barrier
    print(uid, 'waiting')
    barrier.wait()
    print(uid, 'past barrier')    

def main():
    N_PROCS = 10
    with mp.Manager() as man:
        barrier = man.Barrier(N_PROCS)
        with mp.Pool(N_PROCS) as p:
            p.map(procs, ((uid, barrier) for uid in range(N_PROCS)))

if __name__ == '__main__':
    mp.freeze_support()
    main()

相关问题 更多 >