寻找 IO 流的通用拆分+合并语法

Question

我经常想要执行一个需要的操作

1. stdout将管道（我们称之为管道）分成pipeline-before两个并行流；
2. 将生成的流（作为它们的stdin）馈送到两个单独的管道（pipeline-between-0和pipeline-between-1）；
3. 合并两个结果stdout流按严格的顺序;
4. 将生成的合并流提供stdin给另一个管道 ( pipeline-after)。

在 (3) 中，“按严格的顺序”是指来自的所有输出，例如，pipeline-between-0应该出现在合并的输出流中，位于来自 的任何输出之前pipeline-between-1。

整个事情可以用图表来表示：

pipeline-before --.--- pipeline-between-0 -.
                   \                        \
                    `- pipeline-between-1 ---`-- pipeline-after

pipeline-between-0这样的/对的一个例子pipeline-between-1可以是：

0. head -n 1 | tr 'a-z' 'A-Z'
1. tail -n +2 | sort -t $'\t' -k1,1

在英语中，人们会将这种组合描述为“使第一行全部大写，并按第一列对其余行进行排序”。

问：是否有通用的 shell 语法来表达这样的操作？

我对这个问题的答案感兴趣zsh和bash。

---

一般来说，这是一种语法不是工作：

$ pipeline-before | tee >( pipeline-between-0 ) | pipeline-between-1 | pipeline-after

此语法失败的原因有两个：

1. 的某些输出pipeline-between-1经常出现在 的某些输出之前pipeline-between-0。
2. 最终输出被截断（我怀疑这是因为SIGPIPE信号）。

我确实尝试了以下方法演习（我承认，我不能 100% 理解）：

{
  pipeline-before |
  { tee >( pipeline-between-0 4>&1 1>&3 ); } |
  pipeline-between-1
} 3>&1 | pipeline-after

AFAICT，这种语法似乎解决了上面列出的第一个问题（即基于一些非正式测试，pipeline-between-0和的输出pipeline-between-1以正确的顺序显示）。但不幸的是，最终的输出仍然被截断，至少在某些情况下是这样。

Answer 1

也许您只需要命名管道（FIFO）？

以下示例：

 { seq 1 100000 | grep 1$ & seq 1 100000 | grep 2$ ; } > unsorted

应返回 file 中分别以 1 和 2 结尾的数字的混合结果unsorted。我们希望它们排序（所有数字都以 1 结尾，然后所有数字都以 2 结尾），因此现在创建两个命名管道，每个结果一个，然后按照所需的顺序将它们连接起来。

mkfifo stream{1,2}
{ seq 1 100000 | grep 1$ >stream1 & seq 1 100000 | grep 2$ > stream2 ; } &\
cat stream1 stream2 > sorted

检查文件的顺序是否相同：

diff -q {un,}sorted 

（它们应该不同）并查看排序后的排序是否按预期排序：

sed 1,10000q sorted | grep 2$

（应该没有结果，而unsorted文件应该返回数据）

bash命名管道对于和 的工作方式应该相同zsh。

---

或者用伪代码最通用的表达式：

1. tee输入流（您的“之前的管道”）通过和 FIFO复制到任意数量的并行流中in-i。
2. 每个流由不同的命令并行处理，并将其输出发送到流out-i（您的“i 之间的管道”）。
3. 按所需顺序连接输出流并转发到下一个命令（您的“管道之后”）。

---

mkfifo {in,out}-{0..n}

pre-cmd | tee in-0 in-1 ... in-n | cat >/dev/null &
cmd-0 <in-0 >out-0 &
cmd-1 <in-0 >out-0 &
....
cmd-i <in-n >out-n &
cat out-0 out-1 ... out-n | after-cmd

我使用cat >/dev/null而不是cat >in-n出于泛化的原因，同样的（可能）cat最终是多余的。

例子：

mkfifo {in,out}-{0..2}
seq 0 100 | tee in-{0..2} | cat >/dev/null &
grep '33$' <in-0 >out-0 &
awk '$1<2' <in-1 >out-1 &
sed '/^.\{1,2\}$/d' <in-2 >out-2 &
cat out-2 out-0 out-1 | tr '\n' '-'

结果：100-33-0-1-

Answer 2

你在找吗parallel --tee？只要您有足够的可用磁盘空间/tmp用于输出，它就可以轻松处理任何大小的输入。

(printf "Header1\tHeader2\n"; paste <(seq 20 -1 11) <(seq 10) ) |
  parallel -k --pipe --tee ::: "head -n  1 | tr 'a-z' 'A-Z'" "tail -n +2 | sort -t $'\t' -k1,1"

或者与 bash 函数相同：

pipeline-before() {
    printf "Header1\tHeader2\n"
    paste <(seq 20 -1 11) <(seq 10)
}
pipeline-between-0() {
    head -n  1 | tr 'a-z' 'A-Z'
}
pipeline-between-1() {
    tail -n +2 | sort -t $'\t' -k1,1
}
pipeline-after() {
    echo "This is pipeline-after"
    cat
    echo "Done"
}
export -f pipeline-before pipeline-between-0 pipeline-between-1 pipeline-after

pipeline-before |
  parallel -k --pipe --tee ::: pipeline-between-0 pipeline-between-1 |
  pipeline-after

如果没有，您能否详细说明更多输入、输出以及 pipeline-* 可能是什么的示例？