Распространение stdin, чтобы быть параллельными процессам

Я так не думаю. В моем любимом журнале была статья однажды о программировании удара, которое сделало то, что Вы хотите. Я готов полагать что, если бы были инструменты, чтобы сделать это, они упомянули бы их. Таким образом, Вы хотите что-то вроде:

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

Очевидно, можно изменить вызов на фактический рабочий сценарий к симпатии. Журнал я mentionen первоначально делает вещи как установка каналов и на самом деле стартовых рабочих потоков. Выезд mkfifo для этого, но того маршрута намного более сложно, поскольку рабочие процессы должны сигнализировать об основном процессе, что они готовы получить больше данных. Таким образом, Вам нужен тот первым прибыл, первым обслужен, чтобы каждый рабочий процесс отправил ему данные и один FIFO для основного процесса для получения материала от рабочих.

ПРАВОВАЯ ОГОВОРКА я записал что сценарий из вершины головы. Это может иметь некоторые проблемы синтаксиса.

Для Параллели GNU можно установить использование размера блока - блок. Это действительно, однако, требует, чтобы у Вас было достаточно памяти для хранения 1 блока в памяти для каждого из рабочих процессов.

Я понимаю, что это не точно, что Вы ищете, но это может быть приемлемое обходное решение на данный момент.

Если Ваши задачи в среднем занимают то же время, то Вы смогли использовать mbuffer:

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

Это не выглядит возможным в таком общем случае. Это подразумевает, что у Вас есть буфер для каждого процесса, и можно наблюдать буферы снаружи для решения, куда поместить следующую запись (планирование)... Конечно, Вы могли бы записать что-то (или использовать пакетную систему как slurm),

Но в зависимости от того, каков процесс, Вы смогли предварительно обрабатывать вход. Например, если Вы хотите загрузить файлы, записи обновления от DB, или подобный, но 50% из них закончат тем, что были пропущены (и для этого у Вас есть большое различие в обработке в зависимости от входа), затем, просто установите препроцессор, который проверяет, какие записи собираются занять много времени (файл существует, данные были изменены, и т.д.), таким образом, то независимо от того, что прибывает из другой стороны, как гарантируют, займет довольно равное количество времени. Даже если эвристика не прекрасна, Вы могли бы закончить со значительным улучшением. Вы могли бы вывести другие в файл и процесс впоследствии таким же образом.

Но это зависит от Вашего варианта использования.

Попробуйте это:

mkfifo для каждого процесса.

Затем зависните tail -f | myjob на каждом FIFO.

Например, настраивая рабочих (myjob процессы)

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

В зависимости от Вашего приложения (myjob) Вы мог бы eb способный использовать задания-s для нахождения остановленных заданий. Иначе перечислите процессы, отсортированные по ЦП, и выберите одно потребление наименьшее количество ресурсов. Из имеют сам отчет о задании, например, путем установки флага в файловой системе, когда требуется больше работы.

Принятие остановок задания при ожидании входа, использовать

jobs -sl узнать pid остановленного задания и поручить ему работу, например

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

Я протестировал это с

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

Это, которое я должен допустить, было просто придумано так ymmv.

Что действительно необходимо для решения, это - механизм очереди некоторого типа.

Действительно ли возможно иметь задания, читая их вход из Очереди, такие как очередь сообщений SYSV, и затем программы работали параллелью, просто продвигают значения на очередь?

Другая возможность состоит в том, чтобы использовать каталоги для очереди, как это:

1. вывод находки создает символьную ссылку на каждый файл для обработки в каталоге, pending
2. каждый процесс задания выполняет a mv из первого файла это видит в каталоге к одноуровневому каталогу pending, именованный inprogress.
3. если задание успешно перемещает файл, оно выполняет обработку; иначе это возвращается, чтобы найти и переместить другое имя файла от pending

Нет, нет универсального решения. Ваш диспетчер должен знать, когда каждая программа готова считать другую строку, и нет никакого стандарта, о котором я знаю, который допускает это. Все, что можно сделать, помещается строка на STDOUT, и ожидайте чего-то для потребления его; нет действительно хорошего пути к производителю на конвейере, чтобы сказать, готов ли следующий потребитель или нет.

рассуждая об ответе @ash, можно использовать очередь сообщений SYSV для распределения работы. Если Вы не хотите писать свою собственную программу в C существует названная утилита ipcmd это может помочь. Вот то, что я соединил для передачи вывода find $DIRECTORY -type f кому: $PARALLEL количество процессов:

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

Вот тестовый прогон:

$ 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

Если Вы не можете оценить, сколько времени конкретный входной файл будет обработан, и рабочие процессы не имеют способа сообщить планировщику (как они делают в нормальных сценариях параллельных вычислений - часто через MPI), Вы обычно не повезли - любая плата штраф некоторого входа обработки рабочих дольше, чем другие (из-за неравенства входа) или платите штраф порождения единственного нового процесса для каждого входного файла.

GNU Parallel изменился за последние 7 лет. Так что сегодня он может это сделать:

Этот пример показывает, что больше блоков дается процессам 11 и 10, чем процессам 4 и 5, потому что 4 и 5 читаются медленнее:

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