使用 C 库¶

注意

此页面使用两种不同的语法变体

Cython 特定的 cdef 语法，旨在使类型声明简洁且易于从 C/C++ 的角度阅读。
纯 Python 语法，允许在纯 Python 代码中进行静态 Cython 类型声明，遵循 PEP-484 类型提示和 PEP 526 变量注释。

要使用 Python 语法中的 C 数据类型，您需要在要编译的 Python 模块中导入特殊的 cython 模块，例如
```
import cython
```
如果您使用纯 Python 语法，我们强烈建议您使用最新的 Cython 3 版本，因为与 0.29.x 版本相比，这里已经进行了重大改进。

除了编写快速代码之外，Cython 的主要用例之一是从 Python 代码调用外部 C 库。由于 Cython 代码本身编译为 C 代码，因此直接在代码中调用 C 函数实际上是微不足道的。以下给出了在 Cython 代码中使用（和包装）外部 C 库的完整示例，包括适当的错误处理以及有关为 Python 和 Cython 代码设计合适 API 的注意事项。

假设您需要一种有效的方式将整数值存储在 FIFO 队列中。由于内存确实很重要，并且这些值实际上来自 C 代码，因此您无法负担得起在列表或双端队列中创建和存储 Python int 对象。因此，您需要在 C 中寻找队列实现。

经过一些网络搜索，您找到了 C-algorithms 库 [CAlg] 并决定使用其双端队列实现。但是，为了便于处理，您决定将其包装在 Python 扩展类型中，该类型可以封装所有内存管理。

[CAlg]

Simon Howard，C 算法库，https://fragglet.github.io/c-algorithms/

定义外部声明¶

您可以从这里下载 CAlg。

队列实现的 C API 定义在头文件 c-algorithms/src/queue.h 中，本质上看起来像这样

queue.h¶

/* queue.h */

typedef struct _Queue Queue;
typedef void *QueueValue;

Queue *queue_new(void);
void queue_free(Queue *queue);

int queue_push_head(Queue *queue, QueueValue data);
QueueValue queue_pop_head(Queue *queue);
QueueValue queue_peek_head(Queue *queue);

int queue_push_tail(Queue *queue, QueueValue data);
QueueValue queue_pop_tail(Queue *queue);
QueueValue queue_peek_tail(Queue *queue);

int queue_is_empty(Queue *queue);

要开始，第一步是在 .pxd 文件中重新定义 C API，例如，cqueue.pxd

cqueue.pxd¶

cdef extern from "c-algorithms/src/queue.h":
    ctypedef struct Queue:
        pass
    ctypedef void* QueueValue

    Queue* queue_new()
    void queue_free(Queue* queue)

    int queue_push_head(Queue* queue, QueueValue data)
    QueueValue  queue_pop_head(Queue* queue)
    QueueValue queue_peek_head(Queue* queue)

    int queue_push_tail(Queue* queue, QueueValue data)
    QueueValue queue_pop_tail(Queue* queue)
    QueueValue queue_peek_tail(Queue* queue)

    bint queue_is_empty(Queue* queue)

请注意，这些声明与头文件声明几乎相同，因此您通常可以将它们复制过来。但是，您不需要提供上面所有声明，只需要您在代码或其他声明中使用的那些声明，以便 Cython 能够看到它们的一个足够且一致的子集。然后，考虑对其进行一些调整，使其在 Cython 中更易于使用。

具体来说，您应该注意为 C 函数选择合适的参数名称，因为 Cython 允许您将它们作为关键字参数传递。稍后更改它们将是向后不兼容的 API 修改。立即选择合适的名称将使这些函数在 Cython 代码中更易于使用。

我们上面使用的头文件的一个值得注意的差异是第一行中 Queue 结构的声明。在这种情况下，Queue 用作不透明句柄；只有被调用的库才知道里面到底是什么。由于没有 Cython 代码需要知道结构的内容，因此我们不需要声明其内容，因此我们只需提供一个空定义（因为我们不想声明在 C 头文件中引用的 _Queue 类型）[1].

另一个例外是最后一行。函数 queue_is_empty() 的整数返回值实际上是一个 C 布尔值，即它唯一有趣的地方在于它是否为非零或零，表示队列是否为空。这最好用 Cython 的 bint 类型来表示，它在 C 中使用时是普通的 int 类型，但在转换为 Python 对象时映射到 Python 的布尔值 True 和 False。这种在 .pxd 文件中收紧声明的方式通常可以简化使用它们的代码。

良好的做法是为使用的每个库定义一个 .pxd 文件，有时甚至为每个头文件（或功能组）定义一个文件，如果 API 很大。这简化了它们在其他项目中的重用。有时，您可能需要使用标准 C 库中的 C 函数，或者想要直接调用 CPython 中的 C-API 函数。对于像这样的常见需求，Cython 附带了一组标准的 .pxd 文件，这些文件以一种易于使用的方式提供这些声明，并适应它们在 Cython 中的使用。主要包是 cpython、libc 和 libcpp。NumPy 库也有一个标准的 .pxd 文件 numpy，因为它经常在 Cython 代码中使用。有关提供的 .pxd 文件的完整列表，请参阅 Cython 的 Cython/Includes/ 源代码包。

编写包装类¶

声明完 C 库的 API 后，我们可以开始设计应该包装 C 队列的 Queue 类。它将存放在名为 queue.pyx/queue.py 的文件中。 [2]

以下是 Queue 类的第一个开始

queue.py¶

from cython.cimports import cqueue

@cython.cclass
class Queue:
    _c_queue: cython.pointer(cqueue.Queue)

    def __cinit__(self):
        self._c_queue = cqueue.queue_new()

queue.pyx¶

cimport cqueue


cdef class Queue:
    cdef cqueue.Queue* _c_queue

    def __cinit__(self):
        self._c_queue = cqueue.queue_new()

请注意，它写的是 __cinit__ 而不是 __init__。虽然 __init__ 也可用，但不能保证它会被执行（例如，可以创建一个子类并忘记调用祖先的构造函数）。由于未初始化 C 指针通常会导致 Python 解释器崩溃，因此 Cython 提供了 __cinit__，它在构造时始终被立即调用，甚至在 CPython 考虑调用 __init__ 之前，因此它是初始化新实例的静态属性（cdef 字段）的正确位置。但是，由于 __cinit__ 在对象构造期间被调用，因此 self 尚未完全构造，因此必须避免对 self 做任何操作，而只能为静态属性（cdef 字段）赋值。

还要注意，上述方法不接受任何参数，尽管子类型可能希望接受一些参数。无参数的 __cinit__() 方法在这里是一个特殊情况，它只是不接收传递给构造函数的任何参数，因此它不会阻止子类添加参数。如果在 __cinit__() 的签名中使用参数，则它们必须与类层次结构中用于实例化类型的任何声明的 __init__ 方法的参数匹配。

内存管理¶

在我们继续实现其他方法之前，重要的是要了解上述实现是不安全的。如果在调用 queue_new() 时出现任何错误，这段代码只会吞下错误，因此我们很可能会在稍后遇到崩溃。根据 queue_new() 函数的文档，上述代码失败的唯一原因是内存不足。在这种情况下，它将返回 NULL，而通常它会返回指向新队列的指针。

Python 中解决此问题的办法是抛出 MemoryError [3]。因此，我们可以更改 init 函数，如下所示

queue.py¶

from cython.cimports import cqueue

@cython.cclass
class Queue:
    _c_queue = cython.declare(cython.pointer(cqueue.Queue))

    def __cinit__(self):
        self._c_queue = cqueue.queue_new()
        if self._c_queue is cython.NULL:
            raise MemoryError()

queue.pyx¶

cimport cqueue


cdef class Queue:
    cdef cqueue.Queue* _c_queue

    def __cinit__(self):
        self._c_queue = cqueue.queue_new()
        if self._c_queue is NULL:
            raise MemoryError()

接下来要做的是在 Queue 实例不再使用时进行清理（即对它的所有引用都被删除）。为此，CPython 提供了一个回调，Cython 将其作为特殊方法 __dealloc__() 提供。在我们的例子中，我们只需要释放 C Queue，但前提是我们成功地在 init 方法中初始化了它

def __dealloc__(self):
    if self._c_queue is not cython.NULL:
        cqueue.queue_free(self._c_queue)

def __dealloc__(self):
    if self._c_queue is not NULL:
        cqueue.queue_free(self._c_queue)

编译和链接¶

此时，我们拥有一个可以测试的工作 Cython 模块。要编译它，我们需要为 setuptools 配置一个 setup.py 脚本。以下是编译 Cython 模块的最基本脚本

from setuptools import Extension, setup
from Cython.Build import cythonize

setup(
    ext_modules = cythonize([Extension("queue", ["queue.py"])])
)

from setuptools import Extension, setup
from Cython.Build import cythonize

setup(
    ext_modules = cythonize([Extension("queue", ["queue.pyx"])])
)

为了针对外部 C 库进行构建，我们需要确保 Cython 能够找到必要的库。实现这一点有两种方法。首先，我们可以告诉 setuptools 在哪里找到用于编译 queue.c 实现的 C 源代码。或者，我们可以将 C-Alg 构建并安装为系统库，并动态链接它。如果其他应用程序也使用 C-Alg，则后者很有用。

静态链接¶

为了自动构建 C 代码，我们需要在 queue.pyx/queue.py 中包含编译指令。

# distutils: sources = c-algorithms/src/queue.c
# distutils: include_dirs = c-algorithms/src/

import cython
from cython.cimports import cqueue

@cython.cclass
class Queue:
    _c_queue = cython.declare(cython.pointer(cqueue.Queue))

    def __cinit__(self):
        self._c_queue = cqueue.queue_new()
        if self._c_queue is cython.NULL:
            raise MemoryError()

    def __dealloc__(self):
        if self._c_queue is not cython.NULL:
            cqueue.queue_free(self._c_queue)

# distutils: sources = c-algorithms/src/queue.c
# distutils: include_dirs = c-algorithms/src/


cimport cqueue


cdef class Queue:
    cdef cqueue.Queue* _c_queue

    def __cinit__(self):
        self._c_queue = cqueue.queue_new()
        if self._c_queue is NULL:
            raise MemoryError()

    def __dealloc__(self):
        if self._c_queue is not NULL:
            cqueue.queue_free(self._c_queue)

sources 编译指令提供了 setuptools 将要编译并链接（静态）到生成的扩展模块中的 C 文件的路径。通常，所有相关的头文件都应该在 include_dirs 中找到。现在我们可以使用以下命令构建项目

$ python setup.py build_ext -i

并测试我们的构建是否成功

$ python -c 'import queue; Q = queue.Queue()'

动态链接¶

如果我们要包装的库已经安装在系统上，则动态链接很有用。要执行动态链接，我们首先需要构建并安装 c-alg。

要在您的系统上构建 c-算法

$ cd c-algorithms
$ sh autogen.sh
$ ./configure
$ make

要安装 CAlg，请运行

$ make install

之后，文件 /usr/local/lib/libcalg.so 应该存在。

注意

此路径适用于 Linux 系统，在其他平台上可能有所不同，因此您需要根据 libcalg.so 或 libcalg.dll 在您系统上的路径调整本教程的其余部分。

在这种方法中，我们需要告诉安装脚本链接外部库。为此，我们需要扩展安装脚本以更改扩展安装，从

ext_modules = cythonize([Extension("queue", ["queue.py"])])

ext_modules = cythonize([Extension("queue", ["queue.pyx"])])

到

ext_modules = cythonize([
    Extension("queue", ["queue.py"],
              libraries=["calg"])
    ])

ext_modules = cythonize([
    Extension("queue", ["queue.pyx"],
              libraries=["calg"])
    ])

现在我们应该能够使用以下命令构建项目

$ python setup.py build_ext -i

如果 libcalg 未安装在“正常”位置，用户可以通过传递适当的 C 编译器标志（例如）来从外部提供所需的参数

CFLAGS="-I/usr/local/otherdir/calg/include"  \
LDFLAGS="-L/usr/local/otherdir/calg/lib"     \
    python setup.py build_ext -i

在运行模块之前，我们还需要确保 libcalg 位于 LD_LIBRARY_PATH 环境变量中，例如通过设置

$ export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/local/lib

首次编译模块后，我们现在可以导入它并实例化一个新的队列

$ export PYTHONPATH=.
$ python -c 'import queue; Q = queue.Queue()'

但是，到目前为止，这就是我们的 Queue 类所能做的全部，所以让我们让它更易于使用。

映射功能¶

在实现此类的公共接口之前，最好查看 Python 提供的哪些接口，例如在它的 list 或 collections.deque 类中。由于我们只需要一个 FIFO 队列，因此提供 append()、peek() 和 pop() 方法，以及一个 extend() 方法来一次添加多个值就足够了。此外，由于我们已经知道所有值都将来自 C，因此最好现在只提供 cdef/@cfunc 方法，并为它们提供一个直接的 C 接口。

在 C 中，数据结构通常将数据存储为指向任何数据项类型的 void*。由于我们只想存储 int 值，这些值通常适合指针类型的尺寸，我们可以通过一个技巧避免额外的内存分配：我们将我们的 int 值强制转换为 void* 反之亦然，并将值直接存储为指针值。

以下是 append() 方法的简单实现

@cython.cfunc
def append(self, value: cython.int):
    cqueue.queue_push_tail(self._c_queue, cython.cast(cython.p_void, value))

cdef append(self, int value):
    cqueue.queue_push_tail(self._c_queue, <void*>value)

同样，与 __cinit__() 方法相同的错误处理注意事项适用，因此我们最终得到以下实现

@cython.cfunc
def append(self, value: cython.int):
    if not cqueue.queue_push_tail(self._c_queue,
                                  cython.cast(cython.p_void, value)):
        raise MemoryError()

cdef append(self, int value):
    if not cqueue.queue_push_tail(self._c_queue,
                                  <void*>value):
        raise MemoryError()

现在添加 extend() 方法应该很简单

@cython.cfunc
def extend(self, values: cython.p_int, count: cython.size_t):
    """Append all ints to the queue.
    """
    value: cython.int
    for value in values[:count]:  # Slicing pointer to limit the iteration boundaries.
        self.append(value)

cdef extend(self, int* values, size_t count):
    """Append all ints to the queue.
    """
    cdef int value
    for value in values[:count]:  # Slicing pointer to limit the iteration boundaries.
        self.append(value)

例如，在从 C 数组读取值时，这将非常方便。

到目前为止，我们只能向队列添加数据。下一步是编写两个方法来获取第一个元素：peek() 和 pop()，它们分别提供只读和破坏性读取访问。为了避免在将 void* 直接转换为 int 时出现编译器警告，我们使用一个足够大的中间数据类型来容纳 void*。这里，Py_ssize_t

@cython.cfunc
def peek(self) -> cython.int:
    return cython.cast(cython.Py_ssize_t, cqueue.queue_peek_head(self._c_queue))

@cython.cfunc
def pop(self) -> cython.int:
    return cython.cast(cython.Py_ssize_t, cqueue.queue_pop_head(self._c_queue))

cdef int peek(self):
    return <Py_ssize_t>cqueue.queue_peek_head(self._c_queue)

cdef int pop(self):
    return <Py_ssize_t>cqueue.queue_pop_head(self._c_queue)

通常，在 C 中，当我们在没有检查边界的情况下将较大的整数类型转换为较小的整数类型时，我们有可能会丢失数据，并且 Py_ssize_t 可能比 int 类型更大。但是，由于我们控制着值如何添加到队列中，我们已经知道队列中的所有值都适合 int，因此上述从 void* 到 Py_ssize_t 到 int（返回类型）的转换在设计上是安全的。

处理错误¶

现在，当队列为空时会发生什么？根据文档，这些函数返回一个 NULL 指针，这通常不是一个有效的值。但是，由于我们只是在 int 之间进行转换，我们无法再区分返回值是 NULL 因为队列为空，还是因为存储在队列中的值为 0。在 Cython 代码中，我们希望第一种情况引发异常，而第二种情况应该简单地返回 0。为了解决这个问题，我们需要对这个值进行特殊处理，并检查队列是否真的为空。

@cython.cfunc
def peek(self) -> cython.int:
    value: cython.int = cython.cast(cython.Py_ssize_t, cqueue.queue_peek_head(self._c_queue))
    if value == 0:
        # this may mean that the queue is empty, or
        # that it happens to contain a 0 value
        if cqueue.queue_is_empty(self._c_queue):
            raise IndexError("Queue is empty")
    return value

cdef int peek(self):
    cdef int value = <Py_ssize_t>cqueue.queue_peek_head(self._c_queue)
    if value == 0:
        # this may mean that the queue is empty, or
        # that it happens to contain a 0 value
        if cqueue.queue_is_empty(self._c_queue):
            raise IndexError("Queue is empty")
    return value

请注意，我们如何在方法中有效地创建了一条快速路径，在希望常见的返回值不为 0 的情况下。只有这种特殊情况需要额外检查队列是否为空。

如果 peek 函数是一个返回 Python 对象值的 Python 函数，CPython 将在内部简单地返回 NULL 而不是 Python 对象来指示异常，这将立即由周围的代码传播。问题是返回类型是 int，任何 int 值都是有效的队列项值，因此没有办法明确地向调用代码发出错误信号。

调用代码处理这种情况的唯一方法是在从函数返回时调用 PyErr_Occurred() 来检查是否引发了异常，如果是，则传播异常。这显然会影响性能。因此，Cython 使用一个专用值，它在发生异常时隐式返回，以便周围的代码只需要在收到此确切值时检查异常。

默认情况下，值 -1 用作异常返回值。所有其他返回值将几乎无损地传递，从而再次为“正常”值创建一条快速路径。有关更多详细信息，请参阅错误返回值。

现在 peek() 方法已实现，pop() 方法也需要调整。但是，由于它从队列中删除了一个值，因此仅在删除后测试队列是否为空是不够的。相反，我们必须在进入时进行测试

@cython.cfunc
def pop(self) -> cython.int:
    if cqueue.queue_is_empty(self._c_queue):
        raise IndexError("Queue is empty")
    return cython.cast(cython.Py_ssize_t, cqueue.queue_pop_head(self._c_queue))

cdef int pop(self):
    if cqueue.queue_is_empty(self._c_queue):
        raise IndexError("Queue is empty")
    return <Py_ssize_t>cqueue.queue_pop_head(self._c_queue)

异常传播的返回值与 peek() 的返回值完全相同。

最后，我们可以通过实现 __bool__() 特殊方法（请注意，Python 2 将此方法称为 __nonzero__，而 Cython 代码可以使用任一名称）以正常 Python 方式为 Queue 提供一个空指示器。

def __bool__(self):
    return not cqueue.queue_is_empty(self._c_queue)

请注意，此方法返回 True 或 False，因为我们在 cqueue.pxd 中将 queue_is_empty() 函数的返回类型声明为 bint。

测试结果¶

现在实现已经完成，您可能想要编写一些测试来确保它正常工作。特别是 doctests 非常适合此目的，因为它们同时提供了一些文档。但是，要启用 doctests，您需要一个可以调用的 Python API。C 方法在 Python 代码中不可见，因此无法从 doctests 调用。

为该类提供 Python API 的一种快速方法是将方法从 cdef/@cfunc 更改为 cpdef/@ccall。这将使 Cython 生成两个入口点，一个可以通过使用 Python 调用语义和 Python 对象作为参数的普通 Python 代码调用，另一个可以通过使用快速 C 语义的 C 代码调用，而无需进行从 Python 类型到 Python 类型的中间参数转换。请注意，cpdef/@ccall 方法确保它们即使从 Cython 调用也能被 Python 方法适当地覆盖。与 cdef/@cfunc 方法相比，这会增加一点开销。

现在我们已经为我们的类提供了 C 接口和 Python 接口，我们应该确保这两个接口是一致的。Python 用户会期望一个接受任意可迭代对象的 extend() 方法，而 C 用户则希望有一个允许传递 C 数组和 C 内存的方法。这两个签名不兼容。

我们将通过考虑在 C 中，API 也可能想要支持其他输入类型（例如 long 或 char 的数组）来解决此问题，这通常通过不同命名的 C API 函数（例如 extend_ints()、extend_longs()、extend_chars() 等）来支持。这使我们能够释放方法名称 extend() 用于鸭子类型的 Python 方法，该方法可以接受任意可迭代对象。

以下列表显示了使用 cpdef/@ccall 方法的完整实现，这些方法在可能的情况下使用

queue.py¶

from cython.cimports import cqueue
from cython import cast

@cython.cclass
class Queue:
    """A queue class for C integer values.

    >>> q = Queue()
    >>> q.append(5)
    >>> q.peek()
    5
    >>> q.pop()
    5
    """
    _c_queue = cython.declare(cython.pointer(cqueue.Queue))
    def __cinit__(self):
        self._c_queue = cqueue.queue_new()
        if self._c_queue is cython.NULL:
            raise MemoryError()

    def __dealloc__(self):
        if self._c_queue is not cython.NULL:
            cqueue.queue_free(self._c_queue)

    @cython.ccall
    def append(self, value: cython.int):
        if not cqueue.queue_push_tail(self._c_queue,
                cast(cython.p_void, cast(cython.Py_ssize_t, value))):
            raise MemoryError()

    # The `cpdef` feature is obviously not available for the original "extend()"
    # method, as the method signature is incompatible with Python argument
    # types (Python does not have pointers).  However, we can rename
    # the C-ish "extend()" method to e.g. "extend_ints()", and write
    # a new "extend()" method that provides a suitable Python interface by
    # accepting an arbitrary Python iterable.
    @cython.ccall
    def extend(self, values):
        for value in values:
            self.append(value)

    @cython.cfunc
    def extend_ints(self, values: cython.p_int, count: cython.size_t):
        value: cython.int
        for value in values[:count]:  # Slicing pointer to limit the iteration boundaries.
            self.append(value)

    @cython.ccall
    @cython.exceptval(-1, check=True)
    def peek(self) -> cython.int:
        value: cython.int = cast(cython.Py_ssize_t, cqueue.queue_peek_head(self._c_queue))

        if value == 0:
            # this may mean that the queue is empty,
            # or that it happens to contain a 0 value
            if cqueue.queue_is_empty(self._c_queue):
                raise IndexError("Queue is empty")
        return value

    @cython.ccall
    @cython.exceptval(-1, check=True)
    def pop(self) -> cython.int:
        if cqueue.queue_is_empty(self._c_queue):
            raise IndexError("Queue is empty")
        return cast(cython.Py_ssize_t, cqueue.queue_pop_head(self._c_queue))

    def __bool__(self):
        return not cqueue.queue_is_empty(self._c_queue)

queue.pyx¶

cimport cqueue



cdef class Queue:
    """A queue class for C integer values.

    >>> q = Queue()
    >>> q.append(5)
    >>> q.peek()
    5
    >>> q.pop()
    5
    """
    cdef cqueue.Queue* _c_queue
    def __cinit__(self):
        self._c_queue = cqueue.queue_new()
        if self._c_queue is NULL:
            raise MemoryError()

    def __dealloc__(self):
        if self._c_queue is not NULL:
            cqueue.queue_free(self._c_queue)


    cpdef append(self, int value):
        if not cqueue.queue_push_tail(self._c_queue,
                                      <void*> <Py_ssize_t> value):
            raise MemoryError()

    # The `cpdef` feature is obviously not available for the original "extend()"
    # method, as the method signature is incompatible with Python argument
    # types (Python does not have pointers).  However, we can rename
    # the C-ish "extend()" method to e.g. "extend_ints()", and write
    # a new "extend()" method that provides a suitable Python interface by
    # accepting an arbitrary Python iterable.

    cpdef extend(self, values):
        for value in values:
            self.append(value)


    cdef extend_ints(self, int* values, size_t count):
        cdef int value
        for value in values[:count]:  # Slicing pointer to limit the iteration boundaries.
            self.append(value)



    cpdef int peek(self) except? -1:
        cdef int value = <Py_ssize_t> cqueue.queue_peek_head(self._c_queue)

        if value == 0:
            # this may mean that the queue is empty,
            # or that it happens to contain a 0 value
            if cqueue.queue_is_empty(self._c_queue):
                raise IndexError("Queue is empty")
        return value



    cpdef int pop(self) except? -1:
        if cqueue.queue_is_empty(self._c_queue):
            raise IndexError("Queue is empty")
        return <Py_ssize_t> cqueue.queue_pop_head(self._c_queue)

    def __bool__(self):
        return not cqueue.queue_is_empty(self._c_queue)

现在我们可以使用 python 脚本测试我们的 Queue 实现，例如这里 test_queue.py

import time

import queue

Q = queue.Queue()

Q.append(10)
Q.append(20)
print(Q.peek())
print(Q.pop())
print(Q.pop())
try:
    print(Q.pop())
except IndexError as e:
    print("Error message:", e)  # Prints "Queue is empty"

i = 10000

values = range(i)

start_time = time.time()

Q.extend(values)

end_time = time.time() - start_time

print("Adding {} items took {:1.3f} msecs.".format(i, 1000 * end_time))

for i in range(41):
    Q.pop()

Q.pop()
print("The answer is:")
print(Q.pop())

作为在作者机器上使用 10000 个数字的快速测试表明，从 Cython 代码使用 C int 值使用此 Queue 比从 Cython 代码使用 Python 对象值快约五倍，比从 Python 代码在 Python 循环中使用它快近八倍，并且仍然比从 Cython 代码使用 Python 整数使用 Python 的高度优化的 collections.deque 类型快两倍多。

回调¶

假设您想提供一种方法让用户从队列中弹出值，直到发生某个用户定义的事件。为此，您希望允许他们传递一个谓词函数来确定何时停止，例如

def pop_until(self, predicate):
    while not predicate(self.peek()):
        self.pop()

现在，为了论证起见，假设 C 队列提供了这样一个函数，该函数将 C 回调函数作为谓词。API 可能如下所示

/* C type of a predicate function that takes a queue value and returns
 * -1 for errors
 *  0 for reject
 *  1 for accept
 */
typedef int (*predicate_func)(void* user_context, QueueValue data);

/* Pop values as long as the predicate evaluates to true for them,
 * returns -1 if the predicate failed with an error and 0 otherwise.
 */
int queue_pop_head_until(Queue *queue, predicate_func predicate,
                         void* user_context);

C 回调函数通常有一个通用的 void* 参数，允许通过 C-API 将任何类型的上下文或状态传递到回调函数中。我们将使用它来传递我们的 Python 谓词函数。

首先，我们必须定义一个具有预期签名的回调函数，我们可以将其传递到 C-API 函数中。

@cython.cfunc
@cython.exceptval(check=False)
def evaluate_predicate(context: cython.p_void, value: cqueue.QueueValue) -> cython.int:
    "Callback function that can be passed as predicate_func"
    try:
        # recover Python function object from void* argument
        func = cython.cast(object, context)
        # call function, convert result into 0/1 for True/False
        return bool(func(cython.cast(int, value)))
    except:
        # catch any Python errors and return error indicator
        return -1

注意

@cfunc 纯 Python 中的函数默认定义为 @exceptval(-1, check=True)。由于 evaluate_predicate() 应该作为参数传递给函数，因此我们需要完全关闭异常检查。

cdef int evaluate_predicate(void* context, cqueue.QueueValue value):
    "Callback function that can be passed as predicate_func"
    try:
        # recover Python function object from void* argument
        func = <object>context
        # call function, convert result into 0/1 for True/False
        return bool(func(<int>value))
    except:
        # catch any Python errors and return error indicator
        return -1

主要思想是将指向函数对象的指针（也称为借用引用）作为用户上下文参数传递。我们将按如下方式调用 C-API 函数：

def pop_until(self, python_predicate_function):
    result = cqueue.queue_pop_head_until(
        self._c_queue, evaluate_predicate,
        cython.cast(cython.p_void, python_predicate_function))
    if result == -1:
        raise RuntimeError("an error occurred")

def pop_until(self, python_predicate_function):
    result = cqueue.queue_pop_head_until(
        self._c_queue, evaluate_predicate,
        <void*>python_predicate_function)
    if result == -1:
        raise RuntimeError("an error occurred")

通常的模式是首先将 Python 对象引用强制转换为 void* 以将其传递到 C-API 函数中，然后在 C 谓词回调函数中将其强制转换回 Python 对象。强制转换为 void 会创建一个借用引用。在强制转换为 <object> 时，Cython 会增加对象的引用计数，从而将借用引用转换回拥有引用。在谓词函数结束时，拥有引用将再次超出范围，Cython 会将其丢弃。

上面代码中的错误处理有点简单。具体来说，谓词函数引发的任何异常基本上都会被丢弃，并且只会导致在事后引发一个简单的 RuntimeError()。可以通过将异常存储在通过上下文参数传递的对象中，并在 C-API 函数返回 -1 以指示错误后重新引发异常来改进这一点。

使用 C 库¶

定义外部声明¶

编写包装类¶

内存管理¶

编译和链接¶

静态链接¶

动态链接¶

映射功能¶

处理错误¶

测试结果¶

回调¶

目录

上一主题

下一主题

此页面