Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

transport: Fix deadlock in transport caused by GOAWAY race with new stream creation #5652

Merged
merged 4 commits into from Sep 21, 2022
Merged

transport: Fix deadlock in transport caused by GOAWAY race with new stream creation #5652

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6004b0d
Fixed deadlock in transport
zasweq Sep 13, 2022
3e5eae7
Switched transport level test to e2e test
zasweq Sep 15, 2022
09bdda8
Responded to Doug's comment
zasweq Sep 19, 2022
f9a0d75
Responded to Doug's comments
zasweq Sep 21, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
13 changes: 10 additions & 3 deletions internal/transport/http2_client.go
View file
Open in desktop
Expand Up @@ -1232,16 +1232,23 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
if upperLimit == 0 { // This is the first GoAway Frame.
upperLimit = math.MaxUint32 // Kill all streams after the GoAway ID.
}

activeStreams := make(map[uint32]*Stream)
for streamID, stream := range t.activeStreams {
activeStreams[streamID] = stream
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use a slice; only add streams to be closed to it. Early exit if t.activeStreams is zero?

t.prevGoAwayID = id

if len(t.activeStreams) == 0 {
	t.mu.Unlock()
	t.Close(...)
	return
}

streamsToClose := make([]*Stream, 0, len(t.activeStreams))
for streamID, stream := range t.activeStreams {
	if streamID > id && streamID <= upperLimit {
		streamsToClose = append(streamsToClose, stream)
	}
}
t.mu.Unlock()
for _, stream := range(streamsToClose) {
	atomic.StoreUint32()
	t.closeStream()
}

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is it important to have the cap(streamsToClose) len(t.activeStreams)?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, sorry, I tried this way and for some reason it still induced deadlock. I'm going to keep it as is. I couldn't figure out why your way wouldn't work.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. Verified on my e2e_test I checked in and also on the transport test which I didn't check in since it wrote directly to framer.


t.prevGoAwayID = id
t.mu.Unlock()
for streamID, stream := range activeStreams {
if streamID > id && streamID <= upperLimit {
// The stream was unprocessed by the server.
atomic.StoreUint32(&stream.unprocessed, 1)
t.closeStream(stream, errStreamDrain, false, http2.ErrCodeNo, statusGoAway, nil, false)
}
}
t.prevGoAwayID = id
active := len(t.activeStreams)
t.mu.Unlock()

active := len(activeStreams)
if active == 0 {
t.Close(connectionErrorf(true, nil, "received goaway and there are no active streams"))
}
Expand Down
41 changes: 41 additions & 0 deletions internal/transport/transport_test.go
View file
Open in desktop
Expand Up @@ -2501,3 +2501,44 @@ func (s) TestPeerSetInServerContext(t *testing.T) {
}
server.mu.Unlock()
}

// TestGoAwayCloseStreams tests the scenario where a client has many streams
// created, and the server sends a GOAWAY frame with a stream id less than some
// of them, while the client is still creating new streams. This should not
// induce a deadlock.
func (s) TestGoAwayCloseStreams(t *testing.T) {
server, ct, cancel := setUp(t, 0, math.MaxUint32, normal)
defer cancel()
defer server.stop()
defer ct.Close(fmt.Errorf("closed manually by test"))
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
for i := 0; i < 5; i++ {
_, err := ct.NewStream(ctx, &CallHdr{})
if err != nil {
t.Fatalf("error creating stream: %v", err)
}
}

waitWhileTrue(t, func() (bool, error) {
server.mu.Lock()
defer server.mu.Unlock()

if len(server.conns) == 0 {
return true, fmt.Errorf("timed-out while waiting for connection to be created on the server")
}
return false, nil
})

var st *http2Server
server.mu.Lock()
for k := range server.conns {
st = k.(*http2Server)
}
server.mu.Unlock()

st.framer.fr.WriteGoAway(5, http2.ErrCodeNo, []byte{})
for i := 0; i < 10; i++ {
ct.NewStream(ctx, &CallHdr{})
}
}