python - 由多个子进程锁定互斥体

Question

互斥量是否在这段代码中发挥作用？即 sys.stdout.write 是否 protected ？

import sys
from multiprocessing import Pool, Lock

class ParentApp():
    mutex=Lock()
    def report(self,msg):
        with ParentApp.mutex:
            sys.stdout.write(msg)

class ChildApp1(ParentApp):
    def print_report(self):
        for i in xrange(100):
            ParentApp.report(self, 'BLABLA')


class ChildApp2(ParentApp):
    def print_report(self):
        for i in xrange(100):
            ParentApp.report(self, 'TESTTEST')

def runnable(app):
    app.print_report()

def main():
    app=[]
    app.append(ChildApp1())
    app.append(ChildApp2())

    pool = Pool(len(apps))
    pool.map(runnable, apps)

    exit(0)

if __name__ == '__main__':
    sys.exit(main())

PS:代码也在这里:http://pastebin.com/GyV3w45F

PS:我在Linux主机上运行

Answer 1

只有在类似 Posix 的平台上，它才能发挥作用。如果您使用的是 Windows，每个进程最终都会得到一个完全不同的锁副本。

您可以通过添加一些额外的跟踪和 sleep 语句来看到这一点:

class ParentApp():
    mutex=Lock()
    def report(self,msg):
        print("\nGETTING for {}".format(msg))
        with self.mutex:
            print("GOT for {}".format(msg))
            sys.stdout.write(msg)
            sys.stdout.flush()
            time.sleep(5)

在 Linux 上:

GETTING for BLABLA
GOT for BLABLA
BLABLA
GETTING for TESTTEST
< 5 second delay here>

在 Windows 上:

GETTING for BLABLA
GOT for BLABLA
BLABLA
GETTING for TESTTEST
GOT for TESTTEST
TESTTEST
<5 second delay here>

这是因为 Posix 平台使用 os.fork() 创建新进程，这意味着您在父进程中创建的 Lock() 会自动由子进程继承。但是，Windows 没有 os.fork，因此它需要生成一个新进程，然后在子进程中重新导入您的模块。重新导入模块意味着 ParentApp 被重新导入并重新执行，Lock 类属性也是如此。因此，您的 parent 和两个 child 最终都拥有了自己独特的Lock。

要解决此问题，您需要在父级中创建一个 Lock，并将其传递给子级。对于您当前的体系结构，这实际上不是一项微不足道的任务 - 您将 Lock 对象作为参数传递给 pool.map，这将不允许您传递 Lock 对象。如果你尝试它，你会得到一个异常(exception):

RuntimeError: Lock objects should only be shared between processes through inheritance

您只能在您实际开始 进程 时将普通的Lock 对象传递给子进程。一旦它们启动(就像它们在调用 Pool 方法时一样)，就会出现异常:

l = Lock()
p = Process(target=func, args=(l,)) # ok
p.start()

pool = Pool()
pool.apply(func, args=(l,)) # not ok, raises an exception.

为了将 Lock 传递给 Pool 函数，如 map，您需要使用 multiprocessing.Manager 创建一个共享锁。这是我推荐的做法:

import sys 
from multiprocessing import Pool, Lock, get_context, Manager
import time

class ParentApp():
    def __init__(self, mutex):
        self.mutex = mutex

    def report(self,msg):
        with self.mutex:
            sys.stdout.write(msg)

class ChildApp1(ParentApp):
    def print_report(self):
        for i in range(100):
            ParentApp.report(self, 'BLABLA')


class ChildApp2(ParentApp):
    def print_report(self):
        for i in range(100):
            ParentApp.report(self, 'TESTTEST')

def runnable(app):
    app.print_report()

def main():
    apps=[]
    m = Manager()
    lock = m.Lock()
    apps.append(ChildApp1(lock))
    apps.append(ChildApp2(lock))

    pool = Pool(len(apps))
    pool.map(runnable, apps)

if __name__ == '__main__':
    sys.exit(main())

为了确保共享 Lock，我们需要让 ParentApp 实际将锁对象作为参数。将它完全独立于类中并不好，但我认为这是我们在 Windows 的限制下所能做的最好的事情。