2

I wrote a wrapper for the Corsair Utility Engine SDK, but there is one function that I have not been able to wrap. It's an async function that accepts a callback function, but I cannot seem to figure out how to give it that callback.

The function looks like this:

bool CorsairSetLedsColorsAsync(int size, CorsairLedColor* ledsColors, void (*CallbackType)(void* context, bool result, CorsairError error), void *context)

These are the implementations that I have tried so far:

def SetLedsColorsAsync(self, size, led_color, callback, context):
    c_func = CFUNCTYPE(c_void_p, c_void_p, c_bool, c_int)
    c_callback = c_func(callback)
    self._libcue.CorsairSetLedsColorsAsync.restype = c_bool
    self._libcue.CorsairSetLedsColorsAsync.argtypes = [c_int, POINTER(CorsairLedColor), c_void_p, c_void_p]
    return self._libcue.CorsairSetLedsColorsAsync(size, led_color, c_callback, context)

as well as

def SetLedsColorsAsync(self, size, led_color, callback, context):
    c_func = CFUNCTYPE(None, c_void_p, c_bool, c_int)
    c_callback = c_func(callback)
    self._libcue.CorsairSetLedsColorsAsync.restype = c_bool
    self._libcue.CorsairSetLedsColorsAsync.argtypes = [c_int, POINTER(CorsairLedColor), c_func, c_void_p]
    return self._libcue.CorsairSetLedsColorsAsync(size, led_color, c_callback, context)

The code I'm testing with is

from cue_sdk import *
import time


def test(context, result, error):
    print context, result, error
    return 0

Corsair = CUE("CUESDK.x64_2013.dll")
Corsair.RequestControl(CAM_ExclusiveLightingControl)
Corsair.SetLedsColorsAsync(1, CorsairLedColor(CLK_H, 255, 255, 255), test, 1)

while True:
time.sleep(1)

The time.sleep() is there just to keep the program alive.

When running it, it crashes with error code 3221225477 (STATUS_ACCESS_VIOLATION) on Windows.

If you need to see the actual wrapper, you can find it here.

14
  • @IgnacioVazquez-Abrams I'm not sure what you mean, can you elaborate? Commented Apr 28, 2016 at 21:00
  • How do you define CallbackType? Commented Apr 28, 2016 at 21:02
  • 1
    "Async" in the name implies that the operation is carried out asynchronously. In that case, you don't know when the callback will be called, and it must not be deallocated until then. I suggest you create a permanent callback as a class attribute. To handle the user's callable, you can store a dict in the class that maps it to a unique ID, e.g. id(callback). Pass this ID as the context value. Then when the permanent callback is called, it can use the ID as a key to pop the user's callable from the dict and call it. Commented Apr 29, 2016 at 5:57
  • 1
    BTW, _libcue and all of the function prototypes should be defined only once at either the module or class level. It's a waste of time to do this for every instance and method call. Commented Apr 29, 2016 at 5:59
  • 1
    To support a context parameter for the user's callback function, you can use a tuple such as user_context = (callback, context). Use the ID of this tuple as the key in the class callback mapping, and pass this ID as the callback context, e.g. context = id(user_context); self._callback_map[context] = user_context. Then the permanent callback can use the ID to pop the tuple with the user's callback and context parameter. Commented Apr 29, 2016 at 6:11

1 Answer 1

1

I completely forgot about the issue with garbage collection until eryksun reminded me. He suggested that I create a permanent callback handler that would store all of the callbacks and call + pop them when necessary. This is what I did.

The function prototype looks like this:

self._callback_type = CFUNCTYPE(None, c_void_p, c_bool, c_int)
self._callback = self._callback_type(self._callback_handler)
self._libcue.CorsairSetLedsColorsAsync.restype = c_bool
self._libcue.CorsairSetLedsColorsAsync.argtypes = [c_int, POINTER(CorsairLedColor), self._callback_type, c_void_p]

The _callback_handler function looks like this:

def _callback_handler(self, context, result, error):
    if context is not None and context in self._callbacks:
        self._callbacks.pop(context)(context, result, error)

The actual function looks like this.

def SetLedsColorsAsync(self, size, led_color, callback=None, context=None):
    if callback:
        if context is None:
            context = id(callback)
        self._callbacks[context] = callback
    return self._libcue.CorsairSetLedsColorsAsync(size, led_color, self._callback, context)

_callback_type is the actual CFUNCTYPE that wraps the permanent callback (_callback_handler) which is one of the prototype's argtype. When SetLedsColorsAsync is called, the callback parameter is put into a dictionary (with the context or the ID of the function being the key). Instead of supplying the callback into the function, the permanent callback is passed on instead. Once the permanent callback is called, it will call the proper function and remove it from the dictionary.

The test I used:

#!python3
import time

from cue_sdk import *


def test(context, result, error):
    print(context, result, error)
    assert context == id(test)


Corsair = CUE("CUESDK.x64_2013.dll")
Corsair.RequestControl(CAM_ExclusiveLightingControl)
Corsair.SetLedsColorsAsync(1, CorsairLedColor(CLK_H, 255, 255, 255), test)

while True:
    time.sleep(1)

Example output:

2969710418936 True 0

If I'm not making sense, the commit is here.

Sign up to request clarification or add additional context in comments.

5 Comments

The user context may not be unique. I suggest creating a (callback, context) tuple and setting its id as context. The permanent callback will then pop and unpack the tuple.
The optimization to pass the context as NULL (None) when the user doesn't provide a callback is good, but it needs error checking. Early exceptions will be clearer than asynchronous exceptions in the permanent callback. If callback is None, then raise ValueError if context isn't also None. Else raise TypeError if not callable(callback). Otherwise store the tuple in _callbacks by its id, which becomes the context for the permanent callback.
I also suggest storing _libcue, _callback, and _callbacks on the class itself. There's no reason to set these on every instance.
@eryksun Do you mean setting them as static elements of the class? All three of them are defined in __init__. I'll implement your first two suggestions, however.
Yes, set them as class attributes defined in the body of the class. You can still refer to them using self from an instance method. Also, all of the prototypes should be defined only once after creating _libcue. That way you're not redefining the prototype every time a method is called. I'd loop over a list defined in the module level that has each function name and prototype (error check function, result type, and argument types). Given the name, get the function pointer dynamically via func = getattr(_libcue, name). Then assign its errcheck, restype, and argtypes attributes.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.