gh-109649: Add os.process_cpu_count() function (#109907)

* Refactor os_sched_getaffinity_impl(): move variable definitions to
  their first assignment.
* Fix test_posix.test_sched_getaffinity(): restore the old CPU mask
  when the test completes!
* Doc: Specify that os.cpu_count() counts *logicial* CPUs.
* Doc: Specify that os.sched_getaffinity(0) is related to the calling
  thread.
This commit is contained in:
Victor Stinner 2023-10-01 00:12:51 +02:00 committed by GitHub
parent 2c234196ea
commit c81521020d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 125 additions and 47 deletions

View file

@ -3996,14 +3996,42 @@ class OSErrorTests(unittest.TestCase):
self.fail(f"No exception thrown by {func}")
class CPUCountTests(unittest.TestCase):
def check_cpu_count(self, cpus):
if cpus is None:
self.skipTest("Could not determine the number of CPUs")
self.assertIsInstance(cpus, int)
self.assertGreater(cpus, 0)
def test_cpu_count(self):
cpus = os.cpu_count()
if cpus is not None:
self.assertIsInstance(cpus, int)
self.assertGreater(cpus, 0)
else:
self.check_cpu_count(cpus)
def test_process_cpu_count(self):
cpus = os.process_cpu_count()
self.assertLessEqual(cpus, os.cpu_count())
self.check_cpu_count(cpus)
@unittest.skipUnless(hasattr(os, 'sched_setaffinity'),
"don't have sched affinity support")
def test_process_cpu_count_affinity(self):
ncpu = os.cpu_count()
if ncpu is None:
self.skipTest("Could not determine the number of CPUs")
# Disable one CPU
mask = os.sched_getaffinity(0)
if len(mask) <= 1:
self.skipTest(f"sched_getaffinity() returns less than "
f"2 CPUs: {sorted(mask)}")
self.addCleanup(os.sched_setaffinity, 0, list(mask))
mask.pop()
os.sched_setaffinity(0, mask)
# test process_cpu_count()
affinity = os.process_cpu_count()
self.assertEqual(affinity, ncpu - 1)
# FD inheritance check is only useful for systems with process support.
@support.requires_subprocess()