0 Launching a Process on P-Cores Only in Windows.md

Launching a Process on P-Cores Only in Windows

This Python script is designed to run a given program on Performance Cores (P-cores) of a modern CPU in a Windows environment. It uses Windows API functions through the ctypes library to interact with low-level system features.

---

1. Getting CPU Information

The first part of the script calls the Windows API function GetLogicalProcessorInformationEx to retrieve details about the system’s logical processors.

• It defines structures (PROCESSOR_RELATIONSHIP, SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX) to interpret the raw data returned by the API.

• It checks each processor’s EfficiencyClass.

  • A value of 1 indicates a P-core.

• It builds a bitmask representing which CPU cores are P-cores.

---

2. Creating and Managing a Process

The script defines wrappers around several Windows API functions (CreateProcessW, SetProcessAffinityMask, WaitForSingleObject, CloseHandle) to:

• Launch a new process (CreateProcessW) with the provided command line.
• Set the process affinity mask so the process is restricted to run only on the detected P-cores.
• Wait for the process to finish (WaitForSingleObject).
• Clean up handles after execution.

---

3. Main Execution Flow

When run from the command line:

1. It checks if a command was given; otherwise, it shows usage instructions.
2. It calculates the P-core mask.
3. If P-cores exist, it prints the mask in hexadecimal form.
4. It launches the target program restricted to P-cores.
5. If no P-cores are found, it prints an error message and exits.

---

✅ In short: This script ensures that a specified program runs only on P-cores, which can be useful for performance testing, benchmarking, or managing workloads on hybrid CPU architectures like Intel Alder Lake and later.

main.py

import sys
import subprocess
import ctypes
from ctypes import wintypes

# -------------------------
# CPU Information Retrieval
# -------------------------
RelationProcessorCore = 0

class PROCESSOR_RELATIONSHIP(ctypes.Structure):
    _fields_ = [
        ("Flags", wintypes.BYTE),
        ("EfficiencyClass", wintypes.BYTE),
        ("Reserved", wintypes.BYTE * 20),
        ("GroupCount", wintypes.DWORD),
        # In reality, a variable-length ProcessorGroup follows, but simplified here
    ]

class SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX(ctypes.Structure):
    _fields_ = [
        ("Relationship", wintypes.INT),
        ("Size", wintypes.DWORD),
        ("Processor", PROCESSOR_RELATIONSHIP),
    ]

GetLogicalProcessorInformationEx = ctypes.windll.kernel32.GetLogicalProcessorInformationEx
GetLogicalProcessorInformationEx.argtypes = [wintypes.INT, ctypes.c_void_p, ctypes.POINTER(wintypes.DWORD)]
GetLogicalProcessorInformationEx.restype = wintypes.BOOL

def get_pcore_mask():
    """
    Retrieve the CPU mask corresponding to Performance cores (P-cores).

    This function queries the system for logical processor information and
    generates a bitmask where bits corresponding to P-cores (EfficiencyClass == 1) are set.

    Returns:
        int: A bitmask with P-core CPU indices set.
    
    Raises:
        OSError: If the system call fails to retrieve processor information.
    """
    buffer_size = wintypes.DWORD(0)
    GetLogicalProcessorInformationEx(RelationProcessorCore, None, ctypes.byref(buffer_size))
    buffer = ctypes.create_string_buffer(buffer_size.value)

    if not GetLogicalProcessorInformationEx(RelationProcessorCore, buffer, ctypes.byref(buffer_size)):
        raise ctypes.WinError()

    mask = 0
    offset = 0
    cpu_index = 0
    while offset < buffer_size.value:
        entry = ctypes.cast(ctypes.byref(buffer, offset), ctypes.POINTER(SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX)).contents
        eff_class = entry.Processor.EfficiencyClass
        if eff_class == 1:  # P-core
            mask |= (1 << cpu_index)
        cpu_index += 1
        offset += entry.Size
    return mask


# -------------------------
# Process Launching
# -------------------------
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)

LPVOID = wintypes.LPVOID
LPWSTR = wintypes.LPWSTR
DWORD  = wintypes.DWORD
HANDLE = wintypes.HANDLE
BOOL   = wintypes.BOOL

CREATE_NEW_CONSOLE = 0x00000010
INFINITE = 0xFFFFFFFF
WAIT_OBJECT_0 = 0x00000000

class STARTUPINFO(ctypes.Structure):
    _fields_ = [
        ("cb", DWORD),
        ("lpReserved", LPWSTR),
        ("lpDesktop", LPWSTR),
        ("lpTitle", LPWSTR),
        ("dwX", DWORD),
        ("dwY", DWORD),
        ("dwXSize", DWORD),
        ("dwYSize", DWORD),
        ("dwXCountChars", DWORD),
        ("dwYCountChars", DWORD),
        ("dwFillAttribute", DWORD),
        ("dwFlags", DWORD),
        ("wShowWindow", ctypes.c_ushort),
        ("cbReserved2", ctypes.c_ushort),
        ("lpReserved2", ctypes.c_char_p),
        ("hStdInput", HANDLE),
        ("hStdOutput", HANDLE),
        ("hStdError", HANDLE),
    ]

class PROCESS_INFORMATION(ctypes.Structure):
    _fields_ = [
        ("hProcess", HANDLE),
        ("hThread", HANDLE),
        ("dwProcessId", DWORD),
        ("dwThreadId", DWORD),
    ]

CreateProcessW = kernel32.CreateProcessW
CreateProcessW.argtypes = [
    LPWSTR, LPWSTR, LPVOID, LPVOID, BOOL,
    DWORD, LPVOID, LPWSTR,
    ctypes.POINTER(STARTUPINFO),
    ctypes.POINTER(PROCESS_INFORMATION)
]
CreateProcessW.restype = BOOL

SetProcessAffinityMask = kernel32.SetProcessAffinityMask
SetProcessAffinityMask.argtypes = [HANDLE, DWORD]
SetProcessAffinityMask.restype = BOOL

WaitForSingleObject = kernel32.WaitForSingleObject
WaitForSingleObject.argtypes = [HANDLE, DWORD]
WaitForSingleObject.restype = DWORD

CloseHandle = kernel32.CloseHandle

def launch_with_affinity(cmdline, affinity_mask):
    """
    Launch a process with a specific CPU affinity mask.

    Args:
        cmdline (str): The command line string to execute.
        affinity_mask (int): The CPU affinity mask to assign to the process.

    Raises:
        OSError: If process creation or setting the affinity mask fails.
    """
    si = STARTUPINFO()
    si.cb = ctypes.sizeof(STARTUPINFO)
    pi = PROCESS_INFORMATION()

    success = CreateProcessW(
        None, cmdline, None, None, False,
        0, None, None,
        ctypes.byref(si), ctypes.byref(pi)
    )

    if not success:
        raise ctypes.WinError(ctypes.get_last_error())

    if not SetProcessAffinityMask(pi.hProcess, affinity_mask):
        raise ctypes.WinError(ctypes.get_last_error())

    while True:
        try:
            WaitForSingleObject(pi.hProcess, INFINITE)
            break
        except KeyboardInterrupt:
            pass

    CloseHandle(pi.hThread)
    CloseHandle(pi.hProcess)


# -------------------------
# Main Execution
# -------------------------
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python launch_pcore.py <command_to_execute>")
        sys.exit(1)

    cmdline = subprocess.list2cmdline(sys.argv[1:])
    mask = get_pcore_mask()
    if mask == 0:
        print("No P-core detected.")
        sys.exit(1)

    print(f"P-core mask: {hex(mask)}")
    launch_with_affinity(cmdline, mask)