python 3.4 multiprocessing不能与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，你必须腌制它以咸菜一个其绑定方法（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这unpickleable取出异物。在这种情况下，它是一个名为_Outcome的内部类。我们不关心它在孩子，所以我们可以从腌状态删除它：

​​