Test that cpu_count() respects the taskset command line tool
(affinity)
| 43 | @pytest.mark.skipif(not LINUX, reason="No command line API for CPU affinity") |
| 44 | @pytest.mark.parametrize(("affinity"), [{0}, {1}, {0, 1}, {0, 2}]) |
| 45 | def test_cpu_affinity_taskset(affinity): |
| 46 | """Test that cpu_count() respects the taskset command line tool""" |
| 47 | if (os.cpu_count() or 1) < max(affinity) + 1: |
| 48 | raise pytest.skip("Not enough CPUs") # pragma: no cover |
| 49 | |
| 50 | subprocess.check_call( |
| 51 | [ |
| 52 | "taskset", |
| 53 | "-c", |
| 54 | ",".join(str(i) for i in sorted(affinity)), |
| 55 | sys.executable, |
| 56 | "-c", |
| 57 | f"from dask.system import CPU_COUNT; assert CPU_COUNT == {len(affinity)}", |
| 58 | ] |
| 59 | ) |
| 60 | |
| 61 | |
| 62 | @pytest.mark.skipif( |