☰

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

tag-icon • tag-icon pipe cat tee concurrency
tee + cat：多次使用输出，然后连接结果

如果我调用某个命令，例如，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 中的读取和写入似乎重叠，导致一切爆炸。

答案1

pee您可以使用 GNU stdbuf 和from的组合更多实用程序:

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

pee popen(3)s 这 3 个 shell 命令行，然后freads 输入并fwrites 全部三个，这将被缓冲到最多 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

但请注意，除了zshNUL 字符的二进制输入之外，其他 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

或者为了避免命名管道，使用 coproc 会更痛苦一些zsh：

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 中的任何一个中流动，因为数据以tee所有 3 个管道中最慢的速率传输。

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

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输出）可以做到这一点。例如，我们可以通过在 和 之间插入可以存储pv -qB 1G多达1G 的数据来等待和读取它们。但这意味着两件事：teecmd2/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

答案2

您提出的建议无法使用任何现有命令轻松完成，并且无论如何都没有多大意义。管道（在 Unix/Linux 中）的整体思想|是（最多）写入输出，直到内存缓冲区填满，然后cmd1 | cmd2从缓冲区（最多）读取数据，直到缓冲区为空。即，并且同时运行，它们之间不需要有超过有限数量的“传输中”数据。如果您想将多个输入连接到一个输出，如果其中一个读取器落后于其他读取器，您要么停止其他读取器（那么并行运行有什么意义？），要么将落后者尚未读取的输出隐藏起来（那么没有中间文件有什么意义呢？）。另外，整个同步得到了cmd1cmd2cmd1cmd2很多更复杂。

在我近 30 年的 Unix 经验中，我不记得有任何情况会真正受益于这样的多输出管道。

cmd1今天，您可以将多个输出组合成一个流，只是不能以任何交错方式（和的输出应该如何cmd2交错？依次一行？轮流写入 10 个字节？以某种方式定义的备用“段落”？如果一个不这样做）长时间不写任何东西？这一切都很难处理）。例如，它是通过(cmd1; cmd2; cmd3) | cmd4程序cmd1、cmd2、 和cmd3依次运行来完成的，输出作为输入发送到cmd4。

答案3

对于您的重叠问题，在 Linux 上（使用bash或zsh不使用ksh93），您可以这样做：

somefunction()
(
  if [ "$1" -eq 1 ]
  then
    echo "Hello world!"
  else
    exec 3> auxfile
    rm -f auxfile
    somefunction "$(($1 - 1))" >&3 auxfile 3>&-
    exec cat <(command1 < /dev/fd/3) \
             <(command2 < /dev/fd/3) \
             <(command3 < /dev/fd/3)
  fi
)

请注意，在每次迭代中使用(...)代替 来{...}获取新进程，这样我们就可以有一个新的 fd 3 指向新的auxfile.< /dev/fd/3是访问现已删除的文件的技巧。它不适用于 Linux 以外的系统，< /dev/fd/3例如dup2(3, 0)，fd 0 将以只写模式打开，光标位于文件末尾。

为了避免嵌套 somefunction 的 fork，您可以将其写为：

somefunction()
{
  if [ "$1" -eq 1 ]
  then
    echo "Hello world!"
  else
    {
      rm -f auxfile
      somefunction "$(($1 - 1))" >&3 auxfile 3>&-
      exec cat <(command1 < /dev/fd/3) \
               <(command2 < /dev/fd/3) \
               <(command3 < /dev/fd/3)
    } 3> auxfile
  fi
}

外壳会照顾备份每次迭代的 fd 3。不过，您最终会很快用完文件描述符。

尽管您会发现这样做更有效：

somefunction() {
  if [ "$1" -eq 1 ]; then
    echo "Hello world!" > auxfile
  else
    somefunction "$(($1 - 1))"
    { rm -f auxfile
      cat <(command1 < /dev/fd/3) \
          <(command2 < /dev/fd/3) \
          <(command3 < /dev/fd/3) > auxfile
    } 3< auxfile
  fi
}
somefunction 12; cat auxfile

也就是说，不要嵌套重定向。

答案4

鉴于这些限制，我看不到解决一般问题的方法：

保持结果按顺序排列很重要，这意味着 command1、command2 和 command3 的输出中的行不应交织在一起，而应按命令排序（就像 cat 中的情况一样）。

:

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

让我们假设每个的输入和输出command*都大于 RAM+磁盘。您以某种方式需要存储command2+3输出时的输出command1，或者需要存储输入。如果输入和输出大于 RAM+磁盘，那么我看不到可以将其存储在哪里。

如果放宽限制并允许存储在磁盘上，同时总输出小于磁盘，则可以执行此操作。

如果输入很大并且您不想冒死锁或内存不足的风险：

echo "Hello world!" | tee >(command1 >out1) >(command2 >out2) >(command3 >out3) >/dev/null
cat out1 out2 out3

这是背后的基本思想parallel --tee：

echo "Hello world!" | parallel --pipe --tee ::: command1 command2 command3
cat bigger-than-ram-file | parallel --pipe --tee ::: command1 command2 command3

parallel还将处理 stderr 并清理临时文件。

如果您放松对交错的限制，则不需要临时文件的空间：您可以拥有比 RAM+磁盘更大的输出。这将缓冲整行，但不是整个输出：

cat bigger-than-ram+disk-file |
  parallel --line-buffer --pipe --tee ::: command1 command2 command3

相关内容