Verwenden Sie das Modul "Multiprocess", um Funktionen in mehreren Prozessen in Python auszuführen.
Die Formel ist leicht zu verstehen, wie man Code verwendet und testet. https://docs.python.org/ja/3/library/multiprocessing.html
Schauen Sie sich nun den folgenden Code an
from multiprocessing import Pool
def f(x):
return x*x
def main():
with Pool(5) as p:
print(p.map(f, [1, 2, 3]))
main()
Ich habe "if \ _ \ _ name \ _ \ _ == '\ _ \ _ main \ _ \ _':" im offiziellen Beispielcode weggelassen. Das zu tun wäre großartig. (Betreten Sie eine Endlosschleife und benötigen Sie einen Prozess-Kill)
Dies scheint eine Spezifikation zu sein, wie sie offiziell erwähnt wird https://docs.python.org/ja/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods
Vorerst ** Lassen Sie "if \ _ \ _ name \ _ \ _ == '\ _ \ _ main \ _ \ _':" im Code mit Multiprozess nicht weg ** Es bedeutet das.
Angenommen, Sie möchten ein Modul erstellen, das mehrere Prozesse verwendet und unittest ausführt.
mp_test.py
from multiprocessing import Pool
def f(x):
return x*x
def main():
with Pool(5) as p:
print(p.map(f, [1, 2, 3]))
if __name__ == '__main__':
pass
test_mp_test.py
"""Tests for `multiprocess_test` package."""
import unittest
from mp_test import mp_test
class TestMultiprocess_test(unittest.TestCase):
"""Tests for `multiprocess_test` package."""
def setUp(self):
"""Set up test fixtures, if any."""
def tearDown(self):
"""Tear down test fixtures, if any."""
def test_000_something(self):
mp_test.main()
if __name__ == "__main__":
unittest.main()
Lass es uns testen.
$python test_mp_test.py
Das kann gemacht werden.
Was passiert also, wenn Sie diesen Test über ** setup.py ** ausführen?
Verzeichnisaufbau.
mp_test
|-mp_test.py
tests
|-test_mp_test.py
setup.py
setup.py
#!/usr/bin/env python
"""The setup script."""
from setuptools import setup, find_packages
setup(
author="rr28",
author_email='[email protected]',
python_requires='>=3.5',
description="multiprocess test.",
entry_points={
'console_scripts': [
'mp_test=mp_test.mp_test:main',
],
},
name='mp_test',
packages=find_packages(include=['mp_test', 'mp_test.*']),
test_suite='tests',
version='0.1.0',
)
Lauf
$python setup.py test
**Korrekt. Obwohl ich "if name == '__ main__':" nicht weggelassen habe, erhalte ich eine Endlosschleife. ** ** **
Ausführungsergebnis
======================================================================
ERROR: test_000_something (tests.test_mp_test.TestMp_test)
Test something.
----------------------------------------------------------------------
Traceback (most recent call last):
~ Ausgelassen ~
RuntimeError:
An attempt has been made to start a new process before the
current process has finished its bootstrapping phase.
This probably means that you are not using fork to start your
child processes and you have forgotten to use the proper idiom
in the main module:
if __name__ == '__main__':
freeze_support()
...
The "freeze_support()" line can be omitted if the program
is not going to be frozen to produce an executable.
----------------------------------------------------------------------
Ran 1 test in 0.007s
FAILED (errors=1)
Test failed: <unittest.runner.TextTestResult run=1 errors=1 failures=0>
test_000_something (tests.test_mp_test.TestMp_test)
Test something. ... ERROR
Die folgende Schleife
・
・
・
Ich denke, dass der Testausführungsprozess von setup.py nicht den Spezifikationen von Multiprocess entspricht. Die Problemumgehung, die ich mir ausgedacht habe, war:
test_mp_test.py
"""Tests for `multiprocess_test` package."""
import unittest
import sys
from mp_test import mp_test
class TestMp_test(unittest.TestCase):
"""Tests for `mp_test` package."""
def setUp(self):
"""Set up test fixtures, if any."""
def tearDown(self):
"""Tear down test fixtures, if any."""
def test_000_something(self):
old_main = sys.modules["__main__"]
old_main_file = sys.modules["__main__"].__file__
sys.modules["__main__"] = sys.modules["mp_test.mp_test"]
sys.modules["__main__"].__file__ = sys.modules["mp_test.mp_test"].__file__
mp_test.main()
sys.modules["__main__"] = old_main
sys.modules["__main__"].__file__ = old_main_file
if __name__ == "__main__":
unittest.main()
https://stackoverflow.com/questions/33128681/how-to-unit-test-code-that-uses-python-multiprocessing
Schreiben Sie das laufende "\ _ \ _ main \ _ \ _" neu. Auf diese Weise können Sie auch unittest von setup.py aus ausführen.
Wenn Sie die Mehrprozessverarbeitung in mehreren Testfällen ausführen möchten, Sie können es zu einer dedizierten Testklasse machen und es in setUp und tearDown beschreiben.
test_mp_test.py
import unittest
import sys
from mp_test import mp_test
class TestMp_test(unittest.TestCase):
"""Tests for `mp_test` package."""
def setUp(self):
"""Set up test fixtures, if any."""
self._old_main = sys.modules["__main__"]
self._old_main_file = sys.modules["__main__"].__file__
sys.modules["__main__"] = sys.modules["mp_test.mp_test"]
sys.modules["__main__"].__file__ = sys.modules["mp_test.mp_test"].__file__
def tearDown(self):
"""Tear down test fixtures, if any."""
sys.modules["__main__"] = self._old_main
sys.modules["__main__"].__file__ = self._old_main_file
def test_000_something(self):
"""Test something."""
mp_test.main()
if __name__=="__main__":
unittest.main()
Ich bemerkte und untersuchte beim Aufbau einer Testumgebung mit Tox. Ich merke es nicht mit VSC oder Unittest alleine, also denke ich, dass es am Ende der Entwicklung zu "Wow!" Werden kann. Es scheint jedoch eine intelligentere Problemumgehung zu geben, aber ich würde es begrüßen, wenn jemand, der es weiß, es mir sagen könnte.
Recommended Posts