パイプライン -- CPUをより高速にするには

クロック周波数を上げずに、CPUは速くなるのでしょうか?答えは「はい」です。パイプライン(Pipeline)技術は、その中でも最も重要な方法の一つです。本講では、Pythonコードとガントチャート風の可視化出力を用いて、パイプラインが命令実行効率を大幅に向上させる仕組みを示します。


日常生活の例え:洗車の流れ作業

あなたが洗車店を経営していると想像してください。経営の仕方は2つあります:

方法1:一人ですべて担当(パイプラインなし)

一人の作業員が最初から最後まで1台の車の全洗浄工程(水洗い、泡立て、ブラシ洗い、拭き取り)を担当します。1台が終わってからでないと次の車を始められません。1台あたり40分かかるため、1時間に洗えるのは1.5台だけです。

方法2:流れ作業で分業(パイプラインあり)

4人の作業員がそれぞれ1つの工程(水洗い、泡立て、ブラシ洗い、拭き取り)を担当します。1台目の車が水洗い工程に入ると、水洗い担当者は2台目の車の水洗いを開始でき、同時に泡立て担当者は1台目の車に泡を付けています。4人が同時に忙しく動き、10分ごとに1台の車が完成します。

2番目の方法は個々の工程を速くしたわけではありません(水洗いはやはり10分のまま)。しかし、異なる車の工程を重ねることで、全体の生産速度は大幅に向上しました。

これがCPUパイプラインの核となる考え方です:異なる命令の異なるステージを重ねて実行するということです。


逐次実行 vs パイプライン実行

逐次実行の問題点

第12講を振り返ると、各命令は4つのステージを経ます:命令フェッチ(IF)、デコード(ID)、実行(EX)、ライトバック(WB)。

に逐次実行(パイプラインなし)モードこのモードでは、CPUは現在の命令の4つのステージをすべて実行し終えるまで、次の命令をフェッチできません。

周期:  1    2    3    4    5    6    7    8    9    10   11   12
I1:   [IF] [ID] [EX] [WB]
I2:                      [IF] [ID] [EX] [WB]
I3:                                          [IF] [ID] [EX] [WB]

3 条指令 = 12 个周期

パイプライン実行の利点

にパイプライン実行モードこのモードでは、命令I1が命令フェッチ(IF)ステージを完了すると、CPUのフェッチユニットは空きます。それを何もせず待たせるよりも、すぐに次の命令I2のフェッチを開始します。

同様に、I1がデコード(ID)を完了して実行ステージに入ると、デコードユニットはI2の処理を開始し、同時にフェッチユニットはI3の処理を開始します。

周期:  1    2    3    4    5    6    7
I1:   [IF] [ID] [EX] [WB]
I2:        [IF] [ID] [EX] [WB]
I3:             [IF] [ID] [EX] [WB]

3 条指令 = 6 个周期(比串行快了一倍!)

重要な数値に注目してください:

  • パイプラインなし:n個の命令 = 4nサイクル
  • パイプラインあり:n個の命令 = 4 + (n-1) = n + 3サイクル(つまりパイプラインが満杯になった後は、各サイクルで1命令が完了する)

nが大きくなると、パイプラインの高速化比は4(パイプラインの段数)に近づきます。

パイプラインは単一の命令の実行を速くするわけではありません(各命令は依然として4サイクル必要です)。しかし、「スループット」——単位時間あたりに完了する命令数——を大幅に向上させます。


パイプラインの「ハザード」問題

パイプラインは優れていますが、コストがかからないわけではありません。パイプラインを停止させる典型的な「ハザード」(Hazard)が3種類あります:

ハザードの種類英語問題の説明身近な例え
構造ハザードStructural Hazard2つのステージが同じハードウェア資源を競合する(例:両方ともメモリへのアクセスが必要)洗車店に高圧洗浄機が1台しかなく、水洗いと最終すすぎの両方でそれを使う必要があるため、順番待ちするしかない
データハザードData Hazard後続の命令が直前の命令の結果に依存しているが、結果がまだライトバックされていない紅焼肉(豚の角煮)を作るには、まず肉を切ってから鍋に入れる必要があります。肉切り担当が切り終わっていなければ、調理担当は待つしかありません。
制御ハザードControl Hazard分岐命令に遭遇したとき、次にどの命令をフェッチすべきか分からないメニューに「お客様が辛いものを注文した場合は、別のレシピに切り替える」と書いてあります。お客様が決めるまで、厨房は何を準備すればよいのかわかりません。

現代のCPUは、さまざまな技術でこれらのハザードを緩和しています:

  • データフォワーディング(Data Forwarding):結果がレジスタにライトバックされるのを待たずに、ALUの出力から直接データを「拝借」して次の命令に渡す
  • 分岐予測(Branch Prediction):分岐方向を予測して、先にフェッチする。当たれば得をし、外れればパイプラインをフラッシュしてやり直す
  • アウト・オブ・オーダー実行(Out-of-Order Execution):プログラムの順序には従わず、「データが準備できた命令から先に実行する」

インタラクティブデモ:パイプラインシミュレータ(ガントチャート風)

以下のPythonコードは、パイプラインありとなしの2つのモードでの命令実行プロセスをシミュレートし、ガントチャート風に出力します。2つの方式の違いを直感的に比較できます。

サンプル

"""
CPUパイプラインシミュレータ(サンプルデモ)
比較:パイプラインなし(逐次) vs パイプラインあり(並行・重複)
各命令の各ステージの実行タイムラインをガントチャート風に出力する
"""


# 命令シーケンスと各段階の略称を定義する
INSTRUCTIONS = ["I1", "I2", "I3", "I4", "I5", "I6", "I7", "I8"]
STAGES = ["IF", "ID", "EX", "WB"]  # 取指、译码、执行、写回
STAGE_NAMES = {"IF": "取指", "ID": "译码", "EX": "执行", "WB": "写回"}


def simulate_serial(instructions, stages):
    """
无流水线模式:每条指令完整执行完 4 个阶段后,才开始下一条
返回: (总周期数, 时间线二维数组)
タイムライン形式: time_grid[サイクル][ステージ列] = 現在実行中の命令名 または None
    """

    num_instr = len(instructions)
    num_stages = len(stages)
    total_cycles = num_instr * num_stages  # n 条指令 × 4 阶段

    # time_grid[cycle][stage_slot] → どの命令がこのサイクルのこのステージにあるか
    time_grid = [[None for _ in range(num_instr)] for _ in range(total_cycles)]

    for i, instr in enumerate(instructions):
        base_cycle = i * num_stages
        for s_idx, stage in enumerate(stages):
            cycle = base_cycle + s_idx
            time_grid[cycle][i] = f"{instr}({stage})"

    return total_cycles, time_grid


def simulate_pipeline(instructions, stages):
    """
パイプライン方式:異なる命令の異なるステージは重ねて実行できる
返回: (总周期数, 时间线二维数组)
    """

    num_instr = len(instructions)
    num_stages = len(stages)
    # 流水线总周期数 = 级数 + (指令数 - 1)
    total_cycles = num_stages + (num_instr - 1)

    time_grid = [[None for _ in range(num_instr)] for _ in range(total_cycles)]

    for i, instr in enumerate(instructions):
        # 命令 i のステージ s はサイクル (i + s) で実行される
        for s_idx, stage in enumerate(stages):
            cycle = i + s_idx
            if cycle < total_cycles:
                time_grid[cycle][i] = f"{instr}({stage})"

    return total_cycles, time_grid


def print_gantt(instructions, stages, time_grid, total_cycles, mode_name):
    """以甘特图风格打印时间线"""
    num_instr = len(instructions)

    # 表头
    header = f"{'周期':>4} "
    for instr in instructions:
        header += f"  {instr:^10}"
    print(header)
    print("-" * (6 + 12 * num_instr))

    for cycle in range(total_cycles):
        row = f"{cycle+1:>4}  "
        for i in range(num_instr):
            cell = time_grid[cycle][i]
            if cell:
                # 给不同阶段配上不同标识
                if "(IF)" in cell:
                    row += f"  [\033[94mIF\033[0m]     "  # 蓝色
                elif "(ID)" in cell:
                    row += f"  [\033[93mID\033[0m]     "  # 黄色
                elif "(EX)" in cell:
                    row += f"  [\033[92mEX\033[0m]     "  # 绿色
                elif "(WB)" in cell:
                    row += f"  [\033[91mWB\033[0m]     "  # 红色
            else:
                row += f"  {'':10}"
        print(row)

    print("-" * (6 + 12 * num_instr))
    print(f"总周期数: {total_cycles}")
    return total_cycles


def print_simple_gantt(instructions, stages, time_grid, total_cycles, mode_name):
    """简易版甘特图(不使用 ANSI 颜色,兼容所有终端)"""
    num_instr = len(instructions)

    print(f"\n{'='*70}")
    print(f" {mode_name}")
    print(f"{'='*70}")

    # 表头
    header = f"{'周期':>5} |"
    for instr in instructions:
        header += f" {instr:^9} |"
    print(header)
    print("-" * (7 + 11 * num_instr))

    for cycle in range(total_cycles):
        row = f"  {cycle+1:>3} |"
        for i in range(num_instr):
            cell = time_grid[cycle][i]
            if cell:
                # 提取阶段标识
                for stage in stages:
                    if f"({stage})" in cell:
                        row += f"  [{stage}]   |"
                        break
            else:
                row += f"          |"
        print(row)

    print("-" * (7 + 11 * num_instr))
    print(f" 总周期数: {total_cycles}")
    return total_cycles


def calculate_metrics(serial_cycles, pipeline_cycles, num_instr, num_stages):
    """计算性能指标"""
    speedup = serial_cycles / pipeline_cycles
    ideal_speedup = num_stages  # 理想加速比 = 流水线级数

    # スループット:各サイクルで完了する命令数
    serial_throughput = num_instr / serial_cycles
    pipeline_throughput = num_instr / pipeline_cycles

    # 流水线效率
    # 所有指令占用槽位总数
    used_slots = num_instr * num_stages
    total_slots = num_instr * pipeline_cycles
    efficiency = (used_slots / total_slots) * 100 if total_slots > 0 else 0

    return {
        'speedup': speedup,
        'ideal_speedup': ideal_speedup,
        'serial_throughput': serial_throughput,
        'pipeline_throughput': pipeline_throughput,
        'efficiency': efficiency,
        'serial_cycles': serial_cycles,
        'pipeline_cycles': pipeline_cycles,
    }


# ===== 主程序 =====
print("=" * 70)
print(CPU パイプラインシミュレータ — ガントチャートスタイル対比 (サンプル デモ))
print("=" * 70)
print(f" 指令序列: {', '.join(INSTRUCTIONS[:8])}")
print(f" 流水线级数: 4 (IF=取指, ID=译码, EX=执行, WB=写回)")
print()

# 1. 无流水线(串行)
serial_cycles, serial_grid = simulate_serial(INSTRUCTIONS, STAGES)
print_simple_gantt(INSTRUCTIONS, STAGES, serial_grid, serial_cycles, "模式 A: 无流水线(串行执行)")

# 2. 有流水线
pipeline_cycles, pipeline_grid = simulate_pipeline(INSTRUCTIONS, STAGES)
print_simple_gantt(INSTRUCTIONS, STAGES, pipeline_grid, pipeline_cycles, "模式 B: 有流水线(4级流水线)")

# 3. 性能比較
metrics = calculate_metrics(serial_cycles, pipeline_cycles, len(INSTRUCTIONS), len(STAGES))

print(f"\n{'='*70}")
print(f" 性能比較分析")
print(f"{'='*70}")
print(f" 指令数量: {len(INSTRUCTIONS)} 条")
print(f" 流水线级数: {len(STAGES)} 级")
print(f"")
print(f" 无流水线总周期: {metrics['serial_cycles']} 周期")
print(f" 有流水线总周期: {metrics['pipeline_cycles']} 周期")
print(f"")
print(f" 实际加速比: {metrics['speedup']:.2f}x")
print(f" 理想加速比: {metrics['ideal_speedup']}x (等于流水线级数)")
print(f"")
print(f" 串行吞吐率: {metrics['serial_throughput']:.3f} 指令/周期")
print(f" 流水线吞吐率: {metrics['pipeline_throughput']:.3f} 指令/周期")
print(f" パイプライン効率: {metrics['efficiency']:.1f}%")
print(f"")
print(f" 結論: {len(STAGES)} 段パイプラインを使用後、同じ 8 命令")
print(f" {metrics['serial_cycles']} サイクルから {metrics['pipeline_cycles']} サイクルに短縮")
print(f" {metrics['speedup']:.1f} 倍高速化!")
print(f"")
print(f" 注意: パイプラインには「充填」と「排出」のオーバーヘッドがあります。")
print(f" 最初の {len(STAGES)-1} サイクルはパイプラインが満たされていません(充填期)。")
print(f" 最後の {len(STAGES)-1} サイクルでパイプラインは徐々に排出されます。")
print(f" 命令数が多いほど、充填/排出の相対的なオーバーヘッドが小さくなり、")
print(f" 加速比は理想値 {len(STAGES)}x に近づきます。")


# 4. 拡張デモ:異なる命令数での加速比の傾向
print(f"\n{'='*70}")
print(f" 拡張分析:異なる命令数での加速比")
print(f"{'='*70}")
print(f" {'命令数':>6} | {'直列サイクル':>8} | {'パイプラインサイクル':>10} | {'加速比':>8} | {'加速比/理想値':>14}")
print(f" {'-'*6}-+-{'-'*8}-+-{'-'*10}-+-{'-'*8}-+-{'-'*14}")

for n in [1, 2, 4, 8, 16, 32, 64, 128, 256]:
    test_instrs = [f"I{i}" for i in range(1, n+1)]
    s_cycles, _ = simulate_serial(test_instrs, STAGES)
    p_cycles, _ = simulate_pipeline(test_instrs, STAGES)
    sp = s_cycles / p_cycles
    ideal = len(STAGES)
    ratio = (sp / ideal) * 100
    print(f" {n:>6} | {s_cycles:>8} | {p_cycles:>10} | {sp:>7.2f}x | {ratio:>13.1f}%")

print()
print(f" 結論: 命令が多いほど、加速比は理想値 {len(STAGES)}x(つまりパイプラインの段数)に近づきます。")
print(f" 命令が 1 つだけの場合、パイプラインにはまったく高速化効果がありません。")
print(f" しかし、現代のプログラムには通常数百万の命令が含まれており、パイプライン効率は 100% に近くなります。")

"上記のコードを実行すると、次のことがわかります。"

  • "ガントチャート比較"":パイプラインなしモードでは、各命令が連続した 4 サイクルを占有し、命令間に大量の空白があります。パイプラインありモードでは、各ステージが密に配置され、空白サイクルはほとんどありません。"
  • "加速比の傾向"":命令が多いほど、加速比はパイプラインの段数(ここでは 4)に近づきます。ただし、命令が 1 つだけの場合は、充填期と排出期のオーバーヘッドが 100% を占めるため、パイプラインにはまったく高速化効果がありません。"
  • "スループット":パイプラインが「満たされた」後(4 サイクル目から)、毎サイクル 1 つの命令を完了できます。これがパイプラインの核心のな価値です。

インタラクティブデモ:パイプラインのガントチャート比較

以下では、Chart.js の横方向スタックバーチャートを使用してガントチャートを描画し、直感的に比較します。"パイプラインなし(直列)"と"パイプラインあり(4段)""2 つのモードにおける、4 つの命令の各ステージの時間分布。各色のブロックは 1 つのステージを表します(IF=命令フェッチ, ID=デコード, EX=実行, WB=ライトバック)。x 軸はクロックサイクルです。"

パイプラインのガントチャート比較(サンプルデモ)

"IF 命令フェッチ"
"ID デコード"
"EX 実行"
"WB ライトバック"

"パイプラインなし(直列実行)— 合計 16 サイクル"

"パイプラインあり(4段パイプライン)— 合計 7 サイクル"

"パイプラインなしの総サイクル"
16
"パイプラインありの総サイクル"
7
"加速比"
2.29x
"スループット向上"
129%
"重要な観察:"
"パイプラインなしモード:各命令が連続した 4 サイクルを占有し、命令間は完全に直列で、4 つの命令に 16 サイクルかかります。"
"パイプラインありモード:異なる命令の異なるステージが重なって実行され、4 つの命令はわずか 7 サイクル(= 4 + (4-1))で完了します。"
パイプラインが「満たされた」後(サイクル 4 以降)、毎サイクル 1 つの命令が完了します——これがスループットが大幅に向上する理由です。

パイプラインの段数ごとの比較

"パイプラインの「段数」が理論上の高速化の上限を決定しますが、段数が多くなるほどハザード問題も深刻になります:"

"CPU モデル""パイプライン段数""特徴"
"古典的な RISC(例:MIPS)""5 段"シンプルで効率的、教育に適しています。古典のな 5 段パイプライン:IF、ID、EX、MEM、WB
Intel Pentium 4(NetBurst)"20〜31 段""超長パイプラインで、高周波数を追求。ただし分岐予測失敗時の損失が大きく、最終的に放棄されました"
"Intel Core シリーズ(2006 年以降)""14〜19 段""短めのパイプライン + 高い IPC で、周波数と効率のバランスを取りました"
Apple M1(2020)"8 段(デコード)""超幅広アーキテクチャ(8 発行)、低周波数・高 IPC、電力効率に非常に優れています"

"わかるように、パイプライン深度はトレードオフです:段数が多すぎると、1 回の分岐予測失敗のコストが大きすぎます;段数が少なすぎると、周波数を上げられません。"

"現代 CPU 設計の主流のトレンドは""「広く浅い」""——適度なパイプライン深度 + 超幅広の発行幅(毎サイクル複数の命令を開始して実行できる)であり、むやみに段数を追求するのではありません。"

"その他の拡張"