Repository: grpc/grpc-go
File: interop/soak_tests.go (Go)

Code before:
	soakIterationsPerWorker := config.Iterations / config.NumWorkers
	for i := 0; i < soakIterationsPerWorker; i++ {
		if ctx.Err() != nil {
			return
		}
		if time.Since(startTime) >= timeoutDuration {
			fmt.Printf("Test exceeded overall timeout of %v, stopping...\n", config.OverallTimeout)
			return
		}
		earliestNextStart := time.After(config.MinTimeBetweenRPCs)
		currentChannel, cleanup := config.ChannelForTest()
		defer cleanup()
		client := testgrpc.NewTestServiceClient(currentChannel)
		var p peer.Peer
		iterationConfig := SoakIterationConfig{
			RequestSize:  config.RequestSize,
			ResponseSize: config.ResponseSize,
			Client:       client,
			CallOptions:  []grpc.CallOption{grpc.Peer(&p)},
		}
		latency, err := doOneSoakIteration(ctx, iterationConfig)
		if p.Addr != nil {
			fmt.Fprintf(os.Stderr, "Peer address: %v\n", p.Addr)
		} else {
			fmt.Fprintf(os.Stderr, "No peer address available for this RPC.\n")
		}
		if err != nil {
			fmt.Fprintf(os.Stderr, "Worker %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s failed: %s\n", workerID, i, 0, p.Addr, config.ServerAddr, err)
			workerResults.Failures++
			<-earliestNextStart
			continue
		}
		if latency > config.PerIterationMaxAcceptableLatency {
			fmt.Fprintf(os.Stderr, "Worker %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s exceeds max acceptable latency: %d\n", workerID, i, latency, p.Addr, config.ServerAddr, config.PerIterationMaxAcceptableLatency.Milliseconds())
			workerResults.Failures++
			<-earliestNextStart
			continue
		}
		// Success: log the details of the iteration.
		workerResults.Latencies.Add(latency.Milliseconds())
		workerResults.IterationsDone++
		fmt.Fprintf(os.Stderr, "Worker %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s succeeded\n", workerID, i, latency, p.Addr, config.ServerAddr)
		<-earliestNextStart
	}
}

// DoSoakTest runs large unary RPCs in a loop for a configurable number of times, with configurable failure thresholds.
// If resetChannel is false, then each RPC will be performed on tc. Otherwise, each RPC will be performed on a new
// stub that is created with the provided server address and dial options.
// TODO(mohanli-ml): Create SoakTestOptions as a parameter for this method.
func DoSoakTest(ctx context.Context, soakConfig SoakTestConfig) {

Review comment (bug):
What is this for?

We log the peer in all branches below already.

Also we do it unconditionally, as `p.Addr == nil` would be a bug in gRPC.

Code after:
			fmt.Printf("Test exceeded overall timeout of %v, stopping...\n", config.OverallTimeout)
			return
		}
		earliestNextStart := time.After(config.MinTimeBetweenRPCs)
		currentChannel, cleanup := config.ChannelForTest()
		defer cleanup()
		client := testgrpc.NewTestServiceClient(currentChannel)
		var p peer.Peer
		iterationConfig := SoakIterationConfig{
			RequestSize:  config.RequestSize,
			ResponseSize: config.ResponseSize,
			Client:       client,
			CallOptions:  []grpc.CallOption{grpc.Peer(&p)},
		}
		latency, err := doOneSoakIteration(ctx, iterationConfig)
		if err != nil {
			fmt.Fprintf(os.Stderr, "Worker %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s failed: %s\n", workerID, i, 0, p.Addr, config.ServerAddr, err)
			soakWorkerResults.Failures++
			<-earliestNextStart
			continue
		}
		if latency > config.PerIterationMaxAcceptableLatency {
			fmt.Fprintf(os.Stderr, "Worker %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s exceeds max acceptable latency: %d\n", workerID, i, latency, p.Addr, config.ServerAddr, config.PerIterationMaxAcceptableLatency.Milliseconds())
			soakWorkerResults.Failures++
			<-earliestNextStart
			continue
		}
		// Success: log the details of the iteration.
		soakWorkerResults.Latencies.Add(latency.Milliseconds())
		soakWorkerResults.IterationsDone++
		fmt.Fprintf(os.Stderr, "Worker %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s succeeded\n", workerID, i, latency, p.Addr, config.ServerAddr)
		<-earliestNextStart
	}
}

// DoSoakTest runs large unary RPCs in a loop for a configurable number of times, with configurable failure thresholds.
// If resetChannel is false, then each RPC will be performed on tc. Otherwise, each RPC will be performed on a new
// stub that is created with the provided server address and dial options.
// TODO(mohanli-ml): Create SoakTestOptions as a parameter for this method.
func DoSoakTest(ctx context.Context, soakConfig SoakTestConfig) {
	if soakConfig.Iterations%soakConfig.NumWorkers != 0 {
		fmt.Fprintf(os.Stderr, "soakIterations must be evenly divisible by soakNumWThreads\n")
	}
	startTime := time.Now()
	var wg sync.WaitGroup
	soakWorkerResults := make([]SoakWorkerResults, soakConfig.NumWorkers)
	for i := 0; i < soakConfig.NumWorkers; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			executeSoakTestInWorker(ctx, soakConfig, startTime, workerID, &soakWorkerResults[workerID])