子类 ctypes.c_void_p 表示具有用于构造函数和析构函数/自由的自定义 C 函数的自定义处理程序

Question

我有一个有点奇怪的问题。

想象有某种带有类似操作的 C 接口（实际上我正在考虑 webrtc vad API）：

Handler* create();
int process(Handler*);
void free(Handler*);

我知道如何使用

ctypes

和

ctypes.c_void_p

来表示这些函数来表示指向处理程序的指针。

现在是异国情调的问题。可以将此模式表示为派生自

ctypes.c_void_p

的类吗？

• 从

  ctypes.c_void_p

  推导出来总体上正确吗？
• 此类派生类的构造函数是否有任何限制？ ctypes 会使用构造函数来构造编组返回的句柄吗？
• 在派生类

  self.value

  中修改

  __init__

  是否正确？
• ctypes如何创建

  HandlerPointer

  的对象，它会调用

  __init__

  并且会阻止它正常运行吗？
• ctypes 是否正确支持/编组

  c_void_p

  派生的argtypes/restypes？
• 是否有更自然的方式来表示此类胶囊状对象（指向可能不透明结构的指针以及析构函数）/API？

例如做类似的事情：

import os
import ctypes

class HandlerPointer(ctypes.c_void_p):
  @staticmethod
  def ffi(lib_path = os.path.abspath('mylib.so')):
    lib = ctypes.CDLL(lib_path)
    lib.create.argtypes = []
    lib.create.restype = HandlerPointer
    lib.process.argtypes = [HandlerPointer]
    lib.process.restype = ctypes.c_int
    lib.free.argtypes = [HandlerPointer]
    lib.free.restype = None
    return lib

  def __init__(self):
    super().__init__()
    self.value = self.lib.create().value
  
  def process(self):
    return self.lib.process(self)
  
  def __del__(self):
    self.lib.free(self)

# can't do this var init inside the class as can't refer yet to HandlerPointer inside `.ffi()` if class not initialized yet
HandlerPointer.lib = HandlerPointer.ffi() # otherwise

---

UPD：看起来拥有

__init__

方法是有效的，但

__del__

方法（调用 C 地自定义免费方法）除外，其存在将导致

double free or corruption (!prev)

崩溃。发现问题，修复下面我的评论中描述的问题，很快就会发布完整的解决方案。

Answer 1

在创建

ctypes

派生的输出对象时，

__init__

似乎绕过了自定义的

__new__

和

c_void_p

，因此可以安全地覆盖这些魔法。当使用

ctypes

派生类型作为

ctypes.c_void_p

/

argtypes

的一部分时，基于

restype

的绑定可以正常工作。我在这里粘贴了来自 WebRTC 的 VAD 绑定示例：https://webrtc.googlesource.com/src/+/refs/heads/main，其中包含 5 个正在运行的函数（包括构造函数和析构函数）使用自定义的不透明处理程序：

VadInst* WebRtcVad_Create(void);
void WebRtcVad_Free(VadInst* handle);
int WebRtcVad_Init(VadInst* handle);
int WebRtcVad_set_mode(VadInst* handle, int mode);
int WebRtcVad_Process(VadInst* handle, int fs, const int16_t* audio_frame, size_t frame_length);

此包装解决方案使用：

• ctypes.c_void_p

  派生类来表示处理程序

• __new__

  -绑定/表示来自 C 语言的工厂构造函数并保留单个 Python 对象以确保单次销毁的神奇功能。如果在此对象上执行深度复制（通过在同一底层内存指针上调用两次

  __del__

  ），此解决方案可能会导致崩溃/双重释放。但我没有测试过。

• __init__

  -代表处理程序初始化的魔法

• __del__

  -在Python-land对象变得未被引用时调用C-land析构函数的魔法

如果我们不尝试深度复制此处理程序并将其欺骗为双释放，那么效果会很好。

# a more complete version at https://github.com/vadimkantorov/webrtcvadctypes

import os
import ctypes

class Vad(ctypes.c_void_p):
    lib_path = os.path.abspath('webrtcvadctypesgmm.so')
    _webrtcvad = None
    
    @staticmethod
    def initialize(lib_path):
        Vad._webrtcvad = Vad.ffi(lib_path)

    @staticmethod
    def ffi(lib_path):
        lib = ctypes.CDLL(lib_path)
        lib.WebRtcVad_Create.argtypes = []
        lib.WebRtcVad_Create.restype = Vad
        lib.WebRtcVad_Free.argtypes = [Vad]
        lib.WebRtcVad_Free.restype = None
        lib.WebRtcVad_Init.argtypes = [Vad]
        lib.WebRtcVad_Init.restype = ctypes.c_int
        lib.WebRtcVad_set_mode.argtypes = [Vad, ctypes.c_int]
        lib.WebRtcVad_set_mode.restype = ctypes.c_int
        lib.WebRtcVad_Process.argtypes = [Vad, ctypes.c_int, ctypes.c_void_p, ctypes.c_size_t]
        lib.WebRtcVad_Process.restype = ctypes.c_int
        lib.WebRtcVad_ValidRateAndFrameLength.argtypes = [ctypes.c_int, ctypes.c_size_t]
        lib.WebRtcVad_ValidRateAndFrameLength.restype = ctypes.c_int
        return lib
    
    @staticmethod
    def valid_rate_and_frame_length(rate, frame_length, lib_path = None):
        if Vad._webrtcvad is None:
            Vad.initialize(lib_path or Vad.lib_path)
        return 0 == Vad._webrtcvad.WebRtcVad_ValidRateAndFrameLength(rate, frame_length)
    
    def set_mode(self, mode):
        assert Vad._webrtcvad is not None
        assert mode in [None, 0, 1, 2, 3]
        if mode is not None:
            assert 0 == Vad._webrtcvad.WebRtcVad_set_mode(self, mode)

    def is_speech(self, buf, sample_rate, length=None):
        assert Vad._webrtcvad is not None
        assert sample_rate in [8000, 16000, 32000, 48000]
        length = length or (len(buf) // 2)
        assert length * 2 <= len(buf), f'buffer has {len(buf) // 2} frames, but length argument was {length}'
        return 1 == Vad._webrtcvad.WebRtcVad_Process(self, sample_rate, buf, length)

    def __new__(cls, mode=None, lib_path = None):
        if Vad._webrtcvad is None:
            Vad.initialize(lib_path or Vad.lib_path)
        assert Vad._webrtcvad is not None
        return Vad._webrtcvad.WebRtcVad_Create()

    def __init__(self, mode=None, lib_path = None):
        assert Vad._webrtcvad is not None
        assert 0 == Vad._webrtcvad.WebRtcVad_Init(self)
        if mode is not None:
            self.set_mode(mode)
    
    def __del__(self):
        assert Vad._webrtcvad is not None
        Vad._webrtcvad.WebRtcVad_Free(self)
        self.value = None

---

总的来说，我不确定通过从

c_void_p

继承来表示这种语义是否是最好的方法，但是使用这个和各种相关的 Python 魔术方法很有趣。