Cython与NumPy内存视图相比在性能上较差

Question

我遇到了一个挺奇怪的结果，来自一个基准测试。

这些都是不同版本的冒泡排序实现，而在n=10^4的情况下，最快的方法是把Python列表内部转换成C数组。相比之下，黄色的线代表的是我使用NumPy数组和内存视图的代码。我本来以为结果应该是反过来的。我（和我的同事）重复了这个基准测试好几次，结果总是一样。也许有人知道这是怎么回事……

图中的黑线对应的代码是：

%%cython
cimport cython
from libc.stdlib cimport malloc, free

def cython_bubblesort_clist(a_list):
    """ 
    The Cython implementation of bubble sort with internal
    conversion between Python list objects and C arrays.

    """
    cdef int *c_list
    c_list = <int *>malloc(len(a_list)*cython.sizeof(int))
    cdef int count, i, j # static type declarations
    count = len(a_list)

    # convert Python list to C array
    for i in range(count):
        c_list[i] = a_list[i]

    for i in range(count):
        for j in range(1, count):
            if c_list[j] < c_list[j-1]:
                c_list[j-1], c_list[j] = c_list[j], c_list[j-1]

    # convert C array back to Python list
    for i in range(count):
        a_list[i] = c_list[i]

    free(c_list)
    return a_list

而粉色线对应的代码是：

%%cython
import numpy as np
cimport numpy as np
cimport cython
def cython_bubblesort_numpy(long[:] np_ary):
    """ 
    The Cython implementation of bubble sort with NumPy memoryview.

    """
    cdef int count, i, j # static type declarations
    count = np_ary.shape[0]

    for i in range(count):
        for j in range(1, count):
            if np_ary[j] < np_ary[j-1]:
                np_ary[j-1], np_ary[j] = np_ary[j], np_ary[j-1]

    return np.asarray(np_ary)

Answer 1

你可以对你的代码做一个很简单的改动，看看能否进一步提升性能：

cpdef cython_bubblesort_numpy(long[::1] np_ary):
    # ...

这段代码告诉cython，np_ary是一个C连续数组，利用这个信息，嵌套的for循环生成的代码可以得到更好的优化。

不过，这段代码不接受不连续的数组作为参数，但你可以很简单地通过使用numpy.ascontiguousarray()来处理这个问题。

Answer 2

正如上面评论中提到的，我添加了装饰器

%%cython
import numpy as np
cimport numpy as np
cimport cython
@cython.boundscheck(False) 
@cython.wraparound(False)
cpdef cython_bubblesort_numpy(long[:] np_ary):
    """ 
    The Cython implementation of bubble sort with NumPy memoryview.

    """
    cdef int count, i, j # static type declarations
    count = np_ary.shape[0]

    for i in range(count):
        for j in range(1, count):
            if np_ary[j] < np_ary[j-1]:
                np_ary[j-1], np_ary[j] = np_ary[j], np_ary[j-1]

    return np.asarray(np_ary)

现在结果更接近我预期的样子了 :)