モーセルドリブン並列実行とNUMA対応スケジューリング

多コアを「埋めきれない」問題

現代の分析サーバーは数十コア・複数ソケットを備えますが、クエリ実行モデルを速くしただけでは、この並列度を使い切れません。古典的な並列実行は、プランをコア数ぶんに静的分割し、各スレッドに均等量を割り当てます。しかしデータには偏り（スキュー）があり、ある区画だけ行数が多い・選択率が高いと、その担当スレッドだけ走り続け、他は早々に空いて遊びます。最も遅い1スレッドが全体の完了時刻を決める――この末尾の引きずりが、コアを増やしても速くならない主因です。本稿は、これを動的負荷分散で断つ morsel-driven 並列実行と、複数ソケット機で効かせる NUMA 対応スケジューリングの原理を解説します。

morsel：仕事を小片に割る

morsel-driven の核心は、並列の単位をスレッドではなくデータの小片（morsel）にすることです。各入力（テーブルやスキャン結果）を、数千行程度（典型的には1万行前後）の morsel に区切ります。ワーカースレッドは「自分の担当区画」を持たず、共通のディスパッチャから次の morsel を1つずつ取得して処理し、終えたらまた次を取りに来ます。

# ワーカーは担当を固定せず、空いたら次の morsel を取る（pull 型のタスク取得）
worker():
  while (m = dispatcher.next_morsel(pipeline)) != null:
      process(m)          # この morsel ぶんだけ走る（数千行）
  # 取れる morsel が尽きたら、このパイプラインの自分の仕事は終了

固定分割と違い、速いワーカーは多くの morsel を、遅いワーカーは少しの morsel を、結果として自然に消化します。1スレッドが重い morsel に当たっても、それは morsel 1個ぶん（数千行）の遅延に閉じ込められ、全体を引きずりません。粒度の選び方が要で、morsel が大きすぎると末尾の偏りが残り、小さすぎるとディスパッチャ取得や同期の固定費が嵩みます。CPU キャッシュに乗り、かつ1個あたりの処理時間が取得コストを十分上回る大きさが選ばれます。

ワークスティール：遅れを盗んで均す

ディスパッチャから次々 morsel を配る方式は、それ自体が動的負荷分散ですが、より積極的に偏りを均すのが**ワークスティール（work stealing）**です。各ワーカーは自分のローカルなタスクキューを持ち、まずそこから取り出します。自分のキューが空になったワーカーは、まだ仕事が残っている他ワーカーのキューの末尾から morsel を盗み、代わりに実行します。

# 自分のキューが空なら、忙しいワーカーから盗む
worker(self):
  while true:
      m = self.local_queue.pop()
      if m == null:
          victim = pick_busy_worker()      # 残務のあるワーカーを選ぶ
          m = victim.queue.steal()          # 末尾から1つ盗む
          if m == null: break               # どこにも残務がなければ終了
      process(m)

これにより、ある morsel が想定外に重くても、その隣で空いたワーカーが残りを引き受け、全ワーカーがほぼ同時刻に終わるよう収束します。スティールは「キューが空になったときだけ」発生するため、平常時の同期コストはほぼゼロで、偏りが出た瞬間にだけ調整が働く――この競合の局所化がワークスティールの効きどころです。

NUMA：メモリに「距離」がある

複数ソケットのサーバーは **NUMA（Non-Uniform Memory Access）**構成です。各 CPU ソケットは自分に直結したローカルメモリを持ち、他ソケットのメモリ（リモート）へはインターコネクト（UPI など）を経由します。ローカルアクセスは速く、リモートは遅い――この非対称性が NUMA の本質です。morsel-driven のワークスティールを素朴に回すと、ワーカーがソケットをまたいで他所のデータを盗み、大量のリモートアクセスとインターコネクト渋滞を招きます。せっかくの動的均衡が、メモリ帯域のボトルネックに化けるのです。

NUMA 対応スケジューリング：ローカルを優先する

対策は、morsel をそのデータが載っているソケットのワーカーに優先配分することです。入力データはソケットごとに分けて配置し（あるいは初回アクセス時のファーストタッチでローカルに割り付け）、ディスパッチャは各ソケットローカルな morsel キューを持ちます。ワーカーはまず自分と同じソケットの morselを取り、それが尽きて初めて他ソケットから盗みます。

• ローカル優先のディスパッチ: 各ソケットに morsel キューを分け、ワーカーは原則ローカルキューだけを消化する。大半のアクセスがローカルメモリに収まる。
• NUMA ローカルなハッシュ表: 結合のハッシュ表や集約のテーブルも、ソケットごとに分割して各ローカルメモリに置く。プローブはまず自ソケットの区画を引く。
• スティールは最後の手段: ローカルが空のときだけ他ソケットから盗む。均衡（負荷の偏り解消）とローカリティ（リモート回避）が衝突する場面で、ローカリティをやや優先して帯域を守る。

# NUMA 対応の morsel 取得：ローカル優先、空ならリモートを盗む
next_morsel(worker):
  m = local_queue[worker.socket].pop()      # まず自ソケット
  if m != null: return m
  for s in other_sockets_by_distance():     # 近いソケットから順に
      m = local_queue[s].steal()
      if m != null: return m                # リモートだが遊ばせるよりはマシ
  return null

つまり morsel-driven は、平常時はローカリティで帯域を守り、偏りが出た瞬間だけスティールで均衡を取り戻すという二段構えで、負荷分散とメモリ局所性という相反しがちな要求を両立させます。なおデータの読み出し経路もソケットを意識し、ローカルメモリへ直接読み込むことでリモート転送を初手から避けます。

静的分割との比較

まとめ