Làm thế nào để ghi vào một biến chia sẻ trong python joblib

Question

Đoạn mã sau song song một vòng lặp for.

import networkx as nx; 
import numpy as np; 
from joblib import Parallel, delayed; 
import multiprocessing; 

def core_func(repeat_index, G, numpy_arrary_2D): 
    for u in G.nodes(): 
    numpy_arrary_2D[repeat_index][u] = 2; 
    return; 

if __name__ == "__main__": 
    G = nx.erdos_renyi_graph(100000,0.99); 
    nRepeat = 5000; 
    numpy_array = np.zeros([nRepeat,G.number_of_nodes()]); 
    Parallel(n_jobs=4)(delayed(core_func)(repeat_index, G, numpy_array) for repeat_index in range(nRepeat)); 
    print(np.mean(numpy_array)); 

Như có thể thấy, giá trị dự kiến ​​sẽ được in là 2. Tuy nhiên, khi tôi chạy mã của tôi trên một cụm (đa lõi, bộ nhớ chia sẻ), nó sẽ trả 0.0.

Tôi nghĩ rằng vấn đề là mỗi nhân viên tạo ra một bản sao riêng của đối tượng numpy_array và đối tượng được tạo trong chức năng chính không được cập nhật. Làm thế nào tôi có thể sửa đổi mã sao cho mảng có thể được cập nhật numpy_array?

Answer 1

joblib sử dụng hồ bơi đa của quá trình theo mặc định, như its manual nói:

Dưới mui xe, các đối tượng song song tạo ra một hồ bơi đa mà dĩa trình thông dịch Python trong nhiều quy trình để thực hiện mỗi các mục trong danh sách. Hàm bị trì hoãn là một thủ thuật đơn giản là có thể tạo một bộ (hàm, arg, kwargs) với cú pháp gọi hàm .

Điều đó có nghĩa là mọi quá trình thừa kế trạng thái ban đầu của mảng, nhưng bất cứ điều gì nó viết bên trong nó, đều bị mất khi quá trình thoát. Chỉ có kết quả chức năng được gửi trở lại quá trình gọi (chính). Nhưng bạn không trả lại bất cứ điều gì, vì vậy None được trả lại.

Để làm cho mảng được chia sẻ có thể điều chỉnh được, bạn có hai cách: sử dụng chuỗi và sử dụng bộ nhớ dùng chung.

---

Chủ đề, không giống như quy trình, chia sẻ bộ nhớ. Vì vậy, bạn có thể viết cho mảng và mọi công việc sẽ thấy thay đổi này. Theo hướng dẫn joblib, nó được thực hiện theo cách này:

Parallel(n_jobs=4, backend="threading")(delayed(core_func)(repeat_index, G, numpy_array) for repeat_index in range(nRepeat)); 

Khi bạn chạy nó:

$ python r1.py 
2.0 

Tuy nhiên, khi bạn sẽ được viết những điều phức tạp thành các mảng, chắc chắn bạn xử lý đúng đắn các ổ khóa xung quanh dữ liệu hoặc các mảnh dữ liệu, hoặc bạn sẽ đạt đến điều kiện cuộc đua (google nó).

Cũng đọc kỹ về GIL, vì đa luồng tính toán trong Python bị giới hạn (không giống như đa luồng I/O).

---

Nếu bạn vẫn cần các quá trình (ví dụ vì GIL), bạn có thể đặt mảng đó vào bộ nhớ chia sẻ.

Đây là chủ đề phức tạp hơn một chút, nhưng joblib + numpy shared memory example cũng được hiển thị trong hướng dẫn sử dụng joblib.

Answer 2

Như Sergey đã viết trong câu trả lời của mình, các quy trình không chia sẻ trạng thái và bộ nhớ. Đây là lý do tại sao bạn không thấy câu trả lời mong đợi.

Chủ đề chia sẻ trạng thái và không gian bộ nhớ, khi chúng chạy trong cùng một quá trình. Điều này rất hữu ích nếu bạn có nhiều hoạt động I/O.Nó sẽ không giúp bạn có được sức mạnh xử lý hơn (nhiều CPU) vì GIL

Một kỹ thuật để giao tiếp giữa các quá trình là Proxy Đối tượng sử dụng quản lý. Bạn tạo một đối tượng người quản lý, nó đồng bộ hóa các tài nguyên giữa các tiến trình.

Đối tượng quản lý được trả về bởi Manager() điều khiển quá trình máy chủ chứa đối tượng Python và cho phép các quy trình khác thao tác chúng bằng proxy.

tôi đã không kiểm tra mã này (tôi không có tất cả các module bạn sử dụng), và nó có thể đòi hỏi nhiều sửa đổi mã, nhưng sử dụng đối tượng quản lý nó sẽ giống như thế này

if __name__ == "__main__": 
    G = nx.erdos_renyi_graph(100000,0.99); 
    nRepeat = 5000; 

    manager = multiprocessing.Manager() 
    numpys = manager.list(np.zeros([nRepeat, G.number_of_nodes()]) 

    Parallel(n_jobs=4)(delayed(core_func)(repeat_index, G, numpys, que) for repeat_index in range(nRepeat)); 
    print(np.mean(numpys));