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

Code before:
		if ctx.Err() != nil {
			return
		}
		if time.Since(time.Unix(0, startNs)) >= timeoutDuration {
			fmt.Printf("Test exceeded overall timeout of %d seconds, stopping...\n", config.OverallTimeoutSeconds)
			return
		}
		earliestNextStart := time.After(config.MinTimeBetweenRPCs)
		// Get the channel and client from the provided channelFunc (either shared or new).
		_, client := config.MayCreateNewChannel(currentChannel)
		var p peer.Peer
		iterationConfig := SoakIterationConfig{
			SoakRequestSize:  config.SoakRequestSize,
			SoakResponseSize: config.SoakResponseSize,
			Client:           client,
			CallOptions:      []grpc.CallOption{grpc.Peer(&p)},
		}
		result, 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, "Thread %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s failed: %s\n", threadID, i, 0, p.Addr, config.ServerAddr, err)
			mu.Lock()
			threadResults.Failures++
			mu.Unlock()
			<-earliestNextStart
			continue
		}
		latencyMs := result.LatencyMs
		if latencyMs > config.PerIterationMaxAcceptableLatency.Milliseconds() {
			fmt.Fprintf(os.Stderr, "Thread %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s exceeds max acceptable latency: %d\n", threadID, i, latencyMs, p.Addr, config.ServerAddr, config.PerIterationMaxAcceptableLatency.Milliseconds())
			mu.Lock()
			threadResults.Failures++
			mu.Unlock()
			<-earliestNextStart
			continue
		}
		// Success: log the details of the iteration.
		mu.Lock()
		threadResults.Latencies.Add(latencyMs)
		threadResults.IterationsDone++
		mu.Unlock()
		fmt.Fprintf(os.Stderr, "Thread %d: soak iteration: %d elapsed_ms: %d peer: %v server_uri: %s succeeded\n", threadID, i, latencyMs, p.Addr, config.ServerAddr)
		<-earliestNextStart
	}
}

// DoSoakTest runs large unary RPCs in a loop for a configurable number of times, with configurable failure thresholds.

Review comment (suggestion):
Each goroutine should have exclusive access to its results struct.  So this shouldn't need a lock.

Or just make each instance of this function _return_ a results struct instead of mutating one that's stored elsewhere -- that would likely be a cleaner approach.

Code after:
		}
		if err != nil {
			return err
		}
		p := in.GetPayload().GetBody()
		sum += len(p)
	}
}

func (s *testServer) FullDuplexCall(stream testgrpc.TestService_FullDuplexCallServer) error {
	if md, ok := metadata.FromIncomingContext(stream.Context()); ok {
		if initialMetadata, ok := md[initialMetadataKey]; ok {
			header := metadata.Pairs(initialMetadataKey, initialMetadata[0])
			stream.SendHeader(header)
		}
		if trailingMetadata, ok := md[trailingMetadataKey]; ok {
			trailer := metadata.Pairs(trailingMetadataKey, trailingMetadata[0])
			stream.SetTrailer(trailer)
		}
	}
	hasORCALock := false
	for {
		in, err := stream.Recv()
		if err == io.EOF {
			// read done.
			return nil
		}
		if err != nil {
			return err
		}
		st := in.GetResponseStatus()
		if st != nil && st.Code != 0 {
			return status.Error(codes.Code(st.Code), st.Message)
		}

		if r, orcaData := s.metricsRecorder, in.GetOrcaOobReport(); r != nil && orcaData != nil {
			// Transfer the request's OOB ORCA data to the server metrics recorder
			// in the server, if present.
			if !hasORCALock {
				s.orcaMu.Lock()
				defer s.orcaMu.Unlock()
				hasORCALock = true
			}
			setORCAMetrics(r, orcaData)
		}

		cs := in.GetResponseParameters()
		for _, c := range cs {
			if us := c.GetIntervalUs(); us > 0 {
				time.Sleep(time.Duration(us) * time.Microsecond)
			}