python - 构建 ctypes 类的简洁方式

Question

我已经定义了一个 ctypes 类和一个相关的便利函数，如下所示:

class BNG_FFITuple(Structure):
    _fields_ = [("a", c_uint32),
                ("b", c_uint32)]


class BNG_FFIArray(Structure):
    _fields_ = [("data", c_void_p),
                ("len", c_size_t)]

    # Allow implicit conversions from a sequence of 32-bit unsigned ints
    @classmethod
    def from_param(cls, seq):
        return seq if isinstance(seq, cls) else cls(seq)

    def __init__(self, seq, data_type = c_float):
        array_type = data_type * len(seq)
        raw_seq = array_type(*seq)
        self.data = cast(raw_seq, c_void_p)
        self.len = len(seq)


def bng_void_array_to_tuple_list(array, _func, _args):
    res = cast(array.data, POINTER(BNG_FFITuple * array.len))[0]
    return res

convert = lib.convert_to_bng
convert.argtypes = (BNG_FFIArray, BNG_FFIArray)
convert.restype = BNG_FFIArray
convert.errcheck = bng_void_array_to_tuple_list
drop_array = lib.drop_array 
drop_array.argtypes = (POINTER(BNG_FFIArray),)

然后我定义一个简单的便利函数:

def f(a, b):
    return [(i.a, i.b) for i in iter(convert(a, b))]

大部分都可以完美运行，但我有两个问题:

• 不够灵活；我希望能够使用 c_float 而不是 c_uint32 实例化 BNG_FFITuple (所以字段是 c_float )，反之亦然，所以 BNG_FFIArray data_type 是 c_uint32。不过，我不清楚如何做到这一点。
• 我想释放现在由 Python 拥有的内存，方法是将 POINTER(BNG_FFIArray) 发送回我的 dylib(请参阅 drop_array - 我已经已经在我的 dylib 中定义了一个函数)，但我不确定应该在什么时候调用它。

有没有办法以一种更整洁、更 Python 的方式封装所有这些，也更安全？我担心如果没有以稳健的方式定义内存清理(在 __exit__?__del__?)出现任何问题都会导致内存未释放

Answer 1

由于您对 rust 方面有一定的控制权，因此最简单的做法是在调用之前从 Python 预先分配结果数组，并将所有内容传递到单个结构中。

下面的代码假定了这种修改，但也指定了如果你不能这样做，你将在哪里进行释放。

请注意，如果您进行这种封装，则不需要为库函数指定参数和结果处理等内容，因为您只是从一个地方调用实际函数，并且始终使用完全相同的相同类型的参数。

我不知道 rust(甚至我的 C 有点生锈)，但下面的代码假设您重新定义 rust 以匹配类似这样的内容:

typedef struct FFIParams {
    int32 source_ints;
    int32 len;
    void * a;
    void * b;
    void * result;
} FFIParams;

void convert_to_bng(FFIParams *p) {
}

这里是 Python。最后一点——由于参数结构的重用，这不是线程安全的。如果需要，这很容易解决。

from ctypes import c_uint32, c_float, c_size_t, c_void_p
from ctypes import Structure, POINTER, pointer, cast
from itertools import izip, islice

_test_standalone = __name__ == '__main__'

if _test_standalone:
    class lib(object):
        @staticmethod
        def convert_to_bng(ptr_params):
            params = ptr_params.contents
            source_ints = params.source_ints
            types = c_uint32, c_float
            if not source_ints:
                types = reversed(types)
            length = params.len
            src_type, dst_type = types
            src_type = POINTER(length * src_type)
            dst_type = POINTER(length * 2 * dst_type)
            a = cast(params.a, src_type).contents
            b = cast(params.b, src_type).contents
            result = cast(params.result, dst_type).contents

            # Assumes we are converting int to float or back...
            func = float if source_ints else int
            result[0::2] = map(func, a)
            result[1::2] = map(func, b)

class _BNG_FFIParams(Structure):
    _fields_ = [("source_ints", c_uint32),
                ("len", c_size_t),
                ("a", c_void_p),
                ("b", c_void_p),
                ("result", c_void_p)]

class _BNG_FFI(object):

    int_type = c_uint32
    float_type = c_float
    _array_type = type(10 * int_type)

    # This assumes we want the result to be opposite type.
    # Maybe I misunderstood this -- easily fixable if so.
    _result_type = {int_type: float_type, float_type: int_type}

    def __init__(self):
        my_params = _BNG_FFIParams()
        self._params = my_params
        self._pointer = POINTER(_BNG_FFIParams)(my_params)
        self._converter = lib.convert_to_bng


    def _getarray(self, seq, data_type):
        # Optimization for pre-allocated correct array type
        if type(type(seq)) == self._array_type and seq._type_ is data_type:
            print("Optimized!")
            return seq
        return (data_type * len(seq))(*seq)

    def __call__(self, a, b, data_type=float_type):
        length = len(a)
        if length != len(b):
            raise ValueError("Input lengths must be same")

        a, b = (self._getarray(x, data_type) for x in (a, b))

        # This has the salutary side-effect of insuring we were
        # passed a valid type
        result = (length * 2 * self._result_type[data_type])()

        params = self._params
        params.source_ints = data_type is self.int_type
        params.len = length
        params.a = cast(pointer(a), c_void_p)
        params.b = cast(pointer(b), c_void_p)
        params.result = cast(pointer(result), c_void_p)
        self._converter(self._pointer)

        evens = islice(result, 0, None, 2)
        odds = islice(result, 1, None, 2)
        result = list(izip(evens, odds))

        # If you have to have the converter allocate memory,
        # deallocate it here...

        return result

convert = _BNG_FFI()

if _test_standalone:
    print(convert([1.0, 2.0, 3.0], [4.0, 5.0, 6.0], c_float))
    print(convert([1, 2, 3], [4, 5, 6], c_uint32))
    print(convert([1, 2, 3], (c_uint32 * 3)(4, 5, 6), c_uint32))