並行、效能調校與 Profiling

並行基礎：race condition、同步原語與 deadlock 高頻

Concurrency vs Parallelism

Rob Pike 的說法：concurrency 是關於結構，parallelism 是關於執行；好的並發結構讓平行變得可能，但不保證。CPython 的 threading 有 concurrency 但（因 GIL）沒有 CPU parallelism；Go 的 goroutine 兩者皆有。

Race condition 與 critical section

Race condition：結果取決於多個執行緒操作的交錯順序。最常見兩種型態：

• Read-modify-write：count += 1 = 讀 → 加 → 寫，兩個 thread 同時讀到 5，各自寫回 6，遺失一次更新。
• Check-then-act：if key not in cache: cache[key] = load()，兩個 thread 都判斷不存在，各自 load 一次（快取擊穿的小型版本）；或「檢查餘額足夠 → 扣款」被並發穿透導致超扣。

Critical section 是存取共享資源、同一時間只能有一個執行者進入的程式區段。原則：臨界區越短越好（不要在持有鎖時做 I/O）、所有存取同一資源的路徑都要用同一把鎖保護。

同步原語一覽

import threading, collections

class BoundedQueue:
    """用 Condition 實作阻塞佇列：面試常要求手寫"""
    def __init__(self, capacity: int):
        self.buf = collections.deque()
        self.cap = capacity
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)    # 兩個 condition 共用同一把鎖
        self.not_empty = threading.Condition(self.lock)

    def put(self, item):
        with self.not_full:
            while len(self.buf) >= self.cap:     # 一定要用 while，不是 if：防止 spurious wakeup 與被別人搶先
                self.not_full.wait()
            self.buf.append(item)
            self.not_empty.notify()

    def get(self):
        with self.not_empty:
            while not self.buf:
                self.not_empty.wait()
            item = self.buf.popleft()
            self.not_full.notify()
            return item

q = BoundedQueue(2)
t = threading.Thread(target=lambda: [print("got", q.get()) for _ in range(3)])
t.start()
for i in range(3):
    q.put(i)
t.join()

Atomic 操作與 CAS

Compare-And-Swap：「如果記憶體位置目前的值等於 expected，就改成 new，回傳是否成功」，由 CPU 一條指令完成（x86 的 LOCK CMPXCHG）。lock-free 演算法以 CAS 迴圈實作：讀舊值 → 計算新值 → CAS，失敗就重試。

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
)

func main() {
	var hits atomic.Int64 // Go 1.19+ 型別化 atomic
	var maxLatency atomic.Int64
	var wg sync.WaitGroup
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func(lat int64) {
			defer wg.Done()
			hits.Add(1)
			for { // CAS 迴圈：更新最大值
				cur := maxLatency.Load()
				if lat <= cur || maxLatency.CompareAndSwap(cur, lat) {
					break
				}
			}
		}(int64(i % 97))
	}
	wg.Wait()
	fmt.Println(hits.Load(), maxLatency.Load()) // 1000 96
}

注意：(1) CAS 在高競爭下大量重試會浪費 CPU；(2) ABA 問題：值從 A 變 B 又變回 A，CAS 誤以為沒變——lock-free stack 中可能導致錯誤，常用版本號（tagged pointer）解決；(3) 分散式版本的 CAS 就是資料庫的 optimistic locking（UPDATE ... SET v = v + 1 WHERE id = ? AND version = ?）與 Redis 的 WATCH/MULTI。

Deadlock：四個必要條件與預防

Coffman 條件，四個同時成立才會死結：

1. Mutual exclusion：資源一次只能被一個執行者持有。
2. Hold and wait：持有資源的同時等待其他資源。
3. No preemption：資源不能被強制搶走，只能由持有者釋放。
4. Circular wait：存在等待環 A 等 B、B 等 A。

預防 = 打破任一條件：

import threading

class Account:
    def __init__(self, aid, balance):
        self.id, self.balance, self.lock = aid, balance, threading.Lock()

def transfer(a: Account, b: Account, amount: int) -> bool:
    first, second = (a, b) if a.id < b.id else (b, a)   # 固定順序，打破 circular wait
    with first.lock, second.lock:
        if a.balance < amount:
            return False
        a.balance -= amount
        b.balance += amount
        return True

Livelock

執行者沒有阻塞、一直在「動」，卻都在互相讓步而沒有進展。例：兩個 thread 都 try_lock 失敗後立刻釋放再重試，步調一致地永遠衝突。解法：隨機化 backoff（jitter）。乙太網路 CSMA/CD 的指數退避就是這個概念。

Starvation（飢餓）

某個執行者長期拿不到資源。例：reader 優先的 RWLock 在讀流量不斷時 writer 永遠等不到；優先權排程中低優先任務永遠不被執行。解法：公平鎖（FIFO）、aging（等越久優先權越高）、writer-preferring RWLock。

Priority inversion

高優先權任務等待被低優先權任務持有的鎖，而低優先權任務又被中優先權任務搶占。解法：priority inheritance。（火星探路者號的經典案例）

Producer-Consumer 與 Thread Pool 大小 進階

Producer-consumer 用一個有界佇列把「產生工作」與「處理工作」解耦：平滑突發流量、讓兩邊用不同並發度、並透過佇列滿時阻塞達成 backpressure（背壓）。無界佇列看似不會阻塞，實際上只是把問題延後成記憶體爆掉與延遲無限增長。

import queue, threading, time, random

SENTINEL = object()
q: "queue.Queue[object]" = queue.Queue(maxsize=100)   # 有界：提供背壓

def producer(n: int):
    for i in range(n):
        q.put({"event_id": i})            # 佇列滿時阻塞；也可用 put(timeout=0.01) 失敗就丟棄/回錯

def consumer(cid: int, batch_size: int = 20):
    batch = []
    while True:
        item = q.get()
        if item is SENTINEL:
            q.task_done()
            break
        batch.append(item)
        if len(batch) >= batch_size:
            time.sleep(random.uniform(0.005, 0.01))   # 模擬批次寫入 DB：一次 I/O 處理 20 筆
            batch.clear()
        q.task_done()
    if batch:
        pass                                          # flush 剩餘
    print(f"consumer {cid} exit")

consumers = [threading.Thread(target=consumer, args=(i,)) for i in range(4)]
for c in consumers: c.start()
producer(1000)
for _ in consumers: q.put(SENTINEL)                   # 每個 consumer 一個毒丸
for c in consumers: c.join()

import asyncio

async def main():
    q: asyncio.Queue[int] = asyncio.Queue(maxsize=100)

    async def worker(name: str):
        while True:
            item = await q.get()
            try:
                await asyncio.sleep(0.001)     # 處理
            finally:
                q.task_done()

    workers = [asyncio.create_task(worker(f"w{i}")) for i in range(8)]
    for i in range(1000):
        await q.put(i)                         # 滿了會 await，自然背壓
    await q.join()                             # 等所有 item 被 task_done
    for w in workers: w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

asyncio.run(main())

Thread pool 要開多大？

• CPU-bound：約等於核心數 N_cores（或 +1 以填補偶發的 page fault 等停頓）。開更多只會增加 context switch 與 cache 失效。Python 要用 process pool 才有意義。
• IO-bound（Brian Goetz《Java Concurrency in Practice》公式）：
  N_threads = N_cores × U_target × (1 + W / C)
  其中 W 是等待時間、C 是計算時間、U_target 是目標 CPU 使用率。例：8 核、每個請求 CPU 計算 5ms、等待下游 45ms，目標使用率 80% → 8 × 0.8 × (1 + 45/5) = 64。
• 用 Little's law 反推：需要的並發數 = 目標吞吐 × 每個請求的平均延遲。要撐 2,000 RPS、下游延遲 50ms → 至少 2000 × 0.05 = 100 個並發中的請求（thread、coroutine 或連線）。
• 真正的上限常常不是 CPU 而是下游：DB 連線數、對方 API 的 rate limit。pool 大小要跟連線池大小、下游容量一起設計，否則只是把排隊從 app 移到 DB。

I/O 模型：blocking、non-blocking、epoll 與 reactor 高頻

select vs poll vs epoll

Level-triggered：只要 buffer 還有資料，每次 epoll_wait 都會回報。Edge-triggered：只在狀態「變化」時回報一次，所以必須搭配 non-blocking fd 並一直讀到 EAGAIN，否則剩下的資料不會再被通知。ET 減少重複通知，但更難寫對。

import selectors, socket

sel = selectors.DefaultSelector()           # Linux 上自動選 EpollSelector

def accept(server: socket.socket):
    conn, _ = server.accept()
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, handle)

def handle(conn: socket.socket):
    data = conn.recv(4096)
    if data:
        conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")   # 簡化：假設 send 不會被阻塞
    else:
        sel.unregister(conn)
        conn.close()

server = socket.socket()
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("127.0.0.1", 8080)); server.listen(1024); server.setblocking(False)
sel.register(server, selectors.EVENT_READ, accept)

def run_forever():                           # 這就是 reactor：一個 loop + 事件分派到 handler
    while True:
        for key, _ in sel.select(timeout=1):
            callback = key.data
            callback(key.fileobj)
# run_forever()

Reactor vs Proactor

• Reactor：通知「可以讀了」，由應用程式自己讀。Nginx、Redis（單執行緒 reactor；6.0 起網路 I/O 可多執行緒）、Netty、asyncio、libuv（Node.js）。
• Multi-reactor：一個 main reactor 負責 accept，多個 sub reactor（每核一個）負責連線讀寫——Nginx 的多 worker process、Netty 的 boss/worker group。
• Proactor：通知「讀完了」。Windows IOCP、Linux io_uring。

C10K 與 C10M

C10K（Dan Kegel，1999）：如何讓一台伺服器同時處理一萬個連線。答案是放棄「一連線一 thread/process」，改用 non-blocking I/O + epoll/kqueue 的事件驅動架構，並調整 OS 參數（fd 上限 ulimit -n、somaxconn、ephemeral port 範圍）。今天 C10K 已是常態。

C10M（Robert Graham，2013）：一千萬並發連線。瓶頸變成 kernel 本身（中斷、系統呼叫、kernel 網路堆疊的鎖與複製），解法是kernel bypass（DPDK、user-space TCP stack）、多佇列網卡與 RSS 把封包分散到各核、CPU pinning、zero-copy、hugepages。了解即可，但它說明了一個觀念：規模每上升一個數量級，瓶頸就會移動到下一層。

效能方法論：USE、RED、Amdahl、Little's law 與 tail latency 高頻

效能調校最大的錯誤是「憑感覺改程式」。正確流程：定義目標（SLO）→ 量測 → 找瓶頸 → 假設 → 改一個變數 → 再量測。以下是幫你系統化找瓶頸的框架。

USE method（看資源）

Brendan Gregg 提出：對每一個資源（CPU、記憶體、磁碟、網卡、以及軟體資源如連線池、thread pool、鎖）檢查：

Utilization

資源忙碌的時間比例（或使用容量比例）。例：CPU 85%、磁碟 %util 95%。

Saturation

排隊中、無法被服務的工作量。例：CPU run queue 長度（vmstat 的 r 大於核心數）、磁碟 queue size、連線池等待數、swap。

Errors

錯誤事件數。例：網卡 drop、TCP retransmit、磁碟錯誤、OOM kill。

重點是 saturation 往往比 utilization 更早反映問題：CPU 平均 60% 但 run queue 一直很長，代表短時間內有突發；連線池使用率 100% 且有大量等待者，延遲必然上升。

RED method（看服務）

Tom Wilkie 提出，針對每個服務/endpoint：Rate（每秒請求數）、Errors（失敗率）、Duration（延遲分布，看 p50/p90/p99/p999）。Google SRE 的 Four Golden Signals 是 latency、traffic、errors、saturation，等於 RED + saturation。實務上：RED 做服務儀表板與告警，USE 做資源層的根因分析。詳見第 9 章。

Amdahl's law：平行化的上限

若程式中可平行的比例為 p，用 N 個處理器的加速比：

Speedup(N) = 1 / ((1 − p) + p / N)

例：p = 0.9（90% 可平行），N = 8 → 1 / (0.1 + 0.1125) ≈ 4.7 倍；N → ∞ 也最多 10 倍。結論：序列部分決定上限。優化時先找「每個請求都必須經過的序列化點」：全域鎖、單一 DB primary、單執行緒的聚合器、同步的 log flush。Amdahl 也適用於單機優化：某函式佔總時間 20%，就算把它優化到 0，總時間也只降 20%——先 profile 找出佔比最大的部分再動手。（Gustafson's law 從另一角度補充：問題規模隨處理器增加而擴大時，可達到的加速更樂觀。）

Little's law：並發、吞吐與延遲

L = λ × W：系統中平均的工作數 L = 平均到達率 λ × 平均停留時間 W。對穩定系統成立，與分布無關。

• 服務 1,000 RPS、平均延遲 200ms → 系統中平均有 200 個同時進行中的請求，你的 worker/連線/coroutine 至少要能容納這麼多。
• 下游 DB 變慢（W 從 5ms 變 50ms），同樣 RPS 下同時佔用的連線數變 10 倍 → 連線池耗盡 → 排隊 → 延遲更高：這就是延遲上升 → 並發上升 → 資源耗盡的連鎖反應。
• 反推容量：一個 worker 同時只能處理 1 個請求、延遲 50ms → 每 worker 上限 20 RPS；要 2,000 RPS 需要 100 個 worker。

Latency vs Throughput

兩者的關係由排隊理論決定：使用率接近 100% 時，排隊時間會急遽上升（M/M/1 模型中平均等待時間與 ρ / (1 − ρ) 成正比，ρ 為使用率；ρ 從 50% 到 90%，該項從 1 變成 9）。所以延遲敏感的服務要保留 headroom，常見目標是 CPU 平均使用率 50–70%，而不是榨到 95%。

Tail latency：為什麼 p99 比平均重要

• 平均值會騙人：99 個請求 10ms、1 個請求 2 秒，平均 30ms 看起來很好，但那 1% 的使用者等了 2 秒。
• 重度使用者更常遇到 tail：一個頁面載入發出 50 個請求，至少一個碰到 p99 的機率是 1 − 0.99⁵⁰ ≈ 39%。
• Fan-out 放大（Dean & Barroso,《The Tail at Scale》）：一個請求要等 N 個後端都回應才能完成，整體延遲由最慢的那個決定。若每個後端有 1% 機率超過 1 秒，fan-out 到 100 個後端時，1 − 0.99¹⁰⁰ ≈ 63% 的請求會超過 1 秒。

import random, statistics

def simulate(fanout: int, n: int = 100_000) -> float:
    """每個後端：99.5% 在 5~15ms，0.5% 是 200ms 的慢請求；回傳整體 p99"""
    def backend():
        return 200.0 if random.random() < 0.005 else random.uniform(5, 15)
    totals = sorted(max(backend() for _ in range(fanout)) for _ in range(n // fanout + 1000))
    return totals[int(len(totals) * 0.99)]

for f in (1, 10, 50, 100):
    print(f"fan-out={f:3d}  整體 p99 ≈ {simulate(f):.0f} ms  慢請求比例 ≈ {1 - 0.995 ** f:.1%}")
# fan-out=1 時 p99 仍在 15ms 左右；fan-out=10 起 p99 就是 200ms

對抗 tail latency 的手法：

• Hedged requests：送出請求後若超過 p95 時間仍未回應，就對另一個副本再送一次，取先回來的結果（只多約 5% 負載）。
• Tied requests：同時送給兩個副本並互相告知，一方開始處理就取消另一方。
• 嚴格的 timeout + 部分結果：超時的分片直接放棄，用已有結果回應（降級）。
• 減少變異來源：GC 停頓、背景任務（compaction、cron）、noisy neighbor、CPU throttling（K8s CPU limit 的 CFS quota 造成的週期性卡頓是常見 p99 殺手）、排隊。

Latency numbers every programmer should know 高頻

源自 Jeff Dean 與 Peter Norvig 的整理，數值隨硬體世代變動，以下為現代伺服器的數量級近似值。面試重點不是背精確數字，而是知道各層之間差幾個數量級，用來做 back-of-the-envelope 估算。

Python Profiling 工具實戰 高頻

先選對工具。Profiler 分兩大類：deterministic / tracing（每次函式呼叫都記錄，精確但 overhead 大，會扭曲結果）與 sampling（每隔固定時間抓一次 call stack，overhead 低、可用於 production，但對極短函式不精確）。

timeit：比較兩種寫法

python -m timeit -s "d = {i: i for i in range(1000)}" "999 in d"
# 50000000 loops, best of 5: 8.35 nsec per loop
python -m timeit -s "l = list(range(1000))" "999 in l"
# 100000 loops, best of 5: 3.09 usec per loop     ← list 線性掃描，慢了約 370 倍

timeit 自動選擇迴圈次數、取 best of N（排除系統雜訊），並暫時關閉 GC。只適合微基準；整體服務的效能要用 profiler 與壓測看，不要因為 microbenchmark 快了 20% 就以為服務會快 20%（Amdahl）。

cProfile + pstats

範例：一個「彙整廣告事件」的報表函式很慢。

# slow_report.py
import json, random

def load_events(n=200_000):
    random.seed(1)
    return [json.dumps({"user_id": f"u{random.randint(1, 5000)}", "campaign_id": f"c{random.randint(1, 50)}",
                        "type": random.choice(["view", "click"]), "price": random.random()}) for _ in range(n)]

def parse(lines):
    return [json.loads(l) for l in lines]

def is_valid(e, blocked):
    return e["user_id"] not in blocked          # blocked 是 list → O(n)

def aggregate(events):
    blocked = [f"u{i}" for i in range(0, 5000, 2)]    # 2500 個被封鎖 user（list！）
    stats = {}
    for e in events:
        if not is_valid(e, blocked):
            continue
        key = e["campaign_id"]
        s = stats.setdefault(key, {"views": 0, "clicks": 0, "spend": 0.0})
        if e["type"] == "click":
            s["clicks"] += 1
        else:
            s["views"] += 1
        s["spend"] += e["price"]
    return stats

# 方法 1：命令列  python -m cProfile -s cumulative slow_report.py
# 方法 2：只 profile 關心的區段（避免把啟動、資料準備算進來）
import cProfile, pstats
from slow_report import load_events, parse, aggregate

lines = load_events()
with cProfile.Profile() as pr:                  # 3.8+ 支援 context manager
    stats = aggregate(parse(lines))
ps = pstats.Stats(pr).strip_dirs().sort_stats("cumulative")
ps.print_stats(8)                               # 前 8 名
ps.sort_stats("tottime").print_stats(5)
ps.print_callers("is_valid")                    # 誰呼叫了它
pr.dump_stats("report.prof")                    # 存檔，可用 snakeviz report.prof 視覺化

         2300178 function calls in 2.484 seconds

   Ordered by: cumulative time
   List reduced from 14 to 8 due to restriction <8>

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
        1    0.053    0.053    1.959    1.959 slow_report.py:14(aggregate)
   200000    1.896    0.000    1.896    0.000 slow_report.py:11(is_valid)
        1    0.033    0.033    0.525    0.525 slow_report.py:8(parse)
   200000    0.082    0.000    0.492    0.000 __init__.py:304(loads)
   200000    0.145    0.000    0.382    0.000 decoder.py:340(decode)
   200000    0.136    0.000    0.136    0.000 decoder.py:351(raw_decode)
   400000    0.062    0.000    0.062    0.000 {method 'match' of 're.Pattern' objects}

ncalls

呼叫次數；遞迴函式會顯示成 7/1（總次數 / 非遞迴的原始呼叫次數）。

tottime

函式自身花的時間，不含它呼叫的子函式。找「真正在燒 CPU 的地方」看這欄。

cumtime

累計時間，含所有子函式。找「哪條呼叫路徑最貴」看這欄。

percall

第一個是 tottime/ncalls，第二個是 cumtime/原始呼叫次數。

怎麼讀：aggregate 的 cumtime 1.959s 佔了大部分，但它的 tottime 只有 0.053s，表示時間都花在子函式；往下看 is_valid 的 tottime 高達 1.896s、被呼叫 20 萬次——一個只做 in 判斷的函式不該這麼貴，原因是 blocked 是 list（O(n) 掃描 2,500 個元素）。修正：blocked = {...} 改成 set，is_valid 降到數十毫秒等級。

line_profiler：定位到行

uv pip install line_profiler
# 在要分析的函式上加 @profile（kernprof 執行時會注入這個名字），然後：
kernprof -l -v slow_report.py

Timer unit: 1e-06 s

Total time: 2.20776 s
Function: aggregate at line 14

Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================
    14                                           @profile
    15                                           def aggregate(events):
    16         1        149.1    149.1      0.0      blocked = [f"u{i}" for i in range(0, 5000, 2)]
    17         1          0.2      0.2      0.0      stats = {}
    18    200001      31507.4      0.2      1.4      for e in events:
    19    200000    2059735.8     10.3     93.3          if not is_valid(e, blocked):
    20     99826      15441.5      0.2      0.7              continue
    21    100174      17255.8      0.2      0.8          key = e["campaign_id"]
    22    100174      25331.0      0.3      1.1          s = stats.setdefault(key, {"views": 0, ...})
    ...

一眼看出第 19 行佔 93.3%、每次 10.3µs（其他行約 0.2µs）。line_profiler 適合在 cProfile 已經縮小到某個函式之後使用。

py-spy：production 的首選

py-spy 用 Rust 寫成，從外部讀取目標 process 的記憶體來重建 Python call stack（Linux 用 process_vm_readv），不需要修改程式、不需要重啟、幾乎不拖慢目標 process。需要 ptrace 權限：root、或 CAP_SYS_PTRACE（容器中要加這個 capability）、或 kernel.yama.ptrace_scope=0。

# 1) top：即時看哪些函式最耗 CPU（像 Linux top 但是看 Python 函式）
sudo py-spy top --pid 12345

# 2) record：取樣 30 秒並輸出火焰圖（SVG），--subprocesses 一併取樣 gunicorn 的 worker
sudo py-spy record -o profile.svg --pid 12345 --duration 30 --subprocesses
#    --native        連 C 擴充（NumPy、orjson）內部的 stack 也顯示
#    --idle          連等待中（idle）的 thread 也取樣：排查「慢但 CPU 不高」時很有用
#    --gil           只取樣持有 GIL 的 thread
#    --format speedscope   輸出給 https://www.speedscope.app 互動檢視
#    --rate 100      取樣頻率（預設 100 Hz）

# 3) dump：印出「此刻」每個 thread 的 stack。process 卡住 / 死結 / CPU 100% 時第一個要做的事
sudo py-spy dump --pid 12345 --locals

# 也可以直接啟動程式並取樣
py-spy record -o profile.svg -- python slow_report.py

$ py-spy dump --pid 12345
Process 12345: /usr/bin/python3 -m gunicorn app:app
Python v3.12.3 (/usr/bin/python3.12)

Thread 12345 (idle): "MainThread"
    wait (threading.py:355)
    acquire (app/budget.py:41)
    reserve_budget (app/budget.py:58)
    handle_bid (app/bid.py:102)
Thread 12360 (active): "ThreadPoolExecutor-0_3"
    is_valid (app/filters.py:12)
    aggregate (app/report.py:18)

讀法：每個 thread 由上到下是「最內層 → 最外層」的呼叫。上例主執行緒卡在 budget.py:41 的鎖上，另一個 thread 正在跑 is_valid——若多次 dump 都看到同樣畫面，就有了鎖競爭或死結的強烈證據。

scalene

uv pip install scalene
scalene --cpu --memory --cli slow_report.py      # 也可產生 HTML 報告

scalene 的特色是逐行區分 Python 時間 vs native 時間 vs system 時間，以及每行的記憶體成長與「copy volume」（Python 與 native 之間的資料複製量）。若某行 Python 時間比例很高，代表值得向量化或改寫；若主要是 native 時間，代表已經在 C 裡跑，再優化 Python 層效益有限。

tracemalloc：找記憶體成長點

import tracemalloc, json

_cache = {}                                   # 沒有上限的「快取」→ 洩漏

def handle_request(req_id: int) -> str:
    payload = {"id": req_id, "features": [0.1] * 200}
    body = json.dumps(payload)
    _cache[req_id] = body                     # 每個 request_id 都不同，永遠不會命中也不會淘汰
    return body

tracemalloc.start(25)                         # 保存 25 層 traceback
for i in range(1_000): handle_request(i)      # 暖機
snap1 = tracemalloc.take_snapshot()
for i in range(1_000, 21_000): handle_request(i)
snap2 = tracemalloc.take_snapshot()

for stat in snap2.compare_to(snap1, "lineno")[:3]:
    print(stat)
top = snap2.compare_to(snap1, "traceback")[0]
for line in top.traceback.format(limit=4, most_recent_first=True):
    print(line)

/usr/lib/python3.14/json/encoder.py:263: size=21.4 MiB (+20.4 MiB), count=21001 (+20000), average=1067 B
leak.py:14: size=624 KiB (+624 KiB), count=19984 (+19984), average=32 B
leak.py:8: size=576 KiB (+540 KiB), count=1 (+0), average=576 KiB
  File "/usr/lib/python3.14/json/encoder.py", line 263
    return _iterencode(o, 0)
  ...
  File "leak.py", line 7
    body = json.dumps(payload)

讀法：在兩個 snapshot 之間，json/encoder.py:263 配置的記憶體增加了 20.4 MiB、多了 20,000 個物件（剛好等於請求數）——每個請求留下一個約 1KB 的字串沒被釋放。lineno 只告訴你「在哪裡配置」（標準庫內部），要用 traceback 模式往上追到我們的程式 leak.py:7，再看那個值被誰持有（_cache，第 8 行的 dict 本身也在成長）。

memray

uv pip install memray
memray run -o out.bin app.py            # 追蹤所有配置（含 C 擴充、numpy）
memray flamegraph out.bin               # 產生 HTML 火焰圖：寬度 = 該路徑在峰值時持有的記憶體
memray flamegraph --leaks out.bin       # 只看結束時仍未釋放的配置（建議搭配 PYTHONMALLOC=malloc）
memray summary out.bin                  # 終端機表格：各函式的 total / own memory
memray attach 12345                     # 附加到執行中的 process（需要 gdb/lldb 與權限）

tracemalloc 只看得到 Python 層透過 PyMem 的配置；memray 攔截 malloc/mmap，連 C 擴充的配置都看得到，適合「RSS 一直漲但 tracemalloc 看不出來」的情境。

火焰圖（flame graph）怎麼讀

• Y 軸是 call stack 深度：最底下是進入點（main），往上是被呼叫的函式。（Icicle graph 則是上下顛倒。）
• X 軸寬度 = 該函式（含其子呼叫）出現在多少比例的樣本中 ≈ 佔用的時間比例。X 軸不是時間軸，同層的框依字母排序，左右位置沒有意義。
• 找「寬的高原」：頂端寬而平的框代表函式自身花很多時間（self time 高）——真正的熱點。底部寬但上面被許多子框切碎，代表時間分散在子呼叫。
• 顏色通常沒有意義（隨機暖色），部分工具用顏色區分 Python / native / kernel。
• Differential flame graph：比較優化前後兩份 profile，紅色變寬、藍色變窄，用來驗證優化與找回歸。
• Off-CPU flame graph：取樣的是「在等待」的時間（鎖、I/O、sleep），用來解釋「CPU 不高但很慢」。py-spy 的 --idle 就是類似概念。

# py-spy --format raw 輸出的 collapsed stack（火焰圖的原始資料），最後一個數字是樣本數
<module> (slow_report.py:36);main (slow_report.py:32);aggregate (slow_report.py:18);is_valid (slow_report.py:12) 180
<module> (slow_report.py:36);main (slow_report.py:31);parse (slow_report.py:9);loads (json/__init__.py:352);... 17
<module> (slow_report.py:36);main (slow_report.py:30);load_events (slow_report.py:5);dumps (json/__init__.py:235);... 17
# 總樣本 243（100Hz ≈ 2.4 秒），is_valid 佔 180/243 ≈ 74%

Linux 層的效能工具 進階

應用層 profiler 看不到 kernel、排程、磁碟與網路。Brendan Gregg 的「Linux 效能分析 60 秒」清單是很好的起手式（細節見第 3 章）：

uptime                 # load average 1/5/15 分鐘：趨勢上升還是下降
dmesg -T | tail        # OOM killer、TCP drop、硬體錯誤
vmstat 1               # 整體 CPU / 記憶體 / swap / run queue
mpstat -P ALL 1        # 每顆 CPU 的使用率：是否有單核被打滿（單執行緒瓶頸、中斷不均）
pidstat 1              # 每個 process 的 CPU
iostat -xz 1           # 磁碟
free -m                # 記憶體（看 available，不是 free）
sar -n DEV 1           # 網卡流量
sar -n TCP,ETCP 1      # TCP 連線建立速率、重傳
top                    # 最後總覽

vmstat / iostat / pidstat 重點欄位

$ vmstat 1
procs -----------memory---------- ---swap-- -----io---- -system-- ------cpu-----
 r  b   swpd   free   buff  cache   si   so    bi    bo   in   cs us sy id wa st
12  0      0 812340  20480 4096000    0    0     0    48 9800 45000 78 15  7  0  0

iostat -xz 1      # 看 r/s w/s（IOPS）、rkB/s wkB/s（吞吐）、r_await w_await（每個 I/O 平均延遲 ms）、
                  # aqu-sz（平均佇列長度，saturation）、%util（對 SSD/NVMe 這種可平行的裝置，100% 不代表飽和）
pidstat -u -r -d -w -p 12345 1
                  # -u CPU、-r 記憶體與 page fault、-d 磁碟 I/O、
                  # -w cswch/s（自願切換：在等 I/O 或鎖）與 nvcswch/s（非自願：時間片用完 → CPU 不夠）
top -H -p 12345   # 看該 process 內每個 thread 的 CPU（找出哪個 thread 在燒）

perf、strace、ltrace

# perf：kernel 內建的 sampling profiler（硬體計數器 + call stack）
perf top -p 12345                         # 即時熱點函式（含 kernel）
perf record -F 99 -g -p 12345 -- sleep 30 # 以 99Hz 取樣 30 秒並記錄 call graph
perf report                               # 互動式檢視
perf stat -p 12345 -- sleep 10            # cycles、instructions、IPC、cache-misses、context-switches
# Python 3.12+ 支援 perf trampoline，讓 perf 看得到 Python 函式名稱：
python -X perf app.py                     # 或設定環境變數 PYTHONPERFSUPPORT=1

# strace：追蹤系統呼叫（基於 ptrace，overhead 非常大，production 只能短時間用）
strace -c -f -p 12345                     # 統計摘要：每種 syscall 的次數、總時間、錯誤數；Ctrl-C 結束
strace -f -tt -T -e trace=network -p 12345   # 每次網路 syscall 的時間戳與耗時
# ltrace：追蹤動態函式庫呼叫（malloc、SSL_read...），同樣 overhead 大
ltrace -c -p 12345

$ strace -c -f -p 12345        （示意輸出）
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
 62.10    1.204512          60     20034           write
 21.33    0.413760          20     20601      3120 recvfrom
  9.87    0.191400          95      2015           epoll_wait
  4.02    0.077900          38      2030           futex

讀法範例：write 被呼叫 2 萬次、佔 62% 的 syscall 時間——可能每筆 log 都直接 write 沒有緩衝、或回應被拆成很多小塊寫出；recvfrom 有大量 errors 通常是 non-blocking socket 的 EAGAIN（正常）；futex 很多代表鎖競爭。eBPF 工具（bcc / bpftrace，例如 offcputime、biolatency、tcpretrans）是 overhead 更低的現代替代方案。

Go 的 pprof 與 benchmark 進階

Go 把 profiling 內建在 runtime 與工具鏈中，是它在後端效能工作上的一大優勢。

package main

import (
	"log"
	"net/http"
	_ "net/http/pprof" // 註冊 /debug/pprof/* handler 到 DefaultServeMux
	"runtime"
)

func main() {
	runtime.SetBlockProfileRate(10_000)   // 開啟 block profile（單位 ns：平均每阻塞 10µs 取樣一次，超過的事件必定記錄），預設關閉
	runtime.SetMutexProfileFraction(100)  // 開啟 mutex 競爭 profile（每 100 次競爭取樣 1 次），預設關閉
	go func() {
		// 只綁 localhost 或內網，切勿對外暴露
		log.Println(http.ListenAndServe("127.0.0.1:6060", nil))
	}()
	select {} // 主服務邏輯...
}

# CPU profile：取樣 30 秒
go tool pprof http://127.0.0.1:6060/debug/pprof/profile?seconds=30
# Heap：inuse_space（目前持有，找洩漏）vs alloc_space（累計配置，找 GC 壓力來源）
go tool pprof -sample_index=alloc_space http://127.0.0.1:6060/debug/pprof/heap
# Goroutine：數量一直漲 → goroutine leak；debug=2 印出每個 goroutine 的完整 stack
curl -s http://127.0.0.1:6060/debug/pprof/goroutine?debug=2 | head -50
# Block / Mutex：在 channel、select、鎖上等待的時間
go tool pprof http://127.0.0.1:6060/debug/pprof/block
go tool pprof http://127.0.0.1:6060/debug/pprof/mutex
# 開啟 Web UI（含火焰圖、source view）
go tool pprof -http=:8081 cpu.prof
# 比較兩份 profile（優化前後）
go tool pprof -diff_base=before.prof after.prof
# 執行追蹤：排程延遲、GC 停頓、goroutine 在各狀態的時間
curl -o trace.out http://127.0.0.1:6060/debug/pprof/trace?seconds=5 && go tool trace trace.out

go test -bench

package benchdemo

import (
	"fmt"
	"strconv"
	"strings"
	"testing"
)

func KeySprintf(campaign, user string, slot int) string {
	return fmt.Sprintf("%s:%s:%d", campaign, user, slot)
}

func KeyBuilder(campaign, user string, slot int) string {
	var b strings.Builder
	b.Grow(len(campaign) + len(user) + 12) // 預先配置，避免多次擴容
	b.WriteString(campaign)
	b.WriteByte(':')
	b.WriteString(user)
	b.WriteByte(':')
	b.WriteString(strconv.Itoa(slot))
	return b.String()
}

func BenchmarkKeySprintf(b *testing.B) {
	for i := 0; i < b.N; i++ { // Go 1.24+ 也可寫 for b.Loop() { ... }
		_ = KeySprintf("c123", "u456789", i)
	}
}

func BenchmarkKeyBuilder(b *testing.B) {
	for i := 0; i < b.N; i++ {
		_ = KeyBuilder("c123", "u456789", i)
	}
}

go test -bench=. -benchmem -count=10 -cpuprofile cpu.out | tee new.txt
benchstat old.txt new.txt          # 統計比較兩組結果（含信賴區間），避免被雜訊誤導
go tool pprof -top -nodecount=8 cpu.out

BenchmarkKeySprintf-24    	12948900	        91.28 ns/op	      32 B/op	       2 allocs/op
BenchmarkKeyBuilder-24    	28105770	        42.85 ns/op	      31 B/op	       1 allocs/op

      flat  flat%   sum%        cum   cum%
     250ms  8.47%  8.47%      250ms  8.47%  runtime.nextFreeFast (inline)
     220ms  7.46% 15.93%      360ms 12.20%  runtime.mallocgcTiny
     210ms  7.12% 23.05%      210ms  7.12%  internal/strconv.formatBase10
     190ms  6.44% 36.27%      680ms 23.05%  fmt.(*pp).doPrintf
     120ms  4.07% 45.42%     1010ms 34.24%  runtime.mallocgc

讀法：-24 是 GOMAXPROCS；第二欄是迭代次數；ns/op 每次操作時間；B/op、allocs/op 是每次操作的 heap 配置（配置次數往往比 CPU 指令更影響效能，因為會增加 GC 壓力）。pprof 中 flat 對應 cProfile 的 tottime、cum 對應 cumtime。這裡 runtime.mallocgc 的 cum 佔 34%，表示記憶體配置本身就是主要成本；用 go build -gcflags=-m 可看哪些變數 escape 到 heap。

常見效能優化手法 高頻

優先順序大致是：不做 > 少做 > 批次做 > 換更快的方式做 > 平行做。先做架構與演算法層級的改善，最後才是微優化。

1. 演算法與資料結構

上面的 profiling 範例就是典型：list 換 set，O(n) → O(1)，效益遠大於任何微優化。其他常見：巢狀迴圈 join 換成 dict 索引（O(N×M) → O(N+M)）、重複排序換成 heap、反覆字串串接換成 join、在迴圈中重複計算不變量。

2. 批次化（batching）與減少往返

import redis
r = redis.Redis()

user_ids = [f"u{i}" for i in range(100)]
# 差：100 次網路往返（RTT 0.3ms → 至少 30ms）
# feats = [r.get(f"feat:{u}") for u in user_ids]

# 好：1 次往返
feats = r.mget([f"feat:{u}" for u in user_ids])

# 好：多種指令用 pipeline 一次送出（非 transaction 可設 transaction=False）
with r.pipeline(transaction=False) as p:
    for u in user_ids:
        p.hgetall(f"profile:{u}")
        p.incr(f"freq:{u}")
    results = p.execute()

同樣概念：SQL 用 executemany/多值 INSERT、Kafka producer 的 linger.ms + batch.size、把逐筆 HTTP 呼叫改成批次 API。取捨：批次增加單筆延遲（要等湊滿或等 linger 時間）與失敗時的影響範圍。

3. 快取

多層快取：process 內（lru_cache、cachetools.TTLCache，最快但每個 worker 一份、一致性差）→ 分散式（Redis/Memcached，一次網路 RTT）→ CDN。要設計 TTL、失效策略、並防範 cache stampede（熱 key 過期瞬間大量請求同時回源：用 single-flight / 鎖、提前非同步刷新、TTL 加隨機抖動）。詳見第 6 章。

4. 連線池（connection pooling）

import httpx

# 差：每個請求建立新 client → 每次都做 DNS + TCP + TLS 握手
# async def call(): async with httpx.AsyncClient() as c: return await c.get(URL)

# 好：整個 process 共用一個 client（內建連線池 + keep-alive）
client = httpx.AsyncClient(
    timeout=httpx.Timeout(0.05, connect=0.02),
    limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
)
# 在 app shutdown 時 await client.aclose()

DB 連線池（SQLAlchemy pool_size/max_overflow、pgbouncer）同理。池太小 → 請求排隊等連線（看「等待取得連線的時間」指標）；池太大 → DB 端連線過多、記憶體與 context switch 增加反而變慢。

5. 降低序列化成本

在高 QPS 服務中，JSON 編解碼常常是 CPU 熱點前幾名。以下是一個約 0.5KB 的 bid-request 風格 dict 在本機的實測（Python 3.13，數值僅供數量級參考）：

import json, timeit
import orjson, msgpack          # uv pip install orjson msgpack

doc = {"id": "req-123",
       "imp": [{"id": str(i), "bidfloor": 0.5 + i, "w": 300, "h": 250} for i in range(5)],
       "user": {"id": "u-99", "segments": list(range(50))},
       "device": {"ua": "Mozilla/5.0 ...", "ip": "1.2.3.4"}}
n = 100_000
for name, enc, dec in [("json", json.dumps, json.loads),
                       ("orjson", orjson.dumps, orjson.loads),      # 回傳 bytes
                       ("msgpack", msgpack.packb, msgpack.unpackb)]:
    b = enc(doc)
    te = timeit.timeit(lambda: enc(doc), number=n) / n * 1e6
    td = timeit.timeit(lambda: dec(b), number=n) / n * 1e6
    print(f"{name:8s} size={len(b):4d}B  encode={te:5.2f}µs  decode={td:5.2f}µs")

選擇原則：對外公開 API 用 JSON（相容性）但換成 orjson 實作；內部服務間高流量用 Protobuf（schema 演進 + 體積小）；快取值或佇列訊息可用 msgpack。也要算上網路與壓縮成本，不只是 CPU。

6. 避免 N+1 查詢

# N+1：1 次查 campaigns + N 次查各自的 creatives
# for c in session.query(Campaign).all():
#     print(c.creatives)          # lazy load：每個 campaign 再打一次 DB

# 解法：eager loading（SQLAlchemy）
from sqlalchemy.orm import selectinload
# campaigns = session.scalars(select(Campaign).options(selectinload(Campaign.creatives))).all()
# → 2 次查詢：SELECT campaigns; SELECT creatives WHERE campaign_id IN (...)

偵測：在測試中計算每個請求的 SQL 數量、在 APM trace 中看到同一個 span 重複幾十次。GraphQL 的 resolver 特別容易產生 N+1，用 DataLoader 批次化（見第 7 章）。

7. 預先計算（precompute）與把工作移出熱路徑

能離線算的就不要在請求中算：預先彙整報表（rollup table、materialized view）、預先計算使用者特徵存入 key-value store、啟動時把設定編譯成查找表（例如把 targeting 規則轉成 bitmap / 倒排索引）。不需要同步完成的工作（寫 log、更新統計、發通知）丟到佇列非同步處理。

8. 向量化（NumPy）

import numpy as np, time

bids = np.random.rand(1_000_000)
weights = np.random.rand(1_000_000)

t = time.perf_counter()
s = 0.0
for b, w in zip(bids, weights):        # Python 迴圈：每個元素都要 dispatch、boxing
    s += b * w
print(f"loop   {time.perf_counter() - t:.3f}s")

t = time.perf_counter()
s2 = float(np.dot(bids, weights))      # 一次呼叫進入 C/BLAS，連續記憶體 + SIMD
print(f"numpy  {time.perf_counter() - t:.4f}s")   # 通常快 2–3 個數量級

9. 原生擴充：Cython、Rust（PyO3）、C

當熱點是無法向量化的 Python 迴圈（例如複雜的逐筆規則判斷）：Cython 加型別宣告編譯成 C；Rust + PyO3/maturin 寫擴充（orjson、pydantic-core、polars、ruff 都是這條路）；或直接用現成的 C 實作函式庫。記得在長時間運算中釋放 GIL（Cython 的 with nogil:、PyO3 的 allow_threads），否則仍無法多執行緒平行。

10. Object pooling、減少配置與 GC 壓力

• Python：大量小物件用 __slots__ / tuple / array；避免在熱路徑建立大量暫時物件（例如每個請求複製大 dict）；啟動後 gc.freeze()；調整 gc.set_threshold。
• Go：sync.Pool 重用 bytes.Buffer、預先配置 make([]T, 0, n)、避免不必要的 []byte↔string 轉換、用 escape analysis 讓變數留在 stack。
• Java：減少短命物件、選擇合適的 GC（G1、ZGC）。
• Pooling 的代價：物件歸還時要重設狀態（否則資料外洩到下一個請求）、池本身的同步成本。

11. 壓縮

壓縮用 CPU 換頻寬與儲存。對外 HTTP 回應用 gzip/brotli（Nginx 層處理）；服務間大批次資料或 Kafka 訊息用 LZ4 / Snappy / zstd（速度快，壓縮比適中）；冷資料儲存用 zstd 高等級或 gzip。小訊息（幾百 bytes）壓縮常常得不償失。

負載測試與 benchmark 設計 進階

wrk -t4 -c200 -d60s --latency http://10.0.0.5:8000/bid      # 4 threads、200 連線、60 秒
wrk2 -t4 -c200 -d60s -R 5000 --latency http://10.0.0.5:8000/bid   # 固定 5000 RPS
hey -z 60s -c 100 -m POST -T application/json -D req.json http://10.0.0.5:8000/bid

Running 1m test @ http://10.0.0.5:8000/bid        （示意輸出）
  4 threads and 200 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    12.31ms    8.02ms 310.55ms   91.20%
    Req/Sec     4.12k   350.11     5.02k    70.50%
  Latency Distribution
     50%   10.80ms
     75%   13.90ms
     90%   18.70ms
     99%   45.10ms
  983412 requests in 1.00m, 160.12MB read
Requests/sec:  16390.20

// k6：open model（固定到達率），並設定 SLO 門檻讓 CI 自動判定失敗
import http from "k6/http";
import { check } from "k6";

export const options = {
  scenarios: {
    steady: {
      executor: "constant-arrival-rate",
      rate: 2000, timeUnit: "1s",      // 每秒 2000 個請求，不受回應速度影響
      duration: "5m",
      preAllocatedVUs: 200, maxVUs: 1000,
    },
  },
  thresholds: {
    http_req_duration: ["p(99)<50"],   // p99 小於 50ms
    http_req_failed: ["rate<0.001"],
  },
};

export default function () {
  const res = http.post("http://10.0.0.5:8000/bid", JSON.stringify({ request_id: "r1", user_id: "u1", floor_price: 0.5, slots: ["s1"] }),
    { headers: { "Content-Type": "application/json" } });
  check(res, { "status 200": (r) => r.status === 200 });
}

# locustfile.py   執行：locust -f locustfile.py --host http://10.0.0.5:8000
from locust import HttpUser, task, between

class Advertiser(HttpUser):
    wait_time = between(0.5, 2)

    @task(5)
    def report(self):
        self.client.get("/reports?campaign_id=c1&range=7d", name="/reports")

    @task(1)
    def update_budget(self):
        self.client.patch("/campaigns/c1", json={"daily_budget": 1000})

Coordinated omission：為什麼你的壓測延遲太好看

Gil Tene 提出的問題。Closed-loop 壓測工具（每個連線「等上一個回應回來才送下一個」）在伺服器卡住 1 秒的期間，不會送出原本應該送的請求。結果只有少數幾個請求記錄到 1 秒延遲，而現實中這 1 秒內持續到達的大量使用者請求全都會被延遲，卻沒有被量到——p99 被嚴重低估。

解法：使用 open model / 固定到達率的工具（wrk2 -R、k6 constant-arrival-rate、vegeta），並以「預定送出時間」而非「實際送出時間」計算延遲；HdrHistogram 也有校正功能。另一個意涵：真實流量是 open model（使用者不會因為你慢就不來），所以 closed model 的結果會過度樂觀。

如何設計一個可信的 benchmark

1. 明確的問題與 SLO：「在 p99 小於 50ms 的前提下，單 Pod 最大可承受 RPS 是多少？」而不是「它多快？」
2. 貼近 production 的環境：相同機型、CPU/memory limit、版本、設定（關掉 debug log）、資料量（空 DB 的 index 都在記憶體裡，結果沒有意義）、真實的請求分布（熱點 key、payload 大小）。
3. 壓測端不能成為瓶頸：監控 load generator 的 CPU 與網路，必要時分散式施壓；不要跟受測服務跑在同一台機器。
4. 暖機：JIT、連線池、快取、Python 的 specializing interpreter 都需要暖機，丟掉前段數據。
5. 階梯式增壓：逐步提高 RPS，畫出「吞吐 vs 延遲」曲線，找出延遲開始陡升的拐點（knee），容量規劃取拐點以下並保留 headroom。
6. 看分布不看平均：記錄 p50/p90/p99/p999 與 max、錯誤率；同時收集伺服器端的 USE 指標，才能解釋瓶頸在哪。
7. 可重現與統計：每組跑多次、只改一個變數、用 benchstat / 信賴區間判斷差異是否顯著。
8. Soak test：中等負載長時間（數小時）跑，才看得到記憶體洩漏、連線洩漏、log 塞滿磁碟。

記憶體洩漏排查流程 深入

1. 確認是洩漏：看長時間的記憶體曲線。持續、與請求數成比例地上升才是洩漏；一開始上升後持平可能只是快取暖機。也要分辨 RSS 與 Python heap：CPython 的 pymalloc 以 arena 管理小物件，只要 arena 中還有一個存活物件就無法歸還 OS，所以「峰值後 RSS 沒有下降」不一定是洩漏（碎片化）。
2. 重現：在 staging 用固定的請求反覆打，記錄每 N 個請求後的 RSS，找出是哪個 endpoint 造成。
3. 定位配置點：tracemalloc 前後 snapshot 比較（見前節）；或用 gc.get_objects() 依型別計數，看哪種型別數量一直增加。
4. 找出持有者：gc.get_referrers(obj) 或 objgraph.show_backrefs() 畫出參考鏈，找到是哪個全域結構抓著不放。
5. Native 層：Python 層沒成長但 RSS 漲，用 memray（含 native stack）、PYTHONMALLOC=malloc 搭配 valgrind/heaptrack，或檢查 C 擴充版本的已知問題。glibc malloc 的多 arena 碎片可用 MALLOC_ARENA_MAX=2 或改用 jemalloc 緩解。
6. 修正與驗證：修正後跑 soak test 確認曲線變平。短期止血：gunicorn --max-requests 定期回收 worker、K8s 設 memory limit 讓 Pod 被重啟（但要了解這只是緩解）。

import gc, collections

def top_types(limit=10):
    """定期呼叫並比較：哪種型別的物件數量一直增加？"""
    counts = collections.Counter(type(o).__name__ for o in gc.get_objects())
    return counts.most_common(limit)

class Session:
    def __init__(self, uid): self.uid = uid

before = dict(top_types(50))
leaky = [Session(i) for i in range(10_000)]   # 模擬洩漏
after = dict(top_types(50))
print({k: after[k] - before.get(k, 0) for k in after if after[k] - before.get(k, 0) > 1000})
# {'Session': 10000}
# 注意：gc.get_objects() 只回傳被 GC 追蹤的容器物件；只含 int/str 的 dict、tuple 會被「取消追蹤」，不會出現在這裡

實戰劇本：某 API p99 從 50ms 變成 800ms，怎麼排查？ 高頻

這是面試中非常典型的開放題。評分重點是有沒有系統化的思路：先止血、再定位、從外到內、用數據排除假設，而不是一開始就猜「是 DB 慢」。

1. 確認與止血（先恢復服務，再找原因）：確認告警是真的（不是監控本身的問題），評估影響範圍與 SLO 燃燒速度。若時間點與部署吻合，先回滾；若是流量暴增則擴容或開啟降級（關閉非必要功能、用快取的舊結果、對低優先流量 load shedding）。
2. 縮小範圍：只有 p99 變差、還是 p50 也變差？（只有 tail 變差 → 偏向偶發事件：GC、鎖、少數慢查詢、少數壞節點、重試；整體都變差 → 偏向資源飽和或程式路徑變慢）。所有 endpoint 還是單一 endpoint？所有 instance 還是少數 instance / 某個 AZ？錯誤率與 timeout 有沒有一起上升？
3. 對齊時間線找變更：部署、設定變更、feature flag、依賴服務部署、資料量成長（某個表剛好跨過記憶體大小、某客戶資料突然變大）、流量模式改變（新客戶、爬蟲、重試風暴）、基礎設施事件（節點遷移、雲端事件）。
4. 用 distributed tracing 分段：挑幾個慢的 trace，看時間花在哪一段——
   • 下游 span 變長：DB 看 slow query log、EXPLAIN（是否有 index 失效、統計資訊過期導致執行計畫改變）、鎖等待、連線數；Redis 看 SLOWLOG、大 key、熱 key、KEYS 指令；外部 API 看對方狀態。
   • span 之間有空白（沒被任何 span 涵蓋的時間）：在 app 內部排隊或被阻塞——event loop 被同步程式卡住、等連線池、GIL 競爭、GC 停頓。
   • 請求進來前就在排隊：比較 LB 記錄的延遲與 app 自身記錄的延遲，差值就是排隊時間（worker 全忙、accept backlog）。
5. 看資源（USE）：CPU 使用率與 throttling（nr_throttled）、run queue、記憶體與 swap、GC 頻率與停頓、連線池使用率與等待數、thread/goroutine 數、網路重傳、檔案描述符。
6. Profiling：在慢的 instance 上 py-spy record --idle（同時看 on-CPU 與 off-CPU）或 py-spy dump 數次看 thread 卡在哪；Go 用 pprof 的 CPU、block、mutex profile。
7. 提出假設並驗證：一次只改一個變數，在 staging 重現或在一個 canary 上驗證，確認 p99 回到基線。
8. 收尾：寫 blameless postmortem；補上缺少的監控（例如連線池等待時間）、告警、壓測案例與防回歸（CI benchmark）。

常見根因清單（依經驗出現頻率）

Appier 情境：在 100ms 預算內完成競價 深入

以廣告平台常見做法為例（不代表 Appier 實際架構）：RTB 流程中，ad exchange 發出 bid request，DSP 的 bidder 必須在期限內（常見約 100ms，含網路往返）回應出價。扣掉跨網路的時間，bidder 內部可用的時間往往只有數十毫秒，且要在極高 QPS 下維持穩定的 tail latency。

延遲預算分配（示意）

設計原則

• Deadline propagation：從收到請求的時間點計算剩餘預算，傳給每個下游呼叫（Go 的 context.WithDeadline、gRPC deadline、Python 的 asyncio.timeout）；預算用完就放棄並回 no-bid，遲到的出價毫無價值。
• 熱路徑上不做阻塞 I/O 以外的同步工作：logging、指標、bid log 一律非同步（記憶體 buffer → 背景 flush → Kafka）。
• 以記憶體換延遲：campaign 設定、targeting 索引、黑名單常駐記憶體，透過訂閱變更事件或定期 reload 更新（double buffering：建好新版本再原子替換參考）。
• 控制 tail：限制每個請求評估的候選數、對慢的下游設 timeout 與 circuit breaker、hedged request、避免大型 GC 停頓、K8s 上避免 CPU throttling。
• Load shedding：過載時主動丟棄一部分請求（快速回 no-bid）比讓所有請求都超時更好——超時的請求浪費了全部資源卻沒產生價值。
• 可觀測性：每個階段的延遲直方圖（p50/p99/p999）、no-bid 原因分類、timeout 率、win rate；在 Go 服務常駐 pprof endpoint、在 Python 服務可隨時 py-spy。

import asyncio, time

BUDGET_S = 0.060                      # bidder 內部可用時間（示意）

async def get_features(uid):  await asyncio.sleep(0.004); return {"ctr_hist": 0.02}
async def predict(feats):     await asyncio.sleep(0.008); return 0.031
async def check_budget(cid):  return True

async def handle_bid(req: dict) -> dict:
    start = time.monotonic()
    deadline = start + BUDGET_S

    def remaining() -> float:
        return max(0.0, deadline - time.monotonic())

    try:
        async with asyncio.timeout(min(0.010, remaining())):         # 特徵查詢子預算
            feats = await get_features(req["user_id"])
    except TimeoutError:
        feats = {"ctr_hist": 0.01}                                  # 降級：使用預設特徵，而不是放棄

    try:
        async with asyncio.timeout(min(0.020, remaining())):        # 模型推論子預算
            ctr = await predict(feats)
    except TimeoutError:
        return {"id": req["id"], "nobid": "model_timeout"}           # 沒有分數就不出價

    if not await check_budget(req["campaign_id"]):
        return {"id": req["id"], "nobid": "budget"}
    price = ctr * 1000 * 0.8                                        # 示意：eCPM 的一部分
    return {"id": req["id"], "price": round(price, 4),
            "elapsed_ms": round((time.monotonic() - start) * 1000, 2)}

print(asyncio.run(handle_bid({"id": "r1", "user_id": "u1", "campaign_id": "c1"})))

面試自測題