use*_*060 51 python algorithm performance xor
挑战:
在两个大小相等的缓冲区上执行按位XOR.缓冲区将需要是python str类型,因为传统上它是python中数据缓冲区的类型.将结果值作为a返回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)
Tor*_*rek 36
使用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,而不是使用,这将确保尽快将任何写入流传输到主存储器 - 这样,输出数组不会耗尽宝贵的缓存行.
至于时间,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编译的,我已经验证了编译器实际上使用了SSE指令.
jfs*_*jfs 36
| 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(以安装依赖项,键入: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马立克氏回答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()
Joh*_*ooy 17
这是我对cython的结果
slow_xor 0.456888198853
faster_xor 0.400228977203
cython_xor 0.232881069183
cython_xor_vectorised 0.171468019485
在我的计算机上使用for循环大约25%的折扣在cython中进行矢量化,但是超过一半的时间用于构建python字符串(return语句) - 我不认为额外的副本可以避免(合法地),因为数组可能包含空字节.
非法的方式是传入一个Python字符串并将其变异,并使该函数的速度加倍.
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]
Ale*_*lli 10
一个简单的加速是使用更大的'块':
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也从进口的numpy,当然.我timeit这是4毫秒,而byte版本是6毫秒.
你的问题不是NumPy的xOr方法的速度,而是所有的缓冲/数据类型转换.我个人怀疑这篇文章的观点可能真的是吹嘘Python,因为你在这里所做的是在时间框架中处理与非解释性语言相同的三种数据技术,这些语言本质上更快.
下面的代码表明,即使在我卑微的电脑Python可以XOR"AA"(1MB)和"BB"(1MB)为"C"(1MB)一千倍(共3GB)根据2秒.说真的,你想要多少改进？特别是从解释语言!80%的时间用于调用"frombuffer"和"tostring".实际的xOr-ing在其他20%的时间内完成.在3GB在2秒内,你会捉襟见肘时,要提高基本在C甚至只是使用的memcpy.
如果这是一个真正的问题,而不只是隐蔽吹嘘Python,答案是编码,以尽量减少类型转换的数量,数量和频率,如"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倍的速度改进应该被称为"实质性",不是吗？
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 快，有点难以测试它有多快，在我的机器上看起来像0.018 .. 0.020s。奇怪的是"little"-endian 转换要快一点。
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 的细节留给读者作为练习。