Repository: yarpc/yab
Pull request #319 (2021-01-13 20:24:10)

New: added gRPC streams support for benchmark - 5
## Aim
Support gRPC streams in Yab

## Changes
* Add benchmark handling for stream requests which works with the existing benchmarking parameters
* Reuses the initial stream request dispatcher for benchmarking
* Refactor benchmark handling into different files `benchmark_unary.go` and `benchmark_stream.go` for handlings it's respective benchmark type. 
* Stream benchmark works alongside warmup request (ad-hoc request) by recording the stream requests (via STDIN or passed on CLI) sent during the warmup and reusing the same during the Benchmarks

## How to test?
Download [grpc-stream-test-server.zip](https://github.com/yarpc/yab/files/5826809/grpc-stream-test-server.zip), follow the `readme.md` to run the test server and dispatch streaming yab queries

### Bidi Stream
```
yab  --max-requests=10000 --max-duration=1s --timeout=1m -F testproto --service "test-server" --method uber.yarpc.encoding.protobuf.Test/Stream -p grpc://localhost:55555   -r '{"value": "1"}{"value": "2"}'
// Warmup response
{
  "value": "1"
}

{
  "value": "12"
}

Benchmark parameters:
  CPUs:            12
  Connections:     24
  Concurrency:     1
  Max requests:    10000
  Max duration:    1s
  Max RPS:         0
Latencies:
  0.5000: 922.41µs
  0.9000: 2.495403ms
  0.9500: 3.514855ms
  0.9900: 22.468952ms
  0.9990: 40.554458ms
  0.9995: 40.882234ms
  1.0000: 42.377369ms
Elapsed time (seconds):   0.74
Total requests:           10000
RPS:                      13558.80
```
#### STDIN mode
```
yab  --max-requests=10000 --max-duration=1s --timeout=1m -F testproto --service "test-server" --method uber.yarpc.encoding.protobuf.Test/Stream -p grpc://localhost:55555   -r '-'
{"value":"1"} // STDIN request-1
{
  "value": "1"
}

{"value":"2"} // STDIN request-2
{
  "value": "12"
}

Benchmark parameters:
  CPUs:            12
  Connections:     24
  Concurrency:     1
  Max requests:    10000
  Max duration:    1s
  Max RPS:         0
Latencies:
  0.5000: 625.964µs
  0.9000: 1.353507ms
  0.9500: 1.899145ms
  0.9900: 16.519147ms
  0.9990: 23.410829ms
  0.9995: 23.508979ms
  1.0000: 23.947096ms
Elapsed time (seconds):   0.43
Total requests:           10000
RPS:                      23259.32
```

#### Without Warmup
```
yab --max-requests=10000 --max-duration=1s --timeout=1m -F testproto --service "test-server" --method uber.yarpc.encoding.protobuf.Test/Stream -p grpc://localhost:55555   -r '{"value": "1"}{"value": "2"}' --warmup=0
Benchmark parameters:
  CPUs:            12
  Connections:     24
  Concurrency:     1
  Max requests:    10000
  Max duration:    1s
  Max RPS:         0
Latencies:
  0.5000: 696.727µs
  0.9000: 1.639929ms
  0.9500: 2.154043ms
  0.9900: 17.015733ms
  0.9990: 27.704704ms
  0.9995: 27.858184ms
  1.0000: 28.193725ms
Elapsed time (seconds):   0.48
Total requests:           10000
RPS:                      20980.48
```

### Server Stream
```
yab  --max-requests=10000 --max-duration=1s --timeout=1m -F testproto --service "test-server" --method uber.yarpc.encoding.protobuf.Test/ServerStream -p grpc://localhost:55555   -r '{"value": "2"}'
{
  "value": "2"
}

Benchmark parameters:
  CPUs:            12
  Connections:     24
  Concurrency:     1
  Max requests:    10000
  Max duration:    1s
  Max RPS:         0
Latencies:
  0.5000: 606.604µs
  0.9000: 1.262325ms
  0.9500: 1.879435ms
  0.9900: 19.652758ms
  0.9990: 22.332575ms
  0.9995: 22.619429ms
  1.0000: 22.860896ms
Elapsed time (seconds):   0.44
Total requests:           10000
RPS:                      22716.25
```

### Client Stream
```
yab --max-requests=10000 --max-duration=1s --timeout=1m -F testproto --service "test-server" --method uber.yarpc.encoding.protobuf.Test/ClientStream -p grpc://localhost:55555   -r '{"value": "a"}{"value": "b"}'
{
  "value": "2"
}

Benchmark parameters:
  CPUs:            12
  Connections:     24
  Concurrency:     1
  Max requests:    10000
  Max duration:    1s
  Max RPS:         0
Latencies:
  0.5000: 867.434µs
  0.9000: 1.94393ms
  0.9500: 2.652286ms
  0.9900: 13.673861ms
  0.9990: 21.272389ms
  0.9995: 21.932361ms
  1.0000: 23.618049ms
Elapsed time (seconds):   0.55
Total requests:           10000
RPS:                      18123.10
```

#### STDIN mode

```
yab --max-requests=10000 --max-duration=1s --timeout=1m -F testproto --service "test-server" --method uber.yarpc.encoding.protobuf.Test/ClientStream -p grpc://localhost:55555   -r '-'
{"value": "a"} // STDIN request-1
{"value": "b"} // STDIN request-2
{
  "value": "2"
}

Benchmark parameters:
  CPUs:            12
  Connections:     24
  Concurrency:     1
  Max requests:    10000
  Max duration:    1s
  Max RPS:         0
Latencies:
  0.5000: 783.716µs
  0.9000: 1.671911ms
  0.9500: 2.270289ms
  0.9900: 10.897964ms
  0.9990: 16.759355ms
  0.9995: 16.998659ms
  1.0000: 18.837755ms
Elapsed time (seconds):   0.46
Total requests:           10000
RPS:                      21630.18

```


codecov[bot]: # [Codecov](https://codecov.io/gh/yarpc/yab/pull/319?src=pr&el=h1) Report
> Merging [#319](https://codecov.io/gh/yarpc/yab/pull/319?src=pr&el=desc) (858d265) into [grpc-streams-4](https://codecov.io/gh/yarpc/yab/commit/86f8e09928454ca805a7e680f799ecef713a8b96?el=desc) (86f8e09) will **decrease** coverage by `1.52%`.
> The diff coverage is `35.29%`.

[![Impacted file tree graph](https://codecov.io/gh/yarpc/yab/pull/319/graphs/tree.svg?width=650&height=150&src=pr&token=yEgrqUTCuh)](https://codecov.io/gh/yarpc/yab/pull/319?src=pr&el=tree)

```diff
@@                Coverage Diff                 @@
##           grpc-streams-4     #319      +/-   ##
==================================================
- Coverage           90.47%   88.94%   -1.53%     
==================================================
  Files                  51       51              
  Lines                2508     2561      +53     
==================================================
+ Hits                 2269     2278       +9     
- Misses                169      210      +41     
- Partials               70       73       +3     
```


| [Impacted Files](https://codecov.io/gh/yarpc/yab/pull/319?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [bench\_method.go](https://codecov.io/gh/yarpc/yab/pull/319/diff?src=pr&el=tree#diff-YmVuY2hfbWV0aG9kLmdv) | `74.46% <16.66%> (-19.98%)` | :arrow_down: |
| [main.go](https://codecov.io/gh/yarpc/yab/pull/319/diff?src=pr&el=tree#diff-bWFpbi5nbw==) | `79.42% <39.28%> (-8.24%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/yarpc/yab/pull/319?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/yarpc/yab/pull/319?src=pr&el=footer). Last update [86f8e09...6d31f72](https://codecov.io/gh/yarpc/yab/pull/319?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

jronak: @Dogild is suggesting to hold off this PR until #314, #315, #316, and #317 are merged. Might have to close and re-open this PR to merge into a different branch

Patch:
diff --git a/bench_method.go b/bench_method.go
index 07df6c6..eb008d8 100644
--- a/bench_method.go
+++ b/bench_method.go
@@ -25,6 +25,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/yarpc/yab/encoding"
 	"github.com/yarpc/yab/transport"
 
 	"github.com/opentracing/opentracing-go"
@@ -38,18 +39,37 @@ type peerTransport struct {
 // benchmarkCaller exposes method to dispatch requests for benchmark.
 type benchmarkCaller interface {
 	// Call dispatches a request using the provided transport.
-	Call(transport.Transport) (benchmarkCallLatencyResult, error)
+	Call(transport.Transport) (benchmarkCallReporter, error)
+
+	// CallMethodType returns the type of the RPC method invoked by `Call`.
+	CallMethodType() encoding.MethodType
+}
+
+// benchmarkCallReporter exposes method to access benchmark call report like latency.
+type benchmarkCallReporter interface {
+	// Latency returns the time taken to send request and receive response.
+	Latency() time.Duration
+}
+
+// benchmarkStreamCallReporter exposes method to access benchmark stream call report
+// like stream messages send and received.
+type benchmarkStreamCallReporter interface {
+	// StreamMessagesSent returns number of stream messages sent from the client.
+	StreamMessagesSent() int
+
+	// StreamMessagesReceived returns number of stream messages received from the server.
+	StreamMessagesReceived() int
 }
 
-type benchmarkCallLatencyResult struct {
+type benchmarkCallLatencyReport struct {
 	latency time.Duration
 }
 
-func newBenchmarkCallLatencyResult(latency time.Duration) benchmarkCallLatencyResult {
-	return benchmarkCallLatencyResult{latency}
+func newBenchmarkCallLatencyReport(latency time.Duration) benchmarkCallLatencyReport {
+	return benchmarkCallLatencyReport{latency}
 }
 
-func (r benchmarkCallLatencyResult) Latency() time.Duration {
+func (r benchmarkCallLatencyReport) Latency() time.Duration {
 	return r.latency
 }
 
diff --git a/bench_state.go b/bench_state.go
index 88a7317..f795ead 100644
--- a/bench_state.go
+++ b/bench_state.go
@@ -36,6 +36,9 @@ type benchmarkState struct {
 	totalSuccess  int
 	totalRequests int
 	latencies     []time.Duration
+
+	totalStreamMessagesSent     int
+	totalStreamMessagesReceived int
 }
 
 func newBenchmarkState(statter statsd.Client) *benchmarkState {
@@ -69,6 +72,8 @@ func (s *benchmarkState) merge(other *benchmarkState) {
 	s.totalErrors += other.totalErrors
 	s.totalSuccess += other.totalSuccess
 	s.totalRequests += other.totalRequests
+	s.totalStreamMessagesReceived += other.totalStreamMessagesReceived
+	s.totalStreamMessagesSent += other.totalStreamMessagesSent
 }
 
 func (s *benchmarkState) recordLatency(d time.Duration) {
@@ -79,6 +84,11 @@ func (s *benchmarkState) recordLatency(d time.Duration) {
 	s.statter.Timing("latency", d)
 }
 
+func (s *benchmarkState) recordStreamMessages(sent, received int) {
+	s.totalStreamMessagesSent += sent
+	s.totalStreamMessagesReceived += received
+}
+
 // Returns a mapping of quantiles to latency values
 func (s *benchmarkState) getLatencies() map[float64]time.Duration {
 	sort.Sort(byDuration(s.latencies))
diff --git a/benchmark.go b/benchmark.go
index 9748b65..d6c8654 100644
--- a/benchmark.go
+++ b/benchmark.go
@@ -32,6 +32,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/yarpc/yab/encoding"
 	"github.com/yarpc/yab/limiter"
 	"github.com/yarpc/yab/statsd"
 	"github.com/yarpc/yab/transport"
@@ -65,11 +66,21 @@ type Summary struct {
 	RPS                float64 `json:"rps"`
 }
 
+// StreamSummary stores summary of stream messages sent and received
+type StreamSummary struct {
+	TotalStreamMessagesSent     int `json:"totalStreamMessagesSent"`
+	TotalStreamMessagesReceived int `json:"totalStreamMessagesReceived"`
+}
+
 // BenchmarkOutput stores benchmark settings and results for JSON output
 type BenchmarkOutput struct {
 	Parameters Parameters        `json:"benchmarkParameters"`
 	Latencies  map[string]string `json:"latencies"`
 	Summary    Summary           `json:"summary"`
+
+	// StreamSummary is available only for streaming benchmark. It is nil and
+	// omitted in unary benchmark.
+	StreamSummary *StreamSummary `json:"streamSummary,omitempty"`
 }
 
 // setGoMaxProcs sets runtime.GOMAXPROCS if the option is set
@@ -111,7 +122,7 @@ func (o BenchmarkOptions) enabled() bool {
 
 func runWorker(t transport.Transport, b benchmarkCaller, s *benchmarkState, run *limiter.Run, logger *zap.Logger) {
 	for cur := run; cur.More(); {
-		callResult, err := b.Call(t)
+		callReport, err := b.Call(t)
 		if err != nil {
 			s.recordError(err)
 			// TODO: Add information about which peer specifically failed.
@@ -119,7 +130,11 @@ func runWorker(t transport.Transport, b benchmarkCaller, s *benchmarkState, run
 			continue
 		}
 
-		s.recordLatency(callResult.Latency())
+		s.recordLatency(callReport.Latency())
+
+		if streamCallReport, ok := callReport.(benchmarkStreamCallReporter); ok {
+			s.recordStreamMessages(streamCallReport.StreamMessagesSent(), streamCallReport.StreamMessagesReceived())
+		}
 	}
 }
 
@@ -208,10 +223,10 @@ func runBenchmark(out output, logger *zap.Logger, allOpts Options, resolved reso
 			state := states[i*opts.Concurrency+j]
 
 			wg.Add(1)
-			go func(c transport.Transport) {
+			go func(t transport.Transport) {
 				defer wg.Done()
-				runWorker(c, b, state, run, logger)
-			}(c)
+				runWorker(t, b, state, run, logger)
+			}(c.Transport)
 		}
 	}
 
@@ -246,23 +261,34 @@ func runBenchmark(out output, logger *zap.Logger, allOpts Options, resolved reso
 		RPS:                rps,
 	}
 
+	var streamSummary *StreamSummary
+
+	// create stream summary for streaming calls only.
+	if b.CallMethodType() != encoding.Unary {
+		streamSummary = &StreamSummary{
+			TotalStreamMessagesSent:     overall.totalStreamMessagesSent,
+			TotalStreamMessagesReceived: overall.totalStreamMessagesReceived,
+		}
+	}
+
 	if formatAsJSON {
-		outputJSON(out, parameters, latencyValues, summary)
+		outputJSON(out, parameters, latencyValues, summary, streamSummary)
 	} else {
-		outputPlaintext(out, latencyValues, summary)
+		outputPlaintext(out, latencyValues, summary, streamSummary)
 	}
 }
 
-func outputJSON(out output, parameters Parameters, latencyValues map[float64]time.Duration, summary Summary) {
+func outputJSON(out output, parameters Parameters, latencyValues map[float64]time.Duration, summary Summary, streamSummary *StreamSummary) {
 	latencies := make(map[string]string, len(_quantiles))
 	for _, quantile := range _quantiles {
 		latencies[fmt.Sprintf("%.4f", quantile)] = latencyValues[quantile].String()
 	}
 
 	benchmarkOutput := BenchmarkOutput{
-		Parameters: parameters,
-		Latencies:  latencies,
-		Summary:    summary,
+		Parameters:    parameters,
+		Latencies:     latencies,
+		Summary:       summary,
+		StreamSummary: streamSummary,
 	}
 
 	jsonOutput, err := json.MarshalIndent(&benchmarkOutput, "" /* prefix */, "  " /* indent */)
@@ -272,14 +298,19 @@ func outputJSON(out output, parameters Parameters, latencyValues map[float64]tim
 	out.Printf("%s\n", jsonOutput)
 }
 
-func outputPlaintext(out output, latencyValues map[float64]time.Duration, summary Summary) {
+func outputPlaintext(out output, latencyValues map[float64]time.Duration, summary Summary, streamSummary *StreamSummary) {
 	// Print out latencies
 	printLatencies(out, latencyValues)
 
 	// Print out summary
-	out.Printf("Elapsed time (seconds):   %.2f\n", summary.ElapsedTimeSeconds)
-	out.Printf("Total requests:           %v\n", summary.TotalRequests)
-	out.Printf("RPS:                      %.2f\n", summary.RPS)
+	out.Printf("Elapsed time (seconds):         %.2f\n", summary.ElapsedTimeSeconds)
+	out.Printf("Total requests:                 %v\n", summary.TotalRequests)
+	out.Printf("RPS:                            %.2f\n", summary.RPS)
+
+	if streamSummary != nil {
+		out.Printf("Total stream messages sent:     %v\n", streamSummary.TotalStreamMessagesSent)
+		out.Printf("Total stream messages received: %v\n", streamSummary.TotalStreamMessagesReceived)
+	}
 }
 
 func printParameters(out output, parameters Parameters) {
diff --git a/benchmark_stream.go b/benchmark_stream.go
new file mode 100644
index 0000000..0f2df68
--- /dev/null
+++ b/benchmark_stream.go
@@ -0,0 +1,131 @@
+// Copyright (c) 2021 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package main
+
+import (
+	"io"
+	"time"
+
+	"github.com/yarpc/yab/encoding"
+	"github.com/yarpc/yab/transport"
+)
+
+// benchmarkStreamMethod benchmarks stream requests.
+type benchmarkStreamMethod struct {
+	serializer            encoding.Serializer
+	streamRequest         *transport.StreamRequest
+	streamRequestMessages [][]byte
+}
+
+// Call dispatches stream request on the provided transport.
+func (m benchmarkStreamMethod) Call(t transport.Transport) (benchmarkCallReporter, error) {
+	streamIO := newStreamIOBenchmark(m.streamRequestMessages)
+
+	start := time.Now()
+	err := makeStreamRequest(t, m.streamRequest, m.serializer, streamIO)
+	callReport := newBenchmarkStreamCallReport(time.Since(start), streamIO.streamMessagesReceived(), streamIO.streamMessagesSent())
+
+	if err != nil {
+		return callReport, err
+	}
+
+	// response validation is delayed to avoid consuming benchmark time for
+	// deserializing the responses.
+	for _, res := range streamIO.streamResponses {
+		if err = m.serializer.CheckSuccess(&transport.Response{Body: res}); err != nil {
+			return callReport, err
+		}
+	}
+
+	return callReport, err
+}
+
+func (m benchmarkStreamMethod) CallMethodType() encoding.MethodType {
+	return m.serializer.MethodType()
+}
+
+// streamIOBenchmark provides stream IO methods using the provided stream requests
+// and records the stream responses.
+type streamIOBenchmark struct {
+	streamRequestsIdx int      // current stream request iterator index
+	streamRequests    [][]byte // provided stream requests
+
+	streamResponses [][]byte // recorded stream responses
+}
+
+func newStreamIOBenchmark(streamRequests [][]byte) *streamIOBenchmark {
+	return &streamIOBenchmark{
+		streamRequests: streamRequests,
+	}
+}
+
+// NextRequest returns next stream request from provided requests
+// returns EOF if last index has been reached.
+func (b *streamIOBenchmark) NextRequest() ([]byte, error) {
+	if len(b.streamRequests) == b.streamRequestsIdx {
+		return nil, io.EOF
+	}
+
+	// TODO: support `stream-interval` option which throttles the rate of input.
+	req := b.streamRequests[b.streamRequestsIdx]
+	b.streamRequestsIdx++
+	return req, nil
+}
+
+// HandleResponse records stream response.
+func (b *streamIOBenchmark) HandleResponse(res []byte) error {
+	b.streamResponses = append(b.streamResponses, res)
+	return nil
+}
+
+func (b *streamIOBenchmark) streamMessagesReceived() int {
+	return len(b.streamResponses)
+}
+
+func (b *streamIOBenchmark) streamMessagesSent() int {
+	return b.streamRequestsIdx
+}
+
+type benchmarkStreamCallReport struct {
+	latency                time.Duration
+	streamMessagesReceived int
+	streamMessagesSent     int
+}
+
+func newBenchmarkStreamCallReport(latency time.Duration, streamMessagesReceived, streamMessagesSent int) benchmarkStreamCallReport {
+	return benchmarkStreamCallReport{
+		latency:                latency,
+		streamMessagesReceived: streamMessagesReceived,
+		streamMessagesSent:     streamMessagesSent,
+	}
+}
+
+func (r benchmarkStreamCallReport) Latency() time.Duration {
+	return r.latency
+}
+
+func (r benchmarkStreamCallReport) StreamMessagesReceived() int {
+	return r.streamMessagesReceived
+}
+
+func (r benchmarkStreamCallReport) StreamMessagesSent() int {
+	return r.streamMessagesSent
+}
diff --git a/benchmark_unary.go b/benchmark_unary.go
index 232b510..f0b6cb3 100644
--- a/benchmark_unary.go
+++ b/benchmark_unary.go
@@ -34,13 +34,17 @@ type benchmarkUnaryMethod struct {
 }
 
 // Call dispatches unary request on the provided transport.
-func (m benchmarkUnaryMethod) Call(t transport.Transport) (benchmarkCallLatencyResult, error) {
+func (m benchmarkUnaryMethod) Call(t transport.Transport) (benchmarkCallReporter, error) {
 	start := time.Now()
 	res, err := makeRequest(t, m.req)
-	duration := time.Since(start)
+	latency := time.Since(start)
 
 	if err == nil {
 		err = m.serializer.CheckSuccess(res)
 	}
-	return newBenchmarkCallLatencyResult(duration), err
+	return newBenchmarkCallLatencyReport(latency), err
+}
+
+func (m benchmarkUnaryMethod) CallMethodType() encoding.MethodType {
+	return encoding.Unary
 }
diff --git a/handler.go b/handler.go
index 2c83a79..494ab85 100644
--- a/handler.go
+++ b/handler.go
@@ -100,7 +100,20 @@ func (r requestHandler) handleStreamRequest() {
 		}
 	}
 
-	// TODO: handle streaming benchmark.
+	// Fetch all the request messages from the provided stream request reader.
+	// This is compatible with warmup+benchmark or benchmark only mode where
+	// during warmup all the requests are recorded and the same are returned from
+	// allRequests method.
+	streamRequests, err := streamIO.allRequests()
+	if err != nil {
+		r.out.Fatalf("%v\n", err)
+	}
+
+	runBenchmark(r.out, r.logger, r.opts, r.resolved, streamReq.Request.Method, benchmarkStreamMethod{
+		serializer:            r.serializer,
+		streamRequest:         streamReq,
+		streamRequestMessages: streamRequests,
+	})
 }
 
 // shouldMakeInitialRequest returns true if initial request must be made
@@ -146,7 +159,6 @@ func makeStreamRequest(t transport.Transport, streamReq *transport.StreamRequest
 
 // makeServerStream starts server-side streaming rpc
 func makeServerStream(ctx context.Context, stream *yarpctransport.ClientStream, streamIO StreamIO) error {
-
 	req, err := streamIO.NextRequest()
 	// Use nil body if no initial request input is empty, since request
 	// is mandatory in server streaming rpc.
@@ -183,7 +195,6 @@ func makeServerStream(ctx context.Context, stream *yarpctransport.ClientStream,
 
 // makeClientStream starts client-side streaming rpc
 func makeClientStream(ctx context.Context, stream *yarpctransport.ClientStream, streamIO StreamIO) error {
-
 	var err error
 	for err == nil {
 		var reqBody []byte
@@ -210,7 +221,6 @@ func makeClientStream(ctx context.Context, stream *yarpctransport.ClientStream,
 
 // makeBidiStream starts bi-directional streaming rpc
 func makeBidiStream(ctx context.Context, stream *yarpctransport.ClientStream, streamIO StreamIO) error {
-
 	var wg sync.WaitGroup
 	var sendErr error
 
@@ -313,8 +323,13 @@ func closeSendStream(ctx context.Context, stream *yarpctransport.ClientStream) e
 
 // streamIOInitializer uses provided stream message reader to provide stream
 // IO methods to get next requests and handle responses.
+//
+// It also records all the requests read from provided stream message reader
+// which can be passed to benchmark, useful when user wants warmup + benchmark
+// requests together.
 type streamIOInitializer struct {
-	eofReached bool // flag indicates if streamMsgReader returned EOF
+	eofReached     bool     // flag indicates if streamMsgReader returned EOF
+	streamRequests [][]byte // recorded stream requests
 
 	out             output
 	serializer      encoding.Serializer
@@ -339,6 +354,7 @@ func (s *streamIOInitializer) NextRequest() ([]byte, error) {
 		return nil, fmt.Errorf("Failed while reading stream input: %v", err)
 	}
 
+	s.streamRequests = append(s.streamRequests, msg)
 	return msg, nil
 }
 
@@ -358,6 +374,19 @@ func (s *streamIOInitializer) HandleResponse(body []byte) error {
 	return nil
 }
 
+// allRequests returns all the requests from the stream reader.
+// It reads all the requests until provided reader reaches EOF and then returns
+// all the requests.
+func (s *streamIOInitializer) allRequests() ([][]byte, error) {
+	for !s.eofReached {
+		if _, err := s.NextRequest(); err != nil && err != io.EOF {
+			return nil, err
+		}
+	}
+
+	return s.streamRequests, nil
+}
+
 // newStreamIOInitializer returns streamIO which also records requests.
 func newStreamIOInitializer(out output, serializer encoding.Serializer, streamMsgReader encoding.StreamRequestReader) *streamIOInitializer {
 	return &streamIOInitializer{


Test patch:
diff --git a/benchmark_stream_test.go b/benchmark_stream_test.go
new file mode 100644
index 0000000..bb02f59
--- /dev/null
+++ b/benchmark_stream_test.go
@@ -0,0 +1,262 @@
+// Copyright (c) 2021 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package main
+
+import (
+	"io"
+	"testing"
+	"time"
+
+	"github.com/opentracing/opentracing-go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/yarpc/yab/encoding"
+	"github.com/yarpc/yab/protobuf"
+	"github.com/yarpc/yab/testdata/protobuf/simple"
+	"github.com/yarpc/yab/transport"
+)
+
+func TestBenchmarkMethodWarmTransportGRPCStreams(t *testing.T) {
+	tests := []struct {
+		name                                 string
+		procedure                            string
+		method                               string
+		numConnections                       int
+		requests                             [][]byte
+		returnOutput                         []simple.Foo
+		expectedInput                        []simple.Foo
+		expectedStreamsOpened                int32
+		expectedServerSentStreamMessages     int32
+		expectedServerReceivedStreamMessages int32
+	}{
+		{
+			name:                                 "client stream success",
+			procedure:                            "Bar::ClientStream",
+			method:                               "Bar/ClientStream",
+			numConnections:                       10,
+			requests:                             [][]byte{nil, nil},
+			returnOutput:                         []simple.Foo{{}},
+			expectedInput:                        []simple.Foo{{}, {}},
+			expectedStreamsOpened:                10,
+			expectedServerReceivedStreamMessages: 20, // 10 client streams, each of them send 2 messages
+			expectedServerSentStreamMessages:     10, // server responds with 1 message for each client stream
+		},
+		{
+			name:                                 "bidirectional stream success",
+			procedure:                            "Bar::BidiStream",
+			method:                               "Bar/BidiStream",
+			numConnections:                       10,
+			requests:                             [][]byte{nil, nil},
+			returnOutput:                         []simple.Foo{{}, {}},
+			expectedInput:                        []simple.Foo{{}, {}},
+			expectedStreamsOpened:                10,
+			expectedServerSentStreamMessages:     20, // 10 client streams, each of them send 2 messages
+			expectedServerReceivedStreamMessages: 20, // server responds with 2 messages for each client stream
+		},
+		{
+			name:                                 "server stream success",
+			procedure:                            "Bar::ServerStream",
+			method:                               "Bar/ServerStream",
+			numConnections:                       10,
+			requests:                             [][]byte{nil},
+			returnOutput:                         []simple.Foo{{}},
+			expectedInput:                        []simple.Foo{{}},
+			expectedStreamsOpened:                10,
+			expectedServerSentStreamMessages:     10, // 10 client streams, each of them send 1 message
+			expectedServerReceivedStreamMessages: 10, // server responds with 1 message for each client stream
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc := &simpleService{
+				expectedInput: tt.expectedInput,
+				returnOutput:  tt.returnOutput,
+			}
+			lis, server := setupGRPCServer(t, svc)
+			defer server.Stop()
+
+			req := &transport.Request{
+				TargetService: "foo",
+				Method:        tt.procedure,
+				Timeout:       time.Second,
+			}
+
+			source, err := protobuf.NewDescriptorProviderFileDescriptorSetBins("./testdata/protobuf/simple/simple.proto.bin")
+			require.NoError(t, err)
+
+			serializer, err := encoding.NewProtobuf(tt.method, source)
+			require.NoError(t, err)
+
+			bench := benchmarkStreamMethod{
+				serializer:            serializer,
+				streamRequest:         &transport.StreamRequest{Request: req},
+				streamRequestMessages: tt.requests,
+			}
+
+			transports, err := warmTransports(bench, tt.numConnections, TransportOptions{
+				ServiceName: "foo",
+				CallerName:  "test",
+				Peers:       []string{"grpc://" + lis.String()},
+			}, _resolvedGrpcProto, 1)
+			require.NoError(t, err)
+
+			for i, transport := range transports {
+				assert.NotNil(t, transport, "transports[%d] must not be nil", i)
+			}
+
+			assert.Equal(t, tt.expectedStreamsOpened, svc.streamsOpened.Load())
+			assert.Equal(t, tt.expectedServerSentStreamMessages, svc.serverSentStreamMessages.Load())
+			assert.Equal(t, tt.expectedServerReceivedStreamMessages, svc.serverReceivedStreamMessages.Load())
+		})
+	}
+}
+
+func TestStreamBenchmarkCallMethod(t *testing.T) {
+	tests := []struct {
+		name                                 string
+		procedure                            string
+		method                               string
+		requests                             [][]byte
+		returnOutput                         []simple.Foo
+		expectedInput                        []simple.Foo
+		expectedStreamsOpened                int32
+		expectedServerSentStreamMessages     int32
+		expectedServerReceivedStreamMessages int32
+	}{
+		{
+			name:                                 "client stream success",
+			procedure:                            "Bar::ClientStream",
+			method:                               "Bar/ClientStream",
+			requests:                             [][]byte{nil, nil},
+			returnOutput:                         []simple.Foo{{}},
+			expectedInput:                        []simple.Foo{{}, {}},
+			expectedStreamsOpened:                1,
+			expectedServerReceivedStreamMessages: 2, // client stream sends 2 messages
+			expectedServerSentStreamMessages:     1, // server responds with 1 message
+		},
+		{
+			name:                                 "bidirectional stream success",
+			procedure:                            "Bar::BidiStream",
+			method:                               "Bar/BidiStream",
+			requests:                             [][]byte{nil, nil},
+			returnOutput:                         []simple.Foo{{}, {}},
+			expectedInput:                        []simple.Foo{{}, {}},
+			expectedStreamsOpened:                1,
+			expectedServerSentStreamMessages:     2, // client stream sends 2 messages
+			expectedServerReceivedStreamMessages: 2, // server responds with 2 messages
+		},
+		{
+			name:                                 "server stream success",
+			procedure:                            "Bar::ServerStream",
+			method:                               "Bar/ServerStream",
+			requests:                             [][]byte{nil},
+			returnOutput:                         []simple.Foo{{}},
+			expectedInput:                        []simple.Foo{{}},
+			expectedStreamsOpened:                1,
+			expectedServerSentStreamMessages:     1, // client stream sends 1 message
+			expectedServerReceivedStreamMessages: 1, // server responds with 1 message
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc := &simpleService{
+				expectedInput: tt.expectedInput,
+				returnOutput:  tt.returnOutput,
+			}
+			lis, server := setupGRPCServer(t, svc)
+			defer server.Stop()
+
+			req := &transport.Request{
+				TargetService: "foo",
+				Method:        tt.procedure,
+				Timeout:       time.Second,
+			}
+
+			source, err := protobuf.NewDescriptorProviderFileDescriptorSetBins("./testdata/protobuf/simple/simple.proto.bin")
+			require.NoError(t, err)
+
+			serializer, err := encoding.NewProtobuf(tt.method, source)
+			require.NoError(t, err)
+
+			bench := benchmarkStreamMethod{
+				serializer:            serializer,
+				streamRequest:         &transport.StreamRequest{Request: req},
+				streamRequestMessages: tt.requests,
+			}
+
+			grpcTransport, err := transport.NewGRPC(transport.GRPCOptions{
+				Addresses: getHosts([]string{"grpc://" + lis.String()}),
+				Tracer:    opentracing.NoopTracer{},
+				Caller:    "test",
+				Encoding:  _resolvedGrpcProto.enc.String(),
+			})
+
+			callReport, err := bench.Call(grpcTransport)
+			require.NoError(t, err)
+
+			streamCallReport, ok := callReport.(benchmarkStreamCallReporter)
+			require.True(t, ok)
+			assert.Equal(t, int(tt.expectedServerSentStreamMessages), streamCallReport.StreamMessagesReceived())
+			assert.Equal(t, int(tt.expectedServerReceivedStreamMessages), streamCallReport.StreamMessagesSent())
+
+			assert.Equal(t, tt.expectedStreamsOpened, svc.streamsOpened.Load())
+			assert.Equal(t, tt.expectedServerSentStreamMessages, svc.serverSentStreamMessages.Load())
+			assert.Equal(t, tt.expectedServerReceivedStreamMessages, svc.serverReceivedStreamMessages.Load())
+		})
+	}
+}
+
+func TestBenchmarkStreamIO(t *testing.T) {
+	t.Run("next request with eof", func(t *testing.T) {
+		requests := [][]byte{
+			[]byte("1"),
+			[]byte("2"),
+		}
+		streamIO := newStreamIOBenchmark(requests)
+
+		for _, expectedRequest := range requests {
+			req, err := streamIO.NextRequest()
+			require.NoError(t, err)
+			assert.Equal(t, expectedRequest, req)
+		}
+
+		_, err := streamIO.NextRequest()
+		assert.EqualError(t, err, io.EOF.Error())
+		assert.Equal(t, len(requests), streamIO.streamMessagesSent())
+	})
+
+	t.Run("handle response", func(t *testing.T) {
+		responses := [][]byte{
+			[]byte("1"),
+			[]byte("2"),
+		}
+		streamIO := newStreamIOBenchmark(nil)
+
+		for _, response := range responses {
+			require.NoError(t, streamIO.HandleResponse(response))
+		}
+
+		assert.Equal(t, streamIO.streamResponses, responses)
+		assert.Equal(t, len(responses), streamIO.streamMessagesReceived())
+	})
+}
diff --git a/benchmark_unary_test.go b/benchmark_unary_test.go
index b902c7c..4f1d359 100644
--- a/benchmark_unary_test.go
+++ b/benchmark_unary_test.go
@@ -1,4 +1,4 @@
-// Copyright (c) 2016 Uber Technologies, Inc.
+// Copyright (c) 2021 Uber Technologies, Inc.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy
 // of this software and associated documentation files (the "Software"), to deal
@@ -159,7 +159,7 @@ func TestBenchmarkMethodCall(t *testing.T) {
 			m.req.Method = tt.reqMethod
 		}
 
-		callResult, err := m.Call(tp)
+		res, err := m.Call(tp)
 		if tt.wantErr != "" {
 			if assert.Error(t, err, "call should fail") {
 				assert.Contains(t, err.Error(), tt.wantErr, "call should return 0 duration")
@@ -168,7 +168,7 @@ func TestBenchmarkMethodCall(t *testing.T) {
 		}
 
 		assert.NoError(t, err, "call should not fail")
-		assert.True(t, callResult.Latency() > time.Microsecond, "duration was too short, got %v", callResult.Latency())
+		assert.True(t, res.Latency() > time.Microsecond, "duration was too short, got %v", res.Latency())
 	}
 }
 
diff --git a/handler_test.go b/handler_test.go
index aeab700..8e63f99 100644
--- a/handler_test.go
+++ b/handler_test.go
@@ -22,6 +22,7 @@ package main
 
 import (
 	"bytes"
+	"errors"
 	"io"
 	"testing"
 
@@ -83,9 +84,8 @@ func TestStreamRequestRecorder(t *testing.T) {
 	})
 
 	t.Run("handle response success", func(t *testing.T) {
-		var outBuf bytes.Buffer
 		out := testOutput{
-			Buffer: &outBuf,
+			Buffer: &bytes.Buffer{},
 		}
 
 		body := `{
@@ -99,12 +99,31 @@ func TestStreamRequestRecorder(t *testing.T) {
 	})
 
 	t.Run("handle response failure", func(t *testing.T) {
-		var outBuf bytes.Buffer
-		out := testOutput{
-			Buffer: &outBuf,
+		streamIO := streamIOInitializer{serializer: encoding.NewJSON("test")}
+		require.Error(t, streamIO.HandleResponse([]byte("a:b")))
+	})
+
+	t.Run("all request failure", func(t *testing.T) {
+		streamIO := streamIOInitializer{
+			streamMsgReader: &mockStreamReader{returnErr: errors.New("test")},
 		}
 
-		streamIO := streamIOInitializer{out: out, serializer: encoding.NewJSON("test")}
-		require.Error(t, streamIO.HandleResponse([]byte("a:b")))
+		_, err := streamIO.allRequests()
+		assert.EqualError(t, err, "Failed while reading stream input: test")
+	})
+
+	t.Run("all requests success", func(t *testing.T) {
+		requests := [][]byte{
+			[]byte("request-a"),
+			[]byte("request-b"),
+		}
+		streamIO := streamIOInitializer{streamMsgReader: &mockStreamReader{requests: requests}}
+
+		_, err := streamIO.NextRequest()
+		require.NoError(t, err)
+
+		gotRequests, err := streamIO.allRequests()
+		require.NoError(t, err)
+		assert.Equal(t, requests, gotRequests)
 	})
 }
diff --git a/integration_test.go b/integration_test.go
index eabdbf5..28330c8 100644
--- a/integration_test.go
+++ b/integration_test.go
@@ -46,6 +46,7 @@ import (
 	"github.com/uber/tchannel-go"
 	"github.com/uber/tchannel-go/testutils"
 	"github.com/uber/tchannel-go/thrift"
+	"go.uber.org/atomic"
 	"go.uber.org/yarpc"
 	ytransport "go.uber.org/yarpc/api/transport"
 	ythrift "go.uber.org/yarpc/encoding/thrift"
@@ -362,6 +363,10 @@ func setupYARPCServer(t *testing.T, inbound ytransport.Inbound, opts ...ythrift.
 }
 
 type simpleService struct {
+	streamsOpened                atomic.Int32 // number of streams opened by the client
+	serverReceivedStreamMessages atomic.Int32 // number of stream messages received by server
+	serverSentStreamMessages     atomic.Int32 // number of stream messages send by server
+
 	expectedInput []simple.Foo // messages expected from the client
 
 	returnOutput []simple.Foo // messages to be streamed back to client
@@ -377,6 +382,8 @@ func (s *simpleService) Baz(c context.Context, in *simple.Foo) (*simple.Foo, err
 }
 
 func (s *simpleService) ClientStream(stream simple.Bar_ClientStreamServer) error {
+	s.streamsOpened.Inc()
+
 	idx := 0
 	for idx < len(s.expectedInput) {
 		msg, err := stream.Recv()
@@ -391,10 +398,13 @@ func (s *simpleService) ClientStream(stream simple.Bar_ClientStreamServer) error
 			return fmt.Errorf("unexpected input found at index: %d", idx)
 		}
 
+		s.serverReceivedStreamMessages.Inc()
 		idx++
 	}
 
 	if len(s.returnOutput) > 0 {
+		s.serverSentStreamMessages.Inc()
+
 		return stream.SendAndClose(&s.returnOutput[0])
 	}
 
@@ -402,6 +412,9 @@ func (s *simpleService) ClientStream(stream simple.Bar_ClientStreamServer) error
 }
 
 func (s *simpleService) ServerStream(req *simple.Foo, stream simple.Bar_ServerStreamServer) error {
+	s.streamsOpened.Inc()
+	s.serverReceivedStreamMessages.Inc()
+
 	if len(s.expectedInput) > 0 && !reflect.DeepEqual(*req, s.expectedInput[0]) {
 		return fmt.Errorf("unexpected input found")
 	}
@@ -410,12 +423,16 @@ func (s *simpleService) ServerStream(req *simple.Foo, stream simple.Bar_ServerSt
 		if err := stream.Send(&req); err != nil {
 			return err
 		}
+
+		s.serverSentStreamMessages.Inc()
 	}
 
 	return s.returnError
 }
 
 func (s *simpleService) BidiStream(stream simple.Bar_BidiStreamServer) error {
+	s.streamsOpened.Inc()
+
 	idx := 0
 	for idx < len(s.expectedInput) {
 		msg, err := stream.Recv()
@@ -430,6 +447,7 @@ func (s *simpleService) BidiStream(stream simple.Bar_BidiStreamServer) error {
 			return fmt.Errorf("unexpected input found at index: %d", idx)
 		}
 
+		s.serverReceivedStreamMessages.Inc()
 		idx++
 	}
 
@@ -437,6 +455,8 @@ func (s *simpleService) BidiStream(stream simple.Bar_BidiStreamServer) error {
 		if err := stream.Send(&req); err != nil {
 			return err
 		}
+
+		s.serverSentStreamMessages.Inc()
 	}
 
 	return s.returnError
@@ -453,6 +473,240 @@ func setupGRPCServer(t *testing.T, svc *simpleService) (net.Addr, *grpc.Server)
 	return ln.Addr(), s
 }
 
+func TestGRPCStreamBenchmark(t *testing.T) {
+	tests := []struct {
+		desc          string
+		opts          Options
+		wantRes       []string
+		wantErr       string
+		returnError   error
+		returnOutput  []simple.Foo
+		expectedInput []simple.Foo
+
+		// warmup (if warmup is not 0) -> 1
+		// benchmark connection warmup -> (total connections * warmup count)
+		// benchmark -> (total benchmark requests)
+		// expectedStreamsOpened = warmup + benchmark connection warmup + benchmark
+		expectedStreamsOpened int32
+
+		// warmup (if warmup is not 0) -> len(returnOutput)
+		// benchmark connection warmup -> (total connections * warmup count * len(returnOutput))
+		// benchmark -> (total benchmark requests * len(returnOutput))
+		// expectedServerSentStreamMessages = warmup + benchmark connection warmup + benchmark
+		expectedServerSentStreamMessages int32
+
+		// warmup (if warmup is not 0) -> len(expectedInput)
+		// benchmark connection warmup -> (total connections * len(expectedInput))
+		// benchmark -> (total benchmark requests * len(expectedInput))
+		// expectedServerReceivedStreamMessages = warmup + benchmark connection warmup + benchmark
+		expectedServerReceivedStreamMessages int32
+	}{
+		{
+			desc: "client streaming - without warmup",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/ClientStream",
+					Timeout:           timeMillisFlag(time.Second),
+					RequestJSON:       `{"test":1}{"test":2} {"test":4}`,
+				},
+				TOpts: TransportOptions{
+					ServiceName: "foo",
+				},
+				BOpts: BenchmarkOptions{
+					MaxRequests:    100,
+					WarmupRequests: 0,
+					Connections:    2,
+					Concurrency:    2,
+				},
+			},
+			wantRes: []string{
+				"Total requests:                 100",
+				"Total stream messages sent:     300",
+				"Total stream messages received: 100",
+			},
+			returnOutput:                         []simple.Foo{{Test: 1}},
+			expectedInput:                        []simple.Foo{{Test: 1}, {Test: 2}, {Test: 4}},
+			expectedStreamsOpened:                100,
+			expectedServerReceivedStreamMessages: 300,
+			expectedServerSentStreamMessages:     100,
+		},
+		{
+			desc: "client streaming - with warmup",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/ClientStream",
+					Timeout:           timeMillisFlag(time.Second),
+					RequestJSON:       `{"test":1}{"test":2}{"test":-1} {"test":10}`,
+				},
+				TOpts: TransportOptions{
+					ServiceName: "foo",
+				},
+				BOpts: BenchmarkOptions{
+					MaxRequests:    100,
+					WarmupRequests: 1,
+					Connections:    2,
+					Concurrency:    2,
+				},
+			},
+			wantRes: []string{
+				"Total requests:                 100",
+				"Total stream messages sent:     400",
+				"Total stream messages received: 100",
+			},
+			returnOutput:                         []simple.Foo{{Test: 1}},
+			expectedInput:                        []simple.Foo{{Test: 1}, {Test: 2}, {Test: -1}, {Test: 10}},
+			expectedStreamsOpened:                103,
+			expectedServerReceivedStreamMessages: 412,
+			expectedServerSentStreamMessages:     103,
+		},
+		{
+			desc: "server streaming - without warmup",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/ServerStream",
+					Timeout:           timeMillisFlag(time.Second),
+					RequestJSON:       `{"test":"2"}`,
+				},
+				TOpts: TransportOptions{
+					ServiceName: "foo",
+				},
+				BOpts: BenchmarkOptions{
+					MaxRequests:    100,
+					WarmupRequests: 0,
+					Connections:    2,
+					Concurrency:    2,
+				},
+			},
+			wantRes: []string{
+				"Total requests:                 100",
+				"Total stream messages sent:     100",
+				"Total stream messages received: 200",
+			},
+			returnOutput:                         []simple.Foo{{Test: 1}, {Test: 2}},
+			expectedInput:                        []simple.Foo{{Test: 2}},
+			expectedStreamsOpened:                100,
+			expectedServerSentStreamMessages:     200,
+			expectedServerReceivedStreamMessages: 100,
+		},
+		{
+			desc: "server streaming - with warmup",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/ServerStream",
+					Timeout:           timeMillisFlag(time.Second),
+					RequestJSON:       `{"test":1}`,
+				},
+				TOpts: TransportOptions{
+					ServiceName: "foo",
+				},
+				BOpts: BenchmarkOptions{
+					MaxRequests:    100,
+					WarmupRequests: 1,
+					Connections:    2,
+					Concurrency:    2,
+				},
+			},
+			wantRes: []string{
+				"Total requests:                 100",
+				"Total stream messages sent:     100",
+				"Total stream messages received: 200",
+			},
+			returnOutput:                         []simple.Foo{{Test: 1}, {Test: 2}},
+			expectedInput:                        []simple.Foo{{Test: 1}},
+			expectedStreamsOpened:                103,
+			expectedServerSentStreamMessages:     206,
+			expectedServerReceivedStreamMessages: 103,
+		},
+		{
+			desc: "bidirectional streaming - with warmup",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/BidiStream",
+					Timeout:           timeMillisFlag(time.Second),
+					RequestJSON:       `{"test":1}{"test":2}`,
+				},
+				TOpts: TransportOptions{
+					ServiceName: "foo",
+				},
+				BOpts: BenchmarkOptions{
+					MaxRequests:    100,
+					WarmupRequests: 1,
+					Connections:    2,
+					Concurrency:    2,
+				},
+			},
+			wantRes: []string{
+				"Total requests:                 100",
+				"Total stream messages sent:     200",
+				"Total stream messages received: 200",
+			},
+			returnOutput:                         []simple.Foo{{Test: 1}, {Test: 2}},
+			expectedInput:                        []simple.Foo{{Test: 1}, {Test: 2}},
+			expectedStreamsOpened:                103,
+			expectedServerSentStreamMessages:     206,
+			expectedServerReceivedStreamMessages: 206,
+		},
+		{
+			desc: "bidirectional streaming - without warmup",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/BidiStream",
+					Timeout:           timeMillisFlag(time.Second),
+					RequestJSON:       `{"test":1}{"test":2}`,
+				},
+				TOpts: TransportOptions{
+					ServiceName: "foo",
+				},
+				BOpts: BenchmarkOptions{
+					MaxRequests:    100,
+					WarmupRequests: 0,
+					Connections:    2,
+					Concurrency:    2,
+				},
+			},
+			wantRes: []string{
+				"Total requests:                 100",
+				"Total stream messages sent:     200",
+				"Total stream messages received: 200",
+			},
+			returnOutput:                         []simple.Foo{{Test: 1}, {Test: 2}},
+			expectedInput:                        []simple.Foo{{Test: 1}, {Test: 2}},
+			expectedStreamsOpened:                100,
+			expectedServerSentStreamMessages:     200,
+			expectedServerReceivedStreamMessages: 200,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			svc := &simpleService{
+				expectedInput: tt.expectedInput,
+				returnOutput:  tt.returnOutput,
+				returnError:   tt.returnError,
+			}
+			addr, server := setupGRPCServer(t, svc)
+			defer server.Stop()
+
+			tt.opts.TOpts.Peers = []string{"grpc://" + addr.String()}
+
+			gotOut, gotErr := runTestWithOpts(tt.opts)
+			assert.Contains(t, gotErr, tt.wantErr)
+			for _, wantOut := range tt.wantRes {
+				assert.Contains(t, gotOut, wantOut)
+			}
+
+			assert.Equal(t, tt.expectedStreamsOpened, svc.streamsOpened.Load())
+			assert.Equal(t, tt.expectedServerSentStreamMessages, svc.serverSentStreamMessages.Load())
+			assert.Equal(t, tt.expectedServerReceivedStreamMessages, svc.serverReceivedStreamMessages.Load())
+		})
+	}
+}
+
 func TestGRPCStream(t *testing.T) {
 	tests := []struct {
 		desc    string
@@ -583,7 +837,7 @@ test: 10
 					ServiceName: "foo",
 				},
 			},
-			wantErr: "Failed while making stream call: code:unavailable message:roundrobin peer list timed out waiting for peer: context deadline exceeded\n",
+			wantErr: "Failed while making stream call",
 		},
 		{
 			desc: "server streaming",
@@ -723,7 +977,7 @@ test: 1
 			tt.opts.TOpts.Peers = []string{"grpc://" + addr.String()}
 
 			gotOut, gotErr := runTestWithOpts(tt.opts)
-			assert.Equal(t, tt.wantErr, gotErr)
+			assert.Contains(t, gotErr, tt.wantErr)
 			assert.Contains(t, gotOut, tt.wantRes)
 		})
 	}
diff --git a/utils_for_test.go b/utils_for_test.go
index 97c146a..adb261d 100644
--- a/utils_for_test.go
+++ b/utils_for_test.go
@@ -49,6 +49,7 @@ var _testLogger = zap.NewNop()
 var (
 	_resolvedTChannelThrift = resolvedProtocolEncoding{protocol: transport.TChannel, enc: encoding.Thrift}
 	_resolvedTChannelRaw    = resolvedProtocolEncoding{protocol: transport.TChannel, enc: encoding.Raw}
+	_resolvedGrpcProto      = resolvedProtocolEncoding{protocol: transport.GRPC, enc: encoding.Protobuf}
 )
 
 type testOutput struct {