Expand stream's flow control in case of an active read. #1248

MakMukhi · 2017-05-18T20:22:40Z

Building on original PR by apolcyn@ #1073

Benchmark on scenario:

{  
   "scenarios":[  
      {  
         "name":"go_protobuf_sync_unary_ping_pong_secure_2097152db",
         "warmup_seconds":5,
         "benchmark_seconds":30,
         "num_servers":1,
         "server_config":{  
            "async_server_threads":0,
            "security_params":{  
               "use_test_ca":true,
               "server_host_override":"foo.test.google.fr"
            },
            "payload_config":{  
               "simple_params":{  
                  "resp_size":2097152,
                  "req_size":2097152
               }
            },
            "server_type":"SYNC_SERVER"
         },
         "client_config":{  
            "client_type":"SYNC_CLIENT",
            "security_params":{  
               "use_test_ca":true,
               "server_host_override":"foo.test.google.fr"
            },
            "payload_config":{  
               "simple_params":{  
                  "resp_size":2097152,
                  "req_size":2097152
               }
            },
            "client_channels":1,
            "async_client_threads":1,
            "outstanding_rpcs_per_channel":1,
            "rpc_type":"UNARY",
            "load_params":{  
               "closed_loop":{  

               }
            },
            "histogram_params":{  
               "max_possible":60000000000.0,
               "resolution":0.01
            }
         },
         "num_clients":1
      }
   ]
}

Results before the changes: QPS: 0.1

Results after the changes: QPS: 0.9

apolcyn · 2017-05-19T17:24:54Z

transport/control.go

+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	senderQuota := int32(f.limit - (f.pendingData + f.pendingUpdate))


small wording nit: rename to estimatedSenderQuota and estimatedUntransmittedData, for clarity here?

dfawley · 2017-05-19T17:04:54Z

transport/control.go

@@ -167,14 +169,37 @@ type inFlow struct {
 	// The amount of data the application has consumed but grpc has not sent
 	// window update for them. Used to reduce window update frequency.
 	pendingUpdate uint32
+	// delta is the extra window update given by receiver when an application
+	// is reading data bigger in size than the inFlow limit.
+	delta int32


Should this be unsigned since it can't be negative? (Also, you cast it to uint32 when reading and cast to int32 when assigning.)

dfawley · 2017-05-19T17:11:04Z

transport/control.go

@@ -189,6 +214,14 @@ func (f *inFlow) onRead(n uint32) uint32 {
 		return 0
 	}
 	f.pendingData -= n
+	if f.delta > 0 {
+		f.delta -= int32(n)


I think this would be easier to read without the casts and taking the negative of a negative number:

if n > f.delta { n -= f.delta f.delta = 0 } else { f.delta -= n n = 0 }

(Also would allow f.delta to be uint32 like it seems like it wants to be.)

dfawley · 2017-05-19T17:12:15Z

transport/http2_client.go

@@ -173,9 +173,9 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
 	conn, err := dial(ctx, opts.Dialer, addr.Addr)
 	if err != nil {
 		if opts.FailOnNonTempDialError {
-			return nil, connectionErrorf(isTemporary(err), err, "transport: %v", err)
+			return nil, connectionErrorf(isTemporary(err), err, "transport: Error while dialing %v", err)


Nit: lowercase "e" in "error", and add a colon before the error being appended.

dfawley · 2017-05-19T17:15:36Z

transport/http2_client.go

+	if s.state == streamDone {
+		return
+	}
+	if w := s.fc.maybeAdjust(n); n > 0 {


should the if be checking w?

If not, check n first.

dfawley · 2017-05-19T17:19:21Z

transport/http2_server.go

+	if s.state == streamDone {
+		return
+	}
+	if w := s.fc.maybeAdjust(n); n > 0 {


Same as above re: w and n.

Can this function be shared somehow? Should it be a method on the Stream instead of server/client?

dfawley · 2017-05-19T17:26:26Z

transport/transport.go

-// Read reads all the data available for this Stream from the transport and
+// Read reads all p bytes from the wire for this stream.
+func (s *Stream) Read(p []byte) (n int, err error) {
+	// Don't request a read if there's was an error earlier


nit: "there was"

dfawley · 2017-05-22T22:26:38Z

test/end2end_test.go

 	te := newTest(t, e)
 	te.startServer(&testServer{security: e.security})
 	defer te.tearDown()
 	tc := testpb.NewTestServiceClient(te.clientConn())

-	stream, err := tc.StreamingInputCall(te.ctx)
+	ctx, _ := context.WithTimeout(te.ctx, time.Minute*3)


apolcyn · 2017-05-23T05:21:11Z

rpc_util_test.go

@@ -67,7 +75,7 @@ func TestSimpleParsing(t *testing.T) {
 		// Check that messages with length >= 2^24 are parsed.
 		{append([]byte{0, 1, 0, 0, 0}, bigMsg...), nil, bigMsg, compressionNone},
 	} {
-		buf := bytes.NewReader(test.p)


nit: I think the use of fullReader and other changes to this file can be reverted, since the transport is still an io.Reader.

These test here rely on recvMsg's behavior of reading the full message, (which is true when it interacts with the transport stream). However, here the parser is given a "fake" buffer instead of the stream. Given that we changed recvMsg to go from io.ReadFull to p.r.Read, the "fake" buffer here needs to read the full message just like the transport stream does.

apolcyn · 2017-05-23T06:32:52Z

transport/control.go

+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	estSenderQuota := int32(f.limit - (f.pendingData + f.pendingUpdate))


optional nit, sorry one more naming suggestion:
change to maxPossibleSenderQuota and ~~minPossibleUnTransmittedData~~ (keep estUnTransmitted)?

How about I add comments and leave the names as is? I fear long names take us away from Go-styling.

apolcyn · 2017-05-23T07:08:48Z

transport/transport_test.go

+	inBuf := make([]byte, 1)
+	actualCount, actualErr := s.Read(inBuf)
+	if actualCount != 0 {
+		t.Errorf("actualCount, _ := s.Read(_) differs; want %v; got %v", 0, actualCount)


nit: s/want %v/want 0

…#1248)" This reverts commit 6dff7c5.

…d. (grpc#1248)"" This reverts commit 8f0921a.

MakMukhi added 3 commits May 17, 2017 13:58

First commit

9012be0

Imported tests from the original PR by @apolcyn.

c2d5ec9

Merge branch 'master' into stream_window

4d09482

MakMukhi requested review from menghanl, apolcyn and dfawley May 18, 2017 20:27

MakMukhi added 3 commits May 18, 2017 13:45

Formatting fixes.

722d058

More formating fixes

c9d1fc8

more golint

78f4eb0

MakMukhi mentioned this pull request May 19, 2017

Connection latency significantly affects throughput #1043

Closed

Make logs more informative.

da22a03

apolcyn reviewed May 19, 2017

View reviewed changes

dfawley reviewed May 19, 2017

View reviewed changes

dfawley mentioned this pull request May 19, 2017

Can make default/initial Window Size configurable ? #760

Closed

post-review update

9d8f07c

dfawley assigned dfawley and menghanl May 19, 2017

Added test to check flow control accounts after sending large messages.

04b882f

derekchiang mentioned this pull request May 22, 2017

Uploading data is painfully slow pachyderm/pachyderm#930

Closed

dfawley reviewed May 22, 2017

View reviewed changes

MakMukhi added 3 commits May 22, 2017 15:46

post-review update

d4d90a8

Merge branch 'master' into stream_window

1df7a2a

Empty commit to kickstart travis.

e85309f

apolcyn reviewed May 23, 2017

View reviewed changes

Post-review update.

a96ed0a

dfawley approved these changes May 23, 2017

View reviewed changes

MakMukhi merged commit 6dff7c5 into grpc:master May 23, 2017

jefferai mentioned this pull request May 24, 2017

panic: io.Reader is *transport.recvBufferReader, not *transport.transportReader #1257

Closed

menghanl added a commit to menghanl/grpc-go that referenced this pull request May 24, 2017

Revert "Expand stream's flow control in case of an active read. (grpc…

8f0921a

…#1248)" This reverts commit 6dff7c5.

menghanl mentioned this pull request May 24, 2017

Re-enable handler tests #1259

Merged

menghanl added a commit to menghanl/grpc-go that referenced this pull request May 24, 2017

Revert "Revert "Expand stream's flow control in case of an active rea…

099ad78

…d. (grpc#1248)"" This reverts commit 8f0921a.

menghanl added 1.4 Type: Performance Performance improvements (CPU, network, memory, etc) labels Jun 7, 2017

apolcyn mentioned this pull request Jun 21, 2017

transport: consume per-stream inflow windows ahead of time - large speedup for big messages #1073

Closed

MakMukhi deleted the stream_window branch May 4, 2018 02:10

gwihlidal mentioned this pull request Jul 23, 2018

Poor large binary streaming throughput tower-rs/tower-grpc#75

Closed

lock bot locked as resolved and limited conversation to collaborators Oct 31, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Expand stream's flow control in case of an active read. #1248

Expand stream's flow control in case of an active read. #1248

MakMukhi commented May 18, 2017 •

edited

apolcyn May 19, 2017

dfawley May 19, 2017

dfawley May 19, 2017 •

edited

dfawley May 19, 2017

dfawley May 19, 2017

dfawley May 19, 2017

dfawley May 19, 2017

dfawley May 22, 2017

apolcyn May 23, 2017

MakMukhi May 23, 2017

apolcyn May 23, 2017 •

edited

MakMukhi May 23, 2017

apolcyn May 23, 2017

Expand stream's flow control in case of an active read. #1248

Expand stream's flow control in case of an active read. #1248

Conversation

MakMukhi commented May 18, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dfawley May 19, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apolcyn May 23, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MakMukhi commented May 18, 2017 •

edited

dfawley May 19, 2017 •

edited

apolcyn May 23, 2017 •

edited