清除构造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))]

大部分工作都很完美,但我有两个问题:

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

有没有办法用更整洁的Pythonic方式封装所有这些,这也更安全？我担心,如果没有内存清理以稳健的方式被定义(上__exit__？__del__？),任何的出错会导致unfreed内存

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))