Repository: grpc/grpc-go
File: benchmark/worker/benchmark_client.go (Go)

Code before:
		return nil, err
	}

	return bc, nil
}

func (bc *benchmarkClient) unaryLoop(conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int, poissonLambda *float64) {
	for ic, conn := range conns {
		client := testgrpc.NewBenchmarkServiceClient(conn)
		// For each connection, create rpcCountPerConn goroutines to do rpc.
		for j := range rpcCountPerConn {
			// Create histogram for each goroutine.
			idx := ic*rpcCountPerConn + j
			bc.lockingHistograms[idx].histogram = stats.NewHistogram(bc.histogramOptions)
			// Start goroutine on the created mutex and histogram.
			go func(idx int) {
				// TODO: do warm up if necessary.
				// Now relying on worker client to reserve time to do warm up.
				// The worker client needs to wait for some time after client is created,
				// before starting benchmark.
				if poissonLambda == nil { // Closed loop.
					for {
						start := time.Now()
						if err := benchmark.DoUnaryCall(client, reqSize, respSize); err != nil {
							select {
							case <-bc.stop:
								return
							default:
							}
							continue
						}
						elapse := time.Since(start)
						bc.lockingHistograms[idx].add(int64(elapse))
						select {
						case <-bc.stop:
						default:
						}
					}
				} else { // Open loop.
					timeBetweenRPCs := time.Duration((rand.ExpFloat64() / *poissonLambda) * float64(time.Second))
					time.AfterFunc(timeBetweenRPCs, func() {
						bc.poissonUnary(client, idx, reqSize, respSize, *poissonLambda)
					})
				}

			}(idx)
		}
	}
}

func (bc *benchmarkClient) streamingLoop(conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int, payloadType string, poissonLambda *float64) {

Review comment (suggestion):
Sigh... it would have been so much nicer if this method accepted a context, which when cancelled would automatically lead to any ongoing RPCs getting cancelled and we can use that as a signal to return from this method as well. I guess cleanup work for another day.

Code after:
		return nil, err
	}

	return bc, nil
}

func (bc *benchmarkClient) unaryLoop(conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int, poissonLambda *float64) {
	for ic, conn := range conns {
		client := testgrpc.NewBenchmarkServiceClient(conn)
		// For each connection, create rpcCountPerConn goroutines to do rpc.
		for j := range rpcCountPerConn {
			// Create histogram for each goroutine.
			idx := ic*rpcCountPerConn + j
			bc.lockingHistograms[idx].histogram = stats.NewHistogram(bc.histogramOptions)
			// Start goroutine on the created mutex and histogram.
			go func(idx int) {
				// TODO: do warm up if necessary.
				// Now relying on worker client to reserve time to do warm up.
				// The worker client needs to wait for some time after client is created,
				// before starting benchmark.
				if poissonLambda == nil { // Closed loop.
					for {
						select {
						case <-bc.stop:
							return
						default:
						}
						start := time.Now()
						if err := benchmark.DoUnaryCall(client, reqSize, respSize); err != nil {
							continue
						}
						elapse := time.Since(start)
						bc.lockingHistograms[idx].add(int64(elapse))
					}
				} else { // Open loop.
					timeBetweenRPCs := time.Duration((rand.ExpFloat64() / *poissonLambda) * float64(time.Second))
					time.AfterFunc(timeBetweenRPCs, func() {
						bc.poissonUnary(client, idx, reqSize, respSize, *poissonLambda)
					})
				}

			}(idx)
		}
	}
}

func (bc *benchmarkClient) streamingLoop(conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int, payloadType string, poissonLambda *float64) {
	var doRPC func(testgrpc.BenchmarkService_StreamingCallClient, int, int) error
	if payloadType == "bytebuf" {
		doRPC = benchmark.DoByteBufStreamingRoundTrip
	} else {