I have a task that processes a list of files on stdin. The start-up time of the program is substantial, and the amount of time each file takes varies widely. I want to spawn a substantial number of these processes, then dispatch work to whichever ones are not busy. There are several different commandline tools that almost do what I want, I've narrowed it down to two almost working options:

find . -type f | split -n r/24 -u --filter="myjob"
find . -type f | parallel --pipe -u -l 1 myjob

The problem is that split does a pure round-robin, so one of the processes gets behind and stays behind, delaying the completion of the entire operation; while parallel wants to spawn one process per N lines or bytes of input and I wind up spending way too much time on startup overhead.

Is there something like this that will re-use the processes and feed lines to whichever processes have unblocked stdins?

  • Where is that split command from? The name conflicts with the standard text processing utility. Oct 9 '12 at 22:27
  • @Gilles, it's the GNU one: "split (GNU coreutils) 8.13". Using it as a weird alternative to xargs is probably not the intended use but it's the closest to what I want I've found.
    – BCoates
    Oct 9 '12 at 22:44
  • 2
    I've been thinking about that, and a fundamental problem is knowing that an instance of myjob is ready to receive more input. There is no way to know that a program is ready to process more input, all you can know is that some buffer somewhere (a pipe buffer, an stdio buffer) is ready to receive more input. Can you arrange for your program to send some kind of request (e.g. display a prompt) when it's ready? Oct 10 '12 at 1:21
  • Assuming that the program isn't using bufering on stdin, a FUSE filesystem that reacts to read calls would do the trick. That's a fairly large programming endeavor. Oct 10 '12 at 1:42
  • why are you using -l 1 in the parallel args? IIRC, that tells parallel to process one line of input per job (i.e. one filename per fork of myjob, so lots of startup overhead).
    – cas
    Oct 10 '12 at 4:40

For GNU Parallel you can set the block size using --block. It does, however, require you have enough memory to keep 1 block in memory for each of the running processes.

I understand this is not precisely what you are looking for, but it may be an acceptable work-around for now.

If your tasks on average take the same time, then you might be able to use mbuffer:

find . -type f | split -n r/24 -u --filter="mbuffer -m 2G | myjob"

That doesn't look possible in such a general case. It implies you have a buffer for each process and you can watch the buffers from outside to decide where to put the next entry (scheduling)... Of course you might write something (or use a batch system like slurm)

But depending on what the process is, you might be able to pre-process the input. For example if you want to download files, update entries from a DB, or similar, but 50% of them will end up being skipped (and therefor you have a large processing difference depending on the input) then, just setup a pre-processor that verifies which entries are going to take long (file exists, data was changed, etc), so whatever comes from the other side is guaranteed to take a fairly equal amount of time. Even if the heuristic is not perfect you might end up with a considerable improvement. You might dump the others to a file and process afterwards in the same manner.

But that depends on your use case.


No, there isn't a generic solution. Your dispatcher needs to know when each program is ready to read another line, and there's no standard I'm aware of which allows for that. All you can do is put a line on STDOUT and wait for something to consume it; there's not really a good way for the producer on a pipeline to tell if the next consumer is ready or not.


I don't think so. In my favorite magazine was an article once on bash programming which did what you want. I'm willing to believe that if there were tools to do that they would have mentioned them. So you want something along the lines of:

set -m # enable job control

child_has_ended() { concurrent_processes=$((concurrent_processes - 1)) }

trap child_has_ended SIGCHLD # that's magic calling our bash function when a child processes ends

for i in $(find . -type f)
  # don't do anything while there are max_processes running
  while [ ${concurrent_processes} -ge ${max_processes}]; do sleep 0.5; done 
  # increase the counter
  concurrent_processes=$((concurrent_processes + 1))
  # start a child process to actually deal with one file
  /path/to/script/to/handle/one/file $i &

Obviously you may change the invocation to the actual working script to your liking. The magazine I mentionen initially does things like setting up pipes and actually starting worker threads. Check out mkfifo for that, but that route is far more complicated as the worker processes need to signal the master process that they are ready to receive more data. So you need one fifo for each worker process to send it data and one fifo for the master process to receive stuff from the workers.

DISCLAIMER I wrote that script from the top of my head. It may have some syntax issues.

  • 1
    This doesn't seem to meet the requirements: you're starting a different instance of the program for each item. Oct 9 '12 at 22:26
  • It's usually preferable to use find . -type f | while read i rather than for i in $(find . -type f).
    – user26112
    Jun 30 '13 at 14:39

Try this:

mkfifo for each process.

Then hang tail -f | myjob on each fifo.

For example setting up the workers (myjob processes)

mkdir /tmp/jobs
for X in 1 2 3 4
   mkfifo pipe$X
   tail -f pipe$X | myjob &
   jobs -l| awk '/pipe'$X'/ {print $2, "'pipe$X'"}' >> pipe-job-mapping

Depending on your application (myjob) you might eb able to use jobs -s to find stopped jobs. Otherwise list the processes sorted by CPU and select the one consuming fewest resources. Of have the job report itself, eg by setting a flag in the file system when it wants more work.

Assuming the job stops when waiting for input, use

jobs -sl to find out pid of a stopped job and assign it work, for example

grep "^$STOPPED_PID" pipe-to-job-mapping | while read PID PIPE
   cat workset > $PIPE

I tested this with

garfield:~$ cd /tmp
garfield:/tmp$ mkfifo f1
garfield:/tmp$ mkfifo f2
garfield:/tmp$ tail -f f1 | sed 's/^/1 /' &
[1] 21056
garfield:/tmp$ tail -f f2 | sed 's/^/2 /' &
[2] 21058
garfield:/tmp$ echo hello > f1
1 hello
garfield:/tmp$ echo what > f2
2 what
garfield:/tmp$ echo yes > f1
1 yes

This I must admit was just concocted so ymmv.


What is really needed to solve this is a queue mechanism of some type.

Is it possible to have the jobs reading their input from a Queue, such as a SYSV message queue, and then have the programs run by parallel simply push the values onto the queue?

Another possibility is to use a directories for the queue, like this:

  1. the find output creates a symlink to each file to process in a directory, pending
  2. each job process performs a mv of the first file it sees in the directory to a sibling directory of pending, named inprogress.
  3. if the job succesfully moves the file, it performs the processing; otherwise, it goes back to find and move another filename from pending

expounding on @ash's answer, you can use a SYSV message queue to distribute the work. If you don't want to write your own program in C there is a utility called ipcmd that can help. Here's what I put together to pass the output of find $DIRECTORY -type f to $PARALLEL number of processes:

set -o errexit
set -o nounset

export IPCMD_MSQID=$(ipcmd msgget)


# clean up message queue on exit
trap 'ipcrm -q $IPCMD_MSQID' EXIT

for i in $(seq $PARALLEL); do
      while true
          message=$(ipcmd msgrcv) || exit
          [ -f $message ] || break
          sleep $((RANDOM/3000))
   } &

find "$DIRECTORY" -type f | xargs ipcmd msgsnd

for i in $(seq $PARALLEL); do
   ipcmd msgsnd "/dev/null/bar"

Here's a test run:

$ for i in $(seq 20 10 100) ; do time parallel.sh /usr/lib/ $i ; done
parallel.sh /usr/lib/ $i  0.30s user 0.67s system 0% cpu 1:57.23 total
parallel.sh /usr/lib/ $i  0.28s user 0.69s system 1% cpu 1:09.58 total
parallel.sh /usr/lib/ $i  0.19s user 0.80s system 1% cpu 1:05.29 total
parallel.sh /usr/lib/ $i  0.29s user 0.73s system 2% cpu 44.417 total
parallel.sh /usr/lib/ $i  0.25s user 0.80s system 2% cpu 37.353 total
parallel.sh /usr/lib/ $i  0.21s user 0.85s system 3% cpu 32.354 total
parallel.sh /usr/lib/ $i  0.30s user 0.82s system 3% cpu 28.542 total
parallel.sh /usr/lib/ $i  0.27s user 0.88s system 3% cpu 30.219 total
parallel.sh /usr/lib/ $i  0.34s user 0.84s system 4% cpu 26.535 total

Unless you can estimate how long a particular input file will be processed and the worker processes don't have a way to report back to the scheduler (as they do in normal parallel computing scenarios - often through MPI), you are generally out of luck - either pay the penalty of some workers processing input longer than others (because of inequality of input), or pay the penalty of spawning a single new process for every input file.


GNU Parallel has changed in the past 7 years. So today it can do it:

This example shows that more blocks are given to process 11 and 10 than process 4 and 5 because 4 and 5 read slower:

seq 1000000 |
  parallel -j8 --tag --roundrobin --pipe --block 1k 'pv -qL {}0000 | wc' ::: 11 4 5 6 9 8 7 10

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.