C vanuit Python aanroepen met ctypes: zonder FFI-overhead

We hebben een genetisch algoritme dat in C is geschreven. Het draait snel, heeft geen Python runtime nodig en produceert precies wat we willen. Maar de rest van onze stack is Python: een async worker die jobs uit een Redis queue haalt, voortgang rapporteert en resultaten terugschrijft.

Die twee werelden moeten elkaar spreken. De meest voor de hand liggende aanpak: ctypes. Geen derde library, geen compilatiestap vanuit Python, geen runtime-dependency buiten de gecompileerde .so. Dit is hoe dat in de praktijk werkt.

Library laden: slimmer dan ctypes.CDLL(pad)

Het simpelste dat werkt is:

lib = ctypes.CDLL("./libgeneretic.so")

Maar in productie wil je niet hardcoden waar de library staat. We bouwen vier varianten van de library naast elkaar (debug, asan, ubsan, release) en willen op elk moment kunnen schakelen zonder code te veranderen.

Onze loader probeert op volgorde:

1. GENERETIC_LIB_PATH — expliciet pad via environment variable (CI, Docker mounts)
2. GENERETIC_BUILD — kies een buildvariant (debug, asan, ubsan, release)
3. Auto-detectie op basis van de unified build structuur: build/{config}/lib/libgeneretic.so
4. Legacy locaties voor backward compatibility

def _find_library() -> Path:
    if explicit := os.getenv("GENERETIC_LIB_PATH"):
        return Path(explicit)

    build_variant = os.getenv("GENERETIC_BUILD", "release")
    candidates = [
        Path(f"build/{build_variant}/lib/libgeneretic.so"),
        Path(f"src/c_solver/build/{build_variant}/libgeneretic.so"),
        Path("libgeneretic.so"),
    ]
    for path in candidates:
        if path.exists():
            return path

    raise FileNotFoundError("Geen libgeneretic.so gevonden. Voer 'make release' uit.")

In productie-Docker is er altijd een release build aanwezig. In CI switchen we via GENERETIC_BUILD=asan naar de memory-safe variant. Lokaal werkt het gewoon met make debug en auto-detectie.

Structs definiëren die C begrijpt

Het lastige van ctypes is dat Python en C over geheugenindeling moeten overeenstemmen. Je definieert structs als Python-klassen die ctypes.Structure uitbreiden:

class SolverEmployee(ctypes.Structure):
    _fields_ = [
        ("employee_id", ctypes.c_int),
        ("postcode", ctypes.c_char * 7),
        ("skills", ctypes.c_uint64),
        ("availability", (ctypes.c_uint128 * 7)),  # 7 dagen, 96 blokken per dag
    ]

class SolverInput(ctypes.Structure):
    _fields_ = [
        ("employees", ctypes.POINTER(SolverEmployee)),
        ("employee_count", ctypes.c_int),
        ("shifts", ctypes.POINTER(SolverShift)),
        ("shift_count", ctypes.c_int),
        ("population_size", ctypes.c_int),
        ("max_generations", ctypes.c_int),
        ("mutation_rate", ctypes.c_double),
        ("progress_state", ctypes.POINTER(ProgressState)),
    ]

De C-kant declareert exact dezelfde structs. Als de velden niet overeenstemmen — verkeerde volgorde, verkeerd type, verkeerde padding — krijg je undefined behavior of een segfault. Wij gebruiken _pack_ = 1 waar nodig om alignment-problemen te voorkomen en testen elke build met ubsan.

Het daadwerkelijke aanroepen is dan recht-door-zee:

lib = ctypes.CDLL(str(_find_library()))
lib.run_solver.argtypes = [
    ctypes.POINTER(SolverInput),
    ctypes.POINTER(SolverResult),
]
lib.run_solver.restype = None

solver_input = SolverInput(...)
solver_result = SolverResult()
lib.run_solver(ctypes.byref(solver_input), ctypes.byref(solver_result))

De C solver schrijft het resultaat direct in de struct. Geen return values, geen heap-allocatie van buitenaf.

Async aanroepen zonder de event loop te blokkeren

Het algoritme loopt 2–6 seconden. Als je lib.run_solver(...) gewoon aanroept in een async Python functie, blokkeer je de hele event loop. Dat is niet acceptabel in een worker die meerdere jobs tegelijk verwerkt.

De oplossing: draai de C-aanroep in een thread pool via asyncio.get_event_loop().run_in_executor:

async def solve(self, request: PlanRequestDTO) -> PlanResponseDTO:
    loop = asyncio.get_event_loop()
    solver_input = self._build_input(request)
    solver_result = SolverResult()

    await loop.run_in_executor(
        None,  # default ThreadPoolExecutor
        lib.run_solver,
        ctypes.byref(solver_input),
        ctypes.byref(solver_result),
    )

    return self._parse_result(solver_result)

De C solver draait nu in een OS-thread. De Python event loop blijft vrij voor andere coroutines — voortgangspolling, keepalive pings, andere jobs.

Voortgang vanuit C naar Python: gedeelde state

Terwijl de C solver draait, wil je de gebruiker informeren over de voortgang. De solver weet hoever hij is (generatie X van Y). Python moet die informatie kunnen uitlezen zonder de solver te onderbreken.

We lossen dit op met een gedeelde ProgressState struct die zowel C als Python schrijven en lezen:

class ProgressState(ctypes.Structure):
    _fields_ = [
        ("current_generation", ctypes.c_int),
        ("total_generations", ctypes.c_int),
        ("population_size", ctypes.c_int),
        ("is_complete", ctypes.c_bool),
    ]
    _lock = threading.Lock()

    def read(self):
        with self._lock:
            return (
                self.current_generation,
                self.total_generations,
                self.population_size,
            )

De C solver werkt deze struct bij na elke generatie. Python poll elke 100ms:

async def _poll_progress(
    self,
    state: ProgressState,
    emit_fn: Callable[[int], Awaitable[None]],
) -> None:
    last_emit = 0.0
    solver_start = time.monotonic()

    while not state.is_complete:
        await asyncio.sleep(0.1)

        current, total, population = state.read()
        now = time.monotonic()

        # Maximaal één update per seconde naar Redis sturen
        if now - last_emit >= 1.0 and total > 0:
            pct = 10 + int((current / total) * 80)  # 10–90%
            elapsed = now - solver_start
            eta = (elapsed / current * (total - current)) if current > 0 else 0

            await emit_fn(pct, current, total, population, eta)
            last_emit = now

De throttle op één seconde voorkomt dat we Redis overspoelen met updates. De 10–90% range laat ruimte voor "start" (10%) en "verwerken resultaat" (90–100%) in de omliggende logica.

Thread safety: minimale locking

De C solver schrijft vanuit één thread. Python leest vanuit één coroutine. Strict genomen is locking niet noodzakelijk voor atomaire int-writes op x86_64, maar we voegen het toch toe: het maakt de code aantoonbaar correct ongeacht architectuur of compilatiesettings.

Wat we bewust niet doen: de lock vasthouden terwijl we naar Redis schrijven. Dat zou de C solver kunnen blokkeren. Lees snel, laat los, doe het zware werk daarna.

Foutafhandeling aan de C-grens

C heeft geen exceptions. Fouten zijn return codes of out-parameter flags. Wij gebruiken een error_code veld in SolverResult:

if solver_result.error_code != 0:
    message = lib.get_error_message(solver_result.error_code)
    raise SolverException(
        f"C solver fout {solver_result.error_code}: {message.decode()}"
    )

get_error_message is een simpele C-functie die een statische string teruggeeft. Geen heap-allocatie, dus geen memory management nodig aan de Python-kant.

Als de C solver crasht (segfault, stack overflow), vangt Python dat op als een OSError of Segmentation fault signaal. In de worker hebben we een top-level try/except die zulke fouten omzet naar een job-failure met fallback naar de greedy solver.

Wat je ermee wint

De keuze voor ctypes tegenover alternatieven als Cython of CFFI:

• Geen compilatiestap vanuit Python: je bouwt de C library los, Python laadt hem gewoon.
• Geen extra dependencies: ctypes zit in de stdlib.
• Directe controle over geheugenindeling: je ziet precies hoe data de C-grens oversteekt.
• Begrijpelijk voor iedereen die C en Python kent: geen aparte DSL of buildtool om te leren.

De keerzijde: als structs niet overeenstemmen, is de foutmelding cryptisch. Dat ondervang je door ubsan te draaien in CI en je structs goed te documenteren.

Voor onze use case — één gecompileerde library, een vast interface, geen dynamisch gegenereerde bindings — is ctypes de juiste keuze. Eenvoudig, snel, transparant.