27 lines
1.1 KiB
Python
27 lines
1.1 KiB
Python
"""Test affinity only on this short-lived Python worker thread."""
|
|
import ctypes as c
|
|
from ctypes import wintypes as w
|
|
import json
|
|
import threading
|
|
k=c.WinDLL("kernel32",use_last_error=True)
|
|
k.GetCurrentThread.restype=w.HANDLE
|
|
k.GetCurrentProcess.restype=w.HANDLE
|
|
k.GetProcessAffinityMask.argtypes=[w.HANDLE,c.POINTER(c.c_size_t),c.POINTER(c.c_size_t)]
|
|
k.SetThreadAffinityMask.argtypes=[w.HANDLE,c.c_size_t]
|
|
k.SetThreadAffinityMask.restype=c.c_size_t
|
|
def worker():
|
|
pm,sm=c.c_size_t(),c.c_size_t()
|
|
k.GetProcessAffinityMask(k.GetCurrentProcess(),c.byref(pm),c.byref(sm))
|
|
print(json.dumps({"process_mask":hex(pm.value),"system_mask":hex(sm.value)}))
|
|
for mask in [1,2,3,4,8,0xc,0xf,0xfc,0xfff,0xffffc,0xfffff,0xfffffffc]:
|
|
c.set_last_error(0)
|
|
old=k.SetThreadAffinityMask(k.GetCurrentThread(),mask)
|
|
error=c.get_last_error()
|
|
print(json.dumps({"requested":hex(mask),"previous":hex(old),"error":error}))
|
|
if old:
|
|
if not k.SetThreadAffinityMask(k.GetCurrentThread(),old):
|
|
raise c.WinError(c.get_last_error())
|
|
t=threading.Thread(target=worker)
|
|
t.start()
|
|
t.join()
|