53

挑战:

对两个大小相等的缓冲区执行按位异或。缓冲区将被要求为 pythonstr类型,因为这通常是 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)
4

11 回答 11

37

第一次尝试

使用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.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 指令。

于 2010-02-01T20:07:58.887 回答
37

性能比较:numpy vs. Cython vs. C vs. Fortran vs. 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.Pythonpyublas不包括在内,因为它们需要手动干预才能工作)

在哪里:

并且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):

xor.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()
于 2010-04-02T10:15:59.323 回答
17

这是我对 cython 的结果

slow_xor   0.456888198853
faster_xor 0.400228977203
cython_xor 0.232881069183
cython_xor_vectorised 0.171468019485

cython 中的矢量化在我的计算机上的 for 循环中减少了大约 25%,但是超过一半的时间用于构建 python 字符串(return语句) - 我认为不能(合法地)避免额外的副本,因为数组可能包含空字节。

非法的方法是传入一个 Python 字符串并在适当的位置对其进行变异,从而使函数的速度加倍。

异或.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]
于 2010-02-01T00:22:05.327 回答
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 毫秒。

于 2010-01-22T19:39:34.363 回答
7

您的问题不在于 NumPy 的 xOr 方法的速度,而在于所有缓冲/数据类型转换。就我个人而言,我怀疑这篇文章的目的可能真的是吹嘘 Python,因为您在这里所做的是在与非解释语言相当的时间范围内处理 3 GB 的数据,这些语言本质上更快。

下面的代码表明,即使在我不起眼的计算机上,Python 也可以在两秒钟内将“aa”(1MB)和“bb”(1MB)xOr“c”(1MB)一千次(总共 3GB)。说真的,你还想要多少改进?特别是从解释语言!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”完全相同的xOr-ing数量,但时间只有1/5。5 倍的速度提升应该被认为是“实质性的”,不是吗?

于 2010-01-31T21:06:25.457 回答
5

最快的按位异或是“^”。我可以“bitwise_xor”更快地输入;-)

于 2010-02-02T23:44:17.257 回答
5

Python3 有int.from_bytesand 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 details are left as exercise to the reader.

于 2014-05-06T12:55:34.290 回答
1

如果您想对数组数据类型进行快速操作,那么您应该尝试 Cython (cython.org)。如果你给它正确的声明,它应该能够编译成纯 c 代码。

于 2010-01-23T21:22:26.403 回答
1

您需要字符串形式的答案有多严重?请注意,该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 + 复制)到较快的 1(只是 xor)一致。

此外,FWIWa ^ 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)
于 2010-04-02T20:55:44.557 回答
0

您可以尝试 sage 位集的对称差异。

http://www.sagemath.org/doc/reference/sage/misc/bitset.html

于 2010-01-22T19:51:21.077 回答
0

最快的方式(speedwise)将做 Max. S 推荐。在 C 中实现它。

这个任务的支持代码应该写起来相当简单。它只是模块中的一个函数,它创建一个新字符串并执行 xor。就这样。当你实现了一个这样的模块时,将代码作为模板很简单。或者,您甚至可以从其他人那里实现一个为 Python 实现简单增强模块的模块,然后丢弃您的任务不需要的所有内容。

真正复杂的部分只是正确地执行 RefCounter-Stuff。但是一旦意识到它是如何工作的,它就可以管理了——同样因为手头的任务非常简单(分配一些内存,然后返回它——参数不会被触及(Ref-wise))。

于 2010-01-29T22:01:25.747 回答