transport: drain client transport when streamID approaches maxStreamID #5889

arvindbr8 · 2022-12-22T20:20:06Z

From #5358 (comment), we noticed that users may be working around the max stream ID issue by using max connection age, which is not really an intended use case of that feature. This change sets MaxStreamID to 75% of math.uint32 and drain client transport when next stream ID > MaxStreamID.

Fixes #5600

RELEASE NOTES:

transport: drain client transport when streamID approaches MaxStreamID

dfawley · 2022-12-22T22:14:44Z

internal/transport/defaults.go

+// MaxStreamIDForTesting is the upper bound for the stream ID before the current
+// transport gracefully closes and a new transport is created for subsequent RPCs.
+// This is set to 75% of math.MaxUint32. It's exported so that tests can override it.
+var MaxStreamIDForTesting = uint32(float32(math.MaxUint32) * 0.75)


Nit: functions that do stuff for testing are ForTesting. This can be just MaxStreamID, and some tests happen to override it, which seems fine to me.

sg! i'm updating this

dfawley · 2022-12-22T22:16:31Z

internal/transport/http2_client.go

+			// 75% of math.MaxUint32, which then signals gRPC to restart transport
+			// for subsequent RPCs.


maybe "which signals gRPC that the connection is closed and a new one must be created for subsequent RPCs"? "Restart transport" doesn't really imply the right things, and also this doesn't happen automatically (the LB policy would need to request it).

dfawley · 2022-12-22T22:17:12Z

internal/transport/http2_client.go

+			if t.nextID > MaxStreamIDForTesting {
+				transportDrainRequired = true
+			}


Shorten: transportDrainRequired = t.nextID > MaxStreamIDForTesting

dfawley · 2022-12-22T22:18:12Z

internal/transport/http2_client.go

@@ -862,6 +869,12 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream,
 			sh.HandleRPC(s.ctx, outHeader)
 		}
 	}
+	if transportDrainRequired {
+		if logger.V(logLevel) {
+			logger.Infof("t.nextID > MaxStreamID. transport: draining")


"transport" is supposed to be the prefix of the whole log message. How about transport: t.nextID > MaxStreamID. Draining?

yes! that makes sense

dfawley · 2022-12-22T22:18:52Z

internal/transport/transport_test.go

@@ -536,6 +536,40 @@ func (s) TestInflightStreamClosing(t *testing.T) {
 	}
 }

+// TestClientTransportDrainsAfterStreamIdExhausted tests that when
+// streamID > MaxStreamId, the current client transport drains.
+func (s) TestClientTransportDrainsAfterStreamIdExhausted(t *testing.T) {


Nit: in Go style we write "ID" not "Id".

dfawley · 2022-12-22T22:27:19Z

test/end2end_test.go

+		transport.MaxStreamIDForTesting = originalMaxStreamID
+	}()
+
+	// setting up StubServer.


Similar nits.

removing...

dfawley · 2022-12-22T22:29:16Z

test/end2end_test.go

+			return nil
+		},
+	}
+	testpb.RegisterTestServiceServer(s, ss)


It's unfortunate you have to have a ton of the boilerplate that StubServer is intended to help with.

Two options:

use WithContextDialer on the client to detect when we're (re)connecting.

Add some stuff to StubServer to facilitate wrapping the listener.

This will save you manually creating the server, registering the service, dialing the client, wrapping the client with the grpc stub, and dealing with extra shutdown tasks. The resulting test code will be the more interesting code which will make it easier to read and understand what it's doing that's unique.

I was also bouncing around different things to get ListenerWrapper to work. I'm doing ahead with option 2 to facilitate StubServer with ListenerWrapper.. let me know what you think of the approach @dfawley

dfawley · 2022-12-22T22:33:36Z

test/end2end_test.go

+		if _, err = stream.Recv(); err != nil && err != io.EOF {
+			t.Fatalf("stream.Recv() = _, %v want: nil or EOF", err)
+		}


Don't we know what we should receive? Shouldn't it always be nil since the server sends 10 messages?

yup, should never error

dfawley · 2022-12-22T22:34:58Z

test/end2end_test.go

+	ss := &stubserver.StubServer{
+		FullDuplexCallF: func(stream testpb.TestService_FullDuplexCallServer) error {
+			for i := 0; i < 10; i++ {
+				m := &grpc.PreparedMsg{}


Why a PreparedMsg? That seems unnecessarily complex. Just do stream.SendMsg(testpb.WhateverThisIs{}).

sure. ive updating this in the rev

dfawley · 2022-12-22T22:44:06Z

test/end2end_test.go

+		t.Fatal("timeout expired when waiting to create new conn channel")
+	}
+
+	// verifying the connection to the old one is drained and closed.


Interestingly you haven't fully received off the streams that were active. So this only works because the flow control is enough that the server finished the stream.

I'd recommend avoiding "coincidences" like this (even if they're literally never going to not be true).

Instead you should make the client & server behavior align.

It also could be interesting to test that the stream is still active even after the new connection is created, which is how "draining" behavior should work. E.g.

Create a stream that ping-pongs back and forth.

Use it and make sure it works.

Create a stream on the next connection that does the same. (Verify that the new connection was used for it somehow.)

Use it and make sure it works.

Verify that the stream from (1) still works.

Close streams from (1) and (3). Client-side context cancelation should be easy and fine here.

Expect the original connection to close.

One way to validate which connection the stream is using: you could use a custom credentials (its AuthInfo ends up in the Peer) that counts connections and increments some number for each one. Creds could also be used to determine when connections are closed, since they return a wrapped net.Conn back to us that is what we really close.

thanks for the detailed comment. I've updated the test to align to your comment. lemme know what you think.

dfawley · 2022-12-22T22:47:58Z

internal/transport/http2_client.go

+			if t.nextID > MaxStreamIDForTesting {
+				transportDrainRequired = true
+			}


Oh whoops. You have a race. This bool needs to be set inside checkForStreamQuota then read here (or below as is and ignore here).

im made the change now

dfawley · 2022-12-22T23:04:12Z

internal/transport/defaults.go

+// MaxStreamIDForTesting is the upper bound for the stream ID before the current
+// transport gracefully closes and a new transport is created for subsequent RPCs.
+// This is set to 75% of math.MaxUint32. It's exported so that tests can override it.
+var MaxStreamIDForTesting = uint32(float32(math.MaxUint32) * 0.75)


Also a nit: I'd simplify this to something like uint32(3*1024*1024*1024) or 3_221_225_472 or 0xc0000000. (Basically, extra casts and floats (in general) should be avoided.)

i missed this one.. updating this now

arvindbr8 · 2022-12-28T02:34:21Z

I just merged 3 of my commits by

git reset --soft HEAD^1 x3
and
git commit && git push

but now it shows up as forced push on this PR 😢 .. im going to refrain for any commit amends in the future. Luckily i see all the past comments here.

internal/transport/transport_test.go

internal/transport/http2_client.go

internal/transport/transport_test.go

internal/transport/http2_client.go

test/end2end_test.go

easwars

Mostly looks good. Just some minor comments this time around.

easwars · 2022-12-29T19:34:02Z

test/end2end_test.go

+// connWrapperWithCloseCh wraps a net.Conn and pushes on a channel when closed.
+type connWrapperWithCloseCh struct {
+	net.Conn
+	CloseCh *testutils.Channel


This one needs to unexported as well.

mmm.. isn't this unexported right now?

CloseCh field is still exported, which is immaterial given that the struct is unexported now. But it is better to not have this anomaly.

got it. thanks. updating this

test/end2end_test.go

easwars · 2022-12-29T19:42:46Z

test/end2end_test.go

+
+	// Setting up 3 streams and calling Send() and Recv() once on each.
+	for i := 0; i < 3; i++ {
+		var p peer.Peer


Optional: Since p is always used as a pointer, you could instead do p := new(peer.Peer) here and replace &p with p down below.

sure. updating this

test/end2end_test.go

dfawley · 2023-01-10T16:51:00Z