Day 05 · LAB 进程与调度动手 🔧

步骤

# mac 上
top -o cpu

# Linux 上更好用的是 htop（要先 brew install htop / apt install htop）
htop

在 top 顶部找到类似 Load Avg: 1.84, 1.42, 1.36 三个数。这是过去 1 分钟、5 分钟、15 分钟的平均 load。Day 4 第 9 节讲过：

• 1.84 是 1 分钟均值，正在上涨说明负载在变重
• 三个数若是 5.0, 1.5, 1.0，说明刚刚突发了高负载，但 15 分钟前正常
• 三个数若是 1.5, 1.5, 1.5，说明稳定状态

用 sysctl -n hw.ncpu（mac）或 nproc（Linux）看你机器核数。load / 核数 > 1 才算真过载。

# mac top 交互键
o cpu     # 按 CPU 排序
o mem     # 按内存排序
e    # 列出某 PID 的详细
q         # 退出

# htop（Linux/mac brew 装）的快捷键更丰富
F2  设置
F3  搜索
F4  过滤
F9  发送信号（kill 一个进程）
F10 退出

步骤

ps -p $$ -o pid,ppid,user,stat,nice,pri,vsz,rss,command
# $$ 是当前 shell 的 PID

ps aux | sort -k 6 -rn | head -5
# 第 6 列是 RSS（实际物理内存 KB）

# 等价的 Linux 写法（mac 也能用）：
ps -eo pid,user,rss,command --sort=-rss | head -5

# 所有非睡眠的进程
ps aux | awk '$8 !~ /S/ {print}'

# 找 Z(zombie) 进程
ps aux | awk '$8 ~ /Z/ {print}'  # 通常 0 个，但如果你 Python 多进程编程有 bug 就会有

# Linux:
pstree -p $$

# mac 没自带 pstree，用 ps 模拟
ps -ef | grep $$
# 或者 brew install pstree

步骤

cat > /tmp/fork_bench.py <<'EOF'
import os, sys
N = 10000
for _ in range(N):
    pid = os.fork()
    if pid == 0:
        os._exit(0)
    else:
        os.waitpid(pid, 0)
EOF

time python3 /tmp/fork_bench.py

cat > /tmp/thread_bench.py <<'EOF'
import threading, time
N = 100000
e = threading.Event()
count = [0]

def worker():
    for _ in range(N):
        count[0] += 1
        # 主动让出 CPU（不一定真切换，但提示调度器）
        time.sleep(0)

threads = [threading.Thread(target=worker) for _ in range(4)]
t0 = time.perf_counter()
for t in threads: t.start()
for t in threads: t.join()
print(f"took {time.perf_counter()-t0:.2f}s")
EOF

python3 /tmp/thread_bench.py

# Linux 上
cat /proc/$$/status | grep -E "voluntary|nonvoluntary"
# voluntary_ctxt_switches: 自己主动让出（等 IO、调 sleep）
# nonvoluntary_ctxt_switches: 被时间片打断强制切走

# mac 上没有 /proc，但可以用：
ps -o pid,minflt,majflt -p $$
# minflt/majflt 不是切换次数，是 page fault 次数，但能感受调度器的"动作"

步骤（Linux / Linux Docker）

# 简单跟踪
strace -e openat,read,write python3 -c "print('hi')" 2>&1 | head -30

# 统计每种 syscall 的次数
strace -c python3 -c "print('hi')" 2>&1 | tail -20

strace -e openat -f python3 -c "
import json
with open('/tmp/test.json', 'w') as f:
    json.dump({'a': 1}, f)
" 2>&1 | grep -E "openat|tmp" | head -20

这种"看进程到底碰了哪些文件"在安全审计里非常关键——比如你怀疑某个 Python 工具偷偷读了你的 SSH key，strace 一跑直接看清楚。

# mac 上看一个进程打开的文件（不是 syscall 级，但有用）
lsof -p $$ | head -20

# fs_usage 看文件系统级活动（需要 sudo）
sudo fs_usage -w -f filesys python3

步骤（Linux）

taskset -p $$
# 输出: pid 12345's current affinity mask: ff
# ff = 11111111 二进制，说明能在 8 个核全部上跑

taskset -c 0 python3 -c "
while True: pass" &
PID=$!

# 看它真的只在核 0 上
ps -o pid,psr,command -p $PID
# psr 列就是当前在哪个核

# 用 top 按 1 键展开每个核的 CPU%，看是不是只核 0 在 100%
top
# 在 top 里按 1，能看到 CPU0..CPUN 分别的利用率

# 清理
kill $PID

# 起两个一样的任务，一个 nice=0，一个 nice=15
python3 -c "while True: pass" &
NORMAL=$!
nice -n 15 python3 -c "while True: pass" &
LOWPRI=$!

# 把它们绑到同一个核（这样才能看到调度竞争）
taskset -cp 0 $NORMAL
taskset -cp 0 $LOWPRI

sleep 5
# 看两个进程的 CPU%
ps -o pid,ni,pcpu,command -p $NORMAL,$LOWPRI

# 清理
kill $NORMAL $LOWPRI

步骤

cat > /tmp/mini_shell.py <<'EOF'
"""一个最小化的 shell 实现：fork + exec + wait"""
import os, sys

def run_one(cmd):
    pid = os.fork()
    if pid == 0:
        # ===== 子进程 =====
        print(f"  [子] my pid is {os.getpid()}, my parent is {os.getppid()}")
        # 用 exec 替换地址空间
        try:
            os.execvp(cmd[0], cmd)
        except FileNotFoundError:
            print(f"  [子] command not found: {cmd[0]}")
            os._exit(127)
    else:
        # ===== 父进程 =====
        print(f"[父] forked child pid={pid}")
        finished_pid, status = os.waitpid(pid, 0)
        exit_code = os.WEXITSTATUS(status)
        print(f"[父] child {finished_pid} exited with code {exit_code}")

if __name__ == "__main__":
    while True:
        line = input("minish> ").strip()
        if line in ("exit", "quit", ""): break
        run_one(line.split())
EOF

python3 /tmp/mini_shell.py

minish> date
minish> ls /
minish> echo hello
minish> false   # 这个会以 exit code 1 退出
minish> xyz123  # 找不到的命令，child 报错
minish> exit

cat > /tmp/orphan.py <<'EOF'
import os, time, sys
pid = os.fork()
if pid == 0:
    # 子进程
    print(f"  child pid={os.getpid()} parent={os.getppid()}")
    time.sleep(3)
    print(f"  child pid={os.getpid()} now parent={os.getppid()}  ← 应该是 init/launchd!")
    sys.exit(0)
else:
    # 父进程立即退出，不 wait
    print(f"父 pid={os.getpid()} 立即退出，留下子进程")
    sys.exit(0)
EOF

python3 /tmp/orphan.py
sleep 4   # 等子进程打印