Wie benutzt manMehrfachverarbeitung anzugehenpeinlich parallele Probleme?

Peinlich parallele Probleme bestehen typischerweise aus drei Grundelementen:

Lesen Eingabedaten (aus einer Datei, Datenbank, TCP-Verbindung usw.).Lauf Berechnungen auf den Eingabedaten, wo sich jede Berechnung befindetunabhängig von anderen Berechnungen.Schreiben Berechnungsergebnisse (zu einer Datei, einer Datenbank, einer TCP-Verbindung usw.).

Wir können das Programm in zwei Dimensionen parallelisieren:

Teil 2 kann auf mehreren Kernen ausgeführt werden, da jede Berechnung unabhängig ist. Reihenfolge der Verarbeitung spielt keine Rolle.Jeder Teil kann unabhängig laufen. Teil 1 kann Daten in eine Eingabewarteschlange stellen, Teil 2 kann Daten aus der Eingabewarteschlange abrufen und Ergebnisse in eine Ausgabewarteschlange stellen, und Teil 3 kann Ergebnisse aus der Ausgabewarteschlange abrufen und ausschreiben.

Dies scheint ein grundlegendstes Muster bei der gleichzeitigen Programmierung zu sein, aber ich bin immer noch nicht in der Lage, es zu lösenSchreiben wir ein kanonisches Beispiel, um zu veranschaulichen, wie dies mit Multiprocessing geschieht.

Hier ist das Beispielproblem: Gegeben aCSV-Datei Berechnen Sie mit Zeilen von Ganzzahlen als Eingabe ihre Summen. Teilen Sie das Problem in drei Teile auf, die alle parallel ablaufen können:

Verarbeiten Sie die Eingabedatei in Rohdaten (Listen / Iterablen von ganzen Zahlen)Berechnen Sie parallel die Summen der DatenGeben Sie die Summen aus

Nachfolgend finden Sie ein traditionelles Python-Programm mit einem einzigen Prozess, das diese drei Aufgaben löst:

#!/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:])

Nehmen wir dieses Programm und schreiben es neu, um die drei oben beschriebenen Teile mithilfe von Multiprocessing zu parallelisieren. Nachfolgend finden Sie ein Skelett dieses neuen, parallelen Programms, das ausgearbeitet werden muss, um die Teile in den Kommentaren zu behandeln:

#!/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:])

Diese Teile des Codes sowieein weiteres Stück Code, mit dem Beispiel-CSV-Dateien generiert werden können zu Testzwecken kann seinauf Github gefunden.

Ich würde mich über einen Einblick darüber freuen, wie Sie als Nebenläufige Gurus dieses Problem angehen würden.

Hier sind einige Fragen, die ich hatte, als ich über dieses Problem nachdachte. Bonuspunkte für die Ansprache eines / aller:

Sollte ich untergeordnete Prozesse zum Einlesen und Einreihen der Daten in die Warteschlange haben, oder kann der Hauptprozess dies tun, ohne zu blockieren, bis alle Eingaben gelesen wurden?Sollte ich auch einen untergeordneten Prozess zum Schreiben der Ergebnisse aus der verarbeiteten Warteschlange haben, oder kann der Hauptprozess dies tun, ohne auf alle Ergebnisse warten zu müssen?Soll ich einProzesspool für die Summenoperationen?Wenn ja, welche Methode muss ich für den Pool aufrufen, damit er mit der Verarbeitung der in die Eingabewarteschlange eingehenden Ergebnisse beginnt, ohne auch die Eingabe- und Ausgabeprozesse zu blockieren?apply_async ()? map_async ()? imap ()? imap_unordered ()?Angenommen, wir müssten die Eingabe- und Ausgabewarteschlangen nicht während der Dateneingabe löschen, sondern könnten warten, bis alle Eingaben analysiert und alle Ergebnisse berechnet wurden (z. B. weil wir wissen, dass alle Eingaben und Ausgaben in den Systemspeicher passen). Sollten wir den Algorithmus in irgendeiner Weise ändern (z. B. keine Prozesse gleichzeitig mit E / A ausführen)?