挑战:
对两个等大小的缓冲区执行按位异或操作。这些缓冲区需要是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)
使用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::string中。frombuffer不会复制，所以这很好，但是内存没有对齐到16字节，因此我们需要使用_mm_loadu_si128而不是更快的_mm_load_si128。
我们使用_mm_stream_si128而不是_mm_store_si128，这将确保任何写入尽可能快地流式传输到主存储器---这样，输出数组就不会占用宝贵的缓存行。
至于时间，首次编辑中的slow_xor条目是指我改进后的版本(内联位异或，uint64)，我删除了那个混淆。 slow_xor指原始问题中的代码。所有计时都是针对1000次运行进行的。
slow_xor：1.85秒(1x)faster_slow_xor：1.25秒(1.48x)inline_xor：0.95秒(1.95x)inline_xor_nocopy：0.32秒(5.78x)该代码是使用gcc 4.4.3编译的，我已经验证编译器实际上使用了SSE指令。
| 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
make（要安装依赖项，请键入：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）：#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()
slow_xor 0.456888198853
faster_xor 0.400228977203
cython_xor 0.232881069183
cython_xor_vectorised 0.171468019485
return语句）上 - 我认为额外的复制无法避免（合法），因为数组可能包含空字节。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]
一个简单的加速方法是使用更大的“chunk”：
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
当然也从numpy导入了uint64。我用timeit测得其执行时间为4毫秒，而使用byte版本的执行时间为6毫秒。
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 只运行了一次 c.tostring，而 test_it 运行了一千次，所以才会这样？ - just somebodyPython3有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更快，很难测试它的速度有多快，在我的机器上看起来是0.018 .. 0.020秒。奇怪的是，"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的细节留给读者自行处理。
最快的按位异或运算符是“^”。我可以比输入“bitwise_xor”更快地输入它;-)
你需要把答案作为字符串吗？ 注意，c.tostring() 方法必须将 c 中的数据复制到新字符串中，因为 Python 字符串是不可变的（而 c 是可变的）。Python 2.6 和 3.1 有一个 bytearray 类型，它的用法类似于 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%)
附带此文章末尾的代码。特别注意，使用预分配缓冲区的方法在操作4字节（uint64）块时速度是创建新对象的两倍。这与较慢的方法每个块执行两个操作（xor + copy）相一致，而更快的方法只需要一个操作（仅仅是xor）。
另外，顺便说一下，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)