Take the 2-minute tour ×
Unix & Linux Stack Exchange is a question and answer site for users of Linux, FreeBSD and other Un*x-like operating systems.. It's 100% free, no registration required.

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?

share|improve this question
    
Where is that split command from? The name conflicts with the standard text processing utility. –  Gilles 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? –  Gilles 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. –  Gilles 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
show 2 more comments

8 Answers 8

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.

share|improve this answer
add comment

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.

share|improve this answer
add comment

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
max_processes=8
concurrent_processes=0

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)
do
  # 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 &
done

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.

share|improve this answer
1  
This doesn't seem to meet the requirements: you're starting a different instance of the program for each item. –  Gilles 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). –  paraxor Jun 30 '13 at 14:39
add comment

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"
share|improve this answer
add comment

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
do
   mkfifo pipe$X
   tail -f pipe$X | myjob &
   jobs -l| awk '/pipe'$X'/ {print $2, "'pipe$X'"}' >> pipe-job-mapping
done

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
do
   cat workset > $PIPE
done

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.

share|improve this answer
add comment

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
share|improve this answer
add comment

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)

DIRECTORY=$1
PARALLEL=$2

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

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

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

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

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
share|improve this answer
add comment

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.

share|improve this answer
add comment

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

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