Thực hiện một loại hàng đợi xử lý đặc biệt trong Python

Question

Hãy tưởng tượng cây nhị phân ngược với các nút A, B, C, D, E, F trên cấp 0. nút G, H, I ở mức 1, nút J ở mức 2, và nút K vào mức độ 3.

Level 1: G = func (A, B), H = func (C, D), I = func (E, F)

Level 2: J = func (G, H)

Cấp 3: K = func (J, I). Mỗi cặp nút trên Mức 0 phải được xử lý theo thứ tự, Mỗi cặp nút trên Cấp 1 có thể được xử lý theo bất kỳ thứ tự nào nhưng kết quả phải ở cấp độ tiếp theo phải được xử lý như được hiển thị, v.v. cho đến khi chúng tôi kết thúc với kết quả cuối cùng, K.

Vấn đề thực tế là một vấn đề hình học tính toán trong đó một chuỗi các chất rắn được hợp nhất với nhau. A tiếp giáp với B tiếp giáp với C, v.v. Cầu chì kết quả của A và B (G) tiếp giáp với cầu chì của C và D (H). Cầu chì kết quả của J và I (K) là kết quả cuối cùng. Vì vậy bạn không thể hợp nhất G và tôi vì chúng không liền kề nhau. Nếu số lượng các nút trên một mức không phải là một sức mạnh của 2, bạn kết thúc với một thực thể lủng lẳng phải được xử lý thêm một cấp.

Vì quá trình cầu chì có tính toán tốn kém và bộ nhớ chuyên sâu nhưng rất song song, tôi muốn sử dụng gói đa xử lý Python và một số hình thức xếp hàng. Sau khi tính toán G = func (A, B), tôi muốn đẩy kết quả G vào hàng đợi cho phép tính J = func (G, H) tiếp theo. Khi hàng đợi trống, kết quả cuối cùng là kết quả cuối cùng. Hãy nhớ rằng mp.queue sẽ không nhất thiết tạo ra kết quả FIFO, vì I = func (E, F) có thể kết thúc trước H = func (C, D)

Tôi đã đưa ra một vài giải pháp nhưng tôi chắc chắn có một giải pháp thanh lịch ngay ngoài tầm tay của tôi. Gợi ý?

Answer 1

Tôi không thể tìm ra thiết kế thông minh cho hàng đợi, nhưng bạn có thể dễ dàng thay thế hàng đợi bằng một quy trình khác, trong ví dụ của tôi, tôi gọi là WorkerManager. Quá trình này thu thập kết quả từ tất cả các quy trình Worker và chỉ bắt đầu công nhân mới nếu có hai gói dữ liệu liền kề đang chờ xử lý. Bằng cách này, bạn sẽ không bao giờ cố gắng tham gia các kết quả không liền kề, vì vậy bạn có thể bỏ qua "các cấp" và kích hoạt tính toán của cặp tiếp theo ngay khi nó sẵn sàng.

from multiprocessing import Process, Queue 

class Result(object): 
    '''Result from start to end.''' 
    def __init__(self, start, end, data): 
     self.start = start 
     self.end = end 
     self.data = data 


class Worker(Process): 
    '''Joins two results into one result.''' 
    def __init__(self, result_queue, pair): 
     self.result_queue = result_queue 
     self.pair = pair 
     super(Worker, self).__init__() 

    def run(self): 
     left, right = self.pair 
     result = Result(left.start, right.end, 
         '(%s, %s)' % (left.data, right.data)) 
     self.result_queue.put(result) 


class WorkerManager(Process): 
    ''' 
    Takes results from result_queue, pairs them 
    and assigns workers to process them. 
    Returns final result into final_queue. 
    ''' 
    def __init__(self, result_queue, final_queue, start, end): 
     self._result_queue = result_queue 
     self._final_queue = final_queue 
     self._start = start 
     self._end = end 
     self._results = [] 
     super(WorkerManager, self).__init__() 

    def run(self): 
     while True: 
      result = self._result_queue.get() 
      self._add_result(result) 
      if self._has_final_result(): 
       self._final_queue.put(self._get_final_result()) 
       return 
      pair = self._find_adjacent_pair() 
      if pair: 
       self._start_worker(pair) 

    def _add_result(self, result): 
     self._results.append(result) 
     self._results.sort(key=lambda result: result.start) 

    def _has_final_result(self): 
     return (len(self._results) == 1 
       and self._results[0].start == self._start 
       and self._results[0].end == self._end) 

    def _get_final_result(self): 
     return self._results[0] 

    def _find_adjacent_pair(self): 
     for i in xrange(len(self._results) - 1): 
      left, right = self._results[i], self._results[i + 1] 
      if left.end == right.start: 
       self._results = self._results[:i] + self._results[i + 2:] 
       return left, right 

    def _start_worker(self, pair): 
     worker = Worker(self._result_queue, pair) 
     worker.start() 

if __name__ == '__main__': 
    DATA = [Result(i, i + 1, str(i)) for i in xrange(6)] 
    result_queue = Queue() 
    final_queue = Queue() 
    start = 0 
    end = len(DATA) 
    man = WorkerManager(result_queue, final_queue, start, end) 
    man.start() 
    for res in DATA: 
     result_queue.put(res) 
    final = final_queue.get() 
    print final.start 
    # 0 
    print final.end 
    # 6 
    print final.data 
    # For example: 
    # (((0, 1), (2, 3)), (4, 5)) 

Ví dụ của tôi, tôi đã sử dụng một đơn giản Worker trả về dữ liệu được đưa ra trong dấu ngoặc đơn, cách nhau bằng một dấu phẩy, nhưng bạn có thể đặt bất kỳ tính toán trong đó. Trong trường hợp của tôi, kết quả cuối cùng là (((0, 1), (2, 3)), (4, 5)) có nghĩa là thuật toán được tính (0, 1) và (2, 3) trước khi tính toán ((0, 1), (2, 3)) và sau đó tham gia kết quả với (4, 5). Tôi hy vọng đó là điều mà bạn đang tìm kiếm.