並列実行パターン｜xargs -P / BG+wait

この記事の狙い

Bash で安全に並列化するための実用パターンを整理します。
主役は xargs -P と バックグラウンド実行 & + wait( -n )。NUL 安全、終了コード集約、出力競合回避、レート制御までを“コピペ可”で。

前提と対象

• Bash 4+、set -Eeuo pipefail 推奨
• 入力（ファイル名など）はNUL 区切りが基本（-print0 / -0 / -d ''）
• stdout=結果、stderr=ログの分離を前提

TL;DR（最小実装）

#!/usr/bin/env bash
set -Eeuo pipefail; set -o pipefail

# 1) find -print0 | xargs -0 -P$(nproc) -n1
find data -type f -name '*.txt' -print0 \
| xargs -0 -P"$(nproc)" -n1 -I{} bash -c '
  set -Eeuo pipefail
  in="$1"; out="${in%.txt}.out"
  awk "{print NR \":\" \$0}" <"$in" >"$out"
' _ "{}"

# 2) BG + wait -n（細かな制御が必要なとき）
concurrency="${concurrency:-4}"; pids=()
spawn(){ ( set -Eeuo pipefail; "$@") & pids+=($!); }
throttle(){ while (( ${#pids[@]} >= concurrency )); do wait -n || exit 1; pids=($(jobs -p)); done; }

for n in {1..20}; do throttle; spawn curl -fsS "https://example.com/id=$n" -o "o/$n.json"; done
# 残りを待つ
for p in "${pids[@]}"; do wait "$p"; done

パターン1：xargs -P（一番簡単で強い）

基本形

producer | xargs -P4 -n1 -I{} bash -c 'set -Eeuo pipefail; do_work "$@"' _ "{}"

ポイント

• -P4 同時実行数、-n1 1 件ずつ、-I{} で必ずクォートして受ける（"$1"）
• 生成側と消費側は NUL でつなぐ

find in -type f -print0 \
| xargs -0 -P"$(nproc)" -n1 -I{} bash -c 'set -Eeuo pipefail; f="$1"; gzip -9 -- "$f"' _ "{}"

出力の競合を避ける

• 各ジョブは専用ファイルに書く（最後に結合）
• 標準出力を混ぜたい時は並列の順序が乱れる前提で設計

find imgs -type f -name '*.jpg' -print0 \
| xargs -0 -P8 -n1 -I{} sh -c '
  set -Eeuo pipefail
  in="$1"; out="out/$(basename "$1").sha256"
  sha256sum -- "$in" >"$out"
' _ "{}"
cat out/*.sha256 > manifest.txt

終了コードの扱い

• xargs は各子の最悪値を返す（どれかが非0で全体が非0）
• 失敗ファイルを記録したい場合はエラー時に stderr へ記録し、呼び出し側で収集

パターン2：バックグラウンド & + wait（細かな制御用）

基本スケルトン

set -Eeuo pipefail; set -o pipefail
limit="${limit:-4}"; pids=(); fails=0

spawn(){ ( set -Eeuo pipefail; "$@" ) & pids+=($!); }
drain_one(){ wait -n || fails=$((fails+1)); pids=($(jobs -p)); }
throttle(){ while (( ${#pids[@]} >= limit )); do drain_one; done; }

for t in "${tasks[@]}"; do throttle; spawn do_task "$t"; done
while (( ${#pids[@]} )); do drain_one; done
(( fails == 0 )) || exit 1

wait のバリエーション

• wait（PID 指定なし）… 全て待つ（終了コードは最後の子プロセス）
• wait PID … 個別に待つ
• wait -n … 何か 1 つ終わるまで待つ（Bash 4.3+）

レート制御（毎秒 N 件）

rate_limit(){
  local per="${1:-5}"   # 1秒あたり
  local slot=0; local t0; t0="$(date +%s)"
  while read -r line; do
    (( slot=(slot+1)%per ))
    (( slot==0 )) && sleep 1
    printf '%s\n' "$line"
  done
}
# 例：URL列を毎秒5件で流す
printf '%s\n' "${urls[@]}" | rate_limit 5 \
| xargs -P4 -n1 -I{} curl -fsS -o /dev/null "{}"

パターン3：トークン（セマフォ）で同時数を制御

mkfifo と FD を利用してシンプルなセマフォを作る。

sem_init(){ mkfifo "$1"; exec 9<>"$1"; rm -f "$1"; for _ in $(seq "${2:-4}"); do printf . >&9; done; }
sem_acq(){ read -r -n1 _ <&9; }
sem_rel(){ printf . >&9; }

sem_init "/tmp/sem.$$" 4
for job in "${jobs[@]}"; do
  sem_acq
  ( set -Eeuo pipefail; run "$job"; sem_rel ) &
done
wait

入力の作り方（NUL 安全）

# ファイル列挙
find in -type f -print0

# 配列 → NUL 出力
printf '%s\0' "${items[@]}"

# Bash 側で読む（-d ''）
while IFS= read -r -d '' p; do do_work "$p"; done < <(producer)

ロックと一時ファイル（並列安全の必需品）

• 書き込み先の衝突は避ける（tmp に per-job で作り、最後に mv で原子的反映）
• 同一資源を更新する処理は flock で直列化

critical(){
  exec {LFD}>"/tmp/app.lock"; flock -x "$LFD"
  update_shared_state
}

ありがちな落とし穴 → 改善

実務のレシピ

大量画像を並列変換（順不同 OK）

find img -type f -name '*.jpg' -print0 \
| xargs -0 -P"$(nproc)" -n1 -I{} bash -c '
  set -Eeuo pipefail
  in="$1"; out="out/${in##*/}"
  convert "$in" -resize 1024x1024 "$out"
' _ "{}"

API 叩き（同時 4・毎秒 5）

printf '%s\0' "${ids[@]}" \
| xargs -0 -n1 -P4 -I{} bash -c '
  set -Eeuo pipefail
  id="$1"; curl -fsS "https://api/items/$id" -o "out/$id.json"
' _ "{}"
# 上流を rate_limit 5 で間引くとさらに安全

失敗だけを再実行

# 1st run
find in -type f -print0 \
| xargs -0 -P8 -n1 -I{} bash -c '
  set -Eeuo pipefail
  f="$1"; out="out/${f##*/}.ok"
  if myproc "$f"; then : >"$out"; else echo "$f" >>fail.list; exit 1; fi
' _ "{}" || true

# rerun
mapfile -t retry < <(sort -u fail.list)
printf '%s\0' "${retry[@]}" \
| xargs -0 -P4 -n1 -I{} bash -c 'set -Eeuo pipefail; myproc "$1"' _ "{}"

“コピペ可”テストブロック（最小）

#!/usr/bin/env bash
set -Eeuo pipefail; set -o pipefail
TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT
mkdir -p "$TMP/in" "$TMP/out"
printf 'a\nb\nc\n' >"$TMP/in/one.txt"; cp "$TMP/in/one.txt" "$TMP/in/two.txt"

# xargs -P 並列処理
find "$TMP/in" -type f -print0 \
| xargs -0 -P2 -n1 -I{} bash -c '
  set -Eeuo pipefail
  in="$1"; out="$2/$(basename "$1").out"
  awk "{print NR \":\" \$0}" <"$in" >"$out"
' _ "{}" "$TMP/out"

# 検証
n="$(find "$TMP/out" -type f | wc -l)"
[[ "$n" -eq 2 ]] || { echo "parallel NG: $n"; exit 1; }
grep -q '^1:a$' "$TMP/out/one.txt.out" || { echo "content NG"; exit 1; }
echo "PASS"

互換性と移植性

• xargs -P は GNU/BSD 両系で利用可能（古環境で -P 非対応な場合は BG+wait で代替）
• wait -n は Bash 4.3+。古い Bash では PID キューを自前実装
• NUL 系オプション（-print0 -0 -d ''）は GNU/BSD で広く対応

セキュリティと安全設計

• 外部入力をコマンド名やリダイレクト先に使わない（固定コマンド＋-- で終端）
• 破壊的操作はドライランを用意し、対象をログにクォート表示（${var@Q} / printf %q）
• 共有資源はロック、成果物は原子更新（tmp→mv）

参考リンク

• GNU findutils — xargs（-P / -0）
  https://www.gnu.org/software/findutils/manual/html_node/find_html/xargs-options.html
• GNU Bash Manual — Jobs/Wait/Process Substitution
  https://www.gnu.org/software/bash/manual/bash.html
• util-linux flock(1)（排他制御）
  https://man7.org/linux/man-pages/man1/flock.1.html