func (s *statsService) GetClientStats(ctx context.Context, in *testpb.LoadBalancerStatsRequest) (*testpb.LoadBalancerStatsResponse, error) {
	mu.Lock()
	watcherKey := statsWatcherKey{currentRequestID, currentRequestID + in.GetNumRpcs()}
	watcher, ok := watchers[watcherKey]
	if !ok {
		watcher = &statsWatcher{
			rpcsByPeer:    make(map[string]int32),
			numFailures:   0,
			remainingRpcs: in.GetNumRpcs(),
			c:             make(chan *testpb.SimpleResponse),
		}
		watchers[watcherKey] = watcher
	}
	mu.Unlock()

	ctx, cancel := context.WithTimeout(ctx, time.Duration(in.GetTimeoutSec())*time.Second)
	defer cancel()

	defer func() {
		mu.Lock()
		delete(watchers, watcherKey)
		mu.Unlock()
	}()

	// Wait until the requested RPCs have all been recorded or timeout occurs.
	for {
		select {
		case r := <-watcher.c:
			if r != nil {
				watcher.rpcsByPeer[(*r).GetHostname()]++
			} else {
				watcher.numFailures++
			}
			watcher.remainingRpcs--
			if watcher.remainingRpcs == 0 {
				return &testpb.LoadBalancerStatsResponse{NumFailures: watcher.numFailures + watcher.remainingRpcs, RpcsByPeer: watcher.rpcsByPeer}, nil
			}
		case <-ctx.Done():
			grpclog.Info("Timed out, returning partial stats")
			return &testpb.LoadBalancerStatsResponse{NumFailures: watcher.numFailures + watcher.remainingRpcs, RpcsByPeer: watcher.rpcsByPeer}, nil
		}
	}
}
// Wait for the next LoadBalancerStatsRequest.GetNumRpcs to start and complete,
// and return the distribution of remote peers. This is essentially a clientside
// LB reporting mechanism that is designed to be queried by an external test
// driver when verifying that the client is distributing RPCs as expected.
Wait for the next LoadBalancerStatsRequest.GetNumRpcs to start and complete,
and return the distribution of remote peers. This is essentially a clientside
LB reporting mechanism that is designed to be queried by an external test
driver when verifying that the client is distributing RPCs as expected.
// Wait for the next LoadBalancerStatsRequest . GetNumRpcs to start and complete // and return the distribution of remote peers . This is essentially a clientside // LB reporting mechanism that is designed to be queried by an external test // driver when verifying that the client is distributing RPCs as expected .
func (s *statsService) GetClientStats(ctx context.Context, in *testpb.LoadBalancerStatsRequest) (*testpb.LoadBalancerStatsResponse, error) {
	mu.Lock()
	watcherKey := statsWatcherKey{currentRequestID, currentRequestID + in.GetNumRpcs()}
	watcher, ok := watchers[watcherKey]
	if !ok {
		watcher = &statsWatcher{
			rpcsByPeer:    make(map[string]int32),
			numFailures:   0,
			remainingRpcs: in.GetNumRpcs(),
			c:             make(chan *testpb.SimpleResponse),
		}
		watchers[watcherKey] = watcher
	}
	mu.Unlock()
	ctx, cancel := context.WithTimeout(ctx, time.Duration(in.GetTimeoutSec())*time.Second)
	defer cancel()
	defer func() {
		mu.Lock()
		delete(watchers, watcherKey)
		mu.Unlock()
	}()
	for {
		select {
		case r := <-watcher.c:
			if r != nil {
				watcher.rpcsByPeer[(*r).GetHostname()]++
			} else {
				watcher.numFailures++
			}
			watcher.remainingRpcs--
			if watcher.remainingRpcs == 0 {
				return &testpb.LoadBalancerStatsResponse{NumFailures: watcher.numFailures + watcher.remainingRpcs, RpcsByPeer: watcher.rpcsByPeer}, nil
			}
		case <-ctx.Done():
			grpclog.Info("Timed out, returning partial stats")
			return &testpb.LoadBalancerStatsResponse{NumFailures: watcher.numFailures + watcher.remainingRpcs, RpcsByPeer: watcher.rpcsByPeer}, nil
		}
	}
}
func ( s * statsService ) GetClientStats ( ctx context . Context , in * testpb . LoadBalancerStatsRequest ) ( * testpb . LoadBalancerStatsResponse , error ) { mu . Lock ( ) watcherKey := statsWatcherKey { currentRequestID , currentRequestID + in . GetNumRpcs ( ) } watcher , ok := watchers [ watcherKey ] if ! ok { watcher = & statsWatcher { rpcsByPeer : make ( map [ string ] int32 ) , numFailures : 0 , remainingRpcs : in . GetNumRpcs ( ) , c : make ( chan * testpb . SimpleResponse ) , } watchers [ watcherKey ] = watcher } mu . Unlock ( ) ctx , cancel := context . WithTimeout ( ctx , time . Duration ( in . GetTimeoutSec ( ) ) * time . Second ) defer cancel ( ) defer func ( ) { mu . Lock ( ) delete ( watchers , watcherKey ) mu . Unlock ( ) } ( ) for { select { case r := <- watcher . c : if r != nil { watcher . rpcsByPeer [ ( * r ) . GetHostname ( ) ] ++ } else { watcher . numFailures ++ } watcher . remainingRpcs -- if watcher . remainingRpcs == 0 { return & testpb . LoadBalancerStatsResponse { NumFailures : watcher . numFailures + watcher . remainingRpcs , RpcsByPeer : watcher . rpcsByPeer } , nil } case <- ctx . Done ( ) : grpclog . Info ( " " ) return & testpb . LoadBalancerStatsResponse { NumFailures : watcher . numFailures + watcher . remainingRpcs , RpcsByPeer : watcher . rpcsByPeer } , nil } } }
Wait for the next LoadBalancerStatsRequest.GetNumRpcs to start and complete,
and return the distribution of remote peers.
Wait for the next LoadBalancerStatsRequest . GetNumRpcs to start and complete and return the distribution of remote peers .
// Wait until the requested RPCs have all been recorded or timeout occurs.