python - 简单的 Python 挑战 : Fastest Bitwise XOR on Data Buffers

Question

挑战:

对两个大小相等的缓冲区执行按位异或。缓冲区将被要求为 python str 类型，因为这通常是 python 中数据缓冲区的类型。将结果值作为 str 返回。尽快执行此操作。

输入是两个 1 兆字节(2**20 字节)的字符串。

挑战是使用 python 或现有的第三方 python 模块大幅击败我的低效算法(宽松规则:或创建自己的模块。)边际增加是无用的。

from os import urandom
from numpy import frombuffer,bitwise_xor,byte

def slow_xor(aa,bb):
    a=frombuffer(aa,dtype=byte)
    b=frombuffer(bb,dtype=byte)
    c=bitwise_xor(a,b)
    r=c.tostring()
    return r

aa=urandom(2**20)
bb=urandom(2**20)

def test_it():
    for x in xrange(1000):
        slow_xor(aa,bb)

Answer 1

第一次尝试

使用 scipy.weave和 SSE2内在函数提供了边际改进。第一次调用有点慢，因为代码需要从磁盘加载并缓存，后续调用更快:

import numpy
import time
from os import urandom
from scipy import weave

SIZE = 2**20

def faster_slow_xor(aa,bb):
    b = numpy.fromstring(bb, dtype=numpy.uint64)
    numpy.bitwise_xor(numpy.frombuffer(aa,dtype=numpy.uint64), b, b)
    return b.tostring()

code = """
const __m128i* pa = (__m128i*)a;
const __m128i* pend = (__m128i*)(a + arr_size);
__m128i* pb = (__m128i*)b;
__m128i xmm1, xmm2;
while (pa < pend) {
  xmm1 = _mm_loadu_si128(pa); // must use unaligned access 
  xmm2 = _mm_load_si128(pb); // numpy will align at 16 byte boundaries
  _mm_store_si128(pb, _mm_xor_si128(xmm1, xmm2));
  ++pa;
  ++pb;
}
"""

def inline_xor(aa, bb):
    a = numpy.frombuffer(aa, dtype=numpy.uint64)
    b = numpy.fromstring(bb, dtype=numpy.uint64)
    arr_size = a.shape[0]
    weave.inline(code, ["a", "b", "arr_size"], headers = ['"emmintrin.h"'])
    return b.tostring()

第二次尝试

考虑到评论，我重新访问了代码以了解是否可以避免复制。原来我读错了字符串对象的文档，所以这是我的第二次尝试:

support = """
#define ALIGNMENT 16
static void memxor(const char* in1, const char* in2, char* out, ssize_t n) {
    const char* end = in1 + n;
    while (in1 < end) {
       *out = *in1 ^ *in2;
       ++in1; 
       ++in2;
       ++out;
    }
}
"""

code2 = """
PyObject* res = PyString_FromStringAndSize(NULL, real_size);

const ssize_t tail = (ssize_t)PyString_AS_STRING(res) % ALIGNMENT;
const ssize_t head = (ALIGNMENT - tail) % ALIGNMENT;

memxor((const char*)a, (const char*)b, PyString_AS_STRING(res), head);

const __m128i* pa = (__m128i*)((char*)a + head);
const __m128i* pend = (__m128i*)((char*)a + real_size - tail);
const __m128i* pb = (__m128i*)((char*)b + head);
__m128i xmm1, xmm2;
__m128i* pc = (__m128i*)(PyString_AS_STRING(res) + head);
while (pa < pend) {
    xmm1 = _mm_loadu_si128(pa);
    xmm2 = _mm_loadu_si128(pb);
    _mm_stream_si128(pc, _mm_xor_si128(xmm1, xmm2));
    ++pa;
    ++pb;
    ++pc;
}
memxor((const char*)pa, (const char*)pb, (char*)pc, tail);
return_val = res;
Py_DECREF(res);
"""

def inline_xor_nocopy(aa, bb):
    real_size = len(aa)
    a = numpy.frombuffer(aa, dtype=numpy.uint64)
    b = numpy.frombuffer(bb, dtype=numpy.uint64)
    return weave.inline(code2, ["a", "b", "real_size"], 
                        headers = ['"emmintrin.h"'], 
                        support_code = support)

不同之处在于字符串是在 C 代码内部分配的。不可能按照 SSE2 指令的要求将其对齐在 16 字节边界，因此使用按字节访问复制开头和结尾的未对齐内存区域。

输入数据还是使用numpy数组传入的，因为weave坚持将Python的str对象复制到std::strings。 frombuffer 不会复制，所以这很好，但是内存不是在 16 字节对齐的，所以我们需要使用 _mm_loadu_si128 而不是更快的 _mm_load_si128。

我们不使用 _mm_store_si128，而是使用 _mm_stream_si128，这将确保任何写入都尽快流式传输到主内存 --- 这样，输出array 不会用完有值(value)的缓存行。

时间

至于计时，第一次编辑中的 slow_xor 条目引用了我的改进版本(内联按位异或，uint64)，我消除了这种混淆。 slow_xor 指的是原始问题的代码。所有计时都是针对 1000 次运行完成的。

• slow_xor:1.85s (1x)
• faster_slow_xor:1.25s (1.48x)
• inline_xor:0.95s (1.95x)
• inline_xor_nocopy:0.32s (5.78x)

代码是使用 gcc 4.4.3 编译的，我已经验证编译器实际上使用了 SSE 指令。