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

Code before:
				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.
					for {
						start := time.Now()
						if err := doRPC(stream, reqSize, respSize); err != nil {
							return
						}
						elapse := time.Since(start)
						bc.lockingHistograms[idx].add(int64(elapse))
						select {
						case <-bc.stop:
							return
						default:
						}
					}
				}(idx)
			} else { // Open loop.
				timeBetweenRPCs := time.Duration((rand.ExpFloat64() / *poissonLambda) * float64(time.Second))
				time.AfterFunc(timeBetweenRPCs, func() {
					bc.poissonStreaming(stream, idx, reqSize, respSize, *poissonLambda, doRPC)
				})
			}
		}
	}
}

func (bc *benchmarkClient) poissonUnary(client testgrpc.BenchmarkServiceClient, idx int, reqSize int, respSize int, lambda float64) {
	go func() {
		start := time.Now()
		if err := benchmark.DoUnaryCall(client, reqSize, respSize); err != nil {
			return
		}
		elapse := time.Since(start)
		bc.lockingHistograms[idx].add(int64(elapse))
	}()
	timeBetweenRPCs := time.Duration((rand.ExpFloat64() / lambda) * float64(time.Second))
	time.AfterFunc(timeBetweenRPCs, func() {
		bc.poissonUnary(client, idx, reqSize, respSize, lambda)
	})
}

func (bc *benchmarkClient) poissonStreaming(stream testgrpc.BenchmarkService_StreamingCallClient, idx int, reqSize int, respSize int, lambda float64, doRPC func(testgrpc.BenchmarkService_StreamingCallClient, int, int) error) {
	go func() {
		start := time.Now()
		if err := doRPC(stream, reqSize, respSize); err != nil {
			return
		}
		elapse := time.Since(start)

Review comment (suggestion):
Since context should not be propagated to `DoUnaryCall`, there is no point in propagating it to `poissonUnary` as well

Code after:
					for {
						start := time.Now()
						if err := doRPC(stream, reqSize, respSize); err != nil {
							return
						}
						elapse := time.Since(start)
						bc.lockingHistograms[idx].add(int64(elapse))
						if ctx.Err() != nil {
							return
						}
					}
				}(idx)
			} else { // Open loop.
				timeBetweenRPCs := time.Duration((rand.ExpFloat64() / *poissonLambda) * float64(time.Second))
				time.AfterFunc(timeBetweenRPCs, func() {
					bc.poissonStreaming(stream, idx, reqSize, respSize, *poissonLambda, doRPC)
				})
			}
		}
	}
}

func (bc *benchmarkClient) poissonUnary(client testgrpc.BenchmarkServiceClient, idx int, reqSize int, respSize int, lambda float64) {
	go func() {
		start := time.Now()
		if err := benchmark.DoUnaryCall(client, reqSize, respSize); err != nil {
			return
		}
		elapse := time.Since(start)
		bc.lockingHistograms[idx].add(int64(elapse))
	}()
	timeBetweenRPCs := time.Duration((rand.ExpFloat64() / lambda) * float64(time.Second))
	time.AfterFunc(timeBetweenRPCs, func() {
		bc.poissonUnary(client, idx, reqSize, respSize, lambda)
	})
}

func (bc *benchmarkClient) poissonStreaming(stream testgrpc.BenchmarkService_StreamingCallClient, idx int, reqSize int, respSize int, lambda float64, doRPC func(testgrpc.BenchmarkService_StreamingCallClient, int, int) error) {
	go func() {
		start := time.Now()

		if err := doRPC(stream, reqSize, respSize); err != nil {
			return
		}
		elapse := time.Since(start)
		bc.lockingHistograms[idx].add(int64(elapse))
	}()
	timeBetweenRPCs := time.Duration((rand.ExpFloat64() / lambda) * float64(time.Second))
	time.AfterFunc(timeBetweenRPCs, func() {
		bc.poissonStreaming(stream, idx, reqSize, respSize, lambda, doRPC)
	})