INNER CODE UNIT · Python
test_auto_affinity_threads
openHiTLS/openHiTLS · testcode/benchmark/async_perf/test_async_perf.py:123
def test_auto_affinity_threads(self):
with tempfile.TemporaryDirectory(prefix="async-affinity-test-") as directory:
snapshots = []
with tempfile.TemporaryFile(mode="w+") as log:
proc = subprocess.Popen(
[str(BINARY), "-a", "hs-sync-tls13-psk", "--device-mode", "worker",
"-w", "2", "--pin-cpu", "auto", "-s", "1", "--out", directory],
stdout=log, stderr=subprocess.STDOUT)
try:
deadline = time.monotonic() + 15
while proc.poll() is None and time.monotonic() < deadline:
try:
masks = {int(p.name): sorted(os.sched_getaffinity(int(p.name)))
for p in Path(f"/proc/{proc.pid}/task").iterdir()}
if len(masks) == 3 and all(len(mask) == 1 for mask in masks.values()):
snapshots.append(masks)
except (FileNotFoundError, ProcessLookupError):
pass