Skip to content

Commit 20eef7f

Browse files
nicholashusinnicholashusin
committed
quic: clear fast-path receive buffer on stream reset
When a stream is reset by the peer, unread stream data is discarded and its underlying memory buffers are returned to a pool for reuse. However, the stream's fast-path read buffer was not cleared when the stream was reset, leaving a dangling reference to the recycled memory. This surfaced as a data race in one of our HTTP/3 tests. Therefore, make sure that the fast-path read buffer is cleared when a stream is reset. Also do the same when STOP_SENDING frame is sent, where an equivalent bug exists. For golang/go#70914 For golang/go#81349 Change-Id: Ie1d75b646baa59804380feeb84ac0ff26a6a6964 Reviewed-on: https://go-review.googlesource.com/c/net/+/827804 Reviewed-by: Damien Neil <dneil@google.com> Reviewed-by: Nicholas Husin <husin@google.com> LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>
1 parent c23af1b commit 20eef7f

3 files changed

Lines changed: 105 additions & 53 deletions

File tree

‎quic/conn_flow_test.go‎

Lines changed: 67 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -98,45 +98,78 @@ func testConnInflowReturnOnRacingReads(t *testing.T) {
9898
}
9999

100100
func TestConnInflowReturnOnClose(t *testing.T) {
101-
synctest.Test(t, testConnInflowReturnOnClose)
102-
}
103-
func testConnInflowReturnOnClose(t *testing.T) {
104-
tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
105-
c.MaxConnReadBufferSize = 64
106-
})
107-
tc.ignoreFrame(frameTypeStopSending)
108-
tc.writeFrames(packetType1RTT, debugFrameStream{
109-
id: s.id,
110-
data: make([]byte, 64),
111-
})
112-
s.CloseRead()
113-
tc.wantFrame("closing stream updates connection-level flow control",
114-
packetType1RTT, debugFrameMaxData{
115-
max: 128,
101+
for _, read := range []bool{true, false} {
102+
name := "with partial read"
103+
if !read {
104+
name = "without partial read"
105+
}
106+
synctestSubtest(t, name, func(t *testing.T) {
107+
tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
108+
c.MaxConnReadBufferSize = 64
109+
})
110+
tc.ignoreFrame(frameTypeStopSending)
111+
tc.writeFrames(packetType1RTT, debugFrameStream{
112+
id: s.id,
113+
data: make([]byte, 32),
114+
})
115+
if read {
116+
if _, err := s.Read(make([]byte, 16)); err != nil {
117+
t.Fatalf("s.Read() = %v", err)
118+
}
119+
tc.wantFrame("reading stream updates connection-level flow control",
120+
packetType1RTT, debugFrameMaxData{
121+
max: 64 + 32, // 16 read + 16 inbuf = 32 credited.
122+
})
123+
}
124+
tc.writeFrames(packetType1RTT, debugFrameStream{
125+
id: s.id,
126+
off: 32,
127+
data: make([]byte, 32),
128+
})
129+
s.CloseRead()
130+
tc.wantFrame("closing stream updates connection-level flow control",
131+
packetType1RTT, debugFrameMaxData{
132+
max: 128,
133+
})
116134
})
135+
}
117136
}
118137

119138
func TestConnInflowReturnOnReset(t *testing.T) {
120-
synctest.Test(t, testConnInflowReturnOnReset)
121-
}
122-
func testConnInflowReturnOnReset(t *testing.T) {
123-
tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
124-
c.MaxConnReadBufferSize = 64
125-
})
126-
tc.ignoreFrame(frameTypeStopSending)
127-
tc.writeFrames(packetType1RTT, debugFrameStream{
128-
id: s.id,
129-
data: make([]byte, 32),
130-
})
131-
tc.writeFrames(packetType1RTT, debugFrameResetStream{
132-
id: s.id,
133-
finalSize: 64,
134-
})
135-
s.CloseRead()
136-
tc.wantFrame("receiving stream reseet updates connection-level flow control",
137-
packetType1RTT, debugFrameMaxData{
138-
max: 128,
139+
for _, read := range []bool{true, false} {
140+
name := "with partial read"
141+
if !read {
142+
name = "without partial read"
143+
}
144+
synctestSubtest(t, name, func(t *testing.T) {
145+
tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
146+
c.MaxConnReadBufferSize = 64
147+
})
148+
tc.ignoreFrame(frameTypeStopSending)
149+
tc.writeFrames(packetType1RTT, debugFrameStream{
150+
id: s.id,
151+
data: make([]byte, 32),
152+
})
153+
if read {
154+
if _, err := s.Read(make([]byte, 16)); err != nil {
155+
t.Fatalf("s.Read() = %v", err)
156+
}
157+
tc.wantFrame("reading stream updates connection-level flow control",
158+
packetType1RTT, debugFrameMaxData{
159+
max: 64 + 32, // 16 read + 16 inbuf = 32 credited.
160+
})
161+
}
162+
tc.writeFrames(packetType1RTT, debugFrameResetStream{
163+
id: s.id,
164+
finalSize: 64,
165+
})
166+
s.CloseRead()
167+
tc.wantFrame("receiving stream reset updates connection-level flow control",
168+
packetType1RTT, debugFrameMaxData{
169+
max: 128,
170+
})
139171
})
172+
}
140173
}
141174

142175
func TestConnInflowStreamViolation(t *testing.T) {

‎quic/stream.go‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -604,7 +604,11 @@ func (s *Stream) StopSending(code uint64) error {
604604
s.inclosedcode = code
605605
s.inclosed.set()
606606
}
607-
discarded := s.in.end - s.in.start
607+
s.inbufmu.Lock()
608+
discarded := max(0, s.in.end-s.in.start-int64(len(s.inbuf)))
609+
s.inbuf = nil
610+
s.inbufoff = 0
611+
s.inbufmu.Unlock()
608612
s.in.discardBefore(s.in.end)
609613
s.inUnlock()
610614
s.conn.handleStreamBytesReadOffLoop(discarded) // must be done with ingate unlocked
@@ -869,7 +873,12 @@ func (s *Stream) handleReset(code uint64, finalSize int64) error {
869873
return err
870874
}
871875
}
872-
s.conn.handleStreamBytesReadOnLoop(finalSize - s.in.start)
876+
s.inbufmu.Lock()
877+
unread := max(0, finalSize-s.in.start-int64(len(s.inbuf)))
878+
s.inbuf = nil
879+
s.inbufoff = 0
880+
s.inbufmu.Unlock()
881+
s.conn.handleStreamBytesReadOnLoop(unread)
873882
s.in.discardBefore(s.in.end)
874883
s.inresetcode = int64(code)
875884
s.insize = finalSize

‎quic/stream_test.go‎

Lines changed: 27 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -971,21 +971,31 @@ func testStreamReadFromClosedStream(t *testing.T) {
971971
}
972972

973973
func TestStreamCloseReadWithAllDataReceived(t *testing.T) {
974-
synctest.Test(t, testStreamCloseReadWithAllDataReceived)
975-
}
976-
func testStreamCloseReadWithAllDataReceived(t *testing.T) {
977-
tc, s := newTestConnAndRemoteStream(t, serverSide, bidiStream, permissiveTransportParameters)
978-
tc.writeFrames(packetType1RTT, debugFrameStream{
979-
id: s.id,
980-
data: []byte{1, 2, 3},
981-
fin: true,
982-
})
983-
s.CloseRead()
984-
tc.wantIdle("CloseRead in Data Recvd state doesn't need to send STOP_SENDING")
985-
// We had all the data for the stream, but CloseRead discarded it.
986-
wantErr := "read from closed stream"
987-
if n, err := s.Read(make([]byte, 16)); err == nil || !strings.Contains(err.Error(), wantErr) {
988-
t.Errorf("s.Read() = %v, %v; want error %q", n, err, wantErr)
974+
for _, read := range []bool{true, false} {
975+
name := "with partial read"
976+
if !read {
977+
name = "without partial read"
978+
}
979+
synctestSubtest(t, name, func(t *testing.T) {
980+
tc, s := newTestConnAndRemoteStream(t, serverSide, bidiStream, permissiveTransportParameters)
981+
tc.writeFrames(packetType1RTT, debugFrameStream{
982+
id: s.id,
983+
data: []byte{1, 2, 3},
984+
fin: true,
985+
})
986+
if read {
987+
if _, err := s.Read(make([]byte, 1)); err != nil {
988+
t.Fatalf("s.Read() = %v", err)
989+
}
990+
}
991+
s.CloseRead()
992+
tc.wantIdle("CloseRead in Data Recvd state doesn't need to send STOP_SENDING")
993+
// We had all the data for the stream, but CloseRead discarded it.
994+
wantErr := "read from closed stream"
995+
if n, err := s.Read(make([]byte, 16)); err == nil || !strings.Contains(err.Error(), wantErr) {
996+
t.Errorf("s.Read() = %v, %v; want error %q", n, err, wantErr)
997+
}
998+
})
989999
}
9901000
}
9911001

@@ -1299,8 +1309,8 @@ func TestStreamPeerResetsWithUnreadAndUnsentData(t *testing.T) {
12991309
code: sentCode,
13001310
})
13011311
wantErr := StreamError(sentCode)
1302-
if _, err := io.ReadAll(s); !errors.Is(err, wantErr) {
1303-
t.Fatalf("Read reset stream: ReadAll got error %v; want %v", err, wantErr)
1312+
if unread, err := io.ReadAll(s); !errors.Is(err, wantErr) || len(unread) != 0 {
1313+
t.Fatalf("Read reset stream: ReadAll got %x, error %v; want empty, %v", unread, err, wantErr)
13041314
}
13051315
})
13061316
}

0 commit comments

Comments
 (0)