挑戰(zhàn)：對(duì)兩個(gè)相等大小的緩沖區(qū)執(zhí)行按位XOR。緩沖區(qū)將要求是python str類型，因?yàn)閭鹘y(tǒng)上這是python中數(shù)據(jù)緩沖區(qū)的類型。將結(jié)果值作為返回str。盡快執(zhí)行此操作。輸入是兩個(gè)1兆字節(jié)（2 ** 20字節(jié)）字符串。目前的挑戰(zhàn)是基本使用python的或現(xiàn)有的第三方Python模塊打我的低效算法（放寬的規(guī)定：或者創(chuàng)建自己的模塊）的邊際增加是無用的。from os import urandomfrom numpy import frombuffer,bitwise_xor,bytedef slow_xor(aa,bb): a=frombuffer(aa,dtype=byte) b=frombuffer(bb,dtype=byte) c=bitwise_xor(a,b) r=c.tostring() return raa=urandom(2**20)bb=urandom(2**20)def test_it(): for x in xrange(1000): slow_xor(aa,bb)
3 回答
小怪獸愛吃肉
TA貢獻(xiàn)1852條經(jīng)驗(yàn) 獲得超1個(gè)贊
第一次嘗試
使用scipy.weave和SSE2內(nèi)在函數(shù)會(huì)帶來一點(diǎn)改進(jìn)。第一次調(diào)用要慢一些，因?yàn)樾枰獜拇疟P加載代碼并進(jìn)行緩存，隨后的調(diào)用會(huì)更快：
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()
第二次嘗試
考慮到注釋，我重新檢查了代碼，以查明是否可以避免復(fù)制。原來我讀錯(cuò)了字符串對(duì)象的文檔，所以這是我的第二次嘗試：
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)
區(qū)別在于字符串是在C代碼內(nèi)部分配的。不可能按照SSE2指令的要求將其對(duì)齊在16字節(jié)的邊界上，因此使用字節(jié)訪問來復(fù)制開頭和結(jié)尾未對(duì)齊的內(nèi)存區(qū)域。
無論如何，使用numpy數(shù)組傳遞輸入數(shù)據(jù)，因?yàn)閣eave堅(jiān)持將Python str對(duì)象復(fù)制到std::strings。frombuffer不會(huì)復(fù)制，因此很好，但是內(nèi)存未對(duì)齊16字節(jié)，因此我們需要使用_mm_loadu_si128而不是更快的_mm_load_si128。
而不是使用_mm_store_si128，而是使用_mm_stream_si128，它可以確保所有寫入均盡快流式傳輸?shù)街鞔鎯?chǔ)器中-這樣，輸出數(shù)組不會(huì)耗盡寶貴的緩存行。
時(shí)機(jī)
至于時(shí)間安排，slow_xor第一次編輯中的條目指的是我的改進(jìn)版本（內(nèi)聯(lián)按位xor，uint64），我消除了這種困惑。slow_xor指的是來自原始問題的代碼。所有計(jì)時(shí)都完成了1000次運(yùn)行。
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編譯的，我已經(jīng)驗(yàn)證了編譯器實(shí)際上使用了SSE指令。
ibeautiful
TA貢獻(xiàn)1993條經(jīng)驗(yàn) 獲得超6個(gè)贊
這是我的cython結(jié)果
slow_xor 0.456888198853
faster_xor 0.400228977203
cython_xor 0.232881069183
cython_xor_vectorised 0.171468019485
在cython中進(jìn)行矢量化可將計(jì)算機(jī)上的for循環(huán)節(jié)省25％左右的時(shí)間，但是花了一半以上的時(shí)間來構(gòu)建python字符串（該return語句）-我認(rèn)為（合法地）避免多余的拷貝是可能的，因?yàn)閿?shù)組可能包含空字節(jié)。
非法的方法是傳入Python字符串并對(duì)其進(jìn)行適當(dāng)?shù)耐蛔?，這會(huì)使函數(shù)的速度加倍。
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]
添加回答
舉報(bào)
0/150
提交
取消