Câu hỏi Giải quyết các vấn đề song song khi sử dụng Python đa xử lý


Làm thế nào để sử dụng đa xử lý để giải quyết vấn đề song song lúng túng?

Các vấn đề song song phức tạp thường bao gồm ba phần cơ bản:

  1. Đọc dữ liệu đầu vào (từ một tệp, cơ sở dữ liệu, kết nối tcp, v.v.).
  2. Chạy tính toán trên dữ liệu đầu vào, trong đó mỗi phép tính là độc lập với bất kỳ phép tính nào khác.
  3. Viết kết quả tính toán (vào một tệp, cơ sở dữ liệu, kết nối tcp, v.v.).

Chúng tôi có thể song song chương trình theo hai chiều:

  • Phần 2 có thể chạy trên nhiều lõi, vì mỗi phép tính là độc lập; thứ tự xử lý không quan trọng.
  • Mỗi phần có thể chạy độc lập. Phần 1 có thể đặt dữ liệu trên hàng đợi đầu vào, phần 2 có thể kéo dữ liệu ra khỏi hàng đợi đầu vào và đưa kết quả vào hàng đợi đầu ra, và phần 3 có thể kéo kết quả ra khỏi hàng đợi đầu ra và ghi chúng ra.

Điều này có vẻ là một mô hình cơ bản nhất trong lập trình đồng thời, nhưng tôi vẫn bị mất trong việc cố gắng giải quyết nó, vì vậy chúng ta hãy viết một ví dụ kinh điển để minh họa cách điều này được thực hiện bằng cách sử dụng đa xử lý.

Đây là vấn đề ví dụ: Tệp CSV với các hàng số nguyên làm đầu vào, tính toán tổng của chúng. Tách riêng vấn đề thành ba phần, tất cả có thể chạy song song:

  1. Xử lý tệp đầu vào thành dữ liệu thô (danh sách / vòng lặp của số nguyên)
  2. Tính tổng các dữ liệu, song song
  3. Xuất số tiền

Dưới đây là chương trình Python ràng buộc đơn, truyền thống, giải quyết ba nhiệm vụ sau:

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

Hãy lấy chương trình này và viết lại nó để sử dụng đa xử lý để song song ba phần được nêu ở trên. Dưới đây là một bộ xương của chương trình song song mới này, cần phải được xác định để giải quyết các phần trong phần bình luận:

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

Những đoạn mã này, cũng như một đoạn mã khác có thể tạo tệp CSV mẫu cho mục đích thử nghiệm, có thể tìm thấy trên github.

Tôi sẽ đánh giá cao bất kỳ cái nhìn sâu sắc ở đây là làm thế nào bạn rất kinh nghiệm đồng thời sẽ tiếp cận vấn đề này.


Dưới đây là một số câu hỏi tôi có khi nghĩ về vấn đề này. Điểm thưởng để giải quyết bất kỳ / tất cả:

  • Tôi có nên có quá trình con để đọc dữ liệu và đặt nó vào hàng đợi, hoặc quá trình chính có thể làm điều này mà không bị chặn cho đến khi tất cả đầu vào được đọc không?
  • Tương tự như vậy, tôi có nên có một tiến trình con để viết kết quả ra khỏi hàng đợi đã xử lý hay không, quá trình chính có thực hiện được điều này mà không phải đợi tất cả các kết quả không?
  • Tôi có nên sử dụng quy trình hồ bơi cho tổng số hoạt động?
    • Nếu có, phương pháp nào tôi gọi trên hồ bơi để làm cho nó bắt đầu xử lý các kết quả đi vào hàng đợi đầu vào, mà không chặn các quá trình đầu vào và đầu ra, quá? apply_async ()? map_async ()? imap ()? imap_unordered ()?
  • Giả sử chúng ta không cần phải loại bỏ hàng đợi đầu vào và đầu ra như dữ liệu đã nhập, nhưng có thể đợi cho đến khi tất cả đầu vào được phân tích cú pháp và tất cả kết quả được tính toán (ví dụ: vì chúng tôi biết tất cả đầu vào và đầu ra sẽ khớp với bộ nhớ hệ thống). Chúng tôi có nên thay đổi thuật toán theo bất kỳ cách nào (ví dụ: không chạy bất kỳ quy trình nào đồng thời với I / O) không?

76
2018-03-01 21:38


gốc


Haha, tôi thích thuật ngữ lúng túng-song song. Tôi ngạc nhiên rằng đây là lần đầu tiên tôi nghe thuật ngữ này, đó là một cách tuyệt vời để nói đến khái niệm đó. - Tom Neyland


Các câu trả lời:


Giải pháp của tôi có thêm chuông và còi để đảm bảo rằng thứ tự của đầu ra có cùng thứ tự của đầu vào. Tôi sử dụng multiprocessing.queue để gửi dữ liệu giữa các tiến trình, gửi các thông báo dừng để mỗi quá trình biết thoát khỏi việc kiểm tra các hàng đợi. Tôi nghĩ rằng các ý kiến ​​trong nguồn nên làm cho nó rõ ràng những gì đang xảy ra nhưng nếu không cho tôi biết.

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

64
2018-03-02 16:16



Đây là chỉ có câu trả lời thực sự được sử dụng multiprocessing. Tiền thưởng đến với bạn, thưa bạn. - gotgenes
Có thực sự cần thiết để gọi join về quá trình nhập và số-crunching? Bạn không thể lấy đi chỉ với tham gia vào quá trình đầu ra và bỏ qua những người khác? Nếu vậy, vẫn còn một lý do chính đáng để gọi join trên tất cả các quy trình khác? - Ryan Thompson
Câu trả lời hay, tôi phải nói. Được đánh dấu vĩnh viễn! - Blender
"để chủ đề biết để thoát" - - "gửi dữ liệu giữa các chủ đề" - Chủ đề và quy trình rất khác nhau. Tôi thấy rằng điều này có thể gây nhầm lẫn cho người mới. Điều quan trọng hơn là sử dụng thuật ngữ chính xác trong câu trả lời đã được upvoted rất nhiều. Bạn đang bắt đầu các quy trình mới tại đây. Bạn không chỉ sinh ra các luồng trong tiến trình hiện tại. - Jan-Philip Gehrcke
Đủ công bằng. Tôi đã sửa văn bản. - hbar


Tôi nhận ra rằng tôi hơi muộn cho bữa tiệc, nhưng gần đây tôi đã phát hiện ra GNU song songvà muốn thể hiện cách dễ dàng để hoàn thành nhiệm vụ điển hình này với nó.

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

Một cái gì đó như thế này sẽ làm cho 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)

Song song sẽ chạy sum.py cho mỗi dòng trong input.csv (song song, tất nhiên), sau đó xuất kết quả ra sums. Rõ ràng là tốt hơn multiprocessing rắc rối


5
2017-08-23 11:00



Các tài liệu song song của GNU sẽ gọi một trình thông dịch Python mới cho mỗi dòng trong tệp đầu vào. Chi phí cho việc bắt đầu một trình thông dịch Python mới (khoảng 30 mili giây cho Python 2.7 và 40 mili giây cho Python 3.3 trên i7 MacBook Pro của tôi với ổ đĩa trạng thái rắn) có thể lớn hơn thời gian để xử lý một dòng dữ liệu riêng lẻ và dẫn đến rất nhiều thời gian lãng phí và lợi nhuận kém hơn dự kiến. Trong trường hợp của vấn đề ví dụ của bạn, tôi có thể sẽ tiếp cận multiprocessing.Pool. - gotgenes


Sắp đến bữa tiệc ...

joblib có một lớp trên cùng của đa xử lý để giúp tạo song song cho các vòng lặp. Nó cung cấp cho bạn các tiện ích như việc gửi đi công việc lười biếng và báo cáo lỗi tốt hơn ngoài cú pháp rất đơn giản của nó.

Là một tuyên bố từ chối trách nhiệm, tôi là tác giả ban đầu của joblib.


5
2018-01-15 08:13



Vì vậy, Joblib có khả năng xử lý I / O song song hay bạn phải làm điều đó bằng tay? Bạn có thể cung cấp một mẫu mã bằng cách sử dụng Joblib? Cảm ơn! - Roko Mijic


Trường cũ.

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

Đây là cấu trúc cuối cùng đa xử lý.

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

Có, vỏ đã đan chúng lại với nhau ở cấp hệ điều hành. Nó có vẻ đơn giản với tôi và nó hoạt động rất độc đáo.

Có, có thêm một chút chi phí trong việc sử dụng dưa (hoặc cPickle). Việc đơn giản hóa, tuy nhiên, có vẻ đáng để nỗ lực.

Nếu bạn muốn tên tập tin là một đối số p1.py, đó là một thay đổi dễ dàng.

Quan trọng hơn, một chức năng như sau rất tiện dụng.

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

Điều đó cho phép bạn thực hiện việc này:

for item in get_stdin():
     process item

Điều này rất đơn giản, nhưng nó không dễ dàng cho phép bạn có nhiều bản sao của P2.py đang chạy.

Bạn có hai vấn đề: fan-out và fan-in. P1.py phải bằng cách nào đó quạt ra nhiều P2.py. Và P2.py phải bằng cách nào đó kết hợp các kết quả của chúng thành một P3.py duy nhất.

Cách tiếp cận cũ của trường học để fan-out là một "Push" kiến ​​trúc, rất hiệu quả.

Về mặt lý thuyết, nhiều P2.py kéo từ một hàng đợi phổ biến là phân bổ tối ưu các tài nguyên. Điều này thường lý tưởng, nhưng nó cũng là một số lượng công bằng của lập trình. Lập trình có thực sự cần thiết không? Hay chế biến vòng tròn sẽ đủ tốt?

Thực tế, bạn sẽ thấy rằng làm cho P1.py làm một "vòng tròn" giao dịch đơn giản giữa nhiều P2.py có thể khá tốt. Bạn đã có cấu hình P1.py để xử lý n bản sao của P2.py thông qua các đường ống có tên. P2.py sẽ đọc từ ống thích hợp của họ.

Điều gì sẽ xảy ra nếu một P2.py nhận được tất cả dữ liệu "trường hợp xấu nhất" và chạy theo cách sau? Vâng, round-robin không hoàn hảo. Nhưng nó tốt hơn chỉ một P2.py và bạn có thể giải quyết sự thiên vị này với sự ngẫu nhiên đơn giản.

Fan-in từ nhiều P2.py đến một P3.py vẫn phức tạp hơn một chút. Tại thời điểm này, cách tiếp cận trường học cũ dừng lại là thuận lợi. P3.py cần phải đọc từ nhiều ống được đặt tên bằng cách sử dụng select thư viện để xen kẽ các lần đọc.


4
2018-03-01 21:55



Điều này sẽ không trở nên tồi tệ hơn khi tôi muốn ra mắt n các trường hợp của p2.py, khiến chúng tiêu thụ và xử lý m khối của r hàng đầu ra bởi p1.py, và có p3.py nhận được mxr kết quả từ tất cả các n p2.py trường hợp? - gotgenes
Tôi không thấy yêu cầu đó trong câu hỏi. (Có lẽ câu hỏi quá dài và phức tạp để làm cho yêu cầu đó nổi bật.) Điều quan trọng là bạn nên có một lý do thực sự tốt để mong đợi rằng nhiều p2 thực sự giải quyết vấn đề hiệu suất của bạn. Trong khi chúng ta có thể giả thuyết rằng tình huống như vậy có thể tồn tại, kiến ​​trúc * nix chưa bao giờ có và không ai thấy phù hợp để thêm nó. Có thể hữu ích khi có nhiều p2. Nhưng trong 40 năm qua, không ai nhìn thấy đủ nhu cầu để làm cho nó trở thành một phần hạng nhất của vỏ. - S.Lott
Đó là lỗi của tôi, sau đó. Hãy để tôi chỉnh sửa và làm rõ điểm đó. Để giúp tôi cải thiện câu hỏi, liệu sự nhầm lẫn có đến từ việc sử dụng sum()? Đó là vì mục đích minh họa. Tôi có thể đã thay thế nó bằng do_something(), nhưng tôi muốn có một ví dụ cụ thể, dễ hiểu (xem câu đầu tiên). Trong thực tế, tôi do_something() là rất nhiều CPU, nhưng song song với nhau, vì mỗi cuộc gọi là độc lập. Do đó, nhiều lõi nhai trên đó sẽ giúp đỡ. - gotgenes
"sự nhầm lẫn đến từ việc sử dụng tổng ()?" Rõ ràng không. Tôi không chắc tại sao bạn lại nhắc đến điều đó. Bạn nói: "Nó sẽ không nhận được nhiều hơn khi tôi muốn khởi chạy n trường hợp của p2.py". Tôi không thấy yêu cầu đó trong câu hỏi. - S.Lott


Có lẽ cũng có thể giới thiệu một chút về tính song song vào phần 1. Có thể không phải là vấn đề với định dạng đơn giản như CSV, nhưng nếu việc xử lý dữ liệu đầu vào chậm hơn đáng kể so với việc đọc dữ liệu, bạn có thể đọc các khối lớn hơn, sau đó tiếp tục đọc cho đến khi tìm thấy "dấu tách hàng" ( dòng mới trong trường hợp CSV, nhưng lại phụ thuộc vào định dạng đã đọc; không hoạt động nếu định dạng đủ phức tạp).

Những khối này, mỗi phần có thể chứa nhiều mục, sau đó có thể được tạo thành một đám đông các quy trình song song đọc các công việc ngoài hàng đợi, nơi chúng được phân tách và phân tách, sau đó được đặt vào hàng đợi cho giai đoạn 2.


0
2018-03-10 16:39