큰 텍스트를 분석하기 위해 파이썬의 다중 처리를 사용하고 있습니다. 내 코드가 중단 된 이유 (예 : 프로세스가 종료되지 않음)를 파악하려고 며칠 후 다음과 같은 간단한 코드로 문제를 재현 할 수있었습니다.
import multiprocessing as mp
for y in range(65500, 65600):
print(y)
def func(output):
output.put("a"*y)
if __name__ == "__main__":
output = mp.Queue()
process = mp.Process(target = func, args = (output,))
process.start()
process.join()
보시다시피 대기열에 넣을 항목이 너무 커지면 프로세스가 중단됩니다. output.put()실행 후 더 많은 코드를 작성하면 멈추지 않지만 여전히 프로세스가 중지되지 않습니다.
이것은 문자열이 65500 자에 도달 할 때 발생하며 인터프리터에 따라 다를 수 있습니다.
나는 그 알고 있었다 mp.Queue이 maxsize인수를하지만 항목 수의 큐의 크기입니다 내가 발견 약간의 검색을 수행, 상품 자체의하지 크기.
이 문제를 해결할 방법이 있습니까? 원래 코드의 큐에 넣어야하는 데이터는 매우 큽니다.
대기열을 비울 소비자가 없습니다.
정의에서 Queue.put:
선택적 인수 블록이 True (기본값)이고 시간 제한이 없음 (기본값)이면 빈 슬롯을 사용할 수있을 때까지 필요한 경우 차단합니다.
생산자와 소비자 사이에 교착 상태가 없다고 가정하고 (샘플이 없기 때문에 원본 코드에 소비자가 있다고 가정) 결국 생산자는 잠금을 해제하고 종료해야합니다. 소비자의 코드를 확인하십시오 (또는 질문에 추가하여 살펴 보겠습니다).
큐에 최대 크기가 지정되지 않았으므로 메모리가 부족할 때까지 넣어야하므로 이것은 문제가되지 않습니다.
이것은 Queue의 동작이 아닙니다. 이 티켓 에서 자세히 설명했듯이 여기서 차단하는 부분은 대기열 자체가 아니라 기본 파이프입니다. 링크 된 리소스에서 ( "[]"사이의 삽입은 내 것임) :
큐는 다음과 같이 작동합니다.-queue.put (data)를 호출하면 데이터가 deque에 추가되어 영원히 늘어나고 줄어들 수 있습니다. 그런 다음 스레드가 deque에서 요소를 꺼내서 다른 프로세스가 할 수 있도록 전송합니다. 파이프 또는 Unix 소켓 (소켓 쌍을 통해 생성됨)을 통해 수신합니다. 그러나 이것이 중요한 점입니다. 파이프와 유닉스 소켓은 모두 제한된 용량을 가지고 있습니다. . -queue.get ()을 수행 할 때 파이프 / 소켓에서 읽기만 수행합니다.
[..] 크기가 [너무 커지면] 쓰기 스레드가 쓰기 시스템 호출에서 차단됩니다. 그리고 항목을 dequeing하기 전에 조인이 수행되기
process.join때문에 조인이 송신 스레드가 완료 될 때까지 대기하고 파이프 / 소켓이 가득 차서 쓰기를 완료 할 수 없기 때문에 교착 상태 일뿐입니다! 제출자 프로세스를 기다리기 전에 항목을 대기열에서 빼면 모든 것이 제대로 작동합니다.
이해 했어요. 그러나 실제로 소비자가 없습니다 (내가 생각하는 것이라면), 프로세스가 대기열에 넣을 때만 대기열에서 결과를 얻습니다.
네, 이것이 문제입니다. 는 multiprocessing.Queue저장 용기가 아닙니다. "생산자"(대기열에 들어가는 데이터를 생성하는 프로세스)와 "소비자 (해당 데이터를"사용 "하는 프로세스)간에 데이터를 전달하는 데만 사용해야합니다. 이제 알고 있듯이 데이터를 그대로 두는 것은 좋지 않습니다. .
대기열에 항목을 먼저 넣을 수없는 경우 대기열에서 항목을 가져 오려면 어떻게해야합니까?
put그리고 get멀리 숨기기가 파이프를 채우는 경우에만 귀하의 "주"과정에서 루프를 설정해야합니다, 그래서 함께 데이터를 넣는 문제 get목록에 추가, 예를 들어, 큐 중 항목을. 목록은 메인 프로세스의 메모리 공간에 있으며 파이프를 막지 않습니다.
이 기사는 인터넷에서 수집됩니다. 재 인쇄 할 때 출처를 알려주십시오.
침해가 발생한 경우 연락 주시기 바랍니다[email protected] 삭제
몇 마디 만하겠습니다