Como alguém usamultiprocessamento enfrentarproblemas embaraçosamente paralelos?

Problemas paralelos embaraçosamente tipicamente consistem em três partes básicas:

Ler dados de entrada (de um arquivo, banco de dados, conexão TCP, etc.).Corre cálculos nos dados de entrada, onde cada cálculo éindependente de qualquer outro cálculo.Escreva resultados de cálculos (para um arquivo, banco de dados, conexão TCP, etc.).

Podemos paralelizar o programa em duas dimensões:

A parte 2 pode ser executada em vários núcleos, pois cada cálculo é independente; ordem de processamento não importa.Cada parte pode ser executada independentemente. A parte 1 pode colocar dados em uma fila de entrada, a parte 2 pode extrair dados da fila de entrada e colocar resultados em uma fila de saída e a parte 3 pode extrair resultados da fila de saída e gravá-los.

Esse parece ser o padrão mais básico da programação simultânea, mas ainda estou perdido ao tentar resolvê-lo.vamos escrever um exemplo canônico para ilustrar como isso é feito usando o multiprocessamento.

Aqui está o problema de exemplo: Dada umaArquivo CSV com linhas de números inteiros como entrada, calcule suas somas. Separe o problema em três partes, que podem ser executadas em paralelo:

Processar o arquivo de entrada em dados brutos (listas / iterables de números inteiros)Calcular as somas dos dados, em paraleloSaída das somas

Abaixo está o programa Python tradicional, vinculado a um único processo, que resolve essas três tarefas:

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

Vamos pegar esse programa e reescrevê-lo para usar o multiprocessamento para paralelizar as três partes descritas acima. Abaixo está um esqueleto deste novo programa paralelo, que precisa ser aprimorado para abordar as partes nos comentários:

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

Esses trechos de código, bem comooutro pedaço de código que pode gerar arquivos CSV de exemplo para fins de teste, pode serencontrado no github.

Eu apreciaria qualquer insight aqui sobre como os gurus da concorrência abordariam esse problema.

Aqui estão algumas perguntas que eu tive ao pensar sobre esse problema. Pontos de bônus por abordar qualquer um / todos:

Devo ter processos filhos para ler os dados e colocá-los na fila ou o processo principal pode fazer isso sem bloquear até que todas as entradas sejam lidas?Da mesma forma, devo ter um processo filho para gravar os resultados da fila processada ou o processo principal pode fazer isso sem ter que esperar por todos os resultados?Devo usar umpool de processos para as operações de soma?Se sim, qual método eu chamo no pool para que ele comece a processar os resultados que entram na fila de entrada, sem bloquear também os processos de entrada e saída?apply_async ()? map_async ()? imap ()? imap_unordered ()?Suponha que não precisássemos desviar as filas de entrada e saída quando os dados as inserissem, mas poderíamos esperar até que todas as entradas fossem analisadas e todos os resultados fossem calculados (por exemplo, porque sabemos que todas as entradas e saídas caberão na memória do sistema). Devemos alterar o algoritmo de alguma forma (por exemplo, não executar nenhum processo simultaneamente com a E / S)?