Use psutil.cpu_count(logical=False) in auto_detect_cpus()

This commit is contained in:
Ivan Zaikin
2020-07-23 10:18:43 +07:00
parent 2fc21fddbb
commit 77e5fdfef9
3 changed files with 10 additions and 30 deletions

View File

@@ -1,6 +1,6 @@
from setuptools import setup, find_packages
install_requires = ["execnet>=1.1", "pytest>=4.4.0", "pytest-forked", "six"]
install_requires = ["execnet>=1.1", "psutil>=3.0.0", "pytest>=4.4.0", "pytest-forked", "six"]
with open("README.rst") as f:

View File

@@ -1,28 +1,13 @@
import os
import uuid
import psutil
import py
import pytest
def auto_detect_cpus():
try:
from os import sched_getaffinity
except ImportError:
if os.environ.get("TRAVIS") == "true":
# workaround https://bitbucket.org/pypy/pypy/issues/2375
return 2
try:
from os import cpu_count
except ImportError:
from multiprocessing import cpu_count
else:
def cpu_count():
return len(sched_getaffinity(0))
try:
n = cpu_count()
n = psutil.cpu_count(logical=False)
except NotImplementedError:
return 1
return n if n else 1

View File

@@ -35,17 +35,10 @@ def test_dist_options(testdir):
def test_auto_detect_cpus(testdir, monkeypatch):
import os
import psutil
from xdist.plugin import pytest_cmdline_main as check_options
if hasattr(os, "sched_getaffinity"):
monkeypatch.setattr(os, "sched_getaffinity", lambda _pid: set(range(99)))
elif hasattr(os, "cpu_count"):
monkeypatch.setattr(os, "cpu_count", lambda: 99)
else:
import multiprocessing
monkeypatch.setattr(multiprocessing, "cpu_count", lambda: 99)
monkeypatch.setattr(psutil, "cpu_count", lambda logical=True: 99)
config = testdir.parseconfigure("-n2")
assert config.getoption("numprocesses") == 2
@@ -58,10 +51,12 @@ def test_auto_detect_cpus(testdir, monkeypatch):
assert config.getoption("usepdb")
assert config.getoption("numprocesses") == 0
monkeypatch.delattr(os, "sched_getaffinity", raising=False)
monkeypatch.setenv("TRAVIS", "true")
def cpu_count_not_implemented(logical=True):
raise NotImplementedError
monkeypatch.setattr(psutil, "cpu_count", cpu_count_not_implemented)
config = testdir.parseconfigure("-nauto")
assert config.getoption("numprocesses") == 2
assert config.getoption("numprocesses") == 1
def test_boxed_with_collect_only(testdir):