Comment utiliser le multitraitement de s'attaquer parallèle gênant problèmes?
Embarassingly parallèle problèmes sont généralement constitués de trois éléments de base:
- Lire les données d'entrée (à partir d'un fichier, base de données, de connexion tcp, etc.).
- Exécuter des calculs sur les données d'entrée, où chaque calcul est indépendant de tout autre calcul.
- Écrire les résultats de calculs (pour un fichier, base de données, de connexion tcp, etc.).
On peut paralléliser le programme en deux dimensions:
- La partie 2 peut s'exécuter sur plusieurs cœurs, puisque chaque calcul est indépendant; ordre de traitement n'a pas d'importance.
- Chaque partie peut fonctionner de façon indépendante. La partie 1 peut placer des données sur l'entrée de la file d'attente, la partie 2 peut tirer des données de la file d'attente d'entrée et de mettre les résultats sur une file d'attente de sortie, et la partie 3 peut tirer des résultats hors de la file d'attente de sortie et de les écrire.
Cela semble des plus rythme de base de la programmation simultanée, mais je suis toujours perdu dans la tente de le résoudre, de sorte que nous allons écrire un exemple canonique pour illustrer comment cela se fait à l'aide de multitraitement.
Voici l'exemple de problème: étant Donné un fichier CSV avec des rangées de nombres entiers en entrée, calculer leurs sommes. Séparer le problème en trois parties, qui peuvent s'exécuter en parallèle:
- Traiter le fichier d'entrée en données brutes (listes/iterables de nombres entiers)
- Calculer la somme des données, en parallèle
- Sortie le sommes
Ci-dessous est traditionnel, d'un processus unique lié programme en Python qui résout ces trois tâches:
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# basicsums.py
"""A program that reads integer values from a CSV file and writes out their
sums to another CSV file.
"""
import csv
import optparse
import sys
def make_cli_parser():
"""Make the command line interface parser."""
usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV",
__doc__,
"""
ARGUMENTS:
INPUT_CSV: an input CSV file with rows of numbers
OUTPUT_CSV: an output file that will contain the sums\
"""])
cli_parser = optparse.OptionParser(usage)
return cli_parser
def parse_input_csv(csvfile):
"""Parses the input CSV and yields tuples with the index of the row
as the first element, and the integers of the row as the second
element.
The index is zero-index based.
:Parameters:
- `csvfile`: a `csv.reader` instance
"""
for i, row in enumerate(csvfile):
row = [int(entry) for entry in row]
yield i, row
def sum_rows(rows):
"""Yields a tuple with the index of each input list of integers
as the first element, and the sum of the list of integers as the
second element.
The index is zero-index based.
:Parameters:
- `rows`: an iterable of tuples, with the index of the original row
as the first element, and a list of integers as the second element
"""
for i, row in rows:
yield i, sum(row)
def write_results(csvfile, results):
"""Writes a series of results to an outfile, where the first column
is the index of the original row of data, and the second column is
the result of the calculation.
The index is zero-index based.
:Parameters:
- `csvfile`: a `csv.writer` instance to which to write results
- `results`: an iterable of tuples, with the index (zero-based) of
the original row as the first element, and the calculated result
from that row as the second element
"""
for result_row in results:
csvfile.writerow(result_row)
def main(argv):
cli_parser = make_cli_parser()
opts, args = cli_parser.parse_args(argv)
if len(args) != 2:
cli_parser.error("Please provide an input file and output file.")
infile = open(args[0])
in_csvfile = csv.reader(infile)
outfile = open(args[1], 'w')
out_csvfile = csv.writer(outfile)
# gets an iterable of rows that's not yet evaluated
input_rows = parse_input_csv(in_csvfile)
# sends the rows iterable to sum_rows() for results iterable, but
# still not evaluated
result_rows = sum_rows(input_rows)
# finally evaluation takes place as a chain in write_results()
write_results(out_csvfile, result_rows)
infile.close()
outfile.close()
if __name__ == '__main__':
main(sys.argv[1:])
Prenons ce programme et de le réécrire le multitraitement de paralléliser les trois parties décrites ci-dessus. Ci-dessous est un squelette de cette nouvelle, parallélisée programme, qui doit être étoffé à l'adresse de pièces dans les commentaires:
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# multiproc_sums.py
"""A program that reads integer values from a CSV file and writes out their
sums to another CSV file, using multiple processes if desired.
"""
import csv
import multiprocessing
import optparse
import sys
NUM_PROCS = multiprocessing.cpu_count()
def make_cli_parser():
"""Make the command line interface parser."""
usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV",
__doc__,
"""
ARGUMENTS:
INPUT_CSV: an input CSV file with rows of numbers
OUTPUT_CSV: an output file that will contain the sums\
"""])
cli_parser = optparse.OptionParser(usage)
cli_parser.add_option('-n', '--numprocs', type='int',
default=NUM_PROCS,
help="Number of processes to launch [DEFAULT: %default]")
return cli_parser
def main(argv):
cli_parser = make_cli_parser()
opts, args = cli_parser.parse_args(argv)
if len(args) != 2:
cli_parser.error("Please provide an input file and output file.")
infile = open(args[0])
in_csvfile = csv.reader(infile)
outfile = open(args[1], 'w')
out_csvfile = csv.writer(outfile)
# Parse the input file and add the parsed data to a queue for
# processing, possibly chunking to decrease communication between
# processes.
# Process the parsed data as soon as any (chunks) appear on the
# queue, using as many processes as allotted by the user
# (opts.numprocs); place results on a queue for output.
#
# Terminate processes when the parser stops putting data in the
# input queue.
# Write the results to disk as soon as they appear on the output
# queue.
# Ensure all child processes have terminated.
# Clean up files.
infile.close()
outfile.close()
if __name__ == '__main__':
main(sys.argv[1:])
Ces morceaux de code, ainsi qu' un autre morceau de code qui peut générer de l'exemple des fichiers CSV pour des fins de test, peut être trouvé sur github.
J'apprécierais toute réflexion ici à la façon dont vous la simultanéité des gourous de l'approche de ce problème.
Voici quelques questions que j'avais lorsque l'on pense à propos de ce problème. Les points de Bonus pour le traitement de toute/tous:
- Dois-je demander à l'enfant de processus de lecture des données et de la placer dans la file d'attente, ou peut la faire sans blocage jusqu'à ce que toutes les entrées sont lues?
- De même, si j'ai un enfant procédé pour écrire les résultats de la transformation de la file d'attente, ou peut la faire sans avoir à attendre pour les résultats?
- Dois-je utiliser un processus de piscine pour la somme des opérations?
- Si oui, quelle méthode dois-je appeler à la piscine pour il pour commencer à traiter les résultats à venir dans l'entrée de la file d'attente, sans bloquer l'entrée et la sortie des processus, trop? apply_async()? map_async()? imap()? imap_unordered()?
- Supposons que nous n'avons pas besoin de siphonner de l'entrée et la sortie des files d'attente comme les données saisies, mais pourrait attendre jusqu'à ce que tous les commentaires ont été analysés et tous les résultats ont été calculées (par exemple, parce que nous le savons tous, l'entrée et la sortie de l'ajustement de la mémoire système). Devrions-nous modifier l'algorithme de toute manière (par exemple, de ne pas exécuter n'importe quel processus simultanément avec I/O)?
