挑戰：對兩個相等大小的緩沖區執行按位XOR。緩沖區將要求是python str類型，因為傳統上這是python中數據緩沖區的類型。將結果值作為返回str。盡快執行此操作。輸入是兩個1兆字節（2 ** 20字節）字符串。目前的挑戰是基本使用python的或現有的第三方Python模塊打我的低效算法（放寬的規定：或者創建自己的模塊）的邊際增加是無用的。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貢獻1852條經驗 獲得超1個贊
第一次嘗試
使用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.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指令。
ibeautiful
TA貢獻1993條經驗 獲得超6個贊
這是我的cython結果
slow_xor 0.456888198853
faster_xor 0.400228977203
cython_xor 0.232881069183
cython_xor_vectorised 0.171468019485
在cython中進行矢量化可將計算機上的for循環節省25％左右的時間，但是花了一半以上的時間來構建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]
添加回答
舉報
0/150
提交
取消