Rozwiązywanie żenująco równoległych problemów przy użyciu wieloprocesorowości Pythona

W jaki sposób można wykorzystać przetwarzanie wieloprocesowe, aby rozwiązać kłopotliwie równoległe problemy ?

Żenująco równoległe problemy zazwyczaj składają się z trzech podstawowych części:

1. Odczytaj dane wejściowe (z pliku, bazy danych, połączenia TCP itp.).
2. Uruchom obliczenia na danych wejściowych, gdzie każde obliczenie jest niezależne od innych obliczeń .
3. Zapisz wyniki obliczeń (do pliku, bazy danych, połączenia tcp itp.).

Możemy zrównoleglać program w dwóch wymiarach:

• Część 2 może działać na wielu rdzeniach, ponieważ każde obliczenie jest niezależne; kolejność przetwarzania nie ma znaczenia.
• Każda część może działać niezależnie. Część 1 może umieścić dane w kolejce wejściowej, część 2 może pobrać dane z kolejki wejściowej i umieścić wyniki w kolejce wyjściowej, a część 3 może pobrać wyniki z kolejki wyjściowej i zapisać je.

Wydaje się, że jest to najbardziej podstawowy wzorzec w programowaniu współbieżnym, ale wciąż jestem zagubiony, próbując go rozwiązać, więc napiszmy przykład kanoniczny, aby zilustrować, jak to się robi przy użyciu wieloprocesowości .

Oto przykładowy problem: Mając plik CSV z wierszami liczb całkowitych jako danymi wejściowymi, oblicz ich sumy. Podziel problem na trzy części, które mogą działać równolegle:

1. Przetwarzaj plik wejściowy na surowe dane (listy / iterable liczb całkowitych)
2. Oblicz sumy danych równolegle
3. Podaj sumy

Poniżej znajduje się tradycyjny, powiązany z jednym procesem program w języku Python, który rozwiązuje te trzy zadania:

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

Weźmy ten program i przepiszmy go tak, aby używał przetwarzania wieloprocesowego w celu zrównoleglenia trzech części opisanych powyżej. Poniżej znajduje się szkielet tego nowego, równoległego programu, który należy uzupełnić, aby zająć się częściami w komentarzach:

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

Te fragmenty kodu, a także inny fragment kodu, który może generować przykładowe pliki CSV do celów testowych, można znaleźć na github .

Byłbym wdzięczny za wszelkie wglądy tutaj, jak wy, guru współbieżności, podejdą do tego problemu.

Oto kilka pytań, które miałem, myśląc o tym problemie. Punkty bonusowe za zajęcie się dowolnymi / wszystkimi:

• Czy powinienem mieć procesy potomne do wczytywania danych i umieszczania ich w kolejce, czy też może to zrobić główny proces bez blokowania do momentu odczytania wszystkich danych wejściowych?
• Podobnie, czy powinienem mieć proces potomny do wypisywania wyników z przetworzonej kolejki, czy może proces główny może to zrobić bez konieczności czekania na wszystkie wyniki?
• Czy powinienem używać puli procesów dla operacji sumarycznych?
  • Jeśli tak, jaką metodę mam wywołać w puli, aby rozpocząć przetwarzanie wyników przychodzących do kolejki wejściowej, bez blokowania również procesów wejściowych i wyjściowych? apply_async () ? map_async () ? imap () ? imap_unordered () ?
• Załóżmy, że nie musieliśmy wyłączać kolejek wejściowych i wyjściowych podczas wprowadzania danych, ale moglibyśmy poczekać, aż wszystkie dane wejściowe zostaną przeanalizowane i wszystkie wyniki zostaną obliczone (np. Ponieważ wiemy, że wszystkie dane wejściowe i wyjściowe zmieszczą się w pamięci systemowej). Czy powinniśmy w jakikolwiek sposób zmienić algorytm (np. Nie uruchamiać żadnych procesów jednocześnie z I / O)?

Moje rozwiązanie ma dodatkowy dzwonek i gwizdek, aby upewnić się, że kolejność danych wyjściowych jest taka sama, jak kolejność wejść. Używam multiprocessing.queue do przesyłania danych między procesami, wysyłania komunikatów o zatrzymaniu, aby każdy proces wiedział, że musi zakończyć sprawdzanie kolejek. Myślę, że komentarze w źródle powinny jasno określać, co się dzieje, ale jeśli nie, daj mi znać.

#!/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

class CSVWorker(object):
    def __init__(self, numprocs, infile, outfile):
        self.numprocs = numprocs
        self.infile = open(infile)
        self.outfile = outfile
        self.in_csvfile = csv.reader(self.infile)
        self.inq = multiprocessing.Queue()
        self.outq = multiprocessing.Queue()

        self.pin = multiprocessing.Process(target=self.parse_input_csv, args=())
        self.pout = multiprocessing.Process(target=self.write_output_csv, args=())
        self.ps = [ multiprocessing.Process(target=self.sum_row, args=())
                        for i in range(self.numprocs)]

        self.pin.start()
        self.pout.start()
        for p in self.ps:
            p.start()

        self.pin.join()
        i = 0
        for p in self.ps:
            p.join()
            print "Done", i
            i += 1

        self.pout.join()
        self.infile.close()

    def parse_input_csv(self):
            """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.

            The data is then sent over inqueue for the workers to do their
            thing.  At the end the input process sends a 'STOP' message for each
            worker.
            """
            for i, row in enumerate(self.in_csvfile):
                row = [ int(entry) for entry in row ]
                self.inq.put( (i, row) )

            for i in range(self.numprocs):
                self.inq.put("STOP")

    def sum_row(self):
        """
        Workers. Consume inq and produce answers on outq
        """
        tot = 0
        for i, row in iter(self.inq.get, "STOP"):
                self.outq.put( (i, sum(row)) )
        self.outq.put("STOP")

    def write_output_csv(self):
        """
        Open outgoing csv file then start reading outq for answers
        Since I chose to make sure output was synchronized to the input there
        is some extra goodies to do that.

        Obviously your input has the original row number so this is not
        required.
        """
        cur = 0
        stop = 0
        buffer = {}
        # For some reason csv.writer works badly across processes so open/close
        # and use it all in the same process or else you'll have the last
        # several rows missing
        outfile = open(self.outfile, "w")
        self.out_csvfile = csv.writer(outfile)

        #Keep running until we see numprocs STOP messages
        for works in range(self.numprocs):
            for i, val in iter(self.outq.get, "STOP"):
                # verify rows are in order, if not save in buffer
                if i != cur:
                    buffer[i] = val
                else:
                    #if yes are write it out and make sure no waiting rows exist
                    self.out_csvfile.writerow( [i, val] )
                    cur += 1
                    while cur in buffer:
                        self.out_csvfile.writerow([ cur, buffer[cur] ])
                        del buffer[cur]
                        cur += 1

        outfile.close()

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.")

    c = CSVWorker(opts.numprocs, args[0], args[1])

if __name__ == '__main__':
    main(sys.argv[1:])

Spóźniłem się na imprezę ...

Joblib ma warstwę wieloprocesową, która pomaga tworzyć równoległe pętle. Oferuje takie udogodnienia, jak leniwe przydzielanie zadań i lepsze raportowanie błędów, a także bardzo prosta składnia.

Jako zastrzeżenie, jestem oryginalnym autorem pliku joblib.

Zdaję sobie sprawę, że trochę się spóźniłem na imprezę, ale niedawno odkryłem równoległość GNU i chcę pokazać, jak łatwo jest z nim wykonać to typowe zadanie.

cat input.csv | parallel ./sum.py --pipe > sums

Coś takiego wystarczy dla sum.py:

#!/usr/bin/python

from sys import argv

if __name__ == '__main__':
    row = argv[-1]
    values = (int(value) for value in row.split(','))
    print row, ':', sum(values)

Parallel będzie działać sum.pydla każdego wiersza w input.csv(oczywiście równolegle), a następnie wyprowadza wyniki do sums. Wyraźnie lepsze niż multiprocessingkłopoty

Stara szkoła.

p1.py

import csv
import pickle
import sys

with open( "someFile", "rb" ) as source:
    rdr = csv.reader( source )
    for line in eumerate( rdr ):
        pickle.dump( line, sys.stdout )

p2.py

import pickle
import sys

while True:
    try:
        i, row = pickle.load( sys.stdin )
    except EOFError:
        break
    pickle.dump( i, sum(row) )

p3.py

import pickle
import sys
while True:
    try:
        i, row = pickle.load( sys.stdin )
    except EOFError:
        break
    print i, row

Oto ostateczna struktura wieloprocesorowa.

python p1.py | python p2.py | python p3.py

Tak, powłoka połączyła je ze sobą na poziomie systemu operacyjnego. Wydaje mi się to prostsze i działa bardzo ładnie.

Tak, użycie marynaty (lub cPickle) jest nieco większe. Uproszczenie wydaje się jednak warte wysiłku.

Jeśli chcesz, aby nazwa pliku była argumentem p1.py, to łatwa zmiana.

Co ważniejsze, funkcja taka jak poniżej jest bardzo przydatna.

def get_stdin():
    while True:
        try:
            yield pickle.load( sys.stdin )
        except EOFError:
            return

To pozwala ci to zrobić:

for item in get_stdin():
     process item

Jest to bardzo proste, ale nie łatwo pozwalają mieć wiele kopii P2.py biegania.

Masz dwa problemy: rozsuwanie i rozprowadzanie. P1.py musi jakoś rozłożyć się na wiele P2.py. A P2.py muszą w jakiś sposób połączyć swoje wyniki w jeden plik P3.py.

Oldschoolowe podejście do fan-out to architektura typu „push”, która jest bardzo efektywna.

Teoretycznie, wielokrotne pobieranie P2.py ze wspólnej kolejki to optymalna alokacja zasobów. Często jest to idealne rozwiązanie, ale jest to również sporo programowania. Czy programowanie jest naprawdę konieczne? A może przetwarzanie okrężne będzie wystarczająco dobre?

Praktycznie przekonasz się, że zrobienie prostego "round robin" na P1.py pomiędzy wieloma P2.py może być całkiem dobre. Miałbyś konfigurację P1.py tak, aby obsługiwała n kopii P2.py przez nazwane potoki. Każdy z P2.py czytałby z odpowiedniego potoku.

Co się stanie, jeśli jeden P2.py otrzyma wszystkie dane „najgorszego przypadku” i zostanie daleko w tyle? Tak, karuzela nie jest idealna. Ale to lepsze niż tylko jeden P2.py i możesz rozwiązać ten błąd za pomocą prostej randomizacji.

Fan-in z wielu P2.py do jednego P3.py jest jednak nieco bardziej złożony. W tym momencie podejście starej szkoły przestaje być korzystne. P3.py musi czytać z wielu nazwanych potoków używając selectbiblioteki do przeplatania odczytów.

Prawdopodobnie możliwe jest również wprowadzenie paralelizmu do części 1. Prawdopodobnie nie jest to problem z formatem tak prostym jak CSV, ale jeśli przetwarzanie danych wejściowych jest zauważalnie wolniejsze niż odczyt danych, możesz czytać większe fragmenty, a następnie kontynuować czytanie, aż znajdziesz „separator wierszy” ( znak nowej linii w przypadku CSV, ale znowu zależy to od odczytanego formatu; nie działa, jeśli format jest wystarczająco złożony).

Te fragmenty, z których każdy prawdopodobnie zawiera wiele wpisów, można następnie przenieść do tłumu równoległych procesów odczytujących zadania z kolejki, gdzie są analizowane i dzielone, a następnie umieszczane w kolejce do etapu 2.