使用 CPU SSE2 指令集加速字符查找

原创

王杰新

修改于 2020-07-02 14:28:36

1.1K0

修改于 2020-07-02 14:28:36

文章被收录于专栏：粑粑是程序员粑粑是程序员

使用 php-ext-xlswriter 作为测试参考项目，在测试代码中导出一份 50W行 × 20列的xlsx文件，每个单元格均为固定的字符（26字母），并开启内存优化模式（固定内存）。

示例程序

function getMemoryUsage()
{
    $pid = getmypid();

    exec("ps -e -o%mem,rss,pid | grep $pid", $output);

    $outputArray = explode(' ', $output[0]);

    return (doubleval($outputArray[2] ?? 0) / 1024) . 'MB';
}

$startTime = microtime(true);

$config = ['path' => __DIR__ . '/tests'];
$excel = new \Vtiful\Kernel\Excel($config);

$chars = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ';

$filePath = $excel->constMemory('tutorial.xlsx')
    ->header([
        'test1', 'test2', 'test3', 'test4', 'test5', 'test6', 'test7', 'test8', 'test9', 'test10',
        'test11', 'test12', 'test13', 'test14', 'test15', 'test16', 'test17', 'test18', 'test19', 'test20',
    ]);

$sheetIndex = 1;

for ($index = 0; $index < 500000; $index++) {
    $rowIndex = $index % 1000000;

    if ($index > 0 && $rowIndex === 0) {
        $sheetIndex++;
        $filePath->addSheet('sheet' . $sheetIndex);
    }

    $filePath->insertText($rowIndex + 1, 0, $chars);
    $filePath->insertText($rowIndex + 1, 1, $chars);
    $filePath->insertText($rowIndex + 1, 2, $chars);
    $filePath->insertText($rowIndex + 1, 3, $chars);
    $filePath->insertText($rowIndex + 1, 4, $chars);
    $filePath->insertText($rowIndex + 1, 5, $chars);
    $filePath->insertText($rowIndex + 1, 6, $chars);
    $filePath->insertText($rowIndex + 1, 7, $chars);
    $filePath->insertText($rowIndex + 1, 8, $chars);
    $filePath->insertText($rowIndex + 1, 9, $chars);
    $filePath->insertText($rowIndex + 1, 10, $chars);
    $filePath->insertText($rowIndex + 1, 11, $chars);
    $filePath->insertText($rowIndex + 1, 12, $chars);
    $filePath->insertText($rowIndex + 1, 13, $chars);
    $filePath->insertText($rowIndex + 1, 14, $chars);
    $filePath->insertText($rowIndex + 1, 15, $chars);
    $filePath->insertText($rowIndex + 1, 16, $chars);
    $filePath->insertText($rowIndex + 1, 17, $chars);
    $filePath->insertText($rowIndex + 1, 18, $chars);
    $filePath->insertText($rowIndex + 1, 19, $chars);

    if ($index % 100000 === 0) {
        $endTime = microtime(true);
        echo ($endTime - $startTime) . 'S, line:' . $index . ', 内存:' . getMemoryUsage() . PHP_EOL;
    }
}

$endTime = microtime(true);
echo ($endTime - $startTime) . 'S, line:' . $index . ', 内存:' . getMemoryUsage() . PHP_EOL;

$filePath->output();

$endTime = microtime(true);
echo ($endTime - $startTime) . 'S, line:' . $index . ', 内存:' . getMemoryUsage() . PHP_EOL;

示例代码输出

0.002471923828125S, line:0, 内存:0MB
2.8797290325165S, line:100000, 内存:0MB
5.7618429660797S, line:200000, 内存:0MB
8.5462019443512S, line:300000, 内存:0MB
11.41543006897S, line:400000, 内存:0MB
13.46573890989S, line:500000, 内存:0MB
22.752922058105S, line:500000, 内存:0MB

示例代码火焰图

查找可能优化的点

通过火焰图可以直接看到 strpbrk 函数以及zip压缩占用了过多的 CPU 时间，zip 压缩这个世界难题，本渣无能为力，但是 strpbrk 是 C 标准库提供的函数，心想不应该如此慢，于是复盘上层逻辑：

if (strpbrk(string, "\x01\x02\x03\x04\x05\x06\x07\x08\x0B\x0C"
                "\x0D\x0E\x0F\x10\x11\x12\x13\x14\x15\x16"
                "\x17\x18\x19\x1A\x1B\x1C\x1D\x1E\x1F")) {
    //......
}

此方法如果在内存优化模式下，每写入一个单元格，都会存在一次字符查找、判断。

优化思路

减少此函数被调用的次数，对 string 做 hash （此处暂不考虑哈希冲突），并保存至 Map 或 HashTable 中，如果相同的字符只需要一次检索即刻。
在标准库中寻找更优的字符查找检索函数。
秀发乃身外之物，自行强撸。

如果可以轻松从标准库中找到替代函数，那么也就不会有这篇分享，所以第二个方案到此结束。那么再来看下第一个方案，由于 xlsx 单张工作表可以写入 `1048576 * 16834 个单元格`，如果用 Map 或 HashTable，将会造成非常大的内存浪费，即便使用 bitmap 标记。

SSE2 指令集

引用维基百科：SSE2，全名为Streaming SIMD Extensions 2，是一种IA-32架构的SIMD（单一指令多重数据）指令集。SSE2是在 2001年随着Intel发表第一代Pentium 4处理器也一并推出的指令集。它延伸较早的SSE指令集，而且可以完全取代MMX指令集。在2004年，Intel 再度扩展了SSE2指令为 SSE3 指令集。与 70 条指令的 SSE 相比，SSE2新增了144条指令。在2003年，AMD也在发布AMD64的64位处理器时跟进SSE2指令集。

通过复盘上层逻辑，if 中的条件语句只是过滤某几个特殊控制符，不需要像标准库一样考虑通用性，所以可以通过下面代码来等效实现：

unsigned cha
lxw_exists_control_chars(const char *string)
{
    size_t str_len = strlen(string);

#ifdef __SSE2__
    /* If the CPU supports the SSE2 instruction set, use the SSE2 instruction set to quickly filter. */
    /* Filtering 16 characters at a time. */
    if (str_len >= 16) {
        const __m128i _char_nul = _mm_set1_epi8('\x00');
        const __m128i _char_ht = _mm_set1_epi8('\x09');
        const __m128i _char_lf = _mm_set1_epi8('\x0A');
        const __m128i _char_space = _mm_set1_epi8('\x20');

        while (str_len >= 16) {
            __m128i _tm, _eq;
            __m128i _value = _mm_loadu_si128((__m128i *)string);

            /* There are no control characters in the current string */
            _tm = _mm_max_epu8(_value, _char_space);
            _eq = _mm_cmpeq_epi8(_value, _tm);
            if (_eq[0] == -1 && _eq[1] == -1)
                goto next;

            /* There are control characters in the current string */
            /* \x0B\x0C\x0D\x0E\x0F\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1A\x1B\x1C\x1D\x1E\x1F */
            _tm = _mm_min_epu8(_value, _char_lf);
            _eq = _mm_cmpeq_epi8(_char_lf, _tm);
            if (_eq[0] == -1 && _eq[1] == -1)
                return LXW_TRUE;

            /* Continue \x09 */
            _tm = _mm_min_epu8(_value, _char_ht);
            _eq = _mm_cmpeq_epi8(_char_ht, _tm);
            if (_eq[0] && _eq[1])
                goto next;

            /* There are control character in the current string */
            /* \x01\x02\x03\x04\x05\x06\x07\x08 */
            _tm = _mm_min_epu8(_value, _char_nul);
            _eq = _mm_cmpeq_epi8(_char_nul, _tm);
            if (_eq[0] == -1 && _eq[1] == -1)
                return LXW_TRUE;

            next:

            string += 16;
            str_len -= 16;
        }
    }
#endif

    /* Filter the remaining characters. */
    /* If the SSE2 instruction set is not supported, please use the conventional way to filter. */
    /* But currently all x86 architecture CPUs on the market support the SSE2 instruction set. */
    while (str_len > 0) {
        unsigned char _string = *string;

        if (_string < '\x20' && ((_string > '\x00' && _string < '\x09') || _string > '\x0A')) {
                return LXW_TRUE;
        }

        ++string;
        --str_len;
    }

    return LXW_FALSE;
}

如果字符串长度等于或超过16，则使用 SSE2 进行快速处理，反之使用常规的方式处理，其核心代码只有以下几行：

__m128i _value = _mm_loadu_si128((__m128i *)string);

_tm = _mm_max_epu8(_value, _char_space);
_eq = _mm_cmpeq_epi8(_value, _tm);
if (_eq[0] == -1 && _eq[1] == -1)
    goto next;

_tm = _mm_min_epu8(_value, _char_lf);
_eq = _mm_cmpeq_epi8(_char_lf, _tm);
if (_eq[0] == -1 && _eq[1] == -1)
    return LXW_TRUE;

_tm = _mm_min_epu8(_value, _char_ht);
_eq = _mm_cmpeq_epi8(_char_ht, _tm);
if (_eq[0] && _eq[1])
    goto next;

_tm = _mm_min_epu8(_value, _char_nul);
_eq = _mm_cmpeq_epi8(_char_nul, _tm);
if (_eq[0] == -1 && _eq[1] == -1)
    return LXW_TRUE;

第一块代码

__m128i _value = _mm_loadu_si128((__m128i *)string);

一次加载16个字符到CPU缓存中；

第二块代码

_tm = _mm_max_epu8(_value, _char_space);
_eq = _mm_cmpeq_epi8(_value, _tm);
if (_eq[0] == -1 && _eq[1] == -1)
    goto next;

进行无符号8位整数比较，打包返回最大值（是否大于我们需要查找最大字符的ASCII码），并对结果进行检查，打包返回的最大值是否完全等于刚刚加载的16个字符（等于可以得到结果 -1），如果前后8个字符均相等，则可以判断本次加载的16个字符內不含我们需要找的控制符；

res = _mm_max_epu8(a, b);
a   = 116  230  136  145  101    9  115  116   49  102  106  107  100  108  115   97
b   = 32   32   32   32   32   32   32   32   32   32   32   32   32   32   32   32
res = 116  230  136  145  101   32  115  116   49  102  106  107  100  108  115   97

下方的三块代码和第二块代码类似，只是查找的范围不同而已。

Benchmark 对比

ASCII: 

strpbrk                  , loop: 1000, str len: 9，time:0.000122
lxw_exists_control_chars , loop: 1000, str len: 9，time:0.000020
strpbrk                  , loop: 10000, str len: 9，time:0.001174
lxw_exists_control_chars , loop: 10000, str len: 9，time:0.000201
strpbrk                  , loop: 100000, str len: 9，time:0.011563
lxw_exists_control_chars , loop: 100000, str len: 9，time:0.002018
strpbrk                  , loop: 1000, str len: 26，time:0.000296
lxw_exists_control_chars , loop: 1000, str len: 26，time:0.000059
strpbrk                  , loop: 1000, str len: 52，time:0.000564
lxw_exists_control_chars , loop: 1000, str len: 52，time:0.000057
strpbrk                  , loop: 1000, str len: 78，time:0.000854
lxw_exists_control_chars , loop: 1000, str len: 78，time:0.000081
strpbrk                  , loop: 1000000, str len: 26，time:0.246461
lxw_exists_control_chars , loop: 1000000, str len: 26，time:0.048152
strpbrk                  , loop: 1000000, str len: 52，time:0.455256
lxw_exists_control_chars , loop: 1000000, str len: 52，time:0.046717
strpbrk                  , loop: 1000000, str len: 78，time:0.721552
lxw_exists_control_chars , loop: 1000000, str len: 78，time:0.067716

NON ASCII: 

strpbrk                  , loop: 1000, str len: 162，time:0.001447
lxw_exists_control_chars , loop: 1000, str len: 162，time:0.000072
strpbrk                  , loop: 100000, str len: 162，time:0.156455
lxw_exists_control_chars , loop: 100000, str len: 162，time:0.007992

在我们的特殊场景中，当字符串长度小于16时，与标准库strpbrk相比，性能提高了5倍。随着字符串长度的增加，如果字符串只有ASCII时，最多可以提高10倍。但是如果字符不是ASCII 或者不全是 ASCII，则其性能最多可以提高20倍。