bashで名前付きパイプを使用する - データが失われるという問題

Question

オンラインで検索したところ、名前付きパイプを使用する簡単な「チュートリアル」が見つかりました。しかし、私はバックグラウンドジョブで何かをするとき、私は多くのデータを失うようです。

[編集：はるかに簡単な解決策を見つけました。返信を参照してください。だから私が提出した質問は、今や学術的なものになりました - もし仕事サーバが必要な場合に備えて）

Linux 2.6.32-25-汎用＃45-Ubuntu SMPでのUbuntu 10.04の使用10月16日19:52:42 UTC 2010 x86_64 GNU/Linux

GNU bash、バージョン4.1.5（1）-release（x86_64-pc-linux-gnu）。

私のbashの機能は次のとおりです。

function jqs 
{ 
    pipe=/tmp/__job_control_manager__ 
    trap "rm -f $pipe; exit" EXIT SIGKILL 

    if [[ ! -p "$pipe" ]]; then 
     mkfifo "$pipe" 
    fi 

    while true 
    do 
    if read txt <"$pipe" 
    then 
     echo "$(date +'%Y'): new text is [[$txt]]" 

     if [[ "$txt" == 'quit' ]] 
     then 
    break 
     fi 
    fi 
    done 
} 

私はこれをバックグラウンドで実行します。

> jqs& 
[1] 5336 

そして今、私はそれを養う：

for i in 1 2 3 4 5 6 7 8 
do 
    (echo aaa$i > /tmp/__job_control_manager__ && echo success$i &) 
done 

出力が矛盾しています。 私は頻繁にすべての成功のエコーを得ることはありません。 成功エコーと同じくらい多くの新しいテキストエコーが得られます。

「フィード」から「&」を削除した場合、動作するようですが、出力が読み取られるまでブロックされています。したがって、私はサブプロセスがブロックされることを望みますが、メインプロセスはブロックしません。

単純なジョブ制御スクリプトを書くことを目的としているので、最大10個のジョブを並列に実行し、後で処理するために残りのキューをキューに入れることはできますが、

以下の全ジョブマネージャ：

function jq_manage 
{ 
    export __gn__="$1" 

    pipe=/tmp/__job_control_manager_"$__gn__"__ 
    trap "rm -f $pipe" EXIT 
    trap "break"  SIGKILL 

    if [[ ! -p "$pipe" ]]; then 
     mkfifo "$pipe" 
    fi 

    while true 
    do 
    date 
    jobs 
    if (($(jobs | egrep "Running.*echo '%#_Group_#%_$__gn__'" | wc -l) < $__jN__)) 
    then 
     echo "Waiting for new job" 
     if read new_job <"$pipe" 
     then 
    echo "new job is [[$new_job]]" 

    if [[ "$new_job" == 'quit' ]] 
    then 
     break 
    fi 

    echo "In group $__gn__, starting job $new_job" 
    eval "(echo '%#_Group_#%_$__gn__' > /dev/null; $new_job) &" 
     fi 
    else 
     sleep 3 
    fi 
    done 
} 

function jq 
{ 
    # __gn__ = first parameter to this function, the job group name (the pool within which to allocate __jN__ jobs) 
    # __jN__ = second parameter to this function, the maximum of job numbers to run concurrently 

    export __gn__="$1" 
    shift 
    export __jN__="$1" 
    shift 

    export __jq__=$(jobs | egrep "Running.*echo '%#_GroupQueue_#%_$__gn__'" | wc -l) 
    if (($__jq__ '<' 1)) 
    then 
    eval "(echo '%#_GroupQueue_#%_$__gn__' > /dev/null; jq_manage $__gn__) &" 
    fi 

    pipe=/tmp/__job_control_manager_"$__gn__"__ 

    echo $@ >$pipe 
} 

jq <name> <max processes> <command> 
jq abc 2 sleep 20 

を呼び出すと、一つのプロセスを開始します。 その部分は正常に動作します。もう1つ、罰金を開始します。 1人で手作業でうまくいくようです。 しかし、上記の単純な例のように、ループを10回開始するとシステムが失われるようです。

IPCデータのこの明らかな損失を解決するために私ができることについてのヒントは非常に高く評価されます。

よろしく、 Alain。

Answer 1

あなたの問題は、以下のif文です：何が起こっている

while true 
do 
    if read txt <"$pipe" 
    .... 
done 

は、あなたのジョブ・キュー・サーバがループの周りたびに開いてパイプを閉じているということです。これは、パイプに書き込もうとしたときにクライアントの一部が「破損パイプ」エラーを起こしていることを意味します。つまり、パイプのリーダーが書き込み側を開いた後に消えてしまいます。

は、ループ全体のために一度のパイプを開くサーバーであなたのループを変更、この問題を解決するには、次の

while true 
do 
    if read txt 
    .... 
done < "$pipe" 

はパイプが一回開かれ、開いたままに、この方法を完了します。

ループ内のすべての処理で、名前付きパイプにstdinが接続されるため、ループ内で実行する内容に注意する必要があります。ループ内のすべてのプロセスのstdinを別の場所からリダイレクトする必要があります。そうしないと、パイプからデータを消費する可能性があります。

編集：最後のクライアントがパイプを閉じるときにEOFを読んでいるという問題が発生したため、ファイル記述子を削除するjillesメソッドを使用するか、クライアントであることを確認するだけで済みます。オープンパイプの書き込み側を保つ：同じ注意点は、標準入力と同じように、このファイルディスクリプタを適用3.

while true 
do 
    if read txt 
    .... 
done < "$pipe" 3> "$pipe" 

これはFD上のパイプの書き込み側が開いて保持します。子プロセスがそれを継承しないように閉じる必要があります。それはおそらく標準よりも重要ではありませんが、よりきれいになります。

Answer 2

一方で、問題は私が思っていたよりも悪いです： 今、私のより複雑な例（jq_manage）では、同じデータがパイプから何度も何度も読み込まれているようです新しいデータが書き込まれています）。

一方、私は（デニスのコメント次のように編集）簡単な解決策が見つかりました：魔法のように

function jqn # compute the number of jobs running in that group 
{ 
    __jqty__=$(jobs | egrep "Running.*echo '%#_Group_#%_$__groupn__'" | wc -l) 
} 

function jq 
{ 
    __groupn__="$1"; shift # job group name (the pool within which to allocate $__jmax__ jobs) 
    __jmax__="$1"; shift # maximum of job numbers to run concurrently 

    jqn 
    while (($__jqty__ '>=' $__jmax__)) 
    do 
    sleep 1 
    jqn 
    done 

    eval "(echo '%#_Group_#%_$__groupn__' > /dev/null; $@) &" 
} 

作品。 ソケットまたはパイプが使用されていません。 シンプル。

Answer 3

Like camh &デニスウィリアムソンはパイプを壊さないと言います。

は、今私は、コマンドラインで直接小さな例を、持っている：

サーバー：

(
    for i in {0,1,2,3,4}{0,1,2,3,4,5,6,7,8,9}; 
    do 
    if read s; 
     then echo ">>$i--$s//"; 
    else 
     echo "<<$i"; 
    fi; 
    done < tst-fifo 
)& 

クライアント：

(echo "Test-$i" > tst-fifo&); 

：

(
    for i in {%a,#b}{1,2}{0,1}; 
    do 
    echo "Test-$i" > tst-fifo; 
    done 
)& 

はとキーの行を置き換えることができます

すべてのclienパイプに送られたデータは読み込まれますが、クライアントのオプション2では、すべてのデータが読み込まれる前に数回サーバーを起動する必要があります。

しかし、読み込みはパイプ内のデータが始まるのを待っていますが、一度データがプッシュされると、空の文字列は永遠に読み込まれます。

これを停止する方法はありますか？

もう一度洞察に感謝します。

Answer 4

他の回答で述べたように、データの損失を避けるために、いつもFIFOを開いたままにする必要があります。

しかし、すべてのライターがfifoが開いてから（ライターがいるので）、すぐに読み込みが戻ります（そして、poll()はPOLLHUPを返します）。この状態をクリアする唯一の方法は、FIFOを再度開くことです。

POSIXはこれには解決策を提供しませんが、少なくともLinuxとFreeBSDはそうします：読み取りが失敗した場合は、元のディスクリプタを開いたままFIFOを再度開きます。これはLinuxとFreeBSDでは "ハングアップ"状態が特定の開いているファイル記述のローカルなのでPOSIXではFIFOにグローバルなためです。

これは、このようなシェルスクリプトで行うことができます

while :; do 
    exec 3<tmp/testfifo 
    exec 4<&- 
    while read x; do 
     echo "input: $x" 
    done <&3 
    exec 4<&3 
    exec 3<&- 
done 

Answer 5

だけ興味があるかもしれないもののために、[[再編集]] CAMHとjillesのコメント以下、ここでの二つの新しいバージョンがありますテストサーバースクリプト。

どちらのバージョンも、期待通りに動作するようになりました。パイプ管理のための

CAMHのバージョン：パイプ管理のための

function jqs # Job queue manager 
{ 
    pipe=/tmp/__job_control_manager__ 
    trap "rm -f $pipe; exit" EXIT TERM 

    if [[ ! -p "$pipe" ]]; then 
     mkfifo "$pipe" 
    fi 

    while true 
    do 
    if read -u 3 txt 
    then 
     echo "$(date +'%Y'): new text is [[$txt]]" 

     if [[ "$txt" == 'quit' ]] 
     then 
    break 
     else 
     sleep 1 
     # process $txt - remember that if this is to be a spawned job, we should close fd 3 and 4 beforehand 
     fi 
    fi 
    done 3< "$pipe" 4> "$pipe" # 4 is just to keep the pipe opened so any real client does not end up causing read to return EOF 
} 

jilleのバージョン：あなたの助けのためのすべての

function jqs # Job queue manager 
{ 
    pipe=/tmp/__job_control_manager__ 
    trap "rm -f $pipe; exit" EXIT TERM 

    if [[ ! -p "$pipe" ]]; then 
     mkfifo "$pipe" 
    fi 

    exec 3< "$pipe" 
    exec 4<&- 

    while true 
    do 
    if read -u 3 txt 
    then 
     echo "$(date +'%Y'): new text is [[$txt]]" 

     if [[ "$txt" == 'quit' ]] 
     then 
    break 
     else 
     sleep 1 
     # process $txt - remember that if this is to be a spawned job, we should close fd 3 and 4 beforehand 
     fi 
    else 
     # Close the pipe and reconnect it so that the next read does not end up returning EOF 
     exec 4<&3 
     exec 3<&- 
     exec 3< "$pipe" 
     exec 4<&- 
    fi 
    done 
} 

感謝。

Answer 6

ランは並列に10のジョブは、最大で言うと、後の処理のために休みをキューに、しかし確実に、彼らはあなたがGNUパラレルでこれを行うことができます

を実行しないことを知っています。このスクリプトは必要ありません。

http://www.gnu.org/software/parallel/man.html#options

あなたはMAX-procsの設定することができます "jobslotsの数を並列にNジョブまで実行します。"使用するCPUコアの数を設定するオプションがあります。実行されたジョブのリストをログファイルに保存できますが、これはベータ機能です。