python 多处理共享变量安全

Question

我在使用多处理模块时遇到问题。我已经使用了锁，但是下面的代码仍然不安全，我不知道为什么，计数器有时不等于100，我该如何修复代码让它安全？

import random
import threading
from multiprocessing import Pool, Manager

import time

lock = threading.Lock()


def a_complex_operation(counter):
    with lock:
        time.sleep(random.random())
        counter.value += 1


def main():
    pool = Pool(16)
    counter = Manager().Value('i', 0)

    for i in range(100):
        pool.apply_async(a_complex_operation, args=(counter,))

    pool.close()
    pool.join()
    if counter.value != 100:
        print "not equal 100, current value is: "+str(counter.value)


if __name__ == '__main__':
    count = 0
    while True:
        t1 = time.time()
        main()
        count += 1
        print "the " + str(count) + " loop, cost time: " + str(time.time() - t1)

输出将是:

the 1 loop, cost time: 4.1369998455
the 2 loop, cost time: 3.74100017548
the 3 loop, cost time: 3.92299985886
the 4 loop, cost time: 4.05500006676
not equal 100, current value is: 99
the 5 loop, cost time: 4.01900005341
the 6 loop, cost time: 4.14299988747

然后我一起测试 Manager().list() 和 Manager().Value('i', 0)

import random
from multiprocessing import Pool, Manager
import time


def a_complex_operation(list_, counter):
    for x in range(10):
        time.sleep(random.random()/10)
        list_.append(x)
        counter.value += 1


def main():
    pool = Pool(16)
    counter0 = 0
    list_ = Manager().list()
    counter = Manager().Value('i', 0)

    for i in range(100):
        pool.apply_async(a_complex_operation, args=(list_, counter))
        counter0 += 1

    pool.close()
    pool.join()
    if len(list_) != 1000:
        print "length of list is not equal 1000, current is:" + str(len(list_))
    if counter.value != 1000:
        print "value of counter is not equal 1000, current is :" + str(counter.value)


if __name__ == '__main__':
    counter = 0
    while True:
        counter += 1
        t1 = time.time()
        main()
        t2 = time.time()
        print "the " + str(counter) + " loop cost time: " + str(t2 - t1)

输出将是:

value of counter is not equal 1000, current is :916
the 1 loop cost time: 3.92299985886
value of counter is not equal 1000, current is :911
the 2 loop cost time: 3.98500013351
value of counter is not equal 1000, current is :925
the 3 loop cost time: 4.007999897
value of counter is not equal 1000, current is :913
the 4 loop cost time: 3.99399995804
value of counter is not equal 1000, current is :920
the 5 loop cost time: 4.09500002861
value of counter is not equal 1000, current is :915

我发现 Manager().list() 是安全的，Manager().Value('i', 0) 是不安全的，有趣的是，任何人都可以告诉我为什么 Manager().list() 看起来安全？

Answer 1

您的子进程不会继承锁对象。或者它们确实如此，但它们是独立的副本，没有链接在一起，不能用于任何用途。所以存在竞争条件，最终会失败。

您可以通过 Manager().Lock() 解决此问题，因为您已经在使用 Manager。

def a_complex_operation(counter, alock):
    with alock:
        time.sleep(random.random())
        counter.value += 1

def main():
    pool = Pool(16)
    ma = Manager()
    counter = ma.Value('i', 0)
    lock = ma.Lock()
    for i in range(100):
        pool.apply_async(a_complex_operation, args=(counter, lock))

这有效(不过，您的子进程现在会慢得多。预计每次运行大约 50 秒，100 次平均 0.5 秒)。

但现在您的 counter.value 始终为 100。