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

挑战：

在两个相同大小的缓冲区上执行按位XOR。 缓冲区将被要求是python strtypes，因为这是传统的python中的数据缓冲区的types。 返回结果值作为一个str 。 尽可能快地做到这一点。

input是两个1兆字节（2 ** 20字节）string。

我们面临的挑战是使用python或现有的第三方python模块（宽松的规则：或者创build自己的模块） 大幅度地打败我低效的algorithm。边际增长是无用的。

 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内在函数提供了一个边际的改进。 由于代码需要从磁盘加载并caching，所以第一次调用速度稍慢，后续调用速度更快：

 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() 

第二次尝试

考虑到这些意见，我重新访问了代码，以确定是否可以避免复制。 原来我读的string对象的文档错了，所以这里是我的第二次尝试：

 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) 

不同之处在于string是在C代码中分配的。 按照SSE2指令的要求，它不可能在16字节的边界alignment，因此开始和结束的未alignment的内存区域都是按字节访问的。

input的数据无论如何都是用numpy数组交换的，因为weave坚持把Python str对象复制到std::string s。 frombuffer不复制，所以这是好的，但内存不是在16字节alignment，所以我们需要使用_mm_loadu_si128而不是更快的_mm_load_si128 。

而不是使用_mm_store_si128 ，我们使用_mm_stream_si128 ，这将确保任何写入尽快_mm_stream_si128主内存—这样，输出数组不会占用宝贵的caching行。

计时

至于时间，在第一次编辑slow_xor条目提到我的改进版本（内联按位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编译的，我已经validation了编译器实际上使用了SSE指令。

Answer 2

性能比较：numpy与Cython与C与Fortran与Boost.Python（pyublas）

 | function | time, usec | ratio | type | |------------------------+------------+-------+--------------| | slow_xor | 2020 | 1.0 | numpy | | xorf_int16 | 1570 | 1.3 | fortran | | xorf_int32 | 1530 | 1.3 | fortran | | xorf_int64 | 1420 | 1.4 | fortran | | faster_slow_xor | 1360 | 1.5 | numpy | | inline_xor | 1280 | 1.6 | C | | cython_xor | 1290 | 1.6 | cython | | xorcpp_inplace (int32) | 440 | 4.6 | pyublas | | cython_xor_vectorised | 325 | 6.2 | cython | | inline_xor_nocopy | 172 | 11.7 | C | | xorcpp | 144 | 14.0 | boost.python | | xorcpp_inplace | 122 | 16.6 | boost.python | #+TBLFM: $3=@2$2/$2;%.1f 

要重现结果，请下载http://gist.github.com/353005并键入;make （要安装依赖关系，请input： sudo apt-get install build-essential python-numpy python-scipy cython gfortran ， Boost.Python依赖关系， pyublas不包括在内，因为他们需要人工干预才能工作）

哪里：

• slow_xor()来自OP的问题
• faster_slow_xor() ， inline_xor() ， inline_xor_nocopy()来自@Torsten Marek的回答
• cython_xor()和cython_vectorised()来自@ gnibbler的答案

和xor_$type_sig()是：

 ! xorf.f90.template subroutine xor_$type_sig(a, b, n, out) implicit none integer, intent(in) :: n $type, intent(in), dimension(n) :: a $type, intent(in), dimension(n) :: b $type, intent(out), dimension(n) :: out integer i forall(i=1:n) out(i) = ieor(a(i), b(i)) end subroutine xor_$type_sig 

它从Python使用如下：

 import xorf # extension module generated from xorf.f90.template import numpy as np def xor_strings(a, b, type_sig='int64'): assert len(a) == len(b) a = np.frombuffer(a, dtype=np.dtype(type_sig)) b = np.frombuffer(b, dtype=np.dtype(type_sig)) return getattr(xorf, 'xor_'+type_sig)(a, b).tostring() 

xorcpp_inplace() （Boost.Python，pyublas）：

xor.cpp ：

 #include <inttypes.h> #include <algorithm> #include <boost/lambda/lambda.hpp> #include <boost/python.hpp> #include <pyublas/numpy.hpp> namespace { namespace py = boost::python; template<class InputIterator, class InputIterator2, class OutputIterator> void xor_(InputIterator first, InputIterator last, InputIterator2 first2, OutputIterator result) { // `result` migth `first` but not any of the input iterators namespace ll = boost::lambda; (void)std::transform(first, last, first2, result, ll::_1 ^ ll::_2); } template<class T> py::str xorcpp_str_inplace(const py::str& a, py::str& b) { const size_t alignment = std::max(sizeof(T), 16ul); const size_t n = py::len(b); const char* ai = py::extract<const char*>(a); char* bi = py::extract<char*>(b); char* end = bi + n; if (n < 2*alignment) xor_(bi, end, ai, bi); else { assert(n >= 2*alignment); // applying Marek's algorithm to align const ptrdiff_t head = (alignment - ((size_t)bi % alignment))% alignment; const ptrdiff_t tail = (size_t) end % alignment; xor_(bi, bi + head, ai, bi); xor_((const T*)(bi + head), (const T*)(end - tail), (const T*)(ai + head), (T*)(bi + head)); if (tail > 0) xor_(end - tail, end, ai + (n - tail), end - tail); } return b; } template<class Int> pyublas::numpy_vector<Int> xorcpp_pyublas_inplace(pyublas::numpy_vector<Int> a, pyublas::numpy_vector<Int> b) { xor_(b.begin(), b.end(), a.begin(), b.begin()); return b; } } BOOST_PYTHON_MODULE(xorcpp) { py::def("xorcpp_inplace", xorcpp_str_inplace<int64_t>); // for strings py::def("xorcpp_inplace", xorcpp_pyublas_inplace<int32_t>); // for numpy } 

它从Python使用如下：

 import os import xorcpp a = os.urandom(2**20) b = os.urandom(2**20) c = xorcpp.xorcpp_inplace(a, b) # it calls xorcpp_str_inplace() 

Answer 3

这是我对cython的结果

 slow_xor 0.456888198853 faster_xor 0.400228977203 cython_xor 0.232881069183 cython_xor_vectorised 0.171468019485 

在cython中进行vector化可以在计算机上为for循环提供25％的折扣。然而，超过一半的时间用于构buildpythonstring（ return语句） – 我不认为额外的副本可以像数组一样被合法地避免可能包含空字节。

非法的方式是传递一个Pythonstring，并将其改变，并将函数的速度加倍。

xor.py

 from time import time from os import urandom from numpy import frombuffer,bitwise_xor,byte,uint64 import pyximport; pyximport.install() import xor_ 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 def faster_xor(aa,bb): a=frombuffer(aa,dtype=uint64) b=frombuffer(bb,dtype=uint64) c=bitwise_xor(a,b) r=c.tostring() return r aa=urandom(2**20) bb=urandom(2**20) def test_it(): t=time() for x in xrange(100): slow_xor(aa,bb) print "slow_xor ",time()-t t=time() for x in xrange(100): faster_xor(aa,bb) print "faster_xor",time()-t t=time() for x in xrange(100): xor_.cython_xor(aa,bb) print "cython_xor",time()-t t=time() for x in xrange(100): xor_.cython_xor_vectorised(aa,bb) print "cython_xor_vectorised",time()-t if __name__=="__main__": test_it() 

xor_.pyx

 cdef char c[1048576] def cython_xor(char *a,char *b): cdef int i for i in range(1048576): c[i]=a[i]^b[i] return c[:1048576] def cython_xor_vectorised(char *a,char *b): cdef int i for i in range(131094): (<unsigned long long *>c)[i]=(<unsigned long long *>a)[i]^(<unsigned long long *>b)[i] return c[:1048576] 

Answer 4

一个简单的加速是使用一个更大的“块”：

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

与uint64当然也从numpyimport。 我在4毫秒的时间，对比byte版本6毫秒。

Answer 5

你的问题不是NumPy的xOr方法的速度，而是所有的缓冲/数据types转换。 就我个人而言，我怀疑这篇文章的重点可能是真的吹嘘Python，因为你在这里做的是在时间框架内处理三个数据的GIGABYTES，与非parsing语言相比，本质上它更快。

下面的代码显示，即使在我不起眼的计算机上，Python可以在两秒钟内将 “aa”（1MB）和“bb”（1MB）变成“c”（1MB）一千次（总共3GB）。 严重的是，你想要多less改善？ 特别是从解释语言！ 80％的时间花在了“从缓冲区”和“暂停”之间。 另外20％的时间完成了实际的xOr-ing。 在2秒内达到3GB，即使在c中使用memcpy，也很难提高性能。

如果这是一个真正的问题，而不只是秘密地吹嘘Python，答案是编码，以尽量减lesstypes转换的数量，数量和频率，如“frombuffer”和“tostring”。 实际的xOr'ing已经很快了。

 from os import urandom from numpy import frombuffer,bitwise_xor,byte,uint64 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 bb=urandom(2**20) aa=urandom(2**20) def test_it(): for x in xrange(1000): slow_xor(aa,bb) def test_it2(): a=frombuffer(aa,dtype=uint64) b=frombuffer(bb,dtype=uint64) for x in xrange(1000): c=bitwise_xor(a,b); r=c.tostring() test_it() print 'Slow Complete.' #6 seconds test_it2() print 'Fast Complete.' #under 2 seconds 

无论如何，上面的“test_it2”完成了与“test_it”完全相同的xOr-ing量，但在1/5的时间。 5倍的速度提升应该符合“实质性”的要求，否？

Answer 6

最快的按位异或是“^”。 我可以比“bitwise_xor”更快地input;-)

Answer 7

Python3有int.from_bytes和int.to_bytes ，因此：

 x = int.from_bytes(b"a" * (1024*1024), "big") y = int.from_bytes(b"b" * (1024*1024), "big") (x ^ y).to_bytes(1024*1024, "big") 

它比IO快，很难testing它有多快，在我的机器上看起来像0.018 … 0.020s 。 奇怪的是， "little"印度转换速度稍快。

CPython 2.x具有底层函数_PyLong_FromByteArray ，它不会被导出，而是可以通过ctypes访问：

 In [1]: import ctypes In [2]: p = ctypes.CDLL(None) In [3]: p["_PyLong_FromByteArray"] Out[3]: <_FuncPtr object at 0x2cc6e20> 

Python 2的细节留给读者去练习。

Answer 8

如果你想对数组数据types进行快速操作，那么你应该尝试Cython（cython.org）。 如果你给它正确的声明，它应该能够编译成纯C代码。

Answer 9

作为一个string你需要多大的答案？ 请注意， c.tostring()方法必须将 c的数据复制到一个新的string中，因为Pythonstring是不可变的（而c是可变的）。 Python 2.6和3.1有一个bytearraytypes，除了是可变的之外，其行为类似于str （Python 3.x中的bytes ）。

另一个优化是使用out参数bitwise_xor来指定存储结果的位置。

在我的机器上，我得到了

 slow_xor (int8): 5.293521 (100.0%) outparam_xor (int8): 4.378633 (82.7%) slow_xor (uint64): 2.192234 (41.4%) outparam_xor (uint64): 1.087392 (20.5%) 

与本文后面的代码。 特别注意，使用预分配缓冲区的方法比创build一个新对象（当在4字节（ uint64 ）块上操作时快两倍）。 这与较慢的方法每个块（xor + copy）对更快的1（只是xor）执行两个操作一致。

另外，FWIW， a ^ b相当于bitwise_xor(a,b) ， a ^= b相当于bitwise_xor(a, b, a) 。

所以，没有写任何外部模块的5倍加速:)

 from time import time from os import urandom from numpy import frombuffer,bitwise_xor,byte,uint64 def slow_xor(aa, bb, ignore, dtype=byte): a=frombuffer(aa, dtype=dtype) b=frombuffer(bb, dtype=dtype) c=bitwise_xor(a, b) r=c.tostring() return r def outparam_xor(aa, bb, out, dtype=byte): a=frombuffer(aa, dtype=dtype) b=frombuffer(bb, dtype=dtype) c=frombuffer(out, dtype=dtype) assert c.flags.writeable return bitwise_xor(a, b, c) aa=urandom(2**20) bb=urandom(2**20) cc=bytearray(2**20) def time_routine(routine, dtype, base=None, ntimes = 1000): t = time() for x in xrange(ntimes): routine(aa, bb, cc, dtype=dtype) et = time() - t if base is None: base = et print "%s (%s): %f (%.1f%%)" % (routine.__name__, dtype.__name__, et, (et/base)*100) return et def test_it(ntimes = 1000): base = time_routine(slow_xor, byte, ntimes=ntimes) time_routine(outparam_xor, byte, base, ntimes=ntimes) time_routine(slow_xor, uint64, base, ntimes=ntimes) time_routine(outparam_xor, uint64, base, ntimes=ntimes) 

Answer 10

你可以尝试圣人的比特集的对称差异。

http://www.sagemath.org/doc/reference/sage/misc/bitset.html

Answer 11

最快的方法（快速）将做什么最大。 Sbuild议。 在C中实现

这个任务的支持代码应该是相当简单的写。 这只是一个模块创build一个新的string和做异或的一个函数。 就这样。 当你实现了这样一个模块时，以代码为模板很简单。 或者你甚至可以从其他人那里实现一个模块来实现一个简单的Python增强模块，只是抛出一切不需要你的任务。

真正复杂的部分就是做RefCounter-Stuff的权利。 但是一旦意识到它是如何工作的，它是可以pipe理的 – 也是因为手头的任务非常简单（分配一些内存，并返回它 – 参数不会被触摸（reflection））。