Решение смущающе параллельно проблемам с помощью многопроцессорной обработки Python

Как каждый использует многопроцессорную обработку для занятия смущающе параллельными проблемами?

Проблемы параллели Embarassingly обычно состоят из трех базовых деталей:

1. Считайте входные данные (из файла, базы данных, соединения TCP, и т.д.).
2. Выполните вычисления на входных данных, где каждое вычисление независимо от любого другого вычисления.
3. Запишите результаты вычислений (в файл, базу данных, соединение TCP, и т.д.).

Мы можем параллелизировать программу в двух размерах:

• Часть 2 может работать на нескольких ядрах, так как каждое вычисление независимо; порядок обработки не имеет значения.
• Каждая часть может работать независимо. Часть 1 может поместить данные по входной очереди, часть 2 может вытянуть данные от входной очереди и поместить результаты на очередь вывода, и часть 3 может вытянуть результаты от очереди вывода и выписать им.

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

Вот проблема в качестве примера: Учитывая файл CSV со строками целых чисел, как введено, вычислите их суммы. Разделите проблему на три части, которые могут все работать параллельно:

1. Обработайте входной файл в необработанные данные (lists/iterables целых чисел)
2. Вычислите суммы данных, параллельно
3. Произведите суммы

Ниже является традиционным, единственный процесс связал программу Python, которая решает эти три задачи:

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

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

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

Эти части кода, а также другая часть кода, который может генерировать файлы CSV в качестве примера для тестирования, могут быть найдены на GitHub.

Я ценил бы любое понимание здесь относительно того, как Вы гуру параллелизма приблизитесь к этой проблеме.

Вот некоторые вопросы, которые я имел при размышлении об этой проблеме. Бонусные очки для обращения к любому/всему:

• У меня должны быть дочерние процессы для чтения в данных и размещения его в очередь, или основной процесс может сделать это, не блокируясь, пока весь вход не читается?
• Аналогично, у меня должен быть дочерний процесс для записи результатов из обработанной очереди, или основной процесс может сделать это, не имея необходимость ожидать всех результатов?
• Я должен использовать пул процессов для операций суммы?
  • Если да, какой метод я обращаюсь к пулу с просьбой заставить это начинать обрабатывать результаты, входящие во входную очередь, не блокируя входные и выходные процессы, также? apply_async ()? map_async ()? IMAP ()? imap_unordered ()?
• Предположим, что мы не должны были откачивать входные и выходные очереди, поскольку данные ввели их, но могли ожидать, пока весь вход не был проанализирован, и все результаты были вычислены (например, потому что мы знаем, что весь ввод и вывод поместится в системную память). Мы должны изменить алгоритм всегда (например, не выполнить какие-либо процессы одновременно с вводом-выводом)?