Sử dụng chế độ mặc định với đa xử lý?

Question

Chỉ cần thử nghiệm và tìm hiểu và tôi biết cách tạo một từ điển được chia sẻ có thể được truy cập với nhiều trình xử lý nhưng tôi không chắc chắn cách giữ đồng bộ hóa dict. defaultdict, tôi tin, minh họa sự cố tôi đang gặp phải.

from collections import defaultdict 
from multiprocessing import Pool, Manager, Process 

#test without multiprocessing 
s = 'mississippi' 
d = defaultdict(int) 
for k in s: 
    d[k] += 1 

print d.items() # Success! result: [('i', 4), ('p', 2), ('s', 4), ('m', 1)] 
print '*'*10, ' with multiprocessing ', '*'*10 

def test(k, multi_dict): 
    multi_dict[k] += 1 

if __name__ == '__main__': 
    pool = Pool(processes=4) 
    mgr = Manager() 
    multi_d = mgr.dict() 
    for k in s: 
     pool.apply_async(test, (k, multi_d)) 

    # Mark pool as closed -- no more tasks can be added. 
    pool.close() 

    # Wait for tasks to exit 
    pool.join() 

    # Output results 
    print multi_d.items() #FAIL 

print '*'*10, ' with multiprocessing and process module like on python site example', '*'*10 
def test2(k, multi_dict2): 
    multi_dict2[k] += 1 


if __name__ == '__main__': 
    manager = Manager() 

    multi_d2 = manager.dict() 
    for k in s: 
     p = Process(target=test2, args=(k, multi_d2)) 
    p.start() 
    p.join() 

    print multi_d2 #FAIL 

Kết quả đầu tiên làm việc (vì nó không sử dụng multiprocessing), nhưng tôi đang gặp vấn đề nhận được nó để làm việc với multiprocessing. Tôi không chắc chắn làm thế nào để giải quyết nó nhưng tôi nghĩ rằng có thể là do nó không được đồng bộ hóa (và tham gia các kết quả sau này) hoặc có thể bởi vì trong vòng multiprocessing Tôi không thể tìm cách đặt defaultdict(int) vào từ điển.

Bất kỳ trợ giúp hoặc đề xuất nào về cách thực hiện công việc này sẽ tuyệt vời!

Answer 1

Bạn có thể phân lớp BaseManager và đăng ký loại bổ sung để chia sẻ. Bạn cần cung cấp loại proxy phù hợp trong trường hợp loại mặc định AutoProxy không hoạt động. Đối với defaultdict, nếu bạn chỉ cần truy cập các thuộc tính đã có trong dict, bạn có thể sử dụng DictProxy.

from multiprocessing import Pool 
from multiprocessing.managers import BaseManager, DictProxy 
from collections import defaultdict 

class MyManager(BaseManager): 
    pass 

MyManager.register('defaultdict', defaultdict, DictProxy) 

def test(k, multi_dict): 
    multi_dict[k] += 1 

if __name__ == '__main__': 
    pool = Pool(processes=4) 
    mgr = MyManager() 
    mgr.start() 
    multi_d = mgr.defaultdict(int) 
    for k in 'mississippi': 
     pool.apply_async(test, (k, multi_d)) 
    pool.close() 
    pool.join() 
    print multi_d.items() 

Answer 2

Vâng, lớp Manager dường như chỉ cung cấp số lượng cố định cấu trúc dữ liệu được xác định trước có thể được chia sẻ giữa các quy trình và defaultdict không nằm trong số đó. Nếu bạn thực sự chỉ cần rằng một defaultdict, giải pháp đơn giản nhất để thực hiện các hành vi mặc định trên của riêng bạn:

def test(k, multi_dict): 
    if k not in multi_dict: 
     multi_dict[k] = 0 
    multi_dict[k] += 1