Propager stdin aux processus parallèles

J'ai une tâche qui traite une liste de fichiers sur stdin. Le temps de démarrage du programme est important et le temps nécessaire à chaque fichier varie considérablement. Je veux générer un nombre important de ces processus, puis envoyer le travail à ceux qui ne sont pas occupés. Il existe plusieurs outils de ligne de commande différents qui font presque ce que je veux, je l'ai réduit à deux options presque opérationnelles:

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

Le problème est que cela splitfait un round-robin pur, donc l'un des processus prend du retard et reste en arrière, retardant la fin de l'opération entière; tandis que parallelveut générer un processus par N lignes ou octets d'entrée et je finis par passer trop de temps sur les frais généraux de démarrage.

Y a-t-il quelque chose comme ça qui réutilisera les processus et les lignes d'alimentation vers les processus qui ont débloqué les stdins?

Cela ne semble pas possible dans un cas aussi général. Cela implique que vous avez un tampon pour chaque processus et vous pouvez regarder les tampons de l'extérieur pour décider où placer la prochaine entrée (planification) ... Bien sûr, vous pourriez écrire quelque chose (ou utiliser un système de traitement par lots comme slurm)

Mais en fonction du processus, vous pourrez peut-être prétraiter l'entrée. Par exemple, si vous souhaitez télécharger des fichiers, mettre à jour des entrées à partir d'une base de données ou similaire, mais 50% d'entre eux finiront par être ignorés (et donc vous avez une grande différence de traitement en fonction de l'entrée), alors, il suffit de configurer un pré-processeur qui vérifie quelles entrées vont prendre du temps (le fichier existe, les données ont été modifiées, etc.), donc tout ce qui vient de l'autre côté est garanti pour prendre un temps assez égal. Même si l'heuristique n'est pas parfaite, vous pourriez vous retrouver avec une amélioration considérable. Vous pouvez vider les autres dans un fichier et les traiter ensuite de la même manière.

Mais cela dépend de votre cas d'utilisation.

Non, il n'y a pas de solution générique. Votre répartiteur doit savoir quand chaque programme est prêt à lire une autre ligne, et il n'y a pas de norme à ma connaissance qui le permette. Tout ce que vous pouvez faire est de mettre une ligne sur STDOUT et d'attendre que quelque chose le consomme; il n'y a pas vraiment de bon moyen pour le producteur sur un pipeline de dire si le prochain consommateur est prêt ou non.

Je ne pense pas. Dans mon magazine préféré, il y avait une fois un article sur la programmation bash qui faisait ce que vous vouliez. Je suis prêt à croire que s'il y avait eu des outils pour le faire, ils les auraient mentionnés. Vous voulez donc quelque chose dans le sens de:

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

De toute évidence, vous pouvez modifier l'invocation du script de travail réel à votre guise. Le magazine que je mentionne au départ fait des choses comme mettre en place des tuyaux et démarrer réellement les threads de travail. Vérifiez mkfifocela, mais cet itinéraire est beaucoup plus compliqué car les processus de travail doivent signaler au processus maître qu'ils sont prêts à recevoir plus de données. Vous avez donc besoin d'un fifo pour chaque processus de travail pour lui envoyer des données et d'un fifo pour que le processus maître reçoive des informations des travailleurs.

AVERTISSEMENT J'ai écrit ce script du haut de ma tête. Il peut avoir des problèmes de syntaxe.

Pour GNU Parallel, vous pouvez définir la taille du bloc à l'aide de --block. Cependant, cela nécessite que vous ayez suffisamment de mémoire pour conserver 1 bloc en mémoire pour chacun des processus en cours d'exécution.

Je comprends que ce n'est pas précisément ce que vous recherchez, mais cela peut être une solution acceptable pour l'instant.

Si vos tâches prennent en moyenne le même temps, vous pourrez peut-être utiliser mbuffer:

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

Essaye ça:

mkfifo pour chaque processus.

Accrochez-vous ensuite tail -f | myjobà chaque fifo.

Par exemple, configurer les travailleurs (processus 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

En fonction de votre application (myjob), vous pouvez peut-être utiliser des jobs -s pour rechercher des jobs arrêtés. Sinon, répertoriez les processus triés par CPU et sélectionnez celui qui consomme le moins de ressources. D'avoir le rapport de travail lui-même, par exemple en définissant un indicateur dans le système de fichiers quand il veut plus de travail.

En supposant que le travail s'arrête en attendant l'entrée, utilisez

jobs -sl pour découvrir le pid d'un travail arrêté et lui attribuer un travail, par exemple

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

J'ai testé cela avec

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

Je dois admettre que cela vient d'être concocté donc ymmv.

Ce qui est vraiment nécessaire pour résoudre ce problème est un mécanisme de file d'attente d'un certain type.

Est-il possible que les travaux lisent leur entrée à partir d'une file d'attente, telle qu'une file d'attente de messages SYSV, puis que les programmes exécutés en parallèle poussent simplement les valeurs dans la file d'attente?

Une autre possibilité consiste à utiliser un répertoire pour la file d'attente, comme ceci:

1. la sortie de recherche crée un lien symbolique vers chaque fichier à traiter dans un répertoire, pending
2. chaque processus de travail exécute un mvdes premier fichier qu'il voit dans le répertoire vers un répertoire frère de pending, nommé inprogress.
3. si le travail déplace le fichier avec succès, il effectue le traitement; sinon, il retourne pour rechercher et déplacer un autre nom de fichierpending

exposant la réponse de @ ash, vous pouvez utiliser une file d'attente de messages SYSV pour distribuer le travail. Si vous ne voulez pas écrire votre propre programme en C, un utilitaire appelé ipcmdpeut vous aider. Voici ce que j'ai mis en place pour passer la sortie de find $DIRECTORY -type fà un $PARALLELcertain nombre de processus:

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

Voici un essai:

$ 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

À moins que vous puissiez estimer la durée de traitement d'un fichier d'entrée particulier et que les processus de travail n'ont aucun moyen de faire rapport au planificateur (comme ils le font dans les scénarios de calcul parallèle normaux - souvent via MPI ), vous n'avez généralement pas de chance - soit payer la pénalité de certains travailleurs qui traitent les données d'entrée plus longtemps que d'autres (en raison de l'inégalité des données d'entrée), soit payer la peine de générer un nouveau processus unique pour chaque fichier d'entrée.

GNU Parallel a changé au cours des 7 dernières années. Aujourd'hui, il peut le faire:

Cet exemple montre que plus de blocs sont donnés aux processus 11 et 10 qu'aux processus 4 et 5 car 4 et 5 lisent plus lentement:

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