英文:
Pytest with multiprocessing lock not working as expected when running tests in parallel
问题
我正在尝试使用pytest插件parallel-0.1.1和xdist-3.2.1以及**--tests-per-worker n**标志来并行运行我的pytests。我有一组需要预处理步骤的测试，这些步骤必须在关键部分中运行。这个部分由一个多进程锁保护，以避免多个工作进程同时执行。
然而，尽管使用了锁，工作进程仍然同时进入关键部分，导致同步问题。
这是问题代码的简化版本：
测试代码：
import pytest
@pytest.mark.parametrize("preprocess", ["config1"], indirect=True)
def test_example1(preprocess):
# 在测试中使用预处理数据
print(f"使用预处理数据的测试: {preprocess}")
# 对预处理数据执行操作
@pytest.mark.parametrize("preprocess", ["config2"], indirect=True)
def test_example2(preprocess):
# 在测试中使用预处理数据
print(f"使用预处理数据的测试: {preprocess}")
# 对预处理数据执行操作
conftest.py文件：
import pytest
import multiprocessing
lock = multiprocessing.Lock()
@pytest.fixture
def preprocess(request):
with lock:
# 关键部分
为什么我的锁没有阻止并行运行测试时同时进入关键部分？如何解决这个同步问题？
感谢您对此事的任何帮助！
英文:
I am trying to run my pytests in parallel using the pytest plugins parallel-0.1.1 and xdist-3.2.1 along with the --tests-per-worker n flag. I have a set of tests that require a preprocessing step which must be run in a critical section. This section is protected by a multiprocessing lock to avoid simultaneous execution by multiple workers.
However, despite using the lock, the workers enter the critical section simultaneously, leading to synchronization problems.
Here is a simplified version of the problematic code:
Test code:
import pytest
@pytest.mark.parametrize("preprocess", ["config1"], indirect=True)
def test_example1(preprocess):
# Use the preprocessed data in the test
print(f"Test using preprocessed data: {preprocess}")
# do something with preprocess
@pytest.mark.parametrize("preprocess", ["config2"], indirect=True)
def test_example2(preprocess):
# Use the preprocessed data in the test
print(f"Test using preprocessed data: {preprocess}")
# do something with preprocess
conftest.py file:
import pytest
import multiprocessing
lock = multiprocessing.Lock()
@pytest.fixture
def preprocess(request):
with lock:
# critical section
Why is my lock not preventing simultaneous entry into the critical section when running the tests in parallel? How can I resolve this synchronization problem?
I appreciate any assistance with this matter!
答案1
得分: 1
https://pypi.org/project/fasteners/ 这样的库具有更好的锁定机制，适合您试图实现的目标。您希望基于文件创建锁，以便不同的进程不会创建不同的锁。
import fasteners
lock = fasteners.InterProcessLock('path/to/lock.file')
with lock:
... # 独占访问
然后您的代码变得线程安全 / 进程安全。
英文:
A library such as https://pypi.org/project/fasteners/ has better locking mechanisms for the goal you're trying to accomplish. You want a lock based around a file so that different processes don't create different locks.
import fasteners
lock = fasteners.InterProcessLock('path/to/lock.file')
with lock:
... # exclusive access
Then your code becomes thread-safe / process-safe.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论