Python에서 병렬로 공유 데이터에 대한 무거운 계산 수행

Question

Python의 병렬 처리에 대한 간단한 질문입니다. 말하자면 큰 공유 데이터 구조를 가지고 병렬로 여러 함수를 적용하려고합니다. 이 함수는 데이터 구조를 읽을 수는 있지만 결과 개체 변이를 수행하고 있습니다 :

def compute_heavy_task(self): 
    big_shared_object = self.big_shared_object 
    result_refs = self.result_refs 
    for ref in result_refs: 
     some_expensive_task(ref, big_shared_object) 

내가이 병렬 어떻게해야합니까, 한 번에 한 번에 5 말, 또는 10. 한 번에 프로세서의 수는 어떻습니까?

Answer 1

Python의 스레드 (적어도 사용하고있는 CPython 구현은 아님)로 유용하게 사용할 수 없습니다. 전역 통역관 잠금 장치는 800 %에 가까운 효율성 대신에 8 코어를 원한다면 90 % 만 얻을 수 있음을 의미합니다.

하지만 별도의 프로세스로이 작업을 수행 할 수 있습니다. 표준 라이브러리에 내장 된 두 가지 옵션, concurrent.futures 및 multiprocessing이 있습니다. 일반적으로 futures은 단순한 경우 더 간단하며 종종 작성하기가 더 쉽습니다. multiprocessing은 일반적으로 더 유연하고 강력합니다. futures도 Python 3.2 이상에서만 제공되지만, a backport for 2.5-3.1 at PyPI이 있습니다.

큰 공유 데이터 구조가있는 경우 multiprocessing의 유연성을 원하는 경우 중 하나입니다. 자세한 내용은 Sharing state between processes과 그 바로 위, 아래 및 링크에서 참조하십시오. 데이터 구조가 INT의 거대한 배열처럼, 정말 간단 경우

, 이것은 매우 간단하다 :

class MyClass(object): 
    def __init__(self, giant_iterator_of_ints): 
     self.big_shared_object = multiprocessing.Array('i', giant_iterator_of_ints) 
    def compute_heavy_task(self): 
     lock = multiprocessing.Lock() 
     def subtask(my_range): 
      return some_expensive_task(self.big_shared_object, lock, my_range) 
     pool = multiprocessing.pool.Pool(5) 
     my_ranges = split_into_chunks_appropriately(len(self.big_shared_object) 
     results = pool.map_async(subtask, my_ranges) 
     pool.close() 
     pool.join() 

참고 some_expensive_task 기능은 이제 잠금을 소요 객체 그것은을 얻을 수 있는지 확인한다 공유 객체에 대한 모든 액세스 (또는 더 자주, 하나 이상의 액세스로 구성된 모든 "트랜잭션")를 잠급니다. Lock discipline은 까다로울 수 있지만 직접 데이터 공유를 사용하려는 경우에는 주위를 둘러 볼 방법이 없습니다.

또한 my_range이 필요합니다. 같은 객체에서 같은 함수를 5 번 호출하면 5 번 똑같은 일을 할 것입니다. 아마도 그렇게 유용하지는 않을 것입니다. 일을 병렬화하는 일반적인 방법 중 하나는 각 작업에 전체 데이터 집합의 하위 범위를 지정하는 것입니다. (대충 간단히 설명 할 수있을뿐만 아니라 올바른 알고리즘을 사용하면 이런 식으로 많은 잠금을 피할 수 있습니다.)

함수을 데이터 세트으로 변환하려면 some_expensive_task을 반복적으로 사용하는 것보다 분명히 작업 할 함수 모음이 필요합니다. 그런 다음 예를 들어 이러한 함수를 반복하여 각각 apply_async을 호출 할 수 있습니다. 그러나 당신은 또한 주위를 돌릴 수 있습니다 : 단일 어플리 일러 함수를 작성하여 데이터 주변의 클로저로서 걸리는 것은 함수를 취하여 데이터에 적용합니다. 그런 다음 기능 모음에 대해 기능하는 map 만 있습니다.

또한 데이터 구조가 multiprocessing.Array으로 정의 할 수 있다고 가정했습니다. 그렇지 않다면 C 스타일의 데이터 구조를 디자인하고 ctypesArray의 데이터를 Structure으로 구현하거나 그 반대로 구현 한 다음 multiprocessing.sharedctypes 데이터를 사용해야합니다.

또한 결과 개체를 결과가 전달 된 결과로 옮겼습니다. 또한 거대하고 공유해야하는 경우 동일한 트릭을 사용하여 공유 할 수 있습니다.

---

이 작업을 진행하기 전에 실제로 데이터를 공유해야하는지 여부를 자문해야합니다. 이 방법을 사용하면 디버깅, 퍼포먼스 튜닝 등의 80 %를 잠금 시간을 추가하고 제거하는 시간을 소비하거나 더 많거나 적게 만들 수 있습니다. 불변의 데이터 구조를 지나갈 수 있다면, 80 %가 나머지 코드를 처리 할 수있는 파일, 데이터베이스 또는 거의 모든 다른 대안을 사용합니다.