Как использовать multiprocessing для решения смущающих параллельных проблем?
Тревожные параллельные проблемы обычно состоят из трех основных частей:
- Прочитать входные данные (из файла, базы данных, подключения к tcp и т.д.).
- Выполнить расчеты на входных данных, где каждый расчет не зависит от других вычислений.
- Напишите результаты вычислений (к файлу, базе данных, подключению tcp и т.д.).
Мы можем распараллелить программу в двух измерениях:
- Часть 2 может работать на нескольких ядрах, поскольку каждый расчет независим; порядок обработки не имеет значения.
- Каждая часть может работать независимо. Часть 1 может помещать данные в очередь ввода, часть 2 может вытащить данные из входной очереди и поместить результаты в очередь вывода, а часть 3 может вывести результаты из очереди вывода и записать их.
Это похоже на самый базовый шаблон в параллельном программировании, но я все еще теряюсь в попытке его решить, поэтому пусть напишет канонический пример, чтобы проиллюстрировать, как это делается с использованием многопроцессорности.
Вот пример проблемы: Учитывая CSV файл с целыми числами в качестве входных данных, вычислите их суммы. Разделите проблему на три части, которые могут выполняться параллельно:
- Обработка входного файла в необработанные данные (списки/итерации целых чисел)
- Рассчитать суммы данных параллельно
- Вывести суммы
Ниже представлена традиционная однопроцессная программа 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 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.
Я был бы признателен вам за понимание того, как вы concurrency гуру приблизились к этой проблеме.
Вот некоторые вопросы, которые я имел, когда думал об этой проблеме. Бонусные точки для адресации всех/всех:
- Должен ли я иметь дочерние процессы для чтения в данных и размещения их в очереди, или может ли основной процесс сделать это без блокировки, пока не будет прочитан все входные данные?
- Аналогично, должен ли я иметь дочерний процесс для записи результатов из обработанной очереди или может ли основной процесс сделать это без необходимости ждать всех результатов?
- Должен ли я использовать пул процессов для операций с суммой?
- Если да, какой метод я вызываю в пуле, чтобы заставить его начать обработку результатов, поступающих во входную очередь, без блокировки процессов ввода и вывода? apply_async()? map_async()? imap()? imap_unordered()?
- Предположим, нам не нужно было переходить от входных и выходных очередей по мере ввода данных, но можно было дождаться, пока все данные будут проанализированы, и все результаты будут вычислены (например, потому что мы знаем, что все входные и выходные данные будут вписываться в систему Память). Должны ли мы каким-либо образом изменить алгоритм (например, не запускать какие-либо процессы одновременно с I/O)?
