"Python threading barrier"왜이 코드가 작동하고 더 좋은 방법이 있습니까?

Question

파이썬 장벽을 검색했지만 관련 문제가 거의 없습니다. 나는 아직도 barrier.wait()에 대해 혼란 스럽다. 심지어 내 코드가 작동한다.

저는 파이썬 장벽을 사용하여 이러한 기능을 구현합니다 : 메인 스레드와 n 개의 서브 스레드. 각 라운드에서 주 스레드는 현재 작업을 완료 한 모든 하위 스레드를 기다린 다음 모든 조건이 충족 될 때까지 모든 스레드를 다음 라운드로 이동합니다. 따라서 장벽이이 함수를 구현하는 것이 적절하다는 것을 알았습니다. 여기에 주 스레드에 대한 코드가 있습니다.

def superstep(self): 
    workers = [] 
    barrier = threading.Barrier(self.num_workers+1) 
    for vertex in self.vertices: 
     worker = Worker(vertex, barrier) 
     workers.append(worker) 
     worker.start() 

    while self.flag: 
     barrier.wait() 
     self.distributeMessages() 
     self.count += 1 
     print ("superstep: ", self.count) 
     self.flag = self.isTerminated() 

    for worker in workers: 
     worker.flag = False 

    for worker in workers: 
     worker.join() 

1. 루프가 노동자라는 이름 목록 노동자에 저장된 n 개의 쓰레드를 생성 '에 대한'첫 번째.
2. 'while'루프는 다른 하위 스레드를 기다리는 주 스레드이며 self.flag가 False 인 경우 중단됩니다.
3. 두 번째 'for'루프는 각 작업자 (하위 스레드)에서 플래그를 False로 설정하여 루프를 종료하도록 알려줍니다.

여기 내 Worker 클래스입니다.

class Worker(threading.Thread): 
    def __init__(self, vertex, barrier): 
     threading.Thread.__init__(self) 
     self.vertex = vertex 
     self.flag = True 
     self.barrier = barrier 

    def run(self): 
     while self.flag: 
      self.barrier.wait() 
      do something 

모든 스레드가 조인 할 수있는 코드가 잘 작동합니다(). 하지만 python barrier을 보면 모든 스레드가 wait()를 호출 할 때 모든 스레드가 동시에 해제됩니다. 메인 쓰레드가 while 루프에서 깨어지고 다른 모든 쓰레드가 바로 기다리고 있다면, 두 번째 'for'루프는 쓸모없고 서브 쓰레드는 절대 참여하지 않을 것입니다().

이 코드는 어떻게 작동합니까? BrokenBarrierError를 발생시키는 대신 장벽을 종료하는 다른 방법이 있습니까? 또한, 두 번째 'for'루프에 일부 코드를 추가하고 일부 정보 또는 다른 것을 인쇄하면 프로 시저가 차단됩니다. wait()에있는 하위 스레드가 있어야하며 플래그를 확인할 기회가 없어서 스레드의 run()을 종료 할 수 없습니다.

Answer 1

abort을 사용하지 않으려는 경우 각 스레드에 Barrier.wait 번을 두 번 호출 할 수 있습니다. 이렇게하면 작업이 두 부분으로 분리됩니다. 첫 번째 부분에서는 작업자 스레드가 작업을 수행하고 주 스레드가 플래그 상태를 업데이트합니다. 그런 다음 두 번째 부분에서 모든 스레드가 플래그 상태를 확인하고 필요한 경우 루프를 종료합니다. 코드 레벨에

가이 같은 보일 것이다 :

# Main 
def superstep(self): 
    workers = [] 
    barrier = threading.Barrier(self.num_workers+1) 
    for vertex in self.vertices: 
     worker = Worker(vertex, barrier) 
     workers.append(worker) 
     worker.start() 

    while self.flag: 
     barrier.wait() 
     self.distributeMessages() 
     self.count += 1 
     print ("superstep: ", self.count) 
     self.flag = self.isTerminated() 
     for worker in workers: 
      worker.flag = self.flag 
     barrier.wait() 

    for worker in workers: 
     worker.join() 

# Worker 
def run(self): 
    while self.flag: 
     self.barrier.wait() 
     # do something 
     self.barrier.wait() 

Answer 2

당신은 루프에 대한 두 번째 후 대기 근로자를 해제

self.barrier.abort() 

를 호출 할 수 있습니다, 그리고 노동자의 run() 방법에 BrokenBarrierError을 잡을 .