001 Notes
ClamAV SIMD Fast Paths For Parser-Heavy Scanning
0. Abstract
| Optimization | Actual measured result |
|---|---|
| String search | 11x speedup |
| Whitespace skip | 6.2x speedup |
| Base64 decode | 6.15 GB/s, roughly 12x over the scalar reference |
| Hash equality | about 2x |
| Hardware CRC32 | about 4x |
1. SIMD String Search
Original Scalar Shape
const char *cli_memstr(const char *haystack, size_t hs, const char *needle, size_t ns)
{
size_t i, s1, s2;
if (!hs || !ns || hs < ns)
return NULL;
if (needle == haystack)
return haystack;
if (ns == 1)
return memchr(haystack, needle[0], hs);
if (needle[0] == needle[1]) {
s1 = 2;
s2 = 1;
} else {
s1 = 1;
s2 = 2;
}
for (i = 0; i <= hs - ns;) {
if (needle[1] != haystack[i + 1]) {
i += s1;
} else {
if ((needle[0] == haystack[i]) &&
!memcmp(needle + 2, haystack + i + 2, ns - 2))
return &haystack[i];
i += s2;
}
}
return NULL;
}
SIMD Replacement
const char *simd_memstr_neon(const char *haystack, size_t hs,
const char *needle, size_t ns)
{
if (ns == 0) return haystack;
if (hs < ns) return NULL;
if (ns == 1) return memchr(haystack, needle[0], hs);
const char first = needle[0];
const char last = needle[ns - 1];
uint8x16_t first_vec = vdupq_n_u8((uint8_t)first);
uint8x16_t last_vec = vdupq_n_u8((uint8_t)last);
size_t i = 0;
size_t end = hs - ns + 1;
while (i + 16 <= end) {
uint8x16_t block_first = vld1q_u8((const uint8_t *)(haystack + i));
uint8x16_t block_last = vld1q_u8((const uint8_t *)(haystack + i + ns - 1));
uint8x16_t eq_first = vceqq_u8(block_first, first_vec);
uint8x16_t eq_last = vceqq_u8(block_last, last_vec);
uint8x16_t candidates = vandq_u8(eq_first, eq_last);
uint64_t mask_lo = vgetq_lane_u64(vreinterpretq_u64_u8(candidates), 0);
uint64_t mask_hi = vgetq_lane_u64(vreinterpretq_u64_u8(candidates), 1);
while (mask_lo) {
int pos = __builtin_ctzll(mask_lo) / 8;
if (memcmp(haystack + i + pos + 1, needle + 1, ns - 2) == 0) {
return haystack + i + pos;
}
mask_lo &= mask_lo - 1;
}
while (mask_hi) {
int pos = 8 + __builtin_ctzll(mask_hi) / 8;
if (memcmp(haystack + i + pos + 1, needle + 1, ns - 2) == 0) {
return haystack + i + pos;
}
mask_hi &= mask_hi - 1;
}
i += 16;
}
for (; i < end; ++i) {
if (haystack[i] == first && haystack[i + ns - 1] == last) {
if (memcmp(haystack + i + 1, needle + 1, ns - 2) == 0) {
return haystack + i;
}
}
}
return NULL;
}
Why The Vector Shape Wins
old:
position -> compare -> branch -> maybe compare -> branch
new:
16 positions -> compare first bytes in parallel
-> compare last bytes in parallel
-> AND masks
-> scalar verify only surviving candidates
Measured Result
| Needle | Scalar GB/s | SIMD GB/s | Speedup |
|---|---|---|---|
stream | 3.33 | 36.61 | 11.0x |
endstream | 3.31 | 36.51 | 11.0x |
obj | 3.48 | 38.45 | 11.0x |
endobj | 3.31 | 37.02 | 11.2x |
/Length | 3.29 | 36.90 | 11.2x |
2. SIMD Whitespace Skip
Original Scalar Shape
static const char *findNextNonWS(const char *q, const char *end)
{
while (q < end &&
(*q == 0 || *q == 9 || *q == 0xa ||
*q == 0xc || *q == 0xd || *q == 0x20))
q++;
return q;
}
SIMD Replacement
const char *simd_skip_ws(const char *ptr, const char *end)
{
if (ptr >= end) return ptr;
while (ptr + 16 <= end) {
uint8x16_t data = vld1q_u8((const uint8_t *)ptr);
uint8x16_t is_null = vceqq_u8(data, vdupq_n_u8(0x00));
uint8x16_t is_tab = vceqq_u8(data, vdupq_n_u8(0x09));
uint8x16_t is_lf = vceqq_u8(data, vdupq_n_u8(0x0A));
uint8x16_t is_ff = vceqq_u8(data, vdupq_n_u8(0x0C));
uint8x16_t is_cr = vceqq_u8(data, vdupq_n_u8(0x0D));
uint8x16_t is_space = vceqq_u8(data, vdupq_n_u8(0x20));
uint8x16_t is_ws = vorrq_u8(is_null, is_tab);
is_ws = vorrq_u8(is_ws, is_lf);
is_ws = vorrq_u8(is_ws, is_ff);
is_ws = vorrq_u8(is_ws, is_cr);
is_ws = vorrq_u8(is_ws, is_space);
uint8_t min_val = vminvq_u8(is_ws);
if (min_val == 0) {
uint8x16_t not_ws = vmvnq_u8(is_ws);
uint64_t mask_lo = vgetq_lane_u64(vreinterpretq_u64_u8(not_ws), 0);
uint64_t mask_hi = vgetq_lane_u64(vreinterpretq_u64_u8(not_ws), 1);
if (mask_lo) {
return ptr + __builtin_ctzll(mask_lo) / 8;
}
return ptr + 8 + __builtin_ctzll(mask_hi) / 8;
}
ptr += 16;
}
while (ptr < end &&
(*ptr == 0 || *ptr == 9 || *ptr == 0xa ||
*ptr == 0xc || *ptr == 0xd || *ptr == 0x20))
ptr++;
return ptr;
}
Why The Vector Shape Wins
scalar:
1 byte -> 6 serial comparisons
SIMD:
16 bytes -> 6 vector comparisons -> OR masks -> skip whole block
scalar: skip 1 byte
SIMD: skip 16 bytes
Measured Result
| Buffer | Scalar GB/s | SIMD GB/s | Speedup |
|---|---|---|---|
256 bytes | 3.36 | 20.98 | 6.2x |
1 KB | 3.23 | 31.59 | 9.8x |
3. SIMD Hash Equality
Scalar Shape
if (memcmp(hash1, hash2, 16) == 0) {
/* match */
}
SIMD Replacement
bool simd_hash_equal(const uint8_t *a, const uint8_t *b, hash_type_t type)
{
size_t size;
switch (type) {
case HASH_MD5: size = 16; break;
case HASH_SHA1: size = 20; break;
case HASH_SHA256: size = 32; break;
default: return memcmp(a, b, 16) == 0;
}
if (size == 16) {
uint8x16_t va = vld1q_u8(a);
uint8x16_t vb = vld1q_u8(b);
uint8x16_t cmp = vceqq_u8(va, vb);
return vminvq_u8(cmp) == 0xFF;
} else if (size == 32) {
uint8x16_t va0 = vld1q_u8(a);
uint8x16_t va1 = vld1q_u8(a + 16);
uint8x16_t vb0 = vld1q_u8(b);
uint8x16_t vb1 = vld1q_u8(b + 16);
uint8x16_t cmp0 = vceqq_u8(va0, vb0);
uint8x16_t cmp1 = vceqq_u8(va1, vb1);
uint8x16_t cmp = vandq_u8(cmp0, cmp1);
return vminvq_u8(cmp) == 0xFF;
}
return memcmp(a, b, size) == 0;
}
Why The Gain Is Smaller
Measured Result
| Operation | Result |
|---|---|
| MD5 equality | 908 M ops/sec |
| SHA-256 equality | 735 M ops/sec |
| Relative speedup | about 2x |
4. Hardware CRC32
Scalar Shape
#include <zlib.h>
uint32_t checksum = crc32(0, data, length);
Hardware-Assisted Shape
#if defined(__aarch64__) && defined(__ARM_FEATURE_CRC32)
#include <arm_acle.h>
uint32_t simd_crc32c(uint32_t crc, const uint8_t *data, size_t len)
{
crc = ~crc;
while (len >= 8) {
crc = __crc32cd(crc, *(const uint64_t *)data);
data += 8;
len -= 8;
}
while (len >= 4) {
crc = __crc32cw(crc, *(const uint32_t *)data);
data += 4;
len -= 4;
}
while (len > 0) {
crc = __crc32cb(crc, *data);
data++;
len--;
}
return ~crc;
}
#endif
Throughput View
| Method | Throughput |
|---|---|
| Software CRC32 | about 500 MB/s |
| Hardware CRC32 | about 20 GB/s |
Measured Result
| Workload | Result |
|---|---|
| CRC32 | about 4x |
5. SIMD Base64 Decode
Scalar Shape
for (i = 0; i < len; i += 4) {
uint8_t a = decode_table[input[i]];
uint8_t b = decode_table[input[i + 1]];
uint8_t c = decode_table[input[i + 2]];
uint8_t d = decode_table[input[i + 3]];
output[j++] = (a << 2) | (b >> 4);
output[j++] = (b << 4) | (c >> 2);
output[j++] = (c << 6) | d;
}
SIMD Replacement
ssize_t simd_base64_decode(const char *input, size_t len,
uint8_t *output, size_t out_capacity)
{
while (in + 16 <= input + len) {
uint8x16_t data = vld1q_u8((const uint8_t *)in);
uint8x16_t decoded = base64_lookup_neon(data);
uint8x16_t packed = base64_pack_neon(decoded);
vst1q_u8(out, packed);
in += 16;
out += 12;
}
/* scalar tail */
}
Why The Vector Shape Wins
scalar:
4 chars -> 3 output bytes
SIMD:
16 chars -> parallel decode -> parallel repack -> 12 output bytes
Measured Result
| Data size | Throughput |
|---|---|
64 bytes | 2.73 GB/s |
1 KB | 3.38 GB/s |
64 KB | 6.15 GB/s |
| Workload | Result |
|---|---|
| Base64 decode | about 6.15 GB/s |
| Relative speedup | roughly 12x over the scalar reference |
6. Benchmark Summary
Expectation Versus Result
| Optimization | Expected speedup | Actual speedup | Status |
|---|---|---|---|
| String search | 8x to 16x | 11x | met |
| Whitespace skip | 4x to 8x | 6.2x | met |
| Base64 decode | 4x to 8x | about 12x | exceeded |
| Hash equality | 2x to 4x | about 2x | met |
| Hardware CRC32 | 4x to 10x | 4x | met |
Directly Measured
Estimated From The Hotspot Mix
| File type | Estimated improvement |
|---|---|
50% to 70% faster | |
| ZIP | 30% to 50% faster |
| MBOX / email | 40% to 60% faster |
| HTML | 30% to 40% faster |
| PE | 20% to 30% faster |
7. Limitations
8. Bottom Line
Do not only optimize the signature engine.
Optimize the parser-shaped byte work around it.
Comments