[Concurrency in Go] 비정상 고루틴의 치료 (steward & ward)

Background 

• long-lived process의 경우, Goroutine 여러개를 들고 있을 수 있다. 고루틴들은 데이터가 들어올 때까지 기다리며 block되기 때문에 그들은 wakeup, do their work, pass data on을 할 수 있어야 한다. 하지만 control하기 어려운 resource들에 대해서 의존성이 있을 수 있다. 예를 들면 webservice에서 data를 끌어올수도 있고 어떤 일시적인 파일에 대해서 monitoring을 할수도 있다. 이런 경우 고루틴은 bad state에 쉽게 stuck될수도 있고, external help에 의해서가 아니면 회복이 불가능하다. 
• 고루틴이 healthy한 상태로 남아있도록 보장하는 mechanism을 만들수 있는데, 이때 unhealthy한 고루틴을 재시작하는 것을 healing이라고 한다.

스튜어드와 와드

• steward: 관리인, 와드의 하트비트를 받아서 비정상 상태라고 판단이 되면 재시작하는 책임을 갖는다. 해당 고루틴을 시작시킬 수 있는 함수에 대한 참조가 필요하다. 
• ward: 모니터링 대상이다. 작업을 수행하는 역할을 가진다.  

 

• 여기서 중요한게 startGoroutine을 통해서 ward를 전달 받고, wardHeartbeat를 확인하게 되는데, ward를 통해서 들어오는 pulse가 있다면 monitor loop을 다시 돌려서 timeout을 재시작하게 되고, 이 사이에 안들어오면 ward를 새로 띄우게 된다.
• ward goroutine은 steward가 멈추거나, steward 가 ward goroutine을 멈추게 하고 싶거나 둘 중 하나에 해당하면 멈추도록 or function을 사용한다. pulseInterval은 1/2timeout으로 전달되었다 (이 값은 tweak될 수 있다. )

Steward

• startGoroutineFunction (done ← chan interface{}, pulseInterval time.Duration,) (heartbeat ←chan interface{})
• newSteward(timeout, startGoroutineFunction) (startGoroutineFunction)
  • startGoroutineFunction이 들어가고, timeout까지 더해서 들어가면, startGoroutineFunction을 return 하는 구조

 

 

 

• startWard는 monitoring할 고루틴을 시작하는 일관적인 방식으로 정의된 closure이다.
• 5) 우리는 ward 고루틴이 steward가 멈추거나, steward가 ward 고루틴을 멈출 시에 멈추도록 만들고 싶어하기 때문에  wardDone 을 추가해서 or 함수를 사용해서 넣는다.
• pulseInterval 마다 pulse channel에 값이 들어오면, heartbeat에 값을 전달한다. 

 

 

• wardHeartbeat가 도착하면 monitorLoop을 다시 시작하여 timeout 조건을 재시작한다. 만약에 heartbeat가 timeout 시간동안 도착하지 않았을 때는 timeoutSignal 채널에 값이 들어와서 기존 ward를 종료시키고 재시작 시킨뒤 monitorLoop을 재시작한다.
• steward가 하는건 ward가 살아있는지 확인하는 pulse, 그리고 ward의 heartbeat는 더 자주 받아본다. ward의 heartbeat를 받으면, monitor 재시작하면서 timeout을 재시작하는 것이다. 
  • ward가 중간에 막히면 ward의 return 값으로 받는 heartbeat가 없을 테니 재시작이 안된다. 책에선 ward의 pulseInterval을 timeout/2로 전달한다.

Bridge Channel 

• 채널의 채널을 하나의 채널로 분해할 수 있다면, consumer가 value가 channel의 sequence로부터 온다는 것을 신경쓰지 않을 수 있다. 
• bridge를 통해서 오는 모든 value를 리턴하는 채널인 valStream을 만든다. 
• 2) chanStream에서 channel을 빼내고, 그 channel을 3의 nested loop에 제공하는 역할을 한다. 
• 3) channel 에서 값을 빼내고 loop를 도는 stream이 closed되면, loop를 빠져나와서 다른 channel 을 고르는 loop인 2번의 다음 iteration을 돈다. 

 

Ward 의 형태는 어떻게 되어야 할까? (p193) 

•  ward를 steward에 맞게끔 계속 rewrite하는 것은 cumbersome 하고 unnecessary하기 때문에 어떤 형태가 필요하다.

• 여기서 bridge 함수를 사용해서 intChanStream에 들어가는 channel의 값들을 하나의 stream으로 합친다.
• 3번이 이제 가장 중요한 steward에 의해 시작하고 모니터링 되는 closure를 생성하는 부분이다. 
• 4번에서 하는 일은 이 ward안에서 커뮤니케이션을 위해 사용할 intStream을 만드는 것이다. 
• 5번 에서는 intChanStream이 우리가 지금 커뮤니케이션 하는 채널을 알려주는 곳이다. 
• ward를 여러개의 copies 를 시작하는 것을 알고 있다. 그렇기 때문에 bridge channel을 사용해서 하나의 uninterrupted channel을 제공하도록 했다. 

 

아래와 같은 패턴으로 주로 사용된다. 

func ward() startGoroutineFn {
    return func(
        done <-chan struct{},
        pulseInterval time.Duration, 
    ) <-chan struct{} {
    	heartbeat := make(chan struct{}) 
        go func() {
            pulse := time.Tick(pulseInterval)
            for {
                select {
                case <-done:
                	return
                case <-pulse:
                    select {
                    case heartbeat <- struct{}{}:
                    default:
                    }
                // case intStream <- intVal: 
                // 	continue valueLoop
                }
            }
        }()
        return heartbeat
    }
}