Repository: grpc/grpc-go
File: internal/xds/clients/xdsclient/ads_stream.go (Go)

Code before:
	}
	resourceState, ok := state.subscribedResources[resourceName]
	if !ok {
		return xdsresource.ResourceWatchState{}, fmt.Errorf("unknown resource name: %v", resourceName)
	}
	return *resourceState, nil
}

func resourceNames(m map[string]*xdsresource.ResourceWatchState) []string {
	ret := make([]string, len(m))
	idx := 0
	for name := range m {
		ret[idx] = name
		idx++
	}
	return ret
}

// adsFlowControl implements ADS stream level flow control that enables the ADS
// stream to block the reading of the next message until the previous update is
// consumed by all watchers.
//
// The lifetime of the flow control is tied to the lifetime of the stream. When
// the stream is closed, it is the responsibility of the caller to set the
// pending state to false. This ensures that any goroutine blocked on the flow
// control's wait method is unblocked.
type adsFlowControl struct {
	mu      sync.Mutex
	cond    *sync.Cond // signals when the most recent update has been consumed
	pending bool       // indicates if the most recent update is pending consumption
	stopped bool       // indicates if the ADS stream has been stopped
}

// newADSFlowControl returns a new adsFlowControl.
func newADSFlowControl() *adsFlowControl {
	fc := &adsFlowControl{}
	fc.cond = sync.NewCond(&fc.mu)
	return fc
}

// stop marks the flow control as stopped and signals the condition variable to
// unblock any goroutine waiting on it.
func (fc *adsFlowControl) stop() {
	fc.mu.Lock()
	defer fc.mu.Unlock()

	fc.stopped = true
	fc.cond.Broadcast()
}

// setPending changes the internal state to indicate whether there is an update

Review comment (suggestion):
> When the stream is closed, it is the responsibility of the caller to set the pending state to false. This ensures that any goroutine blocked on the flow control's wait method is unblocked.

I have a question about this. My understanding is that calling `adsFlowControl.stop()` unblocks any goroutines waiting on `adsFlowControl.wait()`. Why is it also the caller's responsibility to set the pending state to `false`? It seems like `stop()` would already handle this.


Code after:
	s.mu.Lock()
	defer s.mu.Unlock()

	state, ok := s.resourceTypeState[rType]
	if !ok {
		return xdsresource.ResourceWatchState{}, fmt.Errorf("unknown resource type: %v", rType)
	}
	resourceState, ok := state.subscribedResources[resourceName]
	if !ok {
		return xdsresource.ResourceWatchState{}, fmt.Errorf("unknown resource name: %v", resourceName)
	}
	return *resourceState, nil
}

func resourceNames(m map[string]*xdsresource.ResourceWatchState) []string {
	ret := make([]string, len(m))
	idx := 0
	for name := range m {
		ret[idx] = name
		idx++
	}
	return ret
}

// adsFlowControl implements ADS stream level flow control that enables the ADS
// stream to block the reading of the next message until the previous update is
// consumed by all watchers.
//
// The lifetime of the flow control is tied to the lifetime of the stream. When
// the stream is closed, it is the responsibility of the caller to stop the flow
// control. This ensures that any goroutine blocked on the flow control's wait
// method is unblocked.
type adsFlowControl struct {
	mu sync.Mutex
	// cond is used to signal when the most recent update has been consumed, or
	// the flow control has been stopped (in which case, waiters should be
	// unblocked as well).
	cond    *sync.Cond
	pending bool // indicates if the most recent update is pending consumption
	stopped bool // indicates if the ADS stream has been stopped
}

// newADSFlowControl returns a new adsFlowControl.
func newADSFlowControl() *adsFlowControl {
	fc := &adsFlowControl{}
	fc.cond = sync.NewCond(&fc.mu)
	return fc
}

// stop marks the flow control as stopped and signals the condition variable to
// unblock any goroutine waiting on it.