python - 将Python序列输入Cython数组(并返回)

Question

我第一次成功使用Cython来显着加快将半字节从一个整数列表（字节）打包到另一个整数列表（参见Faster bit-level data packing），例如将两个连续字节0x0A和0x0B打包到0xAB中。

def pack(it):
    """Cythonize python nibble packing loop, typed"""
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    return [ (it[i*2]//16)<<4 | it[i*2+1]//16 for i in range(n) ]

尽管最终的速度令人满意，但我很好奇是否可以通过更好地利用输入和输出列表来进一步提高速度。

cython3 -a pack.cyx生成一个非常“ cythonic”的HTML报告，很遗憾，我没有足够的经验来得出任何有用的结论。

从C的角度来看，循环应“简单地”访问两个无符号的int数组。可能的是，使用更宽的数据类型（16/32位）可以进一步按比例加快此速度。

问题是：（如何）为Cython键入 Python [binary/immutable] sequence types作为 unsigned int array？



按照 How to convert python array to cython array?中的建议使用数组似乎并不能使其更快（并且该数组需要事先从bytes对象创建），也不能将参数键入为 list而不是 object（与无类型相同），或者使用for循环而不是列表理解：

def packx(list it):
    """Cythonize python nibble packing loop, typed"""
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    cdef list r = [0]*n
    for i in range(n):
        r[i] = (it[i*2]//16)<<4 | it[i*2+1]//16
    return r

我认为我之前的测试只是将array.array指定为输入，但是在注释之后，我现在尝试了

from cpython cimport array
import array

def packa(array.array a):
    """Cythonize python nibble packing loop, typed"""
    cdef unsigned int n = len(a)//2
    cdef unsigned int i
    cdef unsigned int b[256*64/2]
    for i in range(n):
        b[i] = (a[i*2]//16)<<4 | a[i*2+1]//16;

    cdef array.array c = array.array("B", b)
    return c

编译但

ima = array.array("B", imd) # unsigned char (1 Byte)
pa = packa(ima)
packed = pa.tolist()

段错误。
我发现文档有些稀疏，因此希望了解有关问题所在以及如何为输出数据分配数组的任何提示。


采用@ead的第一种方法，再加上除法和移位（似乎可以节省一微秒：

#cython: boundscheck=False, wraparound=False

def packa(char[::1] a):
    """Cythonize python nibble packing loop, typed with array"""

    cdef unsigned int n = len(a)//2
    cdef unsigned int i

    # cdef unsigned int b[256*64/2]
    cdef array.array res = array.array('B', [])
    array.resize(res, n)

    for i in range(n):
        res.data.as_chars[i] = ( a[i*2] & 0xF0 ) | (a[i*2+1] >> 4);

    return res

编译时间更长，但运行速度更快：

python3 -m timeit -s 'from pack import packa; import array; data = array.array("B", bytes([0]*256*64))' 'packa(data)'
1000 loops, best of 3: 236 usec per loop

惊人！但是，有了额外的字节到数组和数组到列表的转换

ima = array.array("B", imd) # unsigned char (1 Byte)
pa = packa(ima)
packed = pa.tolist() # bytes would probably also do

现在只需要1.7毫秒-太酷了！



定时至大约150 us。实际0.4毫秒：

from cython cimport boundscheck, wraparound
from cpython cimport array
import array

@boundscheck(False)
@wraparound(False)
def pack(const unsigned char[::1] di):
    cdef:
        unsigned int i, n = len(di)
        unsigned char h, l, r
        array.array do = array.array('B')

    array.resize(do, n>>1)

    for i in range(0, n, 2):
        h = di[i] & 0xF0
        l = di[i+1] >> 4
        r = h | l
        do.data.as_uchars[i>>1] = r

    return do

我不再将结果数组转换为列表，这在写入时由py-spidev自动完成，总时间大约相同：10 ms（@ 10 MHz）。

Answer 1

如果您想和C一样快，则不应在列表中使用python-integers，而应使用array.array。使用cython + array.array可以使您的python + list代码加速140左右。

以下是一些如何使用cython加快代码编写速度的想法。作为基准，我选择了一个包含1000个元素的列表（足够大，并且缓存丢失没有影响）：

import random
l=[random.randint(0,15) for _ in range(1000)]

作为基线，您的python实现列表：

def packx(it):
    n = len(it)//2
    r = [0]*n
    for i in range(n):
        r[i] = (it[i*2]%16)<<4 | it[i*2+1]%16
    return r

%timeit packx(l)
143 µs ± 1.95 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)

顺便说一句，我使用的是 %而不是 //，这可能是您想要的，否则结果将只得到 0（描述中只有低位的数据）。

在对同一函数进行cython运算后（使用 %%cython -magic），我们得到了大约2的加速：

%timeit packx(l)
77.6 µs ± 1.28 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)

让我们看一下由选项 -a生成的html，对于与 for -loop对应的行，我们看到以下内容：

..... 
__pyx_t_2 = PyNumber_Multiply(__pyx_v_i, __pyx_int_2); if (unlikely(!__pyx_t_2)) __PYX_ERR(0, 6, __pyx_L1_error)
 __Pyx_GOTREF(__pyx_t_2);
 __pyx_t_5 = PyObject_GetItem(__pyx_v_it, __pyx_t_2); if (unlikely(!__pyx_t_5)) __PYX_ERR(0, 6, __pyx_L1_error)
 __Pyx_GOTREF(__pyx_t_5);
 __Pyx_DECREF(__pyx_t_2); __pyx_t_2 = 0;
 __pyx_t_2 = __Pyx_PyInt_RemainderObjC(__pyx_t_5, __pyx_int_16, 16, 0); if (unlikely(!__pyx_t_2)) __PYX_ERR(0, 6, __pyx_L1_error)
...

Py_NumberMultiply表示我们使用慢速python乘法， Pyx_DECREF-所有临时对象都是慢速python对象。我们需要改变这一点！

我们不传递列表，而是传递字节的 array.array到我们的函数，然后返回字节的 array.array。列表中包含完整的python对象， array.array较低的原始c数据，速度更快：

%%cython
from cpython cimport array
def cy_apackx(char[::1] it):
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    cdef array.array res = array.array('b', [])
    array.resize(res, n)
    for i in range(n):
        res.data.as_chars[i] = (it[i*2]%16)<<4 | it[i*2+1]%16
    return res

import array
a=array.array('B', l)
%timeit cy_apackx(a)
19.2 µs ± 316 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

更好，但让我们看一下生成的html，仍然有一些慢速的python代码：

 __pyx_t_2 = __Pyx_PyInt_From_long(((__Pyx_mod_long((*((char *) ( /* dim=0 */ ((char *) (((char *) __pyx_v_it.data) + __pyx_t_7)) ))), 16) << 4) | __Pyx_mod_long((*((char *) ( /* dim=0 */ ((char *) (((char *) __pyx_v_it.data) + __pyx_t_8)) ))), 16))); if (unlikely(!__pyx_t_2)) __PYX_ERR(0, 9, __pyx_L1_error)
__Pyx_GOTREF(__pyx_t_2);
 if (unlikely(__Pyx_SetItemInt(((PyObject *)__pyx_v_res), __pyx_v_i, __pyx_t_2, unsigned int, 0, __Pyx_PyInt_From_unsigned_int, 0, 0, 1) < 0)) __PYX_ERR(0, 9, __pyx_L1_error)
 __Pyx_DECREF(__pyx_t_2); __pyx_t_2 = 0;

我们仍然对数组（ __Pax_SetItemInt）使用python-setter，为此需要python objecct __pyx_t_2，为避免这种情况，我们使用 array.data.as_chars：

%%cython
from cpython cimport array
def cy_apackx(char[::1] it):
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    cdef array.array res = array.array('B', [])
    array.resize(res, n)
    for i in range(n):
        res.data.as_chars[i] = (it[i*2]%16)<<4 | it[i*2+1]%16  ##HERE!
return res

%timeit cy_apackx(a)
1.86 µs ± 30.5 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

更好，但是让我们再次看一下html，我们看到对 __Pyx_RaiseBufferIndexError的一些调用-这种安全性需要花费一些时间，因此我们将其关闭：

%%cython
from cpython cimport array    
import cython
@cython.boundscheck(False) # switch of safety-checks
@cython.wraparound(False) # switch of safety-checks
def cy_apackx(char[::1] it):
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    cdef array.array res = array.array('B', [])
    array.resize(res, n)
    for i in range(n):
        res.data.as_chars[i] = (it[i*2]%16)<<4 | it[i*2+1]%16  ##HERE!
    return res

%timeit cy_apackx(a)
1.53 µs ± 11.5 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

当我们查看生成的html时，我们看到：

__pyx_t_7 = (__pyx_v_i * 2);
__pyx_t_8 = ((__pyx_v_i * 2) + 1);
(__pyx_v_res->data.as_chars[__pyx_v_i]) = ((__Pyx_mod_long((*((char *) ( /* dim=0 */ ((char *) (((char *) __pyx_v_it.data) + __pyx_t_7)) ))), 16) << 4) | __Pyx_mod_long((*((char *) ( /* dim=0 */ ((char *) (((char *) __pyx_v_it.data) + __pyx_t_8)) ))), 16));

没有python资料！目前很好。但是，我不确定 __Pyx_mod_long，它的定义是：

static CYTHON_INLINE long __Pyx_mod_long(long a, long b) {
   long r = a % b;
   r += ((r != 0) & ((r ^ b) < 0)) * b;
   return r;
}

因此，C和Python的负数 mod有所不同，必须将其考虑在内。尽管是内联的，但此函数定义将阻止C编译器将 a%16优化为 a&15。我们只有正数，所以不需要关心它们，因此我们需要自己做 a&15技巧：

%%cython
from cpython cimport array
import cython
@cython.boundscheck(False)
@cython.wraparound(False)
def cy_apackx(char[::1] it):
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    cdef array.array res = array.array('B', [])
    array.resize(res, n)
    for i in range(n):
        res.data.as_chars[i] = (it[i*2]&15)<<4 | (it[i*2+1]&15)
    return res

%timeit cy_apackx(a)
1.02 µs ± 8.63 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

我也对生成的C-code / html（仅一行）感到满意：

(__pyx_v_res->data.as_chars[__pyx_v_i]) = ((((*((char *) ( /* dim=0 */ ((char *) (((char *) __pyx_v_it.data) + __pyx_t_7)) ))) & 15) << 4) | ((*((char *) ( /* dim=0 */ ((char *) (((char *) __pyx_v_it.data) + __pyx_t_8)) ))) & 15));

结论：总的来说，这意味着可以提高140（140 µs对1.02 µs）的速度-不错！另一个有趣的一点是：计算本身大约需要2 µs（包括少于最佳边界检查和划分的时间）-138 µs用于创建，注册和删除临时python对象。



如果您需要较高的位并且可以假定较低的位没有污垢（否则 &250可以帮助您），则可以使用：

from cpython cimport array
import cython
@cython.boundscheck(False)
@cython.wraparound(False)
def cy_apackx(char[::1] it):
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    cdef array.array res = array.array('B', [])
    array.resize(res, n)
    for i in range(n):
        res.data.as_chars[i] = it[i*2] | (it[i*2+1]>>4)
    return res

%timeit cy_apackx(a)
819 ns ± 8.24 ns per loop (mean ± std. dev. of 7 runs, 1000000 loops each)

另一个有趣的问题是，如果使用列表，则哪些成本可以操作。如果我们从“改进”版本开始：

%%cython
def cy_packx(it):
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    res=[0]*n
    for i in range(n):
        res[i] = it[i*2] | (it[i*2+1]>>4))
    return res

%timeit cy_packx(l)
20.7 µs ± 450 ns per loop (mean ± std. dev. of 7 runs, 10000 loops each)

我们看到，减少整数运算的数量可以大大提高速度。这是由于以下事实：python-integers是不可变的，并且每个操作都会创建一个新的临时对象，这是很昂贵的。消除运营意味着还消除了临时工。

但是， it[i*2] | (it[i*2+1]>>4)是使用python-integer完成的，下一步，我们将其设置为 cdef -operations：

%%cython   
def cy_packx(it):
    cdef unsigned int n = len(it)//2
    cdef unsigned int i
    cdef unsigned char a,b
    res=[0]*n
    for i in range(n):
        a=it[i*2]
        b=it[i*2+1]  # ensures next operations are fast
        res[i]= a | (b>>4)
    return res   

    %timeit cy_packx(l)
    7.3 µs ± 880 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

我不知道该如何进一步改进，因此列表有7.3 µs， array.array有1 µs。



最后一个问题，清单版本的费用为何？为了避免被C编译器优化，我们使用了稍微不同的基线函数：

%%cython
def cy_packx(it):
        cdef unsigned int n = len(it)//2
        cdef unsigned int i
        cdef unsigned char a,b
        cdef unsigned char s = 0
        res=[0]*n
        for i in range(n):
            a=it[i*2]
            b=it[i*2+1]  # ensures next operations are fast
            s+=a | (b>>4)
            res[i]= s
        return res
%timeit cy_packx(l)

In [79]: %timeit cy_packx(l)
7.67 µs ± 106 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

s变量的用法意味着，在第二个版本中并没有对其进行优化：

%%cython   
def cy_packx(it):
        cdef unsigned int n = len(it)//2
        cdef unsigned int i
        cdef unsigned char a,b
        cdef unsigned char s = 0
        res=[0]*n
        for i in range(n):
            a=it[i*2]
            b=it[i*2+1]  # ensures next operations are fast
            s+=a | (b>>4)
        res[0]=s
        return res 

In [81]: %timeit cy_packx(l)
5.46 µs ± 72.7 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

创建新的整数对象的成本大约为2 µs或大约30％。内存分配的成本是多少？

%%cython   
def cy_packx(it):
        cdef unsigned int n = len(it)//2
        cdef unsigned int i
        cdef unsigned char a,b
        cdef unsigned char s = 0
        for i in range(n):
            a=it[i*2]
            b=it[i*2+1]  # ensures next operations are fast
            s+=a | (b>>4)
        return s 

In [84]: %timeit cy_packx(l)
3.84 µs ± 43.2 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

这导致list-version的以下性能崩溃：

                    Time(in µs)      Percentage(in %)
     all                7.7                 100
     calculation          1                  12
     alloc memory       1.6                  21
     create ints        2.2                  29
     access data/cast   2.6                  38

我必须承认，我希望 create ints发挥更大的作用，并且访问列表中的数据并将其强制转换为 char不会花费那么多钱。