python 3.4 unittest에서 다중 처리가 작동하지 않습니다.

Question

다중 처리를 사용하는 unittest가 있습니다.

파이썬 3.2에서 파이썬 3.4로 업그레이드 한 후 다음과 같은 오류가 발생합니다. 파이썬 내부에서 변경된 사항과 코드를 실행하기 위해 바꿔야 할 힌트를 찾을 수 없습니다.

미리 감사드립니다. 샘플 코드에 따라

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 클래스 자체가 pickleable 더 이상 없다, 당신은 하나를 피클하기 위해 그것을 피클해야한다는 것입니다 (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() 

는 다른 방법으로는 unpickleable 년대 TestCase에서 개체를 제거 할 __setstate__/__getstate__를 사용합니다. 이 경우 _Outcome이라는 내부 클래스입니다. 우리는 아이에게 상관 없으므로 절인 상태에서 삭제할 수 있습니다 :

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)