简单的 Python 挑战：数据缓冲区上最快的按位异或
- 2025-03-21 09:06:00
- admin 原创
- 22
问题描述：
挑战：
对两个大小相等的缓冲区执行按位异或。缓冲区必须是 Pythonstr类型，因为 Python 中数据缓冲区的传统类型就是 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)
解决方案 1：
首次尝试
使用scipy.weaveSSE2内部函数可以带来些许改进。第一次调用速度稍慢，因为需要从磁盘加载代码并缓存，后续调用速度会更快：
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坚持将 Pythonstr对象复制到std::strings。frombuffer不复制，所以这很好，但是内存没有对齐到 16 字节，所以我们需要使用_mm_loadu_si128而不是更快的_mm_load_si128。
_mm_store_si128我们不使用，而是使用_mm_stream_si128，这将确保任何写入都尽快传输到主存储器——这样，输出数组就不会耗尽宝贵的缓存行。
时间安排
至于计时，slow_xor第一次编辑中的条目引用了我的改进版本（内联按位异或，uint64），我消除了这种混淆。slow_xor指的是原始问题中的代码。所有计时都是针对 1000 次运行进行的。
slow_xor：1.85 秒 (1x)faster_slow_xor：1.25 秒（1.48 倍）inline_xor：0.95 秒（1.95 倍）inline_xor_nocopy：0.32 秒（5.78 倍）
该代码是使用 gcc 4.4.3 编译的，并且我已经验证该编译器确实使用了 SSE 指令。
解决方案 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（要安装依赖项，请输入：sudo apt-get install build-essential python-numpy python-scipy cython gfortran，的依赖项不包括在内Boost.Python，pyublas因为它们需要手动干预才能工作）
在哪里：
slow_xor()来自 OP 的问题faster_slow_xor()，，来自@Torsten Marek的inline_xor()回答inline_xor_nocopy()cython_xor()来自@gnibbler 的cython_vectorised()回答
并且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）：
异或.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()
解决方案 3：
以下是我对 cython 的结果
slow_xor 0.456888198853
faster_xor 0.400228977203
cython_xor 0.232881069183
cython_xor_vectorised 0.171468019485
在我的计算机上，cython 中的矢量化可以减少大约 25% 的 for 循环，但是超过一半的时间都花在构建 python 字符串（return语句）上 - 我认为无法避免额外的复制（合法），因为数组可能包含空字节。
非法的方法是传入一个 Python 字符串并就地对其进行变异，这会使函数的速度加倍。
异或
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()
异或.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]
解决方案 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，也导入了numpy。我timeit将其设置为 4 毫秒，而该byte版本则设置为 6 毫秒。
解决方案 5：
你的问题不是 NumPy 的 xOr 方法的速度，而是所有缓冲/数据类型转换的速度。我个人怀疑这篇文章的重点可能真的是吹嘘 Python，因为你在这里所做的是在与非解释型语言相当的时间范围内处理 3 GB 的数据，而非解释型语言本身就更快。
下面的代码表明，即使在我这台简陋的电脑上，Python 也能在两秒钟内将“aa”（1MB）和“bb”（1MB）异或运算一千次（总共 3GB）到“c”（1MB）。说真的，你还想要多少改进？尤其是从解释型语言开始！80% 的时间都花在调用“frombuffer”和“tostring”上。实际的异或运算在另外 20% 的时间内完成。在 2 秒内处理 3GB 的情况下，即使只是在 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”完全相同，但只用了 1/5 的时间。5 倍的速度提升应该算是“实质性的”，不是吗？
解决方案 6：
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 的细节留给读者作为练习。
解决方案 7：
最快的按位异或运算是“^”。我输入它比输入“bitwise_xor”快得多;-)
解决方案 8：
如果您想对数组数据类型进行快速操作，那么您应该尝试 Cython (cython.org)。如果您给它正确的声明，它应该能够编译为纯 C 代码。
解决方案 9：
您有多需要字符串形式的答案？请注意，该c.tostring()方法必须将数据复制c到新字符串中，因为 Python 字符串是不可变的（并且c是可变的）。Python 2.6 和 3.1 有一个bytearray类型，其作用类似于Python 3.x 中的str（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）块时）。这与较慢的方法对每个块执行两次操作（异或 + 复制）对较快的方法执行 1 次操作（仅异或）一致。
此外，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)
解决方案 10：
您可以尝试一下 sage 的位集的对称差异。
http://www.sagemath.org/doc/reference/sage/misc/bitset.html
解决方案 11：
最快的方法（速度方面）是按照 Max. S 的建议去做。用 C 实现。
这项任务的支持代码应该相当容易编写。它只是模块中的一个函数，用于创建新字符串并执行异或。就是这样。当您实现了这样的一个模块时，很容易将代码作为模板。或者，您甚至可以从其他人那里获取一个模块，该模块实现了一个简单的 Python 增强模块，然后扔掉任务不需要的所有内容。
真正复杂的部分只是正确地执行 RefCounter-Stuff。但是一旦了解了它的工作原理，它就很容易管理了——而且因为手头的任务非常简单（分配一些内存并返回它——参数不会被触及（Ref-wise））。
