最小的“任务队列”,使用常用的Linux工具来利用多核CPU

可以将命令列表转换为makefile吗?如果是这样,您可以运行“make-j x”。

GNU并行 http://www.gnu.org/software/parallel/ 是一种比PPS更通用的并行化工具。

如果runfile包含:

command 1 > Logs/1.log
command 2 > Logs/2.log
command 3 > Logs/3.log

你可以做到:

cat runfile | parallel -j+0

每个CPU核心运行一个命令。

如果您的命令如上所述简单,您甚至不需要runfile,但可以:

seq 1 3 | parallel -j+0 'command {} > Logs/{}.log'

如果您有更多的计算机可以进行处理,那么您可能需要查看GNU Parallel的--sshlogin和--trc选项。

好吧,在这里发布问题后,我发现下面的项目看起来很有希望: ppss .

编辑:不完全是我想要的,PPSS专注于处理“目录A中的所有文件”。

不管怎样,这是一个有趣的问题。

这就是我要做的,假设 巴什(1) 当然。

• 找出其中有多少命令可以同时有效地运行。这不仅仅是核心的数量;很多命令都会因为I/O之类的事情而被挂起。打那个号码。 N=15 例如。
• 为SIGCHLD信号设置陷阱信号处理程序,该信号在子进程终止时发生。 trap signalHandler SIGCHLD
• 将命令列表放入管道中
• 编写一个循环,读取stdin并逐个执行命令,使计数器递减。当计数器为0时, wait S.
• 你的信号处理器运行在sigchld上, 增量 那个柜台。

所以现在,它是第一个 N 命令,然后等待。当第一个子项终止时,wait返回,它读取另一行,运行一个新命令,然后再次等待。

现在,这是一个需要处理许多工作的案例,这些工作被紧密地终止在一起。我 犯罪嫌疑人 您可以使用更简单的版本:

 N=15
 COUNT=N
 cat mycommands.sh | 
 while read cmd 
 do
   eval $cmd &
   if $((count-- == 0))
   then
       wait
   fi
 od

现在,这个命令将启动前15个命令,然后在某些命令终止时一次运行其余的命令。

类似的分布式计算乐趣是mapreduce bash脚本:

http://blog.last.fm/2009/04/06/mapreduce-bash-script

感谢您指出PPS!

你可以使用 参数代换 命令,其 ——马克斯普罗斯 做你想做的。例如,charlie martin解决方案与xargs结合使用:

tr '\012' '\000' <mycommands.sh |xargs --null --max-procs=$X bash -c

细节:

• x是最大进程数,例如:x=15。--马克斯·普罗斯在施魔法
• 这里的第一个tr是用空字节终止xargs--null选项的行,这样引号重定向等就不会被错误地扩展。
• bash-c运行命令

我用这个mycommands.sh文件测试了它,例如:

date
date "+%Y-%m-%d" >"The Date".txt
wc -c <'The Date'.txt >'The Count'.txt

这是一种特殊情况,但如果您试图处理一组文件并生成另一组输出文件,则可以启动核心进程数,并在处理之前检查输出文件是否存在。下面的示例将.m4b文件的目录转换为.mp3文件:

只需运行此命令多次,只要您有核心:

ls*m4b_while read f;do test-f$f%m4b_mp3_mencoder-o f rawaudio“$f”-OAC mp3name-ovc copy-o$f%m4b_mp3;done&

您可以在bash上看到我的任务队列: https://github.com/pavelpat/yastq

任务队列+并行+动态添加

使用FIFO,该脚本将自身分叉以处理队列。这样,您就可以动态地向队列添加命令(当队列已经启动时)。

用法:/queue命令[of children][queue name]

示例,1个线程:

./queue "sleep 5; echo ONE"
./queue "echo TWO"

输出:

ONE
TWO

示例,使用2个线程:

./queue "sleep 5; echo ONE" 2
./queue "echo TWO"

输出:

TWO
ONE

例如,有两个队列:

./queue "sleep 5; echo ONE queue1" 1 queue1
./queue "sleep 3; echo ONE queue2" 1 queue2

输出:

ONE queue2
ONE queue1

脚本(另存为“queue”和chmod+x queue):

    #!/bin/bash

    #Print usage
    [[ $# -eq 0 ]] && echo Usage: $0 Command [# of children] [Queue name] && exit

    #Param 1 - Command to execute
    COMMAND="$1"

    #Param 2 - Number of childs in parallel
    MAXCHILD=1
    [[ $# -gt 1 ]] && MAXCHILD="$2"

    #Param 3 - File to be used as FIFO
    FIFO="/tmp/defaultqueue"
    [[ $# -gt 2 ]] && FIFO="$3"

    #Number of seconds to keep the runner active when unused
    TIMEOUT=5

    runner(){
      #Associate file descriptor 3 to the FIFO
      exec 3"$FIFO"

      while read -u 3 -t $TIMEOUT line; do
        #max child check
        while [ `jobs | grep Running | wc -l` -ge "$MAXCHILD" ]; do
          sleep 1
        done

        #exec in backgroud
        (eval "$line")&
      done
      rm $FIFO
    }

    writer(){
      #fork if the runner is not running
      lsof $FIFO >/dev/null || ($0 "QueueRunner" "$MAXCHILD" "$FIFO" &)

      #send the command to the runner
      echo "$COMMAND" > $FIFO
    }

    #Create the FIFO file
    [[ -e "$FIFO" ]] || mkfifo "$FIFO"

    #Start the runner if in the runner fork, else put the command in the queue
    [[ "$COMMAND" == "QueueRunner" ]] && runner || writer