Method streamingLoop

benchmark/worker/benchmark_client.go:296–340

(ctx context.Context, conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int, payloadType string, poissonLambda *float64)

}

func (bc *benchmarkClient) streamingLoop(ctx context.Context, 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 {
		doRPC = benchmark.DoStreamingRoundTrip
	}
	for ic, conn := range conns {
		// For each connection, create rpcCountPerConn goroutines to do rpc.
		for j := 0; j < rpcCountPerConn; j++ {
			c := testgrpc.NewBenchmarkServiceClient(conn)
			stream, err := c.StreamingCall(context.Background())
			if err != nil {
				logger.Fatalf("%v.StreamingCall(_) = _, %v", c, err)
			}
			idx := ic*rpcCountPerConn + j
			bc.lockingHistograms[idx].histogram = stats.NewHistogram(bc.histogramOptions)
			if poissonLambda == nil { // Closed loop.
				// 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.
					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() {