双路多处理器组系统下Python ctypes调用GetLogicalProcessorInformationEx问题

问题根源 #

GetLogicalProcessorInformationEx返回的SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX结构体数组为可变长度结构，每个元素的Size字段根据当前记录类型动态变化，直接按固定结构体大小做指针偏移会导致后续结构体地址计算错误，这就是双路系统下无法识别第二个socket的核心原因。

实现步骤 #

• 定义匹配Windows API规范的ctypes结构体，包含动态长度字段的联合体
• 两次调用GetLogicalProcessorInformationEx：首次调用获取所需缓冲区大小，第二次读取完整数据
• 按每个结构体的Size字段动态计算下一个结构体的偏移地址，遍历所有返回记录
• 筛选Relationship为RelationProcessorPackage的记录，即为对应物理socket的处理器信息

完整可运行代码 #

import ctypes
from ctypes import wintypes

# 常量定义
RELATION_PROCESSOR_PACKAGE = 3

# 结构体定义
class GROUP_AFFINITY(ctypes.Structure):
    _fields_ = [
        ("Mask", wintypes.LPVOID),
        ("Group", wintypes.WORD),
        ("Reserved", wintypes.WORD * 3)
    ]

class PROCESSOR_RELATIONSHIP(ctypes.Structure):
    _fields_ = [
        ("Flags", wintypes.BYTE),
        ("EfficiencyClass", wintypes.BYTE),
        ("Reserved", wintypes.WORD),
        ("GroupCount", wintypes.WORD),
        ("GroupMask", GROUP_AFFINITY * 1)
    ]

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

# 绑定Windows API
kernel32 = ctypes.WinDLL("kernel32.dll", use_last_error=True)
GetLogicalProcessorInformationEx = kernel32.GetLogicalProcessorInformationEx
GetLogicalProcessorInformationEx.argtypes = [
    wintypes.DWORD,
    ctypes.POINTER(SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX),
    wintypes.LPDWORD
]
GetLogicalProcessorInformationEx.restype = wintypes.BOOL

# 首次调用获取所需缓冲区大小
buf_size = wintypes.DWORD(0)
GetLogicalProcessorInformationEx(RELATION_PROCESSOR_PACKAGE, None, ctypes.byref(buf_size))

# 分配缓冲区并读取数据
buf = ctypes.create_string_buffer(buf_size.value)
success = GetLogicalProcessorInformationEx(
    RELATION_PROCESSOR_PACKAGE,
    ctypes.cast(buf, ctypes.POINTER(SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX)),
    ctypes.byref(buf_size)
)
if not success:
    raise ctypes.WinError(ctypes.get_last_error())

# 遍历解析所有socket信息
sockets = []
offset = 0
while offset < buf_size.value:
    addr = ctypes.addressof(buf) + offset
    info = ctypes.cast(addr, ctypes.POINTER(SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX)).contents
    sockets.append({
        "socket_id": len(sockets),
        "processor_group": info.u.Processor.GroupMask[0].Group,
        "logical_core_mask": hex(info.u.Processor.GroupMask[0].Mask)
    })
    # 核心：按当前结构体实际Size动态偏移，而非固定大小
    offset += info.Size

# 输出结果
for sock in sockets:
    print(f"Socket {sock['socket_id']}：处理器组{sock['processor_group']}，逻辑核掩码{sock['logical_core_mask']}")

验证说明 #

代码兼容单路、双路、多路CPU场景，输出逻辑可直接对齐现有C++版本结果，满足一致性要求。

内容的提问来源于stack exchange，提问作者jawalking