파이썬 생산자 - 소비자 잠금없는 접근 방식이 스레드로부터 안전한가요?

Question

나는 최근에 간단한 생산자/소비자 패턴을 사용하는 프로그램을 작성했습니다. 처음에는 스레딩의 부적절한 사용과 관련된 버그가있었습니다. 그러나 그것은 잠김없는 방식으로 생산자/소비자 패턴을 구현할 수 있는지 생각하게했습니다.

• 한 프로듀서 스레드 : 내 경우에는

  요구 사항은 간단했다.

• 소비자 스레드 한 개.
• 대기열에는 하나의 항목에 대한 장소가 있습니다.
• 생산자는 현재 항목이 소비되기 전에 다음 항목을 생성 할 수 있습니다. 따라서 현재 아이템은 없어지지 만 괜찮습니다.
• 소비자는 다음 항목이 생성되기 전에 현재 항목을 소비 할 수 있습니다. 따라서 현재 항목은 두 번 (또는 그 이상) 소비되지만, 괜찮습니다.

그래서 내가 쓴 :

QUEUE_ITEM = None 

# this is executed in one threading.Thread object 
def producer(): 
    global QUEUE_ITEM 
    while True: 
     i = produce_item() 
     QUEUE_ITEM = i 

# this is executed in another threading.Thread object 
def consumer(): 
    global QUEUE_ITEM 
    while True: 
     i = QUEUE_ITEM 
     consume_item(i) 

내 질문은 :이 코드 스레드 - 안전합니까?

즉석 코멘트 :이 코드는 실제로는 잠금이 없습니다. CPython을 사용하며 GIL이 있습니다.

코드를 조금 테스트했는데 제대로 작동하는 것 같습니다. 그것은 GIL 때문에 원자 적 인 일부로드 및 저장 연산으로 변환됩니다. 그러나 x가 __del__ 메소드를 구현할 때 del x 조작이 원자가 아님을 또한 알고 있습니다. 따라서 내 항목에 __del__ 메쏘드가 있고 약간의 불규칙한 스케쥴링이 발생하면 일이 깨질 수 있습니다. 안 그래요?

또 다른 질문은 위의 코드가 제대로 작동하도록하려면 어떤 종류의 제한 사항 (예 : 생산 된 항목 유형)을 사용해야합니까?

내 질문은 이론적으로 CPython 및 GIL의 단점을 악용하여 잠금이 필요 없음 (즉, 코드에서 threading.Lock을 명시 적으로 잠그지 않음) 솔루션을 제시합니다.

Answer 1

예 이것은 당신이 설명하는 방식으로 작동합니다 : 생산자는 건너 뛸 요소를 생성 할 수

1. 있다.
2. 소비자가 동일한 요소를 소비 할 수 있음을 나타냅니다.

은 그러나 나는 또한 델 X 작업이 때 X를 구현 델 방법 원자되지 않았 음을 알 수 있습니다. 따라서 내 항목이 메쏘드이고 약간의 불규칙한 스케쥴링이 발생하면 일이 깨질 수 있습니다.

여기에 "del"이 표시되지 않습니다. 만약 del가 consume_item에서 발생하면 del이 생성자 스레드에서 발생할 수 있습니다. 나는 이것이 "문제"라고 생각하지 않는다.

그래도 사용하지 마십시오. 무의미한 폴링 사이클에서 CPU를 사용하게 될 것이고, 파이썬에는 이미 전역 잠금이 있기 때문에 잠금이있는 대기열을 사용하는 것보다 빠르지 않습니다.

Answer 2

소비자가 그것을 소비하기 전에 생산자가 QUEUE_ITEM을 덮어 쓸 수와 소비자 두 번 QUEUE_ITEM를 소비 할 수 있기 때문하지 정말 스레드 안전합니다. 당신이 언급했듯이, 당신은 그것으로 괜찮습니다. 그러나 대부분의 사람들은 그렇지 않습니다.

cpython 내부에 대한 지식이있는 사람은 이론적 인 질문에 답을해야합니다.

Answer 3

특히 항목이 큰 개체 인 경우 스레드가 생성/소비되는 동안 중단 될 수 있습니다. 편집 : 이것은 단순한 추측입니다. 나는 전문가가 아니다.

또한 스레드는 다른 하나가 실행되기 전에 여러 항목을 생성/소비 할 수 있습니다.

Answer 4

두 항목 모두 원 자성이므로/pop을 추가하는 한 목록을 대기열로 사용할 수 있습니다.

QUEUE = [] 

# this is executed in one threading.Thread object 
def producer(): 
    global QUEUE 
    while True: 
     i = produce_item() 
     QUEUE.append(i) 

# this is executed in another threading.Thread object 
def consumer(): 
    global QUEUE 
    while True: 
     try: 
      i = QUEUE.pop(0) 
     except IndexError: 
      # queue is empty 
      continue 

     consume_item(i) 

아래의 클래스 범위에서 대기열을 지울 수도 있습니다.

class Atomic(object): 
    def __init__(self): 
     self.queue = [] 

    # this is executed in one threading.Thread object 
    def producer(self): 
     while True: 
      i = produce_item() 
      self.queue.append(i) 

    # this is executed in another threading.Thread object 
    def consumer(self): 
     while True: 
      try: 
       i = self.queue.pop(0) 
      except IndexError: 
       # queue is empty 
       continue 

      consume_item(i) 

    # There's the possibility producer is still working on it's current item. 
    def clear_queue(self): 
     self.queue = [] 

생성 된 바이트 코드를보고 원자가되는 목록 연산을 찾아야합니다.

Answer 5

위조가 당신을 물으십시오. 스레드간에 통신하려면 Queue를 사용하기 만하면됩니다.

Answer 6

__del__은 문제가 될 수 있습니다. 새 객체를 QUEUE_ITEM에 할당하기 전에 가비지 컬렉터가 이전 객체의 __del__ 메소드를 호출하지 못하게하는 방법 만 있다면 피할 수 있습니다. 우리는 다음과 같은 것이 필요할 것입니다 :

increase the reference counter on the old object 
assign a new one to `QUEUE_ITEM` 
decrease the reference counter on the old object 

나는 그것이 가능할 지 모르지만, 두려워요.