C++小練習(xí)之高性能實(shí)現(xiàn)字符串分割

更新時(shí)間：2023年10月30日 07:48:04 作者：燭秋

字符串分割是日常工作中比較常見(jiàn)的基礎(chǔ)函數(shù),通常大家會(huì)使用現(xiàn)成的基礎(chǔ)庫(kù),基礎(chǔ)庫(kù)的性能是否是最佳的,本文主要和大家探討一下如何最大限度的提升字符串分割的性能,希望對(duì)大家有所幫助

字符串分割是日常工作中比較常見(jiàn)的基礎(chǔ)函數(shù)，通常大家會(huì)使用現(xiàn)成的基礎(chǔ)庫(kù)，基礎(chǔ)庫(kù)的性能是否是最佳的？本文基于一個(gè)周末小練習(xí)，研究如何最大限度的提升字符串分割的性能。

1、背景

字符串按照分隔符拆成多個(gè)子串在日常工作中很常見(jiàn)，譬如：后臺(tái)服務(wù)對(duì)配置的解析、模型服務(wù)對(duì)輸入特征的拆分、磁盤格式索引文件轉(zhuǎn)內(nèi)存格式等等，通常最簡(jiǎn)單的實(shí)現(xiàn)是使用 boost 庫(kù)：

std::vector<std::string> output_tokens;<br>boost::split(output_tokens, input_sentence, boost::is_any_of(" "), boost::token_compress_off);

這是最簡(jiǎn)單也是性能最差的寫法。在一些頻繁做字符串拆分的場(chǎng)景下，部分開(kāi)發(fā)者會(huì)發(fā)現(xiàn)用 string 會(huì)觸發(fā)字符串拷貝的代價(jià)，于是改為自己使用 string_view，這會(huì)顯著提高性能，除此之外，在使用 string_view 的基礎(chǔ)上仍然有兩個(gè)點(diǎn)需要特別注意：

盡可能的減少分割時(shí)的計(jì)算量
使用 SIMD 指令

本文將對(duì)此做細(xì)致分析。

2、目標(biāo)

針對(duì)單字符分割和任意字符分割兩類場(chǎng)景，本文以代碼案例為主，講解如何寫出高性能的字符串分割函數(shù)。在考慮高性能的情況下，最基本的要求是字符串分割不應(yīng)觸發(fā)拷貝，因此本文實(shí)現(xiàn)的多個(gè)字符串分割函數(shù)，都基于 string_view 消除了拷貝，在此基礎(chǔ)上再進(jìn)行性能優(yōu)化分析。本文兩類場(chǎng)景的函數(shù)簽名如下：

std::vector<std::string_view> SplitString(std::string_view input, char delimiter);
std::vector<std::string_view> SplitString(std::string_view input, std::string_view delimiters);

3、高性能實(shí)現(xiàn)

單字符分割和任意字符分割這兩類場(chǎng)景分開(kāi)介紹。

3.1、單字符分割字符串的5 個(gè)版本

下面提供 5 種單字符分割字符串的實(shí)現(xiàn)，并對(duì)其性能做總結(jié)分析。

（1）版本1-簡(jiǎn)單版本

單字符分割是最常見(jiàn)的場(chǎng)景，譬如用于日志解析、特征解析等，首先我們考慮基于遍歷自行實(shí)現(xiàn)，實(shí)現(xiàn)如下：

std::vector<std::string_view> SplitStringV1(std::string_view input, char delimiter) {
  std::vector<std::string_view> tokens;
  int start_pos = 0;
  int size = 0;
  for (size_t i = 0; i < input.size(); ++i) {
    if (input[i] == delimiter) {
      if (size != 0) {
        tokens.emplace_back(input.data() + start_pos, size);
        size = 0;
      }
      start_pos = i + 1;
    } else {
      ++size;
    }
  }
  if (size > 0) {
    tokens.emplace_back(input.data() + start_pos, size);
  }
  return tokens;
}

這樣的實(shí)現(xiàn)很好理解：遍歷 input 字符串，發(fā)現(xiàn)有分隔符 delimiter 時(shí)，考慮生成結(jié)果子字符串，生成子字符串需要知道起點(diǎn)和長(zhǎng)度，因此定義了 token_start 和 size 兩個(gè)臨時(shí)變量，并在遍歷過(guò)程中維護(hù)這兩臨時(shí)變量。但仔細(xì)觀察這個(gè)函數(shù)的實(shí)現(xiàn)，可以發(fā)現(xiàn)有三個(gè)變量：token_start、size、i，但通常來(lái)說(shuō)，定位一個(gè)子字符串只需要兩個(gè)變量（或者叫游標(biāo)）：start、end 即可，這里存在性能優(yōu)化空間。

（2）版本2-優(yōu)化版本

基于兩個(gè)游標(biāo)的實(shí)現(xiàn)代碼：

std::vector<std::string_view> SplitStringV2(std::string_view input, char delimiter) {
  std::vector<std::string_view> tokens;
  const char* token_start = input.data();
  const char* p = token_start;
  const char* end_pos = input.data() + input.size();
  for (; p != end_pos; ++p) {
    if (*p == delimiter) {
      if (p > token_start) {
        tokens.emplace_back(token_start, p - token_start);
      }
      token_start = p + 1;
      continue;
    }
  }
  if (p > token_start) {
    tokens.emplace_back(token_start, p - token_start);
  }
  return tokens;
}

這里 token_start 作為子串的起始位置，p 作為遞增游標(biāo)，上一份代碼的 size 可以通過(guò) p - token_start 獲得。這個(gè)實(shí)現(xiàn)少維護(hù) size 變量，因此性能更好。

（3）版本3-STL 版本

有時(shí)候，我們也會(huì)考慮使用標(biāo)準(zhǔn)庫(kù)的 find_first_of 函數(shù)實(shí)現(xiàn)，代碼量更少，代碼如下：

std::vector<std::string_view> SplitStringV3(std::string_view input, char delimiter) {
  std::vector<std::string_view> tokens;
  size_t token_start = 0;
  while (token_start < input.size()) {
    auto token_end = input.find_first_of(delimiter, token_start);
    if (token_end > token_start) {
      tokens.emplace_back(input.substr(token_start, token_end - token_start));
    }
    if (token_end == std::string_view::npos) {
      break;
    }
    token_start = token_end + 1;
  }
  return tokens;
}

但上述實(shí)現(xiàn)，性能比我們自己實(shí)現(xiàn)遍歷的 SplitStringV2 版本性能要差，畢竟 find_first_of 的每次查找都要重新初始化其實(shí)位置，相比自己實(shí)現(xiàn)遍歷有性能浪費(fèi)。

（4）版本4-SIMD 最佳版本

字符串分割很重要的一步是逐個(gè)字符對(duì)字符串做比較，是否可以并行比較呢？是可以的，SIMD 指令可以加速這里的比較，當(dāng)前大多數(shù)機(jī)器都已支持 AVX2，但還未普遍支持 AVX512，下面以 AVX2 為例。代碼如下：

// 編譯：g++ split_string_by_char.cc -mavx2 -o split_string_by_char
std::vector<std::string_view> SplitStrintV4(std::string_view input, char delimiter) {
  if (input.size() < 32) {
    return SplitStringV2(input, delimiter);
  }

  std::vector<std::string_view> tokens;

  uint32_t end_pos = input.size() >> 5 << 5;
  __m256i cmp_a = _mm256_set1_epi8(delimiter);  // 8bit的分隔重復(fù)32次擴(kuò)充到256bit
  const char* p = input.data();
  const char* end = p + end_pos;
  uint32_t last_lead_zero = 0;  // 上一輪256bit（32個(gè)字符）處理后剩下的未拷貝進(jìn)結(jié)果集的字符串個(gè)數(shù)
  while (p < end) {
    __m256i cmp_b = _mm256_loadu_si256(reinterpret_cast<const __m256i*>(p));  // 32個(gè)字符加載進(jìn)內(nèi)存
    __m256i cmp = _mm256_cmpeq_epi8(cmp_a, cmp_b);  // 256 bit 一次比較
    uint32_t mask = _mm256_movemask_epi8(cmp);
    if (mask == 0) {
      last_lead_zero += 32;
      p += 32;
      continue;
    }

    // 記錄本次的頭部0個(gè)數(shù)，注：mask的序和字符串序是相反的，所以這里頭部的0對(duì)應(yīng)字符串尾部的不匹配字符
    uint32_t lead_zero = __builtin_clz(mask);

    // 補(bǔ)上一次未拷貝的字符串
    uint32_t tail_zero = __builtin_ctz(mask);
    if (last_lead_zero != 0 || tail_zero != 0) {
      tokens.emplace_back(p - last_lead_zero, last_lead_zero + tail_zero);
    }
    mask >>= (tail_zero + 1);
    p += tail_zero + 1;

    // 補(bǔ)完，繼續(xù)處理
    while (mask != 0) {
      uint32_t tail_zero = __builtin_ctz(mask);
      if (tail_zero != 0) {
        tokens.emplace_back(p, tail_zero);
      }
      mask >>= (tail_zero + 1);
      p += tail_zero + 1;
    }

    last_lead_zero = lead_zero;
    p += lead_zero;
  }

  // 256 bit（32字節(jié)） 對(duì)齊之后剩下的部分
  const char* token_start = input.data() + end_pos - last_lead_zero;
  const char* pp = token_start;
  const char* sentence_end = input.data() + input.size();
  for (; pp != sentence_end; ++pp) {
    if (*pp == delimiter) {
      if (pp > token_start) {
        tokens.emplace_back(token_start, pp - token_start);
      }
      token_start = pp + 1;
      continue;
    }
  }
  if (pp > token_start) {
    tokens.emplace_back(token_start, pp - token_start);
  }
  return tokens;
}

這里使用了 5 個(gè)關(guān)鍵的函數(shù)：

_mm256_loadu_si256，用于加載 256 位（32 字節(jié)）數(shù)據(jù)
_mm256_cmpeq_epi8，用于比較 256 位數(shù)據(jù)
_mm256_movemask_epi8，用于將比較的結(jié)果壓縮到 32 位中，一位代表一個(gè)字節(jié)
__builtin_clz，獲取頭部的 0 bit 個(gè)數(shù)
__builtin_ctz，獲取尾部的 0 bit 個(gè)數(shù)

同時(shí)在代碼實(shí)現(xiàn)上需要考慮多個(gè)細(xì)節(jié)點(diǎn)：

待比較字符串小于 32 字節(jié)，此時(shí)不需要用 SIMD 指令，直接逐個(gè)字符比較
在比較過(guò)程中，
在每一次比較結(jié)果的處理時(shí)，除了逐個(gè)判斷 32 個(gè)字符中的分隔符之外，還要考慮上一輪 32 字節(jié)比較的尾部有部分字符沒(méi)有生成結(jié)果子字符串，要和本輪次的頭部字符串拼成一個(gè)子字符串
多輪的 SIMD 指令比較都完成后，考慮：（1）末尾輪有部分字符沒(méi)有進(jìn)入結(jié)果子字符串；（2）部分字符沒(méi)有對(duì)齊 32 字節(jié)，有尾巴部分的數(shù)據(jù)需要逐個(gè)字符比較垂類

要考慮好這些細(xì)節(jié)點(diǎn)，代碼很復(fù)雜，考慮到 SIMD 指令是單條指令，而我們代碼中對(duì) SIMD 比較完成后的 32 bit（4字節(jié)）的逐個(gè)比較是由多個(gè)單條指令祖?zhèn)鳎虼藝L試讓 SIMD 指令多算一些，而減少對(duì) 32 bit 的輪詢判斷。

（5）版本5- SIMD 較差版本

減少 SIMD 比較指令執(zhí)行完后的 32 bit 遍歷處理，改成每次 SIMD 比較指令完成后，只取頭一個(gè)分隔符的結(jié)果。代碼如下：

// 編譯：g++ split_string_by_char.cc -mavx2 -o split_string_by_char
std::vector<std::string_view> SplitStrintV5(std::string_view input, char delimiter) {
  if (input.size() < 32) {
    return SplitStringV2(input, delimiter);
  }

  std::vector<std::string_view> tokens;
  __m256i cmp_a = _mm256_set1_epi8(delimiter);  // 8bit的分隔重復(fù)32次擴(kuò)充到256bit
  const char* p = input.data();
  uint32_t last_lead_zero = 0;  // 上一輪256bit（32個(gè)字符）處理后剩下的未拷貝進(jìn)結(jié)果集的字符串個(gè)數(shù)
  while (p + 32 < input.data() + input.size()) {
    __m256i cmp_b = _mm256_loadu_si256(reinterpret_cast<const __m256i*>(p));  // 32個(gè)字符加載進(jìn)內(nèi)存
    __m256i cmp = _mm256_cmpeq_epi8(cmp_a, cmp_b);  // 256 bit 一次比較
    uint32_t mask = _mm256_movemask_epi8(cmp);
    if (mask == 0) {
      last_lead_zero += 32;
      p += 32;
      continue;
    }

    // 補(bǔ)上一次未拷貝的字符串
    uint32_t tail_zero = __builtin_ctz(mask);
    if (last_lead_zero != 0 || tail_zero != 0) {
      tokens.emplace_back(p - last_lead_zero, last_lead_zero + tail_zero);
      last_lead_zero = 0;
    }
    p += tail_zero + 1;
  }

  // 不足 256 bit（32字節(jié)）部分
  const char* token_start = p - last_lead_zero;
  const char* sentence_end = input.data() + input.size();
  for (; p != sentence_end; ++p) {
    if (*p == delimiter) {
      if (p > token_start) {
        tokens.emplace_back(token_start, p - token_start);
      }
      token_start = p + 1;
      continue;
    }
  }
  if (p > token_start) {
    tokens.emplace_back(token_start, p - token_start);
  }
  return tokens;
}

在長(zhǎng)文本下評(píng)測(cè)發(fā)現(xiàn)本代碼性能較差。

（6）性能對(duì)比

以上 5 個(gè)版本的代碼，再加上 google 開(kāi)源的 absl 基礎(chǔ)庫(kù) absl::StrSplit 函數(shù)，對(duì) 2K+ 的長(zhǎng)文本循環(huán) 10000 次做壓測(cè)，性能對(duì)比如下：

性能對(duì)比	V1-簡(jiǎn)單版	V2-優(yōu)化版	V3-stl版本	V4-SIMD最佳版本	V5-SIMD較差版本	absl
耗時(shí)(ms)	552	426	508	413	501	709

可以看到，absl::StrSplit 性能最差，可能是因?yàn)槠鋾?huì)返回空字符串導(dǎo)致。性能最佳的是需要復(fù)雜處理的 SIMD 版本，性能次之的是采用兩個(gè)游標(biāo)的非 SIMD 版本，兩者的差異非常小?？紤]到 SIMD 實(shí)現(xiàn)的復(fù)雜性，且性能收益較小，在實(shí)際業(yè)務(wù)場(chǎng)景中，可以在復(fù)雜性和性能收益之間做個(gè)權(quán)衡。

3.2、任意字符分割字符串

任意字符分割字符串也很常見(jiàn)，譬如有些日志支持空格、tab、逗號(hào)任意一個(gè)作為分隔符。和 3.1 章的單字符分割類似，提供三種實(shí)現(xiàn)，并對(duì)其性能做總結(jié)。

（1）STL 版

// 不使用 SIMD，使用標(biāo)準(zhǔn)庫(kù)的查找，是性能最差的方式
std::vector<std::string_view> SplitString(std::string_view input, std::string_view delimiters) {
  if (delimiters.empty()) {
    return {input};
  }
  std::vector<std::string_view> tokens;
  std::string_view::size_type token_start = input.find_first_not_of(delimiters, 0);
  std::string_view::size_type token_end = input.find_first_of(delimiters, token_start);
  while (token_start != std::string_view::npos || token_end != std::string_view::npos) {
    tokens.emplace_back(input.substr(token_start, token_end - token_start));
    token_start = input.find_first_not_of(delimiters, token_end);
    token_end = input.find_first_of(delimiters, token_start);
  }
  return tokens;
}

這個(gè)實(shí)現(xiàn)和 3.1 里的 stl 版本類似，代碼量比較少，性能比較差。

（2）自行遍歷版

// 不使用 SIMD，使用遍歷，是非 SIMD 模式下性能最好的方式
std::vector<std::string_view> SplitStringV2(std::string_view input, std::string_view delimiters) {
  if (delimiters.empty()) {
    return {input};
  }
  std::vector<std::string_view> tokens;
  const char* token_start = input.data();
  const char* p = token_start;
  const char* end_pos = input.data() + input.size();
  for (; p != end_pos; ++p) {
    bool match_delimiter = false;
    for (auto delimiter : delimiters) {
      if (*p == delimiter) {
        match_delimiter = true;
        break;
      }
    }
    if (match_delimiter) {
      if (p > token_start) {
        tokens.emplace_back(token_start, p - token_start);
      }
      token_start = p + 1;
      continue;
    }
  }
  if (p > token_start) {
    tokens.emplace_back(token_start, p - token_start);
  }
  return tokens;
}

自行遍歷版本，和 3.1 里的版本 2 代碼很像，性能也是非 SIMD 模式下最好的。

（3）SIMD 版本

std::vector<std::string_view> SplitStringWithSimd256(std::string_view input, std::string_view delimiters) {
  if (delimiters.empty()) {
    return {input};
  }

  if (input.size() < 32) {
    return SplitStringV2(input, delimiters);
  }

  std::vector<std::string_view> tokens;
  uint32_t end_pos = input.size() >> 5 << 5;
  const char* p = input.data();
  const char* end = p + end_pos;
  uint32_t last_lead_zero = 0;  // 上一輪256bit（32個(gè)字符）處理后剩下的未拷貝進(jìn)結(jié)果集的字符串個(gè)數(shù)
  while (p < end) {
    __m256i cmp_a = _mm256_loadu_si256(reinterpret_cast<const __m256i*>(p));  // 32個(gè)字符加載進(jìn)內(nèi)存
    __m256i cmp_result_a = _mm256_cmpeq_epi8(cmp_a, _mm256_set1_epi8(delimiters[0]));

    for (int i = 1; i != delimiters.size(); ++i) {
      __m256i cmp_result_b = _mm256_cmpeq_epi8(cmp_a, _mm256_set1_epi8(delimiters[i]));
      cmp_result_a = _mm256_or_si256(cmp_result_a, cmp_result_b);
    }
    uint32_t mask = _mm256_movemask_epi8(cmp_result_a);
    if (mask == 0) {
      last_lead_zero += 32;
      p += 32;
      continue;
    }

    // 記錄本次的頭部0個(gè)數(shù)，注：mask的序和字符串序是相反的，所以這里頭部的0對(duì)應(yīng)字符串尾部的不匹配字符
    uint32_t lead_zero = __builtin_clz(mask);

    // 補(bǔ)上一次未拷貝的字符串
    uint32_t tail_zero = __builtin_ctz(mask);
    if (last_lead_zero != 0 || tail_zero != 0) {
      tokens.emplace_back(p - last_lead_zero, last_lead_zero + tail_zero);
    }
    mask >>= (tail_zero + 1);
    p += tail_zero + 1;

    // 補(bǔ)完，繼續(xù)處理
    while (mask != 0) {
      uint32_t tail_zero = __builtin_ctz(mask);
      if (tail_zero != 0) {
        tokens.emplace_back(p, tail_zero);
      }
      mask >>= (tail_zero + 1);
      p += tail_zero + 1;
    }

    last_lead_zero = lead_zero;
    p += lead_zero;
  }

  // 256 bit（32字節(jié)） 對(duì)齊之后剩下的部分
  const char* token_start = input.data() + end_pos - last_lead_zero;
  const char* pp = token_start;
  const char* sentence_end = input.data() + input.size();
  for (; pp != sentence_end; ++pp) {
    bool match_delimiter = false;
    for (auto delimiter : delimiters) {
      if (*pp == delimiter) {
        match_delimiter = true;
        break;
      }
    }
    if (match_delimiter) {
      if (pp > token_start) {
        tokens.emplace_back(token_start, pp - token_start);
      }
      token_start = pp + 1;
      continue;
    }
  }
  if (pp > token_start) {
    tokens.emplace_back(token_start, pp - token_start);
  }
  return tokens;
}

處理任意分隔符和處理單分隔符大部分代碼類似，只有兩部分有變化：

一個(gè)待比較字符串和多個(gè)分隔符比較得到的多個(gè) mask 碼，使用 _mm256_or_si256 做合并
非 SIMD 部分使用 for 循環(huán)遍歷判斷多個(gè)分隔符

（4）性能對(duì)比

以上 3 個(gè)版本的代碼，再加上 google 開(kāi)源的 absl 基礎(chǔ)庫(kù) absl::StrSplit 函數(shù)，對(duì) 2K+ 的長(zhǎng)文本循環(huán) 10000 次做壓測(cè)，性能對(duì)比如下：

性能對(duì)比	stl版	自行遍歷版	SIMD版	absl
耗時(shí)(ms)	768	658	480	1054

absl::StrSplit 性能最差，SIMD 版本有多個(gè)分隔符的場(chǎng)景下，性能提升非常明顯，SIMD 應(yīng)用代碼的高復(fù)雜度付出是值得的。

4、思考

字符串分割是很常見(jiàn)的功能，通常其實(shí)現(xiàn)代碼也很簡(jiǎn)潔，這就使得開(kāi)發(fā)者容易忽略其性能，寫出非最佳性能的代碼，譬如：沒(méi)有使用現(xiàn)代 C++ 中的 string_view、對(duì)遍歷過(guò)程沒(méi)有精細(xì)考慮。通過(guò)精細(xì)的控制計(jì)算量以及應(yīng)用 SIMD 指令可以獲得比較好的收益，特別是 SIMD 指令在任意多分隔符場(chǎng)景下性能優(yōu)化效果非常明顯。

以上就是C++小練習(xí)之高性能實(shí)現(xiàn)字符串分割的詳細(xì)內(nèi)容，更多關(guān)于C++字符串分割的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: