Tôi có cần truyền một cách rõ ràng các phép đa xử lý.Queue cho một tiến trình con thực hiện trên một phương thức cá thể không?

Question

Tôi có vài câu hỏi cơ bản khi nói đến sử dụng mô-đun multiprocessing Python:

class Someparallelworkerclass(object) : 

    def __init__(self): 
     self.num_workers = 4 
     self.work_queue = multiprocessing.JoinableQueue() 
     self.result_queue = multiprocessing.JoinableQueue() 

    def someparallellazymethod(self): 
     p = multiprocessing.Process(target=self.worktobedone).start() 

    def worktobedone(self): 
     # get data from work_queue 
     # put back result in result queue 

Có cần phải vượt qua work_queue và result_queue như args để Process? Câu trả lời có phụ thuộc vào hệ điều hành không? Câu hỏi cơ bản hơn là: liệu tiến trình con có được một không gian địa chỉ được sao chép (COW) từ tiến trình cha hay không, và do đó biết được định nghĩa của lớp/phương thức lớp? Nếu có, làm thế nào để biết rằng các hàng đợi được chia sẻ cho IPC, và rằng nó không nên làm cho bản sao của work_queue và result_queue trong quá trình con? Tôi đã thử tìm kiếm trực tuyến này nhưng hầu hết các tài liệu tôi tìm thấy là mơ hồ, và không đi vào đủ chi tiết như những gì chính xác đang xảy ra bên dưới.

Answer 1

Quy trình con không nhận được không gian địa chỉ được sao chép. Đứa trẻ là một quá trình python hoàn toàn riêng biệt mà không có gì chia sẻ. Có, bạn phải vượt qua hàng đợi cho đứa trẻ. Khi bạn làm như vậy, đa xử lý tự động xử lý việc chia sẻ qua IPC. Xem https://docs.python.org/2/library/multiprocessing.html#exchanging-objects-between-processes.

Answer 2

Thực tế không cần phải bao gồm hàng đợi trong đối số args trong trường hợp này, bất kể bạn đang sử dụng nền tảng nào. Lý do là mặc dù nó không giống như bạn đang rõ ràng thông qua hai trường hợp JoinableQueue cho đứa trẻ, bạn thực sự là - thông qua self. Bởi vì selflà một cách rõ ràng được truyền cho trẻ, và hai hàng đợi là một phần của self, chúng sẽ được truyền cho trẻ.

Trên Linux, điều này xảy ra qua os.fork(), có nghĩa là file descriptor được sử dụng bởi các đối tượng multiprocessing.connection.Connection rằng Queue sử dụng nội bộ cho quá trình liên lạc là thừa hưởng bởi đứa trẻ (không sao chép). Các bộ phận khác của Queue trở thành copy-on-write, nhưng điều đó là ổn; multiprocessing.Queue được thiết kế sao cho không có phần nào cần được sao chép thực sự cần phải được đồng bộ giữa hai quy trình. Trên thực tế, nhiều thuộc tính nội bộ được đặt lại sau fork xảy ra:

Vì vậy, bao gồm Linux. Làm thế nào về Windows? Windows không có fork, vì vậy nó sẽ cần phải chọn self để gửi cho đứa trẻ, và điều đó bao gồm tẩy của chúng tôi Queues. Bây giờ, thông thường nếu bạn cố gắng chọn một số multiprocessing.Queue, nó không thành công:

>>> import multiprocessing 
>>> q = multiprocessing.Queue() 
>>> import pickle 
>>> pickle.dumps(q) 
Traceback (most recent call last): 
    File "<stdin>", line 1, in <module> 
    File "/usr/local/lib/python2.7/pickle.py", line 1374, in dumps 
    Pickler(file, protocol).dump(obj) 
    File "/usr/local/lib/python2.7/pickle.py", line 224, in dump 
    self.save(obj) 
    File "/usr/local/lib/python2.7/pickle.py", line 306, in save 
    rv = reduce(self.proto) 
    File "/usr/local/lib/python2.7/copy_reg.py", line 84, in _reduce_ex 
    dict = getstate() 
    File "/usr/local/lib/python2.7/multiprocessing/queues.py", line 77, in __getstate__ 
    assert_spawning(self) 
    File "/usr/local/lib/python2.7/multiprocessing/forking.py", line 52, in assert_spawning 
    ' through inheritance' % type(self).__name__ 
RuntimeError: Queue objects should only be shared between processes through inheritance 

Nhưng đây thực sự là giới hạn nhân tạo. multiprocessing.Queue đối tượng có thể được chọn trong một số trường hợp - cách khác chúng có thể được gửi đến các quy trình con trong Windows? Và quả thực, chúng ta có thể thấy rằng nếu chúng ta nhìn vào thực hiện:

def __getstate__(self): 
    assert_spawning(self) 
    return (self._maxsize, self._reader, self._writer, 
      self._rlock, self._wlock, self._sem, self._opid) 

def __setstate__(self, state): 
    (self._maxsize, self._reader, self._writer, 
    self._rlock, self._wlock, self._sem, self._opid) = state 
    self._after_fork() 

__getstate__, được gọi là khi tẩy một ví dụ, có một cuộc gọi assert_spawning trong nó, mà làm cho chắc chắn chúng tôi đang thực sự đẻ trứng một quá trình trong khi cố gắng dưa chua *. __setstate__, được gọi là trong khi bỏ chọn, có trách nhiệm gọi số _after_fork.

Vậy các đối tượng Connection được sử dụng bởi hàng đợi được duy trì khi chúng ta phải dưa như thế nào?Hóa ra có một mô-đun phụ multiprocessing thực hiện chính xác điều đó - multiprocessing.reduction. Các bình luận ở phía trên cùng của mô-đun khẳng định nó khá rõ ràng:

# 
# Module to allow connection and socket objects to be transferred 
# between processes 
# 

Trên Windows, các mô-đun cuối cùng sử dụng DuplicateHandle API cung cấp bởi Windows để tạo ra một xử lý trùng lặp quá trình con Connection đối tượng có thể sử dụng. Vì vậy, trong khi mỗi quá trình được xử lý riêng, chúng là các bản sao chính xác - bất kỳ hành động nào được thực hiện trên một được phản ánh trên khác:

Xử lý trùng lặp đề cập đến cùng một đối tượng. Do đó, mọi thay đổi đối tượng được phản ánh qua cả hai tay cầm . Ví dụ, nếu bạn sao chép một tập tin xử lý, tập tin hiện tại vị trí là luôn luôn giống nhau cho cả hai tay cầm.

_{* Xem this answer để biết thêm thông tin về assert_spawning}

Answer 3

Quá trình con không có hàng đợi trong việc đóng cửa của nó. Đó là trường hợp của hàng đợi tham chiếu các khu vực khác nhau của bộ nhớ. Khi sử dụng hàng đợi theo cách bạn dự định, bạn phải chuyển chúng như args cho hàm. một giải pháp tôi thích là sử dụng functools.partial để curry chức năng của bạn với hàng đợi bạn muốn, thêm chúng vĩnh viễn để đóng cửa của nó và cho phép bạn quay lên nhiều chủ đề để thực hiện nhiệm vụ tương tự với cùng một kênh IPC.