使用Python FFI进行回调函数（Callback）：处理C函数调用Python代码的栈切换 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Python FFI 回调函数：C 调用 Python 代码的栈切换

大家好，今天我们来深入探讨一个在使用 Python FFI (Foreign Function Interface) 时经常遇到的难题：C 函数调用 Python 代码时的栈切换。这个问题看似简单，但背后涉及到许多底层机制，理解它对于编写安全、稳定的 FFI 代码至关重要。

什么是回调函数？

在开始之前，我们先简单回顾一下回调函数的概念。回调函数本质上是一种“控制反转”的编程模式。你将一个函数的指针（或者其他等价物）传递给另一个函数，后者在特定事件发生时调用你的函数。在 FFI 的上下文中，这通常意味着 C 代码调用 Python 代码。

为什么栈切换是个问题？

每个线程都有自己的调用栈，用于存储函数调用信息、局部变量等。当 C 代码调用 Python 代码时，我们需要从 C 的栈切换到 Python 的栈，然后在 Python 代码执行完毕后，再切换回 C 的栈。

问题在于，这种栈切换并非总是自动完成的。特别是当 C 代码使用不同的线程或协程模型时，手动管理栈切换就变得不可避免。如果栈切换处理不当，可能导致程序崩溃、数据损坏或其他难以调试的问题。

使用 ctypes 进行回调

让我们先看一个简单的例子，使用 ctypes 库定义一个 C 函数，该函数接受一个函数指针作为参数，并在内部调用该函数指针。

import ctypes
import sys

# 定义C函数的类型
CALLBACK_TYPE = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_int)

# 加载C库 (假设你的C库叫做 libexample.so 或 example.dll)
if sys.platform.startswith('win'):
  lib = ctypes.cdll.LoadLibrary("example.dll")
else:
  lib = ctypes.cdll.LoadLibrary("libexample.so")

# 定义C函数的参数类型和返回类型
lib.call_python_callback.argtypes = [CALLBACK_TYPE, ctypes.c_int]
lib.call_python_callback.restype = ctypes.c_int

# Python回调函数
def python_callback(arg):
  print(f"Python callback called with argument: {arg}")
  return arg * 2

# 将Python函数转换为C函数指针
c_callback = CALLBACK_TYPE(python_callback)

# 调用C函数，并将Python回调函数指针传递给它
result = lib.call_python_callback(c_callback, 10)
print(f"Result from C function: {result}")

对应的 C 代码（example.c）：

#include <stdio.h>
#include <stdlib.h>

typedef int (*callback_func)(int);

int call_python_callback(callback_func callback, int arg) {
  printf("Calling Python callback from Cn");
  int result = callback(arg);
  printf("Returned from Python callback to Cn");
  return result;
}

#ifdef _WIN32
#define EXPORT __declspec(dllexport)
#else
#define EXPORT
#endif

EXPORT int call_python_callback(callback_func callback, int arg);

编译 C 代码：

# Linux
gcc -shared -fPIC -o libexample.so example.c

# Windows (MinGW)
gcc -shared -o example.dll example.c -Wl,--export-all-symbols

在这个例子中，ctypes 自动处理了栈切换。但是，在更复杂的情况下，例如 C 代码使用线程，情况就变得复杂了。

使用 cffi 进行回调

cffi 提供了更灵活和强大的 FFI 功能。它允许你直接在 Python 中定义 C 接口，并使用 C 代码进行交互。

from cffi import FFI
import threading

ffi = FFI()

# 定义C接口
ffi.cdef("""
    typedef int (*callback_func)(int);
    int call_python_callback(callback_func callback, int arg);
""")

# 加载C库
if sys.platform.startswith('win'):
  lib = ffi.dlopen("example.dll")
else:
  lib = ffi.dlopen("libexample.so")

# Python回调函数
def python_callback(arg):
  print(f"Python callback called with argument: {arg} in thread {threading.current_thread().name}")
  return arg * 2

# 将Python函数转换为C函数指针
callback_func = ffi.callback("int(int)")(python_callback)

# 调用C函数
result = lib.call_python_callback(callback_func, 10)
print(f"Result from C function: {result}")

C 代码 (example.c) 保持不变。

在这个例子中，cffi 的 ffi.callback() 函数负责将 Python 函数转换为 C 函数指针，并处理必要的栈切换。

线程问题和 GIL

Python 的全局解释器锁 (GIL) 限制了同一时刻只有一个线程可以执行 Python 字节码。这意味着，如果 C 代码在一个单独的线程中调用 Python 回调函数，GIL 会成为一个瓶颈。

为了解决这个问题，可以考虑释放 GIL。cffi 提供了 ffi.release_gil() 和 ffi.acquire_gil() 函数，允许你在 C 代码中手动控制 GIL 的释放和获取。

案例：多线程回调

以下是一个多线程回调的例子，演示了如何使用 cffi 和 GIL 控制来处理线程安全的回调函数。

from cffi import FFI
import threading
import time

ffi = FFI()

# 定义C接口
ffi.cdef("""
    typedef int (*callback_func)(int);
    int call_python_callback(callback_func callback, int arg);
    void start_threaded_callback(callback_func callback, int arg);
""")

# 加载C库
if sys.platform.startswith('win'):
    lib = ffi.dlopen("example.dll")
else:
    lib = ffi.dlopen("libexample.so")

# Python回调函数
def python_callback(arg):
    print(f"Python callback called with argument: {arg} in thread {threading.current_thread().name}")
    time.sleep(0.1)  # 模拟一些耗时操作
    return arg * 2

# 将Python函数转换为C函数指针
callback_func = ffi.callback("int(int)")(python_callback)

# C函数调用Python回调函数
@ffi.def_extern()
def call_python_from_c(arg):
    with ffi.lock:  # 获取GIL
        print(f"Calling Python function from C in thread {threading.current_thread().name}")
        result = python_callback(arg)
        print(f"Result from Python function: {result}")
        return result

# 设置C函数
ffi.cdef("int call_python_from_c(int);")
lib.call_python_from_c = call_python_from_c

# 调用C函数
def run_threaded_callback(arg):
  lib.start_threaded_callback(callback_func, arg)

# 启动一个线程调用C函数
thread = threading.Thread(target=run_threaded_callback, args=(20,), name="C_Thread")
thread.start()

thread.join()
print("Finished")

对应的 C 代码（example.c）：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

typedef int (*callback_func)(int);

// 线程函数，用于在单独的线程中调用回调函数
void* thread_function(void* arg) {
  struct thread_data {
    callback_func callback;
    int arg;
  };

  struct thread_data *data = (struct thread_data*) arg;
  callback_func callback = data->callback;
  int arg_value = data->arg;

  printf("Calling Python callback from C threadn");
  int result = callback(arg_value);
  printf("Returned from Python callback to C threadn");
  printf("Result from Python callback in C thread: %dn", result);
  free(data); // 释放分配的内存
  pthread_exit(NULL);
}

// 启动一个新线程来调用回调函数
void start_threaded_callback(callback_func callback, int arg) {
  pthread_t thread;
  struct thread_data *data = (struct thread_data*) malloc(sizeof(struct thread_data));
  data->callback = callback;
  data->arg = arg;

  if (pthread_create(&thread, NULL, thread_function, (void*)data) != 0) {
    perror("Failed to create thread");
  }
  pthread_detach(thread); // 分离线程，使其在完成后自动释放资源
}

// 导出函数
#ifdef _WIN32
#define EXPORT __declspec(dllexport)
#else
#define EXPORT
#endif

EXPORT void start_threaded_callback(callback_func callback, int arg);
EXPORT int call_python_from_c(int arg);

编译 C 代码 (与之前相同)。

在这个例子中，start_threaded_callback 函数在 C 代码中创建一个新的线程，并在该线程中调用 Python 回调函数。我们使用了 pthread_detach 分离了线程，避免了资源泄露。

案例：异步回调

如果你的 C 代码使用异步事件循环（例如 libuv），你需要确保 Python 回调函数在正确的线程中执行。一种常见的做法是使用一个队列来传递回调函数和参数，然后在 Python 的主线程中处理这些回调。

import threading
import queue
import time
from cffi import FFI

ffi = FFI()

# 定义C接口
ffi.cdef("""
    typedef void (*callback_func)(int);
    void call_python_callback_async(callback_func callback, int arg);
    void run_event_loop(void);
""")

# 加载C库
if sys.platform.startswith('win'):
    lib = ffi.dlopen("example.dll")
else:
    lib = ffi.dlopen("libexample.so")

# 回调队列
callback_queue = queue.Queue()

# Python回调函数
def python_callback(arg):
    print(f"Python callback called with argument: {arg} in thread {threading.current_thread().name}")
    time.sleep(0.1)  # 模拟一些耗时操作

# 处理回调函数的函数
def process_callbacks():
    while True:
        try:
            callback, arg = callback_queue.get(timeout=0.1)  # 设置超时，避免无限阻塞
            callback(arg)
            callback_queue.task_done()
        except queue.Empty:
            # 如果队列为空，检查C的事件循环是否仍在运行
            if not c_event_loop_running:
                break  # C的事件循环已结束，退出处理回调的循环
            continue  # 队列为空，但C的事件循环仍在运行，继续等待

# 将Python函数转换为C函数指针
callback_func = ffi.callback("void(int)")(python_callback)

# C 代码运行事件循环的标志
c_event_loop_running = True

# C代码异步调用Python回调函数
@ffi.def_extern()
def handle_async_callback(arg):
    callback_queue.put((python_callback, arg))

ffi.cdef("void handle_async_callback(int);")
lib.handle_async_callback = handle_async_callback

# 将回调函数传递给C代码
lib.call_python_callback_async(callback_func, 10)

# 启动处理回调函数的线程
callback_thread = threading.Thread(target=process_callbacks, name="Callback_Thread")
callback_thread.start()

# 启动C的事件循环
lib.run_event_loop()
c_event_loop_running = False  # 设置标志，表明C的事件循环已结束

# 等待回调处理线程结束
callback_thread.join()

print("Finished")

对应的 C 代码（example.c）：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

typedef void (*callback_func)(int);

// 模拟异步事件循环
void run_event_loop() {
    for (int i = 0; i < 5; ++i) {
        printf("Running event loop iteration %d in C threadn", i);
        sleep(1);  // 模拟耗时操作
        // 触发异步回调，这里直接调用Python的handle_async_callback
        handle_async_callback(i * 100);
    }
}

void call_python_callback_async(callback_func callback, int arg) {
    printf("Calling Python callback asynchronously from Cn");
    // 在实际的异步事件循环中，这里会将回调函数和参数添加到事件队列中
    // 为了简化，直接调用Python的handle_async_callback
    //callback(arg); // 错误：不能直接调用，需要在Python线程中处理
}

// 导出函数
#ifdef _WIN32
#define EXPORT __declspec(dllexport)
#else
#define EXPORT
#endif

EXPORT void call_python_callback_async(callback_func callback, int arg);
EXPORT void run_event_loop(void);
EXPORT void handle_async_callback(int arg);

编译 C 代码 (与之前相同)。请注意，你需要包含 handle_async_callback 的声明和定义。

在这个例子中，C 代码模拟了一个简单的事件循环，并使用 call_python_callback_async 函数异步地调用 Python 回调函数。回调函数被添加到 callback_queue 队列中，然后在 Python 的 process_callbacks 线程中被处理。这样可以确保 Python 回调函数在正确的线程中执行，并避免 GIL 带来的问题。

最佳实践

以下是一些使用 Python FFI 进行回调函数的最佳实践：

了解你的 C 代码： 仔细研究 C 代码的线程模型和内存管理方式，确保你理解如何正确地调用回调函数。
使用 cffi： cffi 提供了更灵活和强大的 FFI 功能，可以更好地处理复杂的回调场景。
控制 GIL： 在多线程环境中，手动控制 GIL 的释放和获取，以提高性能。
使用队列： 对于异步回调，使用队列来传递回调函数和参数，以确保在正确的线程中执行。
错误处理： 确保你的回调函数能够处理异常，并将其传递回 C 代码。
资源管理： 仔细管理 C 和 Python 对象的生命周期，避免内存泄漏。

表格：ctypes vs cffi

特性	`ctypes`	`cffi`
易用性	简单易用，适合简单的 FFI 场景	学习曲线较陡峭，但功能更强大
性能	性能较差，因为需要进行类型转换	性能更好，因为可以避免类型转换
灵活性	灵活性较低，不支持复杂的 C 接口	灵活性较高，支持复杂的 C 接口和数据结构
类型检查	类型检查较弱，容易出错	类型检查更严格，可以减少错误
线程安全	需要手动处理线程安全问题	提供 GIL 控制函数，可以更好地处理线程安全问题
异步回调	需要手动实现异步回调机制	可以使用队列等机制实现异步回调

栈切换、GIL 和异步回调的总结

栈切换是 C 函数调用 Python 回调函数时需要考虑的关键问题，特别是涉及多线程或异步操作时。cffi 提供了更灵活的工具来管理栈切换和 GIL。对于异步回调，使用队列来传递回调函数和参数是一种常见的解决方案，可以确保回调函数在正确的线程中执行。理解这些概念并遵循最佳实践，可以帮助你编写安全、稳定的 FFI 代码。

更多IT精英技术系列讲座，到智猿学院

Python FFI 回调函数：C 调用 Python 代码的栈切换

发表回复 取消回复

发表回复取消回复