Kafka のバイナリ検索アルゴリズムの改善

[[356205]]

私は最近、Kafak のソース コードをいくつか研究し、Kafak の改良されたバイナリ検索アルゴリズムを皆さんと共有したいと思います。バイナリ検索はすべてのプログラマーが習得すべき基本的なアルゴリズムであり、Kafka がバイナリ検索をどのように改善して独自のシナリオに適用するかを学ぶことは価値があります。

Kafka はインデックス検索シナリオにバイナリ検索を適用するため、この記事ではまず Kafka のログ構造とインデックスについて簡単に紹介します。 Kafak では、メッセージはログの形式で保存されます。各ログは、実際には複数のログ セグメントを含むフォルダーです。ログ セグメントとは、次の図に示すように、メッセージ ログ ファイルと、同じファイル名 (開始オフセット) を持つ 4 つのインデックス ファイルを指します。

メッセージはメッセージ ログ ファイルに追加形式で保存され、各メッセージには一意のオフセットがあります。メッセージを検索する場合、インデックス ファイルを使用して検索します。クエリがオフセットに基づいている場合、メッセージの検索にはオフセット インデックス ファイルが使用されます。インデックスクエリの議論を容易にするために、以下の議論は変位インデックスの背景に基づいて行われます。変位インデックスの本質は、オフセットと対応するディスクの物理的な位置を格納するバイト配列です。ここで、オフセットとディスクの物理的な位置は 4 バイトに固定されており、次の図に示すように、8 バイトごとにキーと値のペアとして考えることができます。

インデックス構造を明確に理解できたので、この記事のテーマである「バイナリ検索」に正式に進むことができます。インデックス項目の配列とターゲット オフセットを指定すると、次のコードを記述できます。

 private def IndexSlotRangeFor(idx: ByteBuffer, target: Long, searchEntity: IndexSearchEntity): ( Int , Int ) = {
  // _entries はインデックスエントリの数を示します
  // 1. 現在のインデックスが空の場合は、見つからないことを示すために (-1,-1) を直接返します
  _entries == 0 の場合
(-1, -1)を返す
 
  // 2. 検索するオフセットが現在の最小オフセットより小さくないことを確認する
  (compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0)の場合
戻り値(-1, 0) 
   
  // 3. バイナリ検索アルゴリズムを実行してターゲットを見つける
  var lo = 0
  var hi = _entries - 1
  （低<高）の間{
    val mid = ceil(hi / 2.0 + lo / 2.0).toInt
    val found = parseEntry(idx, mid)
    val compareResult = compareIndexEntry(見つかったもの、ターゲット、検索エンティティ)
    比較結果 > 0 の場合
      高 = 中 - 1
そうでない場合 (比較結果 < 0)
      lo = 中
それ以外 
リターン（ミッド、ミッド）
  } 
   
  (lo、if (lo == _entries - 1) -1 else lo + 1)
 }

上記のコードでは、通常のバイナリ検索を使用しています。これがどのような問題を引き起こす可能性があるか見てみましょう。各インデックス項目のサイズは 4B ですが、オペレーティング システムがメモリにアクセスする最小単位はページです。ページは通常 4KB (4096B) で、512 個のインデックス項目が含まれます。インデックス内の指定されたオフセットを見つけることは、オペレーティング システムがメモリにアクセスしたときに指定されたオフセットが配置されているページを見つけることになります。次の図に示すように、インデックス サイズが 13 ページであると仮定します。

Kafka は通常、メッセージを読み取るときに最新のオフセットを読み取るため、クエリされるページは末尾、つまりページ 12 に集中します。次に、上記のコードを組み合わせて、最新のオフセットをクエリするときにどのページにアクセスされるかを確認します。バイナリ検索によれば、ページ 6、9、11、12 が順番に訪問されます。

Kafka が受信するメッセージ数が増えると、インデックス ファイルも 13 ページまで増加します。このとき、バイナリ検索に従って、7、10、12、13 ページが順番にアクセスされます。

訪問したページが前のページとは全く異なっていることがわかります。以前は、ページが 12 しかなかったため、Kafka はインデックスを読み取るときにページ 6、9、11、12 に頻繁にアクセスしていました。ただし、Kafka は速度を上げるために mmap を使用するため、すべての読み取りおよび書き込み操作はオペレーティング システムのページ キャッシュを経由するため、ページ 6、9、11、および 12 はディスクの読み込みを回避するためにページ キャッシュにキャッシュされます。ただし、ページ 13 を増やすと、ページ 7、10、12、13 にアクセスする必要があります。ページ 7 と 10 は長い間アクセスされていないため (最近のオペレーティング システムではページ キャッシュの管理に LRU またはその派生形式が使用されています)、ページ キャッシュ外になっている可能性があり、ページ フォールト割り込みが発生します (スレッドは、ページ キャッシュにキャッシュされていないディスクからのデータの読み込みを待機してブロックされます)。 Kafka の公式テストでは、この状況により数ミリ秒から 1 秒の範囲の遅延が発生します。

上記の状況を考慮して、Kafka はバイナリ検索を改良しました。通常、データはインデックスの末尾から読み取られます。次に、インデックスの最後の 8192B (8KB) を「ホット ゾーン」に、残りを「コールド ゾーン」に分割し、それぞれに対してバイナリ検索を実行します。コードは次のように実装されます。

 private def indexSlotRangeFor(idx: ByteBuffer, target: Long, searchEntity: IndexSearchType): ( Int , Int ) = {
  // 1. 現在のインデックスが空の場合は、見つからないことを示すために (-1,-1) を直接返します
  _entries == 0の場合
(-1, -1)を返す
 
 // メソッドとしてカプセル化されたバイナリ検索
  def binarySearch(開始: Int 、終了: Int ):( Int 、 Int ) = {
    var lo =開始 
    var hi =終了 
    低<高の間{
      中間値 = (下限 + 上限 + 1) >>> 1
      val found = parseEntry(idx, mid)
      val compareResult = compareIndexEntry(見つかったもの、ターゲット、検索エンティティ)
      比較結果が 0 より大きい場合
        高 = 中 - 1
そうでない場合(比較結果 < 0)
        lo = 中
それ以外 
リターン（ミッド、ミッド）
    }
    (lo、if (lo == _entries - 1) -1 else lo + 1)
  } 
 
  /**
   * 2. ホットゾーンの最初のインデックス項目を確認します。 _warmEntries はいわゆる境界線であり、現在は 8192 バイトに固定されています。
   * OffsetIndexの場合、_warmEntries = 8192 / 8 = 1024、つまり1024番目のインデックスエントリ
   * ほとんどのクエリはインデックス項目の末尾に集中しているため、末尾の8192バイトがホットゾーンとして設定されます。
   * クエリ対象がホットゾーンインデックス範囲内にある場合は、ページの中断を避けるためにホットゾーンを直接クエリします。
   */
  val firstHotEntry = Math.最大(0, _entries - 1 - _warmEntries)
  // 3. ターゲットオフセット値がホットゾーンかコールドゾーンかを判断する
  if(compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {
    // ホットゾーンの場合はホットゾーンを検索
binarySearch(firstHotEntry, _entries - 1)を返します
  } 
 
  // 4. 求める変位値が現在の最小変位値より小さくないことを確認する
  if(compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0)
戻り値(-1, 0) 
 
  // 5. 寒冷地帯の場合は寒冷地帯を検索する
  バイナリ検索(0, 最初のホットエントリ)
 }

これの利点は、テールが頻繁にクエリされる場合、テール ページは基本的にページ キャッシュ内にあるため、ページ フォールトによる中断を回避できることです。

前の例を使って見てみましょう。各ページには最大 512 個のインデックス項目が含まれるため、最後の 1024 個のインデックス項目を含むページはホット領域と見なされます。次に、ページ 12 がいっぱいでない場合は、ページ 10、11、および 12 がホット エリアであると判断されます。ページ 12 がちょうどいっぱいになると、ページ 11 と 12 がホット エリアであると判断されます。ページ 13 が増加していっぱいでない場合は、ページ 11、12、および 13 がホット領域であると判断されます。最新のメッセージを読んでいると仮定すると、ホットゾーンでのバイナリ検索は次のようになります。

ページ 12 がいっぱいでない場合は、ページ 11 と 12 が順番にアクセスされます。ページ 12 がいっぱいになると、アクセスしたページにも同じ状況が適用されます。ページ 13 が表示されると、ページ 12 とページ 13 が順番にアクセスされ、長時間アクセスされていないページにはアクセスされなくなるため、ページ フォールトによる中断を効果的に回避できます。

ホットゾーンのサイズが 8192 バイトに設定されている理由については、公式の説明では、これが適切な値であるということになります。

ホット ゾーン内のページ数が 3 以下になるように十分小さい場合、バイナリ検索で使用されるページはページ キャッシュ内にある可能性が高くなります。つまり、設定値が大きすぎると、ホットゾーンのページがページ キャッシュに含まれなくなる可能性があります。

これは、4MB のメッセージ データをカバーするのに十分な大きさ (8192 バイト、または変位インデックスの場合は 1024 個のインデックス エントリ) であり、ほとんどの同期ノードがホット ゾーンでクエリを実行するのに十分です。

最後に一言でまとめると、Kafka インデックスで通常のバイナリ検索を使用すると、ページ欠落の中断が発生し、遅延が発生します。また、ほとんどのクエリが末尾に集中するため、インデックス領域をホット領域とコールド領域に分割し、別々に検索することで、ホット領域のページが可能な限りページキャッシュ内にあるようにすることができ、ページ欠落による中断を回避することができます。