Pipe

tee + cat:多次使用輸出,然後連接結果

  • December 24, 2021

如果我呼叫某個命令,例如,echo我可以在其他幾個命令中使用該命令的結果tee。例子:

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

使用 cat 我可以收集幾個命令的結果。例子:

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

我希望能夠同時做這兩件事,這樣我就可以tee在其他東西的輸出上呼叫這些命令(例如echo我寫的),然後在單個輸出上收集所有結果cat.

保持結果的順序很重要,這意味著 , 的輸出中的行command1command2應該command3交織在一起,而是按照命令的順序排列(就像它發生的那樣cat)。

可能有更好的選擇cattee但這些是我迄今為止所知道的。

我想避免使用臨時文件,因為輸入和輸出的大小可能很大。

我怎麼能這樣做?

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 中的讀取和寫入似乎是重疊的,導致一切都爆炸了。

您可以使用 GNU stdbuf 和peemoreutils的組合

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

但請注意,除了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 中的任何一個中流動,因為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在之間插入teecmd2/3在哪裡pv可以儲存多達 1G 的數據等待cmd2cmd3讀取它們。但這意味著兩件事:

  1. 這可能會使用大量記憶體,而且還會複製它
  2. 這沒有讓所有 3 個命令合作,因為cmd2實際上只會在 cmd1 完成時才開始處理數據。

第二個問題的解決方案是使管道 6 和 7 也更大。假設cmd2cmd3產生與消耗一樣多的輸出,則不會消耗更多記憶體。

避免重複數據(在第一個問題中)的唯一方法是在調度程序本身中實現數據的保留,即實現一種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

引用自:https://unix.stackexchange.com/questions/66853