使用GNU Parallel并行处理TSV列变换时程序冻结的原因与解决方法

问题背景 #

编写Bash脚本对TSV文件列执行转换/映射操作，尝试用GNU Parallel实现并行化，但整合后程序出现冻结。

正常运行的串行代码 #

用cat作为恒等映射器，动态生成三列TSV的串行代码可正常执行：

n=1000000
map=cat    # identity: inp -> out

rm -f tmp.col{1,2}.fifo
mkfifo tmp.col{1,2}.fifo
paste <(seq $n) <(seq $n) <(seq $n) \
    | tee >(cut -f1 | $map > tmp.col1.fifo) \
    | tee >(cut -f2 | $map > tmp.col2.fifo) \
    | cut -f3- \
    | paste tmp.col{1,2}.fifo - \
    | python -m tqdm > /dev/null

注：python -m tqdm > /dev/null用于显示处理速度。

可正常运行的Parallel极简示例 #

单独使用GNU Parallel的--pipe --keep-order参数的并行示例能正常工作：

seq 100 | parallel --pipe -k -j4 -N10 'cat && sleep 1'

冻结的并行整合代码 #

将Parallel整合到TSV列处理逻辑后，程序出现冻结：

n=1000000
map=cat   # identity map: inp -> out
rm -f tmp.col{1,2}.fifo
mkfifo tmp.col{1,2}.fifo
paste <(seq $n) <(seq $n) <(seq $n) \
  | tee >(cut -f1 | parallel --id jobA --pipe -k -j4 -N1000 "$map" > tmp.col1.fifo) \
  | tee >(cut -f2 | parallel --id jobB --pipe -k -j4 -N1000 "$map" > tmp.col2.fifo) \
  | cut -f3- \
  | paste tmp.col{1,2}.fifo - \
  | python -m tqdm > /dev/null

环境：Linux 5.15.0-116-generic，Ubuntu 22.04.4 LTS x86_64

冻结原因 #

冻住的核心问题是FIFO读取阻塞和GNU Parallel的缓冲机制冲突：

• paste tmp.col{1,2}.fifo -启动后会等待读取两个FIFO的内容，但GNU Parallel默认会攒够-N1000行的块才处理输出，导致FIFO始终无内容，paste卡死等待。
• 上游的tee和cut进程因下游阻塞停止写入，最终整个管道链陷入死锁。
  而串行版本中cat是逐行无缓冲输出，能实时向FIFO写入内容，因此不会触发阻塞。

解决方案 #

针对Parallel的缓冲问题，有两种可行解决思路：

思路1：禁用Parallel缓冲，强制逐行处理 #

使用--line-buffer参数让Parallel逐行输出，避免攒块缓冲：

n=1000000
map=cat   # identity map: inp -> out
rm -f tmp.col{1,2}.fifo
mkfifo tmp.col{1,2}.fifo
paste <(seq $n) <(seq $n) <(seq $n) \
  | tee >(cut -f1 | parallel --id jobA --pipe -k -j4 --line-buffer "$map" > tmp.col1.fifo) \
  | tee >(cut -f2 | parallel --id jobB --pipe -k -j4 --line-buffer "$map" > tmp.col2.fifo) \
  | cut -f3- \
  | paste tmp.col{1,2}.fifo - \
  | python -m tqdm > /dev/null

注：去掉了-N1000参数，因为--line-buffer是逐行处理，无需指定块大小；若仍需按块处理，可保留-N，但需确保Parallel能及时输出块内容。

思路2：用stdbuf强制Parallel输出无缓冲 #

通过stdbuf -o0强制Parallel的标准输出无缓冲，让处理后的内容立即写入FIFO：

n=1000000
map=cat   # identity map: inp -> out
rm -f tmp.col{1,2}.fifo
mkfifo tmp.col{1,2}.fifo
paste <(seq $n) <(seq $n) <(seq $n) \
  | tee >(cut -f1 | parallel --id jobA --pipe -k -j4 -N1000 "$map" | stdbuf -o0 > tmp.col1.fifo) \
  | tee >(cut -f2 | parallel --id jobB --pipe -k -j4 -N1000 "$map" | stdbuf -o0 > tmp.col2.fifo) \
  | cut -f3- \
  | paste tmp.col{1,2}.fifo - \
  | python -m tqdm > /dev/null

补充说明 #

如果实际的map是耗时计算任务，推荐保留-N块处理参数配合stdbuf，平衡并行效率与实时输出；若为轻量操作，逐行处理的开销也可接受。

内容的提问来源于stack exchange，提问作者TG Gowda