Search code examples
pythonctypes

Subclass ctypes.c_void_p to represent a custom handler with a custom C function for constructor and destructor / free


I have a somewhat exotic question.

Imagine there is some kind of C interface with ops like (in practice I'm thinking of webrtc vad API):

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

I know how to represent these functions using ctypes and ctypes.c_void_p to represent the pointer to a handler.

Now the exotic question. Can one represent this pattern as a class derived from ctypes.c_void_p?

  • Is it in general correct to derive from ctypes.c_void_p?
  • Are there any constraints for a constructor of such a derived class? Will the constructor be used by ctypes for constructing marshalled returned handles?
  • Is it correct to modify the self.value in derived class __init__?
  • How does ctypes create object of HandlerPointer, will it call the __init__ and will it prevent it from correctly functioning?
  • Does ctypes support/marshal correctly c_void_p-derived argtypes/restypes?
  • Is there a more natural way to represent such capsule-like objects (pointer to a possibly opaque structure along with a destructor function) / APIs?

E.g. to do something like:

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: it appears that having __init__ method works, except for __del__ method (which calls a C-land custom free method) whose existence will lead to a crash with double free or corruption (!prev). Found the problem, fix described in my comment below, will post a complete solution soon.


Solution

  • It appears that ctypes bypasses custom-defined __init__ and __new__ when creating c_void_p-derived output objects, so these magics can be safely overridden. And ctypes-based bindings work fine when using ctypes.c_void_p-derived type as part of argtypes/restype. I'm pasting here an example of my bindings for VAD from WebRTC: https://webrtc.googlesource.com/src/+/refs/heads/main which wraps 5 functions (including a constructor and destructor) working with a custom opaque handler:

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

    This wrapping solution uses:

    • ctypes.c_void_p-derived class to represent a handler
    • __new__-magic to bind/represent the factory constructor from the C-land and keep a single Python object ensuring a single-time destruction. This solution would likely lead to crash/double-free if deepcopy is performed on this object (via __del__ being called twice on the same underlying memory-pointer). But I have not tested it.
    • __init__-magic to represent handler initialization
    • __del__-magic to call C-land destructor on Python-land object becoming unreferenced

    This worked kind of fine if we don't try to deepcopy this handler and trick it into double-free.

    # 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
    

    Overall, I'm not sure if representing this semantics via inheritance from c_void_p is the best way, but it was interesting to play with this and various related Python magic methods.