Chia sẻ biến giữa các quá trình

Question

Tôi có chức năng tải xuống tải xuống nhiều tệp song song. Tôi sử dụng multiprocessing.Pool.map_async để tải xuống các phần khác nhau của cùng một tệp. Tôi muốn hiển thị thanh trạng thái của quá trình tải xuống. Đối với điều này, tôi cần phải biết tổng số byte đã được tải xuống (total_bytes_dl).

pool = multiprocessing.Pool(processes) 
    mapObj = pool.map_async(f, args) 

    while not mapObj.ready(): 
     status = r"%.2f MB/%.2f MB" % (total_bytes_dl/1024.0/1024.0, filesize/1024.0/1024.0,) 
     status = status + chr(8)*(len(status)+1) 
     print status, 
     time.sleep(0.5) 

Có cách nào để đặt biến sẽ được chia sẻ giữa tất cả các quy trình này VÀ quy trình chính, vì vậy, mọi quá trình có thể nối thêm số lượng byte vừa tải xuống?

Answer 1

Giải pháp là intilize quá trình mới và vượt qua ctypes giá trị chia sẻ:

from ctypes import c_int 
import dummy 

shared_bytes_var = multiprocessing.Value(c_int) 

def Func(...): 
    .... 
    pool = multiprocessing.Pool(initializer=_initProcess,initargs=(shared_bytes_var,)) 
    .... 

def _initProcess(x): 
    dummy.shared_bytes_var = x 

Answer 2

Chắc chắn, bạn có thể sử dụng các giá trị được chia sẻ ctypes trong bộ nhớ dùng chung, nếu bạn chỉ muốn tải xuống byte cần làm. chuyển giá trị có liên quan cho mỗi công nhân và quá trình gọi sẽ có quyền truy cập vào nó.

xem: http://docs.python.org/library/multiprocessing.html#shared-ctypes-objects

Answer 3

Bạn có thể sử dụng một đối tượng Queue đa tiến mà người lao động có thể sử dụng để gửi dữ liệu trạng thái trên. Quá trình chính của bạn sẽ phải đọc các mục trạng thái từ hàng đợi và cập nhật trạng thái cho phù hợp.

Answer 4

Sử dụng một đối tượng Queue được phân bổ như thế này:

que = multiprocessing.Manager().Queue() 

Vượt qua biến này cho người lao động, và họ có thể sử dụng que.put(bytes) để định kỳ báo cáo số tiền họ đã tải xuống kể từ báo cáo mới nhất của họ. Bạn sau đó chỉ cần kiểm tra kích thước hàng đợi và kéo trong bất kỳ báo cáo đến:

downloaded = 0 
while not mapObj.ready(): 
    for _ in range(q.qsize()): 
     downloaded += q.get() 
    print downloaded, r"bytes downloaded\r", 
    time.sleep(0.5) 

Lưu ý: Mặc dù module cũng cung cấp một phương pháp multiprocessing.Queue(), nó không phải là hoàn toàn tương đương với multiprocessing.Manager().Queue(). Xem this question và câu trả lời.