使用带有bash的命名管道 - 数据丢失问题

Question

做了一些在线搜索,找到了使用命名管道的简单"教程".但是,当我对后台工作做任何事情时,我似乎丢失了大量数据.

[[编辑:发现一个更简单的解决方案,请参阅回复帖子.所以我提出的问题现在是学术性的 - 如果有人想要一个工作服务器]]

使用Ubuntu 10.04和Linux 2.6.32-25-generic#45-Ubuntu SMP Sat Oct 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

输出不一致.我经常没有得到所有成功的回声.我获得最多的成功回声的新文本回声,有时更少.

如果我从"Feed"中删除"&",它似乎可以正常工作,但我会被阻止,直到读取输出为止.因此,我想让子流程被阻止,而不是主流程.

目的是编写一个简单的作业控制脚本,这样我就可以最多并行运行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

将开始一个过程.那部分工作正常.开始第二个,很好.手工一个一个似乎工作正常.但是在循环中开始10似乎失去了系统,就像上面简单的例子一样.

任何关于我如何解决这一明显的IPC数据丢失的提示都将不胜感激.

此致,阿兰.

Answer 1

您的问题if如下:

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

发生的事情是您的作业队列服务器每次在循环周围打开和关闭管道.这意味着一些客户端在尝试写入管道时会出现"管道损坏"错误 - 也就是说,管道读取器在编写器打开后会消失.

要解决此问题,请在服务器中更改循环,为整个循环打开管道一次:

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

通过这种方式,管道打开一次并保持打开状态.

您需要注意在循环中运行的内容,因为循环内的所有处理都将stdin附加到命名管道.您需要确保从其他位置重定向循环内的所有进程的stdin,否则它们可能会消耗管道中的数据.

编辑:现在的问题是,当最后一个客户端关闭管道时,您正在读取EOF,您可以使用jilles方法复制文件描述符,或者您也可以确保您也是客户端并保持写入端管道打开:

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

这将使管道的写入侧在fd 3上保持打开.对于此文件描述符,与stdin一样适用.您将需要关闭它,以便任何子进程不继承它.它可能比stdin更重要,但它会更清洁.

Answer 2

正如在其他答案中所说,您需要始终保持fifo打开以避免丢失数据.

但是,一旦所有作者在fifo打开后离开(所以有一个作家),读取立即返回(并poll()返回POLLHUP).清除此状态的唯一方法是重新打开fifo.

POSIX没有为此提供解决方案,但至少Linux和FreeBSD会这样做:如果读取开始失败,请在保持原始描述符打开的同时再次打开fifo.这是有效的,因为在Linux和FreeBSD中,"hangup"状态是特定打开文件描述的本地状态,而在POSIX中,它是fifo的全局状态.

这可以在shell脚本中完成,如下所示:

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