マルチプロセッシングを使用して、驚異的並列の問題にどのように取り組むのでしょうか。
驚異的並列問題は通常、次の3つの基本部分で構成されます。
- 入力データを読み取ります（ファイル、データベース、tcp接続などから）。
- 入力データに対して計算を実行します。各計算は他の計算から独立しています。
- 計算結果を（ファイル、データベース、tcp接続などに）書き込みます。
プログラムを2次元で並列化できます。
- 各計算は独立しているため、パート2は複数のコアで実行できます。処理の順序は関係ありません。
- 各パーツは独立して実行できます。パート1はデータを入力キューに配置し、パート2はデータを入力キューからプルして結果を出力キューに配置し、パート3は結果を出力キューからプルして書き出すことができます。
これは並行プログラミングの最も基本的なパターンのようですが、それを解決しようとするとまだ迷っています。そこで、マルチプロセッシングを使用してこれがどのように行われるかを示す標準的な例を書いてみましょう。
問題の例を次に示します。入力として整数の行を含むCSVファイルが与えられた場合、それらの合計を計算します。問題を3つの部分に分けます。これらはすべて、並行して実行できます。
- 入力ファイルを生データ（整数のリスト/反復可能ファイル）に処理します
- 並行してデータの合計を計算します
- 合計を出力します
以下は、これら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:])
このプログラムを書き直して、マルチプロセッシングを使用して上記の3つの部分を並列化してみましょう。以下は、この新しい並列化されたプログラムのスケルトンであり、コメントの部分に対処するために具体化する必要があります。
#!/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（）？
- データが入力されたときに入力キューと出力キューを吸い上げる必要はなかったが、すべての入力が解析され、すべての結果が計算されるまで待つことができたとします（たとえば、すべての入力と出力がシステムメモリに収まることがわかっているため）。アルゴリズムを何らかの方法で変更する必要がありますか（たとえば、I / Oと同時にプロセスを実行しないでください）？