Распространение стандартного ввода на параллельные процессы

13

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

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

Проблема заключается в том, что splitвыполняется чистый циклический перебор, поэтому один из процессов отстает и остается позади, задерживая завершение всей операции; while parallelхочет порождать один процесс на N строк или байтов ввода, и я трачу слишком много времени на издержки запуска.

Есть ли что-то подобное, что будет повторно использовать процессы и линии подачи для тех процессов, которые имеют разблокированные stdins?

BCoates
источник
Откуда эта splitкоманда? Имя конфликтует со стандартной утилитой обработки текста.
Жиль "ТАК ... перестать быть злым"
@ Жиль, это GNU: «split (GNU coreutils) 8.13» . Использование его в качестве странной альтернативы xargs, вероятно, не предназначено, но ближе всего к тому, что я хочу найти.
BCoates
2
Я думал об этом, и фундаментальная проблема заключается в знании того, что экземпляр myjobготов получить больше информации. Нет никакого способа узнать, что программа готова обрабатывать больше входных данных, все, что вы можете знать, это то, что некоторый буфер где-то (буфер канала, буфер stdio) готов получить больше ввода. Можете ли вы сделать так, чтобы ваша программа отправила какой-то запрос (например, отобразить подсказку), когда она будет готова?
Жиль "ТАК - перестань быть злым"
Предполагая, что программа не использует буферизацию на stdin, файловая система FUSE, которая реагирует на readвызовы, сделает свое дело. Это довольно большое программирование.
Жиль "ТАК - перестань быть злым"
почему вы используете -l 1в parallelargs? IIRC, который указывает параллельно обрабатывать одну строку ввода на задание (то есть одно имя файла на разветвление myjob, поэтому много накладных расходов при запуске).
Cas

Ответы:

1

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

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

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

estani
источник
1

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

dannysauer
источник
0

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

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

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

ОТКАЗ ОТ ОТВЕТСТВЕННОСТИ Я написал этот сценарий из головы. У него могут быть некоторые проблемы с синтаксисом.

Bananguin
источник
1
Кажется, это не соответствует требованиям: вы запускаете отдельный экземпляр программы для каждого элемента.
Жиль "ТАК - перестать быть злым"
Обычно предпочтительнее использовать find . -type f | while read i, чем for i in $(find . -type f).
0

Для GNU Parallel вы можете установить размер блока, используя --block. Однако для этого требуется, чтобы у вас было достаточно памяти для хранения 1 блока в памяти для каждого из запущенных процессов.

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

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

find . -type f | split -n r/24 -u --filter="mbuffer -m 2G | myjob"
Оле Танге
источник
0

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

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

Затем повесьте tail -f | myjobкаждый пятерку.

Например, настройка рабочих (процессы 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) вы можете использовать задание -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

Я должен признать, что это было просто придумано.

Johan
источник
0

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

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

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

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

подробно излагая ответ @ 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
kouk
источник
0

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

peterph
источник
0

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
Оле Танге
источник