编程与调试 -- 使用 ctypes 进行 Python 和 C 的混合编程

# 设置函数参数类型为 int, int, int, void*
fun.argtypes = (c_int, c_int, c_int, c_void_p)
# 设置返回值类型为 float
fun.restype  = c_float

import ctypes
import ctypes.wintypes

GENERIC_WRITE = 0x40000000
CREATE_ALWAYS = 0x00000002
FILE_ATTRIBUTE_NORMAL = 0x00000080
LOCKFILE_EXCLUSIVE_LOCK = 0x00000002
LOCKFILE_FAIL_IMMEDIATELY = 0x00000001

class Overlapped(ctypes.Structure):
    """Overlapped is required and used in LockFileEx and UnlockFileEx."""
    _fields_ = [('Internal', ctypes.wintypes.LPVOID),
                ('InternalHigh', ctypes.wintypes.LPVOID),
                ('Offset', ctypes.wintypes.DWORD),
                ('OffsetHigh', ctypes.wintypes.DWORD),
                ('Pointer', ctypes.wintypes.LPVOID),
                ('hEvent', ctypes.wintypes.HANDLE)]

# https://docs.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-createfilew
CreateFileW = ctypes.windll.kernel32.CreateFileW
CreateFileW.argtypes = [
    ctypes.wintypes.LPCWSTR,  # lpFileName
    ctypes.wintypes.DWORD,  # dwDesiredAccess
    ctypes.wintypes.DWORD,  # dwShareMode
    ctypes.wintypes.LPVOID,  # lpSecurityAttributes
    ctypes.wintypes.DWORD,  # dwCreationDisposition
    ctypes.wintypes.DWORD,  # dwFlagsAndAttributes
    ctypes.wintypes.LPVOID,  # hTemplateFile
]
CreateFileW.restype = ctypes.wintypes.HANDLE

# https://docs.microsoft.com/en-us/windows/win32/api/handleapi/nf-handleapi-closehandle
CloseHandle = ctypes.windll.kernel32.CloseHandle
CloseHandle.argtypes = [
    ctypes.wintypes.HANDLE,  # hFile
]
CloseHandle.restype = ctypes.wintypes.BOOL

# https://docs.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-lockfileex
LockFileEx = ctypes.windll.kernel32.LockFileEx
LockFileEx.argtypes = [
    ctypes.wintypes.HANDLE,  # hFile
    ctypes.wintypes.DWORD,  # dwFlags
    ctypes.wintypes.DWORD,  # dwReserved
    ctypes.wintypes.DWORD,  # nNumberOfBytesToLockLow
    ctypes.wintypes.DWORD,  # nNumberOfBytesToLockHigh
    ctypes.POINTER(Overlapped),  # lpOverlapped
]
LockFileEx.restype = ctypes.wintypes.BOOL

# Commonly used functions are listed here so callers don't need to import
# ctypes.
GetLastError = ctypes.GetLastError
Handle = ctypes.wintypes.HANDLE

from __future__ import print_function

import contextlib
import logging
import os
import sys
import time
import traceback

class LockError(Exception):
    pass

if sys.platform.startswith('win'):
    # Windows implementation
    try:
        from . import win32imports
    except ImportError: # attempted relative import with no known parent package
        import win32imports

    BYTES_TO_LOCK = 1

    def _open_file(lockfile):
        fdir = os.path.split(lockfile)[0]
        if not os.path.exists(fdir):
            os.makedirs(fdir)
        cfile = win32imports.CreateFileW(
                lockfile,  # lpFileName
                win32imports.GENERIC_WRITE,  # dwDesiredAccess
                0,  # dwShareMode=prevent others from opening file
                None,  # lpSecurityAttributes
                win32imports.CREATE_ALWAYS,  # dwCreationDisposition
                win32imports.FILE_ATTRIBUTE_NORMAL,  # dwFlagsAndAttributes
                None  # hTemplateFile
            )
        retv = win32imports.Handle(cfile)
        assert cfile and retv, lockfile
        return retv

    def _close_file(handle, lockfile):
        # CloseHandle releases lock too.
        win32imports.CloseHandle(handle)
        try:
            os.remove(lockfile)
        except:
            pass

    def _lock_file(handle, lockfile):
        ret = win32imports.LockFileEx(
            handle,  # hFile
            win32imports.LOCKFILE_FAIL_IMMEDIATELY
            | win32imports.LOCKFILE_EXCLUSIVE_LOCK,  # dwFlags
            0,  # dwReserved
            BYTES_TO_LOCK,  # nNumberOfBytesToLockLow
            0,  # nNumberOfBytesToLockHigh
            win32imports.Overlapped()  # lpOverlapped
        )
        # LockFileEx returns result as bool, which is converted into an integer
        # (1 == successful; 0 == not successful)
        if ret == 0:
            error_code = win32imports.GetLastError()
            if error_code == 6: # 无效的空句柄
                pass
            raise OSError('Failed to lock handle(%r) file(%s) (error code: %d).' % (handle, lockfile, error_code))
else:
    # Unix implementation
    import fcntl

    def _open_file(lockfile):
        open_flags = (os.O_CREAT | os.O_WRONLY)
        return os.open(lockfile, open_flags, 0o644)

    def _close_file(fd, lockfile):
        os.close(fd)
        try:
            os.remove(lockfile)
        except:
            pass

    def _lock_file(fd, lockfile):
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)

def _try_lock(lockfile):
    f = _open_file(lockfile)
    try:
        _lock_file(f, lockfile)
    except Exception:
        _close_file(f, lockfile)
        raise
    return lambda: _close_file(f, lockfile)

def _lock(path, timeout=0):
    """_lock returns function to release the lock if locking was successful.

    _lock also implements simple retry logic."""
    elapsed = 0
    while True:
        try:
            return _try_lock(path + '.lockedfile') # 不要改后缀，funclib 忽略这个文件。
        except (OSError, IOError) as e:
            if elapsed < timeout:
                sleep_time = min(10, timeout - elapsed)
                logging.info(
                    'Could not create git cache lockfile; '
                    'will retry after sleep(%d).', sleep_time)
                elapsed += sleep_time
                time.sleep(sleep_time)
                continue
            traceback.print_exc()
            raise LockError("Error locking %s (err: %s)" % (path, str(e)))

@contextlib.contextmanager
def lock(path, timeout=0):
    """Get exclusive lock to path.

    Usage:
      import lockfile
      with lockfile.lock(path, timeout):
        # Do something
        pass

     """
    release_fn = _lock(path, timeout)
    try:
        yield
    finally:
        release_fn()

class POINT(Structure):
    _fields_ = [('x', c_int), ('y', c_int)]

class RECT(Structure):
    _fields_ = [('a', POINT), ('b', POINT)]

p1 = POINT(1, 2)
p2 = POINT(3, 4)
rc = RECT(p1, p2)

print('rc.a.x =', rc.a.x)
print('rc.a.y =', rc.a.y)
print('rc.b.x =', rc.b.x)
print('rc.b.y =', rc.b.y)

rc.a, rc.b = rc.b, rc.a

print('after swap, bad result due to this is the pointer.')
print('rc.a.x =', rc.a.x)
print('rc.a.y =', rc.a.y)
print('rc.b.x =', rc.b.x)
print('rc.b.y =', rc.b.y)

MarioCallback = ctypes.CFUNCTYPE(
        ctypes.c_int,
        ctypes.c_int, # code
        ctypes.c_int, # subcode
        ctypes.c_int, # taskid
        ctypes.c_wchar_p, # instr
        ctypes.POINTER(ctypes.c_wchar_p), # outstr
        )

def mariotest(debug):
    dllpath, dllx64 = getMarioDll(debug)
    mydll = ctypes.cdll.LoadLibrary(dllpath)
    mydll.MarioPython()

    instr = ctypes.create_unicode_buffer(jsondumps({"key": "中文"}))
    outstr = ctypes.c_wchar_p(0)
    print(mydll.MarioAlloc(ctypes.byref(outstr), instr))
    print(outstr.value)
    print(mydll.MarioRelease(ctypes.byref(outstr)))

    def funMarioCallback(code, subcode, tasdid, instr, outstr):
        print("funMarioCallback", code, subcode, tasdid, instr, outstr)
        print(type(outstr), outstr)
        print(mydll.MarioAlloc2(outstr, instr))
        return 1

    code = MARIO_CODE_TEST
    taskid = 2
    instr = ctypes.create_unicode_buffer(jsondumps({"key": "中文"}))
    outstr = ctypes.c_wchar_p(0)
    # MarioCallback(funMarioCallback) 存在生命周期。
    retv = mydll.MarioFun(MarioCallback(funMarioCallback), code, taskid, instr, ctypes.byref(outstr))
    print(outstr.value)
    mydll.MarioRelease(ctypes.byref(outstr))

if __name__ == "__main__":
    mariotest(DEBUG)

# 变长输入，Python -> C++，直接调用就好了。
# 变长回调，C++ -> Python，回调两次就好了。
def mariotest3(data):
    dllpath, dllx64 = getMarioDll(DEBUG)
    mydll = ctypes.cdll.LoadLibrary(dllpath)
    mydll.MarioPython()

    MarioCallbackTest = ctypes.CFUNCTYPE(
        ctypes.c_int,
        ctypes.c_size_t,
        ctypes.POINTER(ctypes.c_char_p),
        )

    result = None
    def funMarioCallbackTest(size, pdata): # <__main__.LP_c_char_p object at>
        #print("funMarioCallbackTest", size, pdata)

        MarioCallbackRCB = ctypes.CFUNCTYPE(
            ctypes.c_int,
            ctypes.c_size_t,
            ctypes.POINTER(ctypes.c_char * size),
        )

        def funMarioCallbackRCB(size, mdata): # <__main__.LP_c_char_Array_10 object at>
            #print("funMarioCallbackRCB", size, mdata)
            #print(mdata.contents.raw)
            nonlocal result
            result = mdata.contents.raw
            return 0
        # 根据参数，再次构造 Python 回调函数。
        mydll.MarioReCallback(MarioCallbackRCB(funMarioCallbackRCB), pdata, size)
        return 0

    datasize = len(data)
    mydll.MarioTest(MarioCallbackTest(funMarioCallbackTest), data, datasize)
    #print(data) -- Python 内存传入，是可以直接被修改的。
    print("Python print", "\t", result)
    return result

if __name__ == "__main__":
    #mariotest2(b"abc")
    #mariotest2(b"abcdef")
    #mariotest(DEBUG)
    mariotest3(b"ab\x00\x01")