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的內部類。我們不關心它在孩子，所以我們可以從醃狀態刪除它：

​​