Python 3.5 中的实例变量过程安全吗？

Question

测试环境：

• Python 版本：3.5.1
• 操作系统平台：Ubuntu 16.04
• IDE：PyCharm 社区版 2016.3.2

---

我编写了一个简单的程序来测试过程安全。我发现subprocess2 在subprocess1 完成之前不会运行。看来实例变量self.count 是进程安全的。进程如何共享这个变量？他们会直接分享self吗？

另外一个问题是我在使用Queue的时候，必须手动使用multiprocessing.Manager来保证进程的安全，否则程序将无法按预期运行。（如果取消注释self.queue = multiprocessing.Queue()，该程序将无法正常运行，但是使用self.queue = multiprocessing.Manager().Queue() 是可以的。）

最后一个问题是为什么最终结果是900？我觉得应该是102。

很抱歉问了这么多问题，但我确实对这些东西很好奇。非常感谢！

代码：

import multiprocessing
import time
class Test:
    def __init__(self):
        self.pool = multiprocessing.Pool(1)
        self.count = 0
        #self.queue = multiprocessing.Queue()
        #self.queue = multiprocessing.Manager().Queue()

    def subprocess1(self):
        for i in range(3):
            print("Subprocess 1, count = %d" %self.count)
            self.count += 1
            time.sleep(1)
        print("Subprocess 1 Completed")

    def subprocess2(self):
        self.count = 100
        for i in range(3):
            print("Subprocess 2, count = %d" %self.count)
            self.count += 1
            time.sleep(1)
        print("Subprocess 2 Completed")

    def start(self):
        self.pool.apply_async(func=self.subprocess1)
        print("Subprocess 1 has been started")
        self.count = 900
        self.pool.apply_async(func=self.subprocess2)
        print("Subprocess 2 has been started")
        self.pool.close()
        self.pool.join()

    def __getstate__(self):
        self_dict = self.__dict__.copy()
        del self_dict['pool']
        return self_dict

    def __setstate__(self, state):
        self.__dict__.update(state)

if __name__ == '__main__':
    test = Test()
    test.start()
    print("Final Result, count = %d" %test.count)

输出：

Subprocess 1 has been started
Subprocess 2 has been started
Subprocess 1, count = 0
Subprocess 1, count = 1
Subprocess 1, count = 2
Subprocess 1 Completed
Subprocess 2, count = 100
Subprocess 2, count = 101
Subprocess 2, count = 102
Subprocess 2 Completed
Final Result, count = 900

Answer 1

底层细节相当棘手（更多信息请参阅the Python3 documentation，并注意Python2的细节略有不同），但本质上，当您将self.subprocess1或self.subprocess2作为参数传递给self.pool.apply_async时， Python 最终调用：

pickle.dumps(self)

在主进程中——Linux 上forking 之前的初始进程，或在 Windows 上以__main__ 调用的那个——然后，最终，池进程中生成的字节字符串的pickle.loads()。1 pickle.dumps 代码最终会调用您自己的 __getstate__ 函数；该函数的工作是将可以是 serialized 的内容返回到字节字符串。² 随后的 pickle.loads 创建一个适当类型的空白实例，没有调用它的__init__，然后使用它的__setstate__ 函数来填充对象（而不是__init__ing）。

您的__getstate__ 返回包含self 状态的字典，减去pool 对象，这是有充分理由的：

>>> import multiprocessing
>>> x = multiprocessing.Pool(1)
>>> import pickle
>>> pickle.dumps(x)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.5/multiprocessing/pool.py", line 492, in __reduce__
    'pool objects cannot be passed between processes or pickled'
NotImplementedError: pool objects cannot be passed between processes or pickled

由于池对象拒绝被腌制（序列化），我们必须避免尝试这样做。

无论如何，所有这一切都意味着池进程拥有自己的self 副本，它拥有自己的self.count 副本（并且完全缺少self.pool）。这些项目不会以任何方式共享，因此可以安全地在那里修改self.count。

我发现最简单的心理模型是为每个工作进程命名：Alice、Bob、Carol 等等，如果你愿意的话。然后，您可以将主要过程视为“您”：您复制某些内容并将副本交给 Alice，然后将其复制并交给 Bob，依此类推。函数调用（例如 apply 或 apply_async）会复制它们的所有参数，包括用于绑定方法的隐含 self。

使用multiprocessing.Queue 时，您会得到一些知道如何在各个进程之间工作、根据需要共享数据以及适当同步的东西。这使您可以来回传递数据副本。但是，与 pool 实例一样，multiprocessing.Queue 实例无法复制。 multiprocessing 例程 do 让您复制一个 multiprocessing.Manager().Queue() 实例，如果您想要一个复制的私有 Queue() 实例，这很好。 （这个内部细节比较复杂。³）

您得到的最终结果只是900，因为您仅查看原始self 对象。

请注意，每个应用的函数（来自apply 或apply_async）都会返回一个结果。这个结果被复制back，从工作进程到主进程。使用apply_async，您可以选择在结果准备好后立即回叫。如果你想要这个结果you should save it somewhere，或者在你需要的时候使用get函数（如同一个答案所示）等待它。

---

¹我们可以在这里说“池”进程而不必担心是哪一个，因为您将自己限制为一个。但是，无论如何，有一个简单的面向字节的双向通信流，由multiprocessing 代码管理，将每个工作进程与调用它的父进程连接起来。如果您创建两个这样的池进程，每个进程都有自己的字节流连接到主进程。这意味着如果有两个或更多并不重要：行为将是相同的。

²这个“东西”通常是一本字典，但详情请参阅Simple example of use of __setstate__ and __getstate__。

³pickle.dumps 在这样的实例上的输出是：

>>> pickle.dumps(y)
(b'\x80\x03cmultiprocessing.managers\n'
b'RebuildProxy\n'
b'q\x00(cmultiprocessing.managers\n'
b'AutoProxy\n'
b'q\x01cmultiprocessing.managers\n'
b'Token\n'
b'q\x02)\x81q\x03X\x05\x00\x00\x00Queueq\x04X$\x00\x00\x00/tmp/pymp-pog4bhub/listener-0_uwd8c9q\x05X\t\x00\x00\x00801b92400q\x06\x87q\x07bX\x06\x00\x00\x00pickleq\x08}q\tX\x07\x00\x00\x00exposedq\n'
b'(X\x05\x00\x00\x00emptyq\x0bX\x04\x00\x00\x00fullq\x0cX\x03\x00\x00\x00getq\rX\n'
b'\x00\x00\x00get_nowaitq\x0eX\x04\x00\x00\x00joinq\x0fX\x03\x00\x00\x00putq\x10X\n'
b'\x00\x00\x00put_nowaitq\x11X\x05\x00\x00\x00qsizeq\x12X\t\x00\x00\x00task_doneq\x13tq\x14stq\x15Rq\x16.\n')

我做了一些小技巧来将它拆分为换行符，然后手动添加括号，以防止长行变得超长。论据在不同的系统上会有所不同；这个特定的使用一个文件系统对象，它是一个侦听器套接字，它允许协作的 Python 进程在它们之间建立一个 new 字节流。

Answer 2

问题：……为什么最终结果是900？我觉得应该是102。

结果应该是 106，range 是基于 0，你得到 3 次迭代。

你可以得到预期的输出，例如：

class PoolTasks(object):
    def __init__(self):
        self.count = None

    def task(self, n, start):
        import os
        pid = os.getpid()
        count = start
        print("Task %s in Process %s has been started - start=%s" % (n, pid, count))

        for i in range(3):
            print("Task %s in Process %s, count = %d " % (n, pid, count))
            count += 1
            time.sleep(1)

        print("Task %s in Process %s has been completed - count=%s" % (n, pid, count))
        return count

    def start(self):
        with mp.Pool(processes=4) as pool:
            # launching multiple tasks asynchronously using processes
            multiple_results = [pool.apply_async(self.task, (p)) for p in [(1, 0), (2, 100)]]

            # sum result from tasks
            self.count = 0
            for res in multiple_results:
                self.count += res.get()

if __name__ == '__main__':
    pool = PoolTasks()
    pool.start()
    print('sum(count) = %s' % pool.count)

输出：
进程 5601 中的任务 1 已启动 - start=0
进程 5601 中的任务 1，计数 = 0
进程 5602 中的任务 2 已启动 - start=100
进程 5602 中的任务 2，计数 = 100
进程 5601 中的任务 1，计数 = 1
进程 5602 中的任务 2，计数 = 101
进程 5601 中的任务 1，计数 = 2
进程 5602 中的任务 2，计数 = 102
进程 5601 中的任务 1 已完成 - count=3
进程 5602 中的任务 2 已完成 - count=103
总和（计数）= 106

用 Python：3.4.2 测试