tee + cat：多次使用输出，然后连接结果

Question

如果我调用某个命令，例如，echo我可以在其他几个命令中使用该命令的结果tee。例子：

echo "Hello world!" | tee >(command1) >(command2) >(command3)

使用 cat 我可以收集几个命令的结果。例子：

cat <(command1) <(command2) <(command3)

我希望能够同时做这两件事，这样我就可以tee在其他东西的输出上调用这些命令（例如echo我写的），然后在一个输出上收集所有结果cat.

保持结果为了这一点很重要，这意味着输出的线路command1，command2并且command3不应该纠缠在一起，但订购的命令是（因为它与发生cat）。

可能有比cat和更好的选择，tee但这些是我目前所知道的。

我想避免使用临时文件，因为输入和输出的大小可能很大。

我怎么能这样做？

PD：另一个问题是这种情况发生在循环中，这使得处理临时文件变得更加困难。这是我拥有的当前代码，它适用于小型测试用例，但是在以某种我不理解的方式从 auxfile 读取和写入时，它会创建无限循环。

somefunction()
{
  if [ $1 -eq 1 ]
  then
    echo "Hello world!"
  else
    somefunction $(( $1 - 1 )) > auxfile
    cat <(command1 < auxfile) \
        <(command2 < auxfile) \
        <(command3 < auxfile)
  fi
}

auxfile 中的读取和写入似乎重叠，导致所有内容爆炸。

Answer 1

您可以使用 GNU stdbuf 和pee来自moreutils的组合：

echo "Hello world!" | stdbuf -o 1M pee cmd1 cmd2 cmd3 > output

peepopen(3)是这 3 个 shell 命令行，然后fread是输入和fwrite所有三个，这将被缓冲到 1M。

这个想法是有一个至少与输入一样大的缓冲区。这样即使三个命令同时启动，他们也只会pee pclose在三个命令依次执行时看到输入。

在每个pclose，pee将缓冲区刷新到命令并等待其终止。这保证只要这些cmdx命令在收到任何输入之前不开始输出任何内容（并且不要分叉可能在其父级返回后继续输出的进程），三个命令的输出将不会交错。

实际上，这有点像在内存中使用临时文件，缺点是 3 个命令是同时启动的。

为避免同时启动命令，您可以编写pee为 shell 函数：

pee() (
  input=$(cat; echo .)
  for i do
    printf %s "${input%.}" | eval "$i"
  done
)
echo "Hello world!" | pee cmd1 cmd2 cmd3 > out

但请注意，zsh对于带有 NUL 字符的二进制输入，shell会失败。

这避免了使用临时文件，但这意味着整个输入都存储在内存中。

在任何情况下，您都必须将输入存储在内存或临时文件中的某处。

实际上，这是一个非常有趣的问题，因为它向我们展示了 Unix 想法的局限性，即让几个简单的工具协作完成一个任务。

在这里，我们希望有几个工具配合完成任务：

• 一个源命令（这里echo）
• 调度员命令 ( tee)
• 一些过滤器命令 ( cmd1, cmd2, cmd3)
• 和聚合命令 ( cat)。

如果他们能够同时运行，并在数据可用时尽快处理他们打算处理的数据，那将会很好。

在一个过滤器命令的情况下，这很容易：

src | tee | cmd1 | cat

所有命令同时运行，一旦可用就cmd1开始咀嚼数据src。

现在，使用三个过滤器命令，我们仍然可以做同样的事情：同时启动它们并用管道连接它们：

               ????????????????????????????????????
               ?   ?????2??????cmd1??????5?????   ?
               ?   ????????????????????????????   ?
????????????????   ????????????????????????????   ???????????????
?src?????1??????tee?????3??????cmd2??????6?????cat???????????out?
????????????????   ????????????????????????????   ???????????????
               ?   ????????????????????????????   ?
               ?   ?????4??????cmd3??????7?????   ?
               ????????????????????????????????????

我们可以使用命名管道相对轻松地做到这一点：

pee() (
  mkfifo tee-cmd1 tee-cmd2 tee-cmd3 cmd1-cat cmd2-cat cmd3-cat
  { tee tee-cmd1 tee-cmd2 tee-cmd3 > /dev/null <&3 3<&- & } 3<&0
  eval "$1 < tee-cmd1 1<> cmd1-cat &"
  eval "$2 < tee-cmd2 1<> cmd2-cat &"
  eval "$3 < tee-cmd3 1<> cmd3-cat &"
  exec cat cmd1-cat cmd2-cat cmd3-cat
)
echo abc | pee 'tr a A' 'tr b B' 'tr c C'

（以上} 3<&0是为了解决从&重定向的事实，我们用来避免管道的打开阻塞，直到另一端 ( ) 也打开）stdin/dev/null<>cat

或者为了避免命名管道，使用zshcoproc 会更痛苦：

pee() (
  n=0 ci= co= is=() os=()
  for cmd do
    eval "coproc $cmd $ci $co"

    exec {i}<&p {o}>&p
    is+=($i) os+=($o)
    eval i$n=$i o$n=$o
    ci+=" {i$n}<&-" co+=" {o$n}>&-"
    ((n++))
  done
  coproc :
  read -p
  eval tee /dev/fd/$^os $ci "> /dev/null &" exec cat /dev/fd/$^is $co
)
echo abc | pee 'tr a A' 'tr b B' 'tr c C'

现在的问题是：一旦所有程序都启动并连接起来，数据会流动吗？

我们有两个限制：

• tee 以相同的速率馈送所有输出，因此它只能以其最慢的输出管道的速率发送数据。
• cat 当所有数据从第一个 (5) 读取时，才会从第二个管道（上图中的管道 6）开始读取。

这意味着数据在cmd1完成之前不会在管道 6 中流动。而且，就像tr b B上面的情况一样，这可能意味着数据也不会在管道 3 中流动，这意味着它不会在管道 2、3 或 4 中的任何一个中流动，因为它以所有 3 个管道中tee最慢的速率提供。

实际上，这些管道具有非空大小，因此某些数据将设法通过，至少在我的系统上，我可以让它工作：

yes abc | head -c $((2 * 65536 + 8192)) | pee 'tr a A' 'tr b B' 'tr c C' | uniq -c -c

除此之外，与

yes abc | head -c $((2 * 65536 + 8192 + 1)) | pee 'tr a A' 'tr b B' 'tr c C' | uniq -c

我们陷入了僵局，我们处于这种情况：

               ?????????2????????????????5?????????
               ?   ????????????cmd1????????????   ?
               ?   ????????????????????????????   ?
?????????1??????   ?????3????????????????6?????   ???????????????
?src????????????tee????????????cmd2????????????cat???????????out?
????????????????   ????????????????????????????   ???????????????
               ?   ?????4????????????????7?????   ?
               ?   ????????????cmd3????????????   ?
               ????????????????????????????????????

我们已经填充了管道 3 和 6（每个管道 64kiB）。tee已经读取了那个额外的字节，它已将其提供给cmd1，但是

• 它现在被阻止在管道 3 上写入，因为它正在等待cmd2清空它
• cmd2无法清空它，因为它被阻止在管道 6 上写入，正在等待cat清空它
• cat 无法清空它，因为它正在等待直到管道 ​​5 上没有更多输入。
• cmd1无法判断cat没有更多输入，因为它正在等待来自tee.
• 并且tee无法判断cmd1没有更多输入，因为它被阻止了......等等。

我们有一个依赖循环，因此出现了死锁。

现在，解决方案是什么？更大的管道 3 和 4（足够大以包含所有src的输出）可以做到。例如，我们可以通过在可以存储多达 1G 的等待和读取它们的数据pv -qB 1G之间插入tee和插入来做到这一点。但这意味着两件事：cmd2/3pvcmd2cmd3

1. 这可能会使用大量内存，而且还要复制它
2. 这无法让所有 3 个命令合作，因为cmd2实际上只有在 cmd1 完成后才开始处理数据。

第二个问题的解决方案是使管道 6 和 7 也更大。假设cmd2并cmd3产生与消耗一样多的输出，则不会消耗更多内存。

避免重复数据（在第一个问题中）的唯一方法是在调度程序本身中实现数据保留，即实现tee可以以最快输出速率馈送数据的变体（保留数据以馈送较慢的按自己的节奏）。不是很琐碎。

所以，最后，我们可以在没有编程的情况下合理获得的最好的东西可能是这样的（Zsh 语法）：

max_hold=1G
pee() (
  n=0 ci= co= is=() os=()
  for cmd do
    if ((n)); then
      eval "coproc pv -qB $max_hold $ci $co | $cmd $ci $co | pv -qB $max_hold $ci $co"
    else
      eval "coproc $cmd $ci $co"
    fi

    exec {i}<&p {o}>&p
    is+=($i) os+=($o)
    eval i$n=$i o$n=$o
    ci+=" {i$n}<&-" co+=" {o$n}>&-"
    ((n++))
  done
  coproc :
  read -p
  eval tee /dev/fd/$^os $ci "> /dev/null &" exec cat /dev/fd/$^is $co
)
yes abc | head -n 1000000 | pee 'tr a A' 'tr b B' 'tr c C' | uniq -c