python 3.4 多处理不适用于 unittest

Question

我有一个使用多处理的单元测试。

从 Python 3.2 升级到 Python 3.4 后，出现以下错误。 我找不到提示，Python 内部发生了什么变化以及我必须改变什么才能使我的代码运行。

提前致谢。

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Python341_64\lib\multiprocessing\spawn.py", line 106, in spawn_main
    exitcode = _main(fd)
  File "C:\Python341_64\lib\multiprocessing\spawn.py", line 116, in _main
    self = pickle.load(from_parent)
EOFError: Ran out of input

Error
Traceback (most recent call last):
  File "D:\test_multiproc.py", line 46, in testSmallWorkflow
    p.start()
  File "C:\Python341_64\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
  File "C:\Python341_64\lib\multiprocessing\context.py", line 212, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "C:\Python341_64\lib\multiprocessing\context.py", line 313, in _Popen
    return Popen(process_obj)
  File "C:\Python341_64\lib\multiprocessing\popen_spawn_win32.py", line 66, in __init__
    reduction.dump(process_obj, to_child)
  File "C:\Python341_64\lib\multiprocessing\reduction.py", line 59, in dump
    ForkingPickler(file, protocol).dump(obj)
TypeError: cannot serialize '_io.TextIOWrapper' object

按照示例代码，如何重现错误:

import shutil
import traceback
import unittest
import time
from multiprocessing import Process
import os


class MyTest(unittest.TestCase):

    #---------------------------------------------------------------------------
    def setUp(self):
        self.working_dir = os.path.join(os.environ["TEMP"], "Testing")
        os.mkdir(self.working_dir)
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def tearDown(self):
        try:
            time.sleep(5)
            shutil.rmtree(self.working_dir, ignore_errors=True)
        except OSError as err:
            traceback.print_tb(err)
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def info(self, title):
        print(title)
        print('module name:', __name__)
        if hasattr(os, 'getppid'):  # only available on Unix
            print('parent process:', os.getppid())
        print('process id:', os.getpid())
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def f(self, name):
        self.info('function f')
        print('hello', name)
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def testSmallWorkflow(self):
        self.info('main line')
        p = Process(target=self.f, args=('bob',))
        p.start()
        p.join()
    #---------------------------------------------------------------------------

Answer 1

问题是 unittest.TestCase 类本身不再是可 pickle 的，您必须 pickle 它才能 pickle 它的绑定(bind)方法之一 (self.f).一个简单的解决方法是为您需要在子进程中调用的方法创建一个单独的类:

class Tester:
    def info(self, title=None):
        print("title {}".format(title))
        print('module name:', __name__)
        if hasattr(os, 'getppid'):  # only available on Unix
            print('parent process:', os.getppid())
        print('process id:', os.getpid())
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def f(self, name):
        self.info('function f')
        print('hello', name)
    #-------------------------------


class MyTest(unittest.TestCase):

    #---------------------------------------------------------------------------
    def setUp(self):
        self.working_dir = os.path.join(os.environ["TEMP"], "Testing")
        os.mkdir(self.working_dir)
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def tearDown(self):
        try:
            time.sleep(5)
            shutil.rmtree(self.working_dir, ignore_errors=True)
        except OSError as err:
            traceback.print_tb(err)
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def testSmallWorkflow(self):
        t = Tester()
        self.info('main line')
        p = Process(target=t.f, args=('bob',))
        p.start()
        p.join()

或者，您可以使用 __setstate__/__getstate__从 TestCase 中删除不可腌制的对象。在本例中，它是一个名为 _Outcome 的内部类。我们不关心子进程中的它，所以我们可以将它从 pickled 状态中删除:

class MyTest(unittest.TestCase):

    #---------------------------------------------------------------------------
    def setUp(self):
        self.working_dir = os.path.join(os.environ["TEMP"], "Testing")
        os.mkdir(self.working_dir)
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def tearDown(self):
        try:
            time.sleep(2)
            shutil.rmtree(self.working_dir, ignore_errors=True)
        except OSError as err:
            traceback.print_tb(err)
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def info(self, title=None):
        print("title {}".format(title))
        print('module name:', __name__)
        if hasattr(os, 'getppid'):  # only available on Unix
            print('parent process:', os.getppid())
        print('process id:', os.getpid())
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def f(self, name):
        self.info('function f')
        print('hello', name)
    #---------------------------------------------------------------------------

    #---------------------------------------------------------------------------
    def testSmallWorkflow(self):
        t = Tester()
        self.info('main line')
        p = Process(target=self.f, args=('bob',))
        p.start()
        p.join()

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

    def __setstate(self, state):
        self.__dict__.update(self_dict)