一尘不染

简单的Python挑战:数据缓冲区上最快的按位异或

algorithm

挑战:

对两个相等大小的缓冲区执行按位XOR。缓冲区必须是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)

阅读 236

收藏
2020-07-28

共1个答案

一尘不染

第一次尝试

使用scipy.weaveSSE2内在函数会带来一点改进。第一次调用要慢一些,因为需要从磁盘加载代码并进行缓存,随后的调用会更快:

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,它可以确保所有写入均尽快流式传输到主内存中-
这样,输出数组不会耗尽宝贵的缓存行。

时机

至于时间安排,slow_xor第一次编辑中的条目指的是我的改进版本(内联按位xor,uint64),我消除了这种混乱。slow_xor指的是来自原始问题的代码。所有计时都完成了1000次运行。

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

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

2020-07-28