我有一个使用多进程的单元测试。
从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()
#---------------------------------------------------------------------------
TestCase
内部使用的_Outcome
对象包含一个result
属性,该属性是一个unittest.runner.TextTestResult
实例。这个类负责将每个测试的结果写入屏幕。它包含对_io.TextIoWrapper
对象的引用,这些对象无法被 pickle。如果我这周找到一些时间,我可能会深入研究一下 3.2 和 3.4 之间发生了什么变化,并提供一个修补程序,使 TestCase 可以再次被 pickle。 - dano