78

Làm thế nào để một người sử dụng đa xử lý để giải quyết các vấn đề song song đáng xấu hổ ?

Các vấn đề song song rõ ràng thường bao gồm ba phần cơ bản:

  1. Đọc dữ liệu đầu vào (từ tệp, cơ sở dữ liệu, kết nối tcp, v.v.).
  2. Chạy các tính toán trên dữ liệu đầu vào, trong đó mỗi phép tính độ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 ta 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 thành vấn đề.
  • 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.

Đâ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 lạc lõng khi cố gắng giải quyết nó, vì vậy hãy viết một ví dụ kinh điển để minh họa cách thực hiện bằng cách sử dụng đa xử lý .

Đây là vấn đề mẫu: Đưa ra một tệp CSV với các hàng số nguyên làm đầu vào, tính tổng của chúng. Phân tách 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 / iterables của số nguyên)
  2. Tính tổng các dữ liệu, song song
  3. Xuất ra các khoản tiền

Dưới đây là chương trình Python ràng buộc một quy trình truyền thống, giải quyết ba tác 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 để sử dụng đa xử lý để song song ba phần được nêu ở trên. Dưới đây là bộ xương của chương trình song song mới này, cần được bổ sung để giải quyết các phần trong các 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:])

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

Tôi sẽ đánh giá cao bất kỳ cái nhìn sâu sắc nào ở đây về cách các bậc thầy đồ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 cho việc giải quyết bất kỳ / tất cả:

  • Tôi có nên có các quy trình con để đọc dữ liệu và đặt nó vào hàng đợi hay quy trình chính có thể thực hiện việc này mà không 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 quy trình con để viết kết quả từ hàng đợi đã xử lý hay quy trình chính có thể làm điều này mà không phải chờ tất cả các kết quả không?
  • Tôi có nên sử dụng nhóm quy trình cho các hoạt động tổng không?
  • Giả sử chúng ta không cần phải lọc ra hàng đợi đầu vào và đầu ra khi dữ liệu được nhập vào, 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ả các kết quả đã được tính toán (ví dụ: vì chúng ta biết tất cả đầu vào và đầu ra sẽ phù hợp với bộ nhớ hệ thống). Chúng ta 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?
|
  • 2

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

    – Đỗ Ðắc Thành 17:28:30 11/03/2010
65

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 giống với thứ tự của đầu vào. Tôi sử dụng Multiprocessing.queue để gửi dữ liệu giữa các quy trình, gửi thông báo dừng để mỗi quy trình biết thoát khỏi việc kiểm tra hàng đợi. Tôi nghĩ rằng các ý kiến ​​trong nguồn sẽ 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:])
|
  • 1

    Đây là câu trả lời duy nhất thực sự được sử dụng multiprocessing. Tiền thưởng đi đến bạn, thưa ông.

    – Trịnh Yến My 16:59:41 12/03/2010
  • 1

    Có thực sự cần thiết để gọi joincác quá trình đầu vào và số giòn? Bạn không thể thoát khỏi việc chỉ 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 tốt để kêu gọi jointất cả các quá trình khác?

    – Lý Cẩm Hân 03:41:43 04/12/2011
  • 1

    Câu trả lời đẹp, tôi phải nói. Chắc chắn được đánh dấu mãi mãi!

    – Trịnh Viễn Thông 05:19:29 09/03/2012
  • 1

    "Vì vậy, các 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 một câu trả lời đã được nâng cấp rất nhiều. Bạn đang bắt đầu các quy trình mới ở đây. Bạn không chỉ sinh ra các chủ đề trong quy trình hiện tại.

    – Đặng Kiều Minh 17:42:14 02/03/2015
  • 1

    Đủ công bằng. Tôi đã sửa văn bản.

    – Hoàng Minh Thủy 22:07:03 03/03/2015
5

Đến muộn trong 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 phương tiện như lười biếng gửi công việc 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 từ chối trách nhiệm, tôi là tác giả ban đầu của joblib.

|
  • 1

    Vì vậy, Joblib có khả năng xử lý song song I / O 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!

    – Trịnh Yến My 10:31:53 08/08/2017
5

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

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.pycho mọi dòng trong input.csv(song song, tất nhiên), sau đó xuất kết quả sang sums. Rõ ràng tốt hơn multiprocessingrắc rối

|
  • 1

    Các tài liệu song song GNU sẽ gọi trình thông dịch Python mới cho mỗi dòng trong tệp đầu vào. Chi phí khởi động 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 MacBook Pro i7 của tôi với ổ đĩa trạng thái rắn) có thể vượt quá thời gian cần thiết để xử lý một dòng dữ liệu riêng lẻ và dẫn đến một 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 vấn đề ví dụ của bạn, tôi có thể sẽ tiếp cận với đa xử lý.Pool .

    – Trịnh Yến My 16:02:31 23/08/2013
4

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

Vâng, vỏ đã đan chúng lại với nhau ở cấp độ HĐH. Nó có vẻ đơn giản hơn với tôi và nó hoạt động rất độc đáo.

Có, có một chút chi phí sử dụng dưa chua (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 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 làm điều 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 P2.txt đang chạy.

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

Cách tiếp cận trường học cũ đối với fan-out là kiến ​​trúc "Đẩy", rất hiệu quả.

Về mặt lý thuyết, việc kéo nhiều P2.txt từ một hàng đợi chung là sự phân bổ tài nguyên tối ưu. Điều này thường là lý tưởng, nhưng nó cũng là một số lượng lớn lập trình. Là chương trình thực sự cần thiết? Hoặc xử lý vòng tròn sẽ đủ tốt?

Trên thực tế, bạn sẽ thấy rằng làm cho P1.py thực hiện một giao dịch "vòng tròn" đơn giản giữa nhiều P2.txt 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 dẫn được đặt tên. Mỗi P2.txt sẽ đọc từ đường ố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 phía sau? Vâng, vòng tròn không hoàn hảo. Nhưng nó tốt hơn chỉ có một P2.py và bạn có thể giải quyết sự thiên vị này bằng cách ngẫu nhiên đơn giản.

Việc chuyển đổi từ nhiều P2.txt sang một P3.py phức tạp hơn một chút. Tại thời điểm này, phương pháp trường học cũ dừng lại là lợi thế. P3.py cần đọc từ nhiều ống có tên bằng selectthư viện để xen kẽ các lần đọc.

|
  • 1

    Điều này sẽ không nhận được hairier khi tôi muốn khởi chạy các nphiên bản của p2.py, chúng có tiêu thụ và xử lý mcác khối rđầu ra của hàng theo p1.py và p3.py có nhận được kết quả mx rtừ tất cả các phiên bản np2.py không?

    – Trịnh Yến My 22:06:49 01/03/2010
  • 1

    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 đó trở nên 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. Mặc dù chúng ta có thể đưa ra giả thuyết rằng một tình huống như vậy có thể tồn tại, kiến ​​trúc * nix chưa bao giờ có điều đó và không ai thấy phù hợp để thêm nó. Nó có thể hữu ích để có nhiều p2. Nhưng trong 40 năm qua, không ai thấy đủ nhu cầu để biến nó thành một phần hạng nhất của vỏ.

    – Lý Cẩm Hân 22:14:10 01/03/2010
  • 1

    Đó là lỗi của tôi. 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, sự nhầm lẫn đến từ việc sử dụng sum()? Đó là cho mục đích minh họa. Tôi có thể đã thay thế nó bằng do_something(), nhưng tôi muốn một ví dụ cụ thể, dễ hiểu (xem câu đầu tiên). Trong thực tế, tôi do_something()rất chuyên sâu về CPU, nhưng song song hóa một cách rõ ràng, vì mỗi cuộc gọi là độc lập. Do đó, nhiều lõi nhai sẽ giúp.

    – Trịnh Viễn Thông 22:38:04 01/03/2010
  • 1

    "sự nhầm lẫn đến từ việc sử dụng sum ()?" Rõ ràng không. Tôi không chắc tại sao bạn lại đề cập đến nó. Bạn nói: "Điều này sẽ không nhận được hairier khi tôi muốn khởi chạy n phiên bản của p2.py". Tôi không thấy yêu cầu đó trong câu hỏi.

    – Đặng Kiều Minh 23:28:28 01/03/2010
0

Có lẽ cũng có thể giới thiệu một chút 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 đoạn lớn hơn, sau đó tiếp tục đọc cho đến khi bạn tìm thấy "dấu tách hàng" ( dòng mới trong trường hợp CSV, nhưng một lần nữa phụ thuộc vào định dạng đã đọc, không hoạt động nếu định dạng đủ phức tạp).

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

|

Câu trả lời của bạn (> 20 ký tự)

Bằng cách click "Đăng trả lời", bạn đồng ý với Điều khoản dịch vụ, Chính sách bảo mật and Chính sách cookie của chúng tôi.

Không tìm thấy câu trả lời bạn tìm kiếm? Duyệt qua các câu hỏi được gắn thẻ hoặc hỏi câu hỏi của bạn.