C++ のリングバッファ実装と SIMD の相性をリアルタイム音声処理でベンチマークしてみた

はじめに

リアルタイム音声処理を書いていて、オーディオコールバックが重くなったり、 std::queue と std::mutex ベースの実装でプチプチとノイズが出たりして悩んだことはないでしょうか。この記事では、その裏側で動いているリングバッファというパターンに視点を当てて、 C++ の実装と SIMD の効き方をベンチマーク結果とあわせて整理します。

SIMD とは

SIMD (Single Instruction Multiple Data) は、ひとつの命令で複数のデータに同じ演算を同時に適用することで、スループットを高めるための仕組みです。前回の記事 SIMD とは何か - C++ で音声バッファのゲイン処理を高速化してみる では、線形バッファに対するゲイン処理を題材に、 SIMD の効果を実測しました。

対象読者

• C++ でオーディオコールバックやリアルタイム音声処理を書いていて、 CPU 負荷やプチノイズに悩んでいる方
• スレッド間の音声バッファ共有を std::queue や std::mutex で実装していて、本当にこれでよいのか不安な方
• リングバッファという言葉は知っているが、音声処理の中でどのような役割を持つのかを整理しておきたい方

参考

• SIMD 入門（C 言語 SIMD 入門：目次）
• Intro to practical SIMD for graphics
• Circular Buffer - Baeldung on Computer Science

リアルタイム音声処理とリングバッファ

オーディオコールバックで起きがちな悩み

リアルタイム音声処理のコードを書いていると、次のような状況に出会うことがあります。

• std::queue と std::mutex でスレッド間の音声バッファを受け渡ししていたら、たまにプチプチとノイズが乗るようになってしまった
• オーディオコールバックの中でデコードやメモリアロケーションを行っていて、負荷グラフがギザギザと不安定になり、最悪ドロップアウトが発生する
• ネットワークやファイル読み込みが一瞬詰まっただけで、オーディオの再生が一気に破綻してしまう

表面的には別々のトラブルに見えますが、多くの場合は次の 2 つに集約できます。

• オーディオスレッドの処理時間が、許される時間枠をたびたびオーバーしている
• 音声データの受け渡しがスムーズにできるだけのバッファリングと設計になっていない

この 2 つの課題を同時に緩和するための、定番のデザインパターンがリングバッファです。

リングバッファというパターンの概要

リングバッファ (circular buffer) は、固定長の配列を端と端でつないで輪のように扱うデータ構造です。先頭から末尾まで使い切ったら、次は先頭に戻って上書きしながら使います。

リングバッファには次のような特徴があります。

• メモリを固定長で確保しておき、その範囲内で読み書きを回すため、動的な再確保が不要
• キューとして使う場合でも、要素の詰め直しやシフトが不要
• 読み取り位置と書き込み位置を管理するだけで、先入れ先出しのバッファを作れる

リアルタイム処理では、ヒープやロックに起因する予測しづらい遅延を避けたい場面が多いです。このため、あらかじめ決めた領域の中で完結できるリングバッファは相性が良いといえます。

オーディオ処理におけるリングバッファの役割

音声再生においては、次の 2 つの処理があります。

非リアルタイム側は、ネットワークやデコードが一瞬重くなることがあります。オーディオコールバック側は、数ミリ秒ごとに必ず決められたサンプル数を返さないと、すぐにノイズやドロップアウトになります。そこで、非リアルタイム側でデコードしたサンプルをいったんリングバッファに貯めておき、コールバック側はそこから一定サイズずつ取り出す構成にして、両者の時間的なばらつきを吸収します。

ただし、リングバッファの書き方によっては、パターン自体は正しくても CPU 負荷が高くなったり、 SIMD の恩恵を受けづらくなったりします。

素朴なリングバッファ実装と SIMD の相性

前回の記事 では、単純な線形バッファに対してゲイン処理を行い、 SIMD による高速化を確認しました。線形バッファ版の処理は次のような形でした。

void process_linear_block(float* data, int numSamples, float gain)
{
    for (int i = 0; i < numSamples; ++i) {
        data[i] *= gain;
    }
}

これに対して、リングバッファで同じ処理をしようとすると、多くの方がまず次のようなコードを書くと思います。

void process_ring_naive(float* data, int capacity, int& readPos,
                        int numSamples, float gain)
{
    for (int i = 0; i < numSamples; ++i) {
        int idx = (readPos + i) % capacity;
        data[idx] *= gain;
    }
    readPos = (readPos + numSamples) % capacity;
}

ぱっと見は分かりやすく、動作もしますが、 SIMD の観点では次のような問題があります。

• ループの反復ごとに % capacity を実行するため、インデックス計算が重くなる
• リングの末尾で配列アクセスが分断され、連続した大きなチャンクとしてロードしづらくなる
• コンパイラから見ると、ループの各反復が単純な連続添字ではなくなるため、自動ベクトル化の対象にしづらい

結果として、同じゲイン処理でも、線形バッファ版よりずっと遅くなってしまうことがあります。

SIMD から見たリングバッファ設計のポイント

では、リングバッファを使いつつ SIMD の恩恵を受けるにはどうすればよいでしょうか。たとえば、コアの数値処理はできるだけ単純な線形配列に対して書く方法や、リングバッファ特有の巻き戻り処理を外側の薄いラッパーに閉じ込める方法が考えられます。

リングバッファ上の連続した numSamples 分は、readPos から配列末尾までの区間と、配列先頭から残りの区間の 2 つのチャンクに分けることができます。この 2 つをそれぞれ線形ブロックとして扱うことで、内部のループは process_linear_block のような単純な形のままにできます。

void process_ring_chunked(float* data, int capacity, int& readPos,
                          int numSamples, float gain)
{
    int firstChunk  = std::min(numSamples, capacity - readPos);
    int secondChunk = numSamples - firstChunk;

    // 1 つ目のチャンクは [readPos, readPos + firstChunk)
    process_linear_block(data + readPos, firstChunk, gain);

    // 2 つ目のチャンクは先頭から残り
    if (secondChunk > 0) {
        process_linear_block(data, secondChunk, gain);
    }

    readPos = (readPos + numSamples) % capacity;
}

このように設計すると、リングバッファとしての機能と、 SIMD の効きやすさを両立できます。

リングバッファ実装をベンチマークで比較する

ここからは、実際に C++ でいくつかのリングバッファ実装パターンを書き、ベンチマークした結果を見ていきます。

検証環境

検証環境は次のとおりです。

• CPU: Intel(R) Core(TM) i7-11700F @ 2.50GHz
• コンパイラ: MSVC v143 - VS 2022 C++ x64/x86 ビルドツール
• 言語モード:C++ 17
• ビルド構成: Release x64
• コンパイルオプションの一例は次のとおり
  • /O2 /arch:AVX2 /GL /Oi などの最適化オプション
  • ベクトル化レポート出力のために /Qvec-report:2 を追加

ターゲットはサンプルコードの形なので、 CPU の世代や細かいオプションによって絶対値は変わります。ここでは、あくまでパターン間の相対的な傾向を確認することを目的とします。

ベンチマークコード概要

まず、検証に共通で使うリングバッファ構造を定義します。

struct RingBuffer {
    std::vector<float> data;
    int capacity{};
    int readPos{};
};

RingBuffer makeRingBuffer(int capacity)
{
    RingBuffer rb;
    rb.data.resize(static_cast<std::size_t>(capacity));
    rb.capacity = capacity;
    rb.readPos = 0;

    // 擬似ランダムな初期値を入れる
    for (int i = 0; i < capacity; ++i) {
        rb.data[static_cast<std::size_t>(i)] =
            static_cast<float>((i * 13) % 100) / 100.0f;
    }

    return rb;
}

コアとなる線形バッファ向けの処理は、次のような単純なゲイン処理です。

void process_linear_block(float* data, int numSamples, float gain)
{
    for (int i = 0; i < numSamples; ++i) {
        data[i] *= gain;
    }
}

void process_linear_entry(RingBuffer& rb, int numSamples, float gain)
{
    process_linear_block(rb.data.data(), numSamples, gain);
}

ベンチマーク関数は、同じ処理を多数回実行し、 1 サンプルあたりの平均時間を求める形にしました。

using Func = void(*)(RingBuffer&, int, float);

double benchmark(Func f, RingBuffer rb, int numSamples, float gain)
{
    constexpr int NumIterations = 100000;

    auto start = std::chrono::steady_clock::now();

    for (int i = 0; i < NumIterations; ++i) {
        f(rb, numSamples, gain);
    }

    auto end = std::chrono::steady_clock::now();

    std::chrono::duration<double> diff = end - start;
    double totalSec = diff.count();

    double totalSamples = static_cast<double>(NumIterations) * numSamples;
    double nsPerSample = (totalSec * 1e9) / totalSamples;

    // 最適化で処理が消されないようにチェックサムを取る
    volatile float checksum = 0.0f;
    for (float v : rb.data) {
        checksum += v;
    }
    (void)checksum;

    return nsPerSample;
}

capacity は 4096 サンプル (2 の冪) に固定し、ブロックサイズは 64 / 128 / 256 / 512 サンプルのパターンを試しました。 gain は 0.5f としていますが、性能にはほとんど影響しません。

比較対象とする 4 パターン

比較対象とした実装パターンは次の 4 つです。

線形バッファ版

配列の先頭から numSamples 分だけを処理する、最も単純な形です。

void process_linear_entry(RingBuffer& rb, int numSamples, float gain)
{
    process_linear_block(rb.data.data(), numSamples, gain);
}

このループは、 SIMD による自動ベクトル化が最も効きやすい形です。

素朴なリングバッファ版

読み取り位置 readPos から、 % capacity で巻き戻しながら処理する素朴な実装です。

void process_ring_naive(RingBuffer& rb, int numSamples, float gain)
{
    for (int i = 0; i < numSamples; ++i) {
        int idx = (rb.readPos + i) % rb.capacity;
        rb.data[static_cast<std::size_t>(idx)] *= gain;
    }
    rb.readPos = (rb.readPos + numSamples) % rb.capacity;
}

多くのサンプルコードで見かける形ですが、ループの中に % とインデックス計算が入るため、 SIMD 化の観点では不利です。

ビットマスクを使ったリングバッファ版

capacity を 2 の冪にできる前提で、 % の代わりにビットマスクを使う実装です。

void process_ring_mask(RingBuffer& rb, int numSamples, float gain)
{
    int mask = rb.capacity - 1;

    for (int i = 0; i < numSamples; ++i) {
        int idx = (rb.readPos + i) & mask;
        rb.data[static_cast<std::size_t>(idx)] *= gain;
    }
    rb.readPos = (rb.readPos + numSamples) & mask;
}

% よりは軽量ですが、毎回インデックス計算を行う点は変わりません。

チャンク分割方式のリングバッファ版

リングバッファ上の連続した領域を、最大でもふたつのチャンクに分割し、それぞれを線形バッファとして処理する方式です。

void process_ring_chunked(RingBuffer& rb, int numSamples, float gain)
{
    int firstChunk  = std::min(numSamples, rb.capacity - rb.readPos);
    int secondChunk = numSamples - firstChunk;

    // 1 つ目のチャンク
    process_linear_block(rb.data.data() + rb.readPos, firstChunk, gain);

    // 2 つ目のチャンク
    if (secondChunk > 0) {
        process_linear_block(rb.data.data(), secondChunk, gain);
    }

    rb.readPos = (rb.readPos + numSamples) % rb.capacity;
}

内部で呼び出しているのは線形バッファ版の process_linear_block なので、その部分は線形バッファと同じ条件で SIMD 化されることが期待できます。

測定結果

Release x64 ビルドでベンチマークを実行した結果、 1 サンプルあたりの処理時間は次のようになりました。

capacity,4096
blockSize,linear_ns_per_sample,naive_ns_per_sample,mask_ns_per_sample,chunked_ns_per_sample
64,0.0542188,1.96087,0.973609,0.162484
128,0.0346016,1.7579,0.731492,0.124078
256,0.0304609,1.46162,0.573102,0.0586641
512,0.0266738,1.39959,0.526123,0.041375

表にすると次のようになります。

線形バッファ版を 1.0 としたときの倍率で見ると、次のような傾向になります。

この結果から次のことが分かります。

• 素朴な % ベースのリングバッファ実装は、線形バッファ版と比べて常に数十倍遅くなっています
• capacity を 2 の冪にしてビットマスクに置き換えても、依然として十数倍から二十倍程度の差が残っています
• 一方でチャンク分割方式は、特にブロックサイズが大きくなるにつれて線形バッファに近い性能になり、 blockSize 512 では約 1.6 倍程度に収まっています

つまり、どのパターンも論理的には同じゲイン処理をしているにもかかわらず、実装方法によって性能差が 1 桁どころか 2 桁レベルで変わっていることが分かります。

ベクトル化レポートによる裏付け

次に、 /Qvec-report:2 を有効にした状態でビルドし、コンパイラがどのループをベクトル化したかを確認しました。該当部分のログは次のとおりです。

• process_linear_block のループ

  --- Analyzing function: void __cdecl process_linear_block(float * __ptr64,int,float)
  ... info C5001: loop vectorized
  

• process_ring_naive のループ

  --- Analyzing function: void __cdecl process_ring_naive(struct RingBuffer & __ptr64,int,float)
  ... info C5002: loop not vectorized due to reason '1200'
  

• process_ring_mask のループ

  --- Analyzing function: void __cdecl process_ring_mask(struct RingBuffer & __ptr64,int,float)
  ... info C5002: loop not vectorized due to reason '1200'
  

線形バッファ版の process_linear_block は、自動ベクトル化されていることが分かります。一方で、 % や & mask を含むリングバッファ版のループは、理由コード 1200 によりベクトル化されていません。

理由コード 1200
ループ内で移動されるデータの依存関係により、ベクター化が妨げられています。 ループの各繰り返しが互いに干渉しているため、ループのベクター化により間違った結果が生成されています。自動ベクター化自体によりそのようなデータ依存関係を確実になくすことはできません。
(ベクター化と並列化のメッセージ より)

一方、チャンク分割方式のループは、リングバッファ特有の処理を外側の薄いラッパーに押し出しており、実際にサンプルを走査する部分は process_linear_block の中に閉じ込められています。そのため、チャンク分割版も線形バッファ版と同じく、自動ベクトル化されたループを使えていると考えられます。

このことから、今回の性能差は主に次の 2 つが原因だと解釈できます。

• % や & mask を含むインデックス計算がループの内側にあるため、スカラー演算のオーバーヘッドが大きくなっている
• インデックス計算を排除し、処理を線形バッファに寄せることで、コンパイラによる自動ベクトル化が効きやすくなっている

オーディオコールバックは、数百サンプル単位の小さなブロックを何度も処理するため、この差がそのまま CPU 負荷や余裕時間の差として現れます。

まとめ

本記事では、 C++ におけるリングバッファというパターンが、リアルタイム音声処理でどのような役割を担っているかを整理し、その実装方法によって SIMD の効き方と性能が大きく変わることを、ベンチマーク結果とベクトル化レポートを通じて確認しました。素朴な % ベースの実装やビットマスク実装は分かりやすい一方で、線形バッファ版と比べて十倍から数十倍の性能差が出るケースがあり、チャンク分割方式のようにコア処理を線形配列に寄せる設計が有効であることが分かりました。オーディオコールバックがつらいと感じている場合は、リングバッファの実装とループ構造を見直すことで、 SIMD の恩恵を引き出せる余地がないか、今回の結果を参考に検討してみてください。