fix race when stream.Read and CancelRead are called concurrently

integrationtests/self/cancelation_test.go

@@ -25,7 +25,8 @@ var _ = Describe("Stream Cancelations", func() {
 
 		// The server accepts a single session, and then opens numStreams unidirectional streams.
 		// On each of these streams, it (tries to) write PRData.
-		runServer := func() <-chan int32 {
+		// When done, it sends the number of canceled streams on the channel.
+		runServer := func(data []byte) <-chan int32 {
 			numCanceledStreamsChan := make(chan int32)
 			var err error
 			server, err = quic.ListenAddr("localhost:0", getTLSConfig(), getQuicConfig(nil))
@@ -44,7 +45,7 @@ var _ = Describe("Stream Cancelations", func() {
 						defer wg.Done()
 						str, err := sess.OpenUniStreamSync(context.Background())
 						Expect(err).ToNot(HaveOccurred())
-						if _, err := str.Write(PRData); err != nil {
+						if _, err := str.Write(data); err != nil {
 							Expect(err).To(MatchError(&quic.StreamError{
 								StreamID:  str.StreamID(),
 								ErrorCode: quic.StreamErrorCode(str.StreamID()),
@@ -70,7 +71,7 @@ var _ = Describe("Stream Cancelations", func() {
 		})
 
 		It("downloads when the client immediately cancels most streams", func() {
-			serverCanceledCounterChan := runServer()
+			serverCanceledCounterChan := runServer(PRData)
 			sess, err := quic.DialAddr(
 				fmt.Sprintf("localhost:%d", server.Addr().(*net.UDPAddr).Port),
 				getTLSClientConfig(),
@@ -113,7 +114,7 @@ var _ = Describe("Stream Cancelations", func() {
 		})
 
 		It("downloads when the client cancels streams after reading from them for a bit", func() {
-			serverCanceledCounterChan := runServer()
+			serverCanceledCounterChan := runServer(PRData)
 
 			sess, err := quic.DialAddr(
 				fmt.Sprintf("localhost:%d", server.Addr().(*net.UDPAddr).Port),
@@ -159,6 +160,51 @@ var _ = Describe("Stream Cancelations", func() {
 			Expect(clientCanceledCounter).To(BeNumerically(">", numStreams/10))
 			Expect(numStreams - clientCanceledCounter).To(BeNumerically(">", numStreams/10))
 		})
+
+		It("allows concurrent Read and CancelRead calls", func() {
+			// This test is especially valuable when run with race detector,
+			// see https://github.com/lucas-clemente/quic-go/issues/3239.
+			serverCanceledCounterChan := runServer(make([]byte, 100)) // make sure the FIN is sent with the STREAM frame
+
+			sess, err := quic.DialAddr(
+				fmt.Sprintf("localhost:%d", server.Addr().(*net.UDPAddr).Port),
+				getTLSClientConfig(),
+				getQuicConfig(&quic.Config{MaxIncomingUniStreams: numStreams / 2}),
+			)
+			Expect(err).ToNot(HaveOccurred())
+
+			var wg sync.WaitGroup
+			wg.Add(numStreams)
+			var counter int32
+			for i := 0; i < numStreams; i++ {
+				go func() {
+					defer GinkgoRecover()
+					defer wg.Done()
+					str, err := sess.AcceptUniStream(context.Background())
+					Expect(err).ToNot(HaveOccurred())
+
+					done := make(chan struct{})
+					go func() {
+						defer GinkgoRecover()
+						defer close(done)
+						b := make([]byte, 32)
+						if _, err := str.Read(b); err != nil {
+							atomic.AddInt32(&counter, 1)
+							Expect(err.Error()).To(ContainSubstring("canceled with error code 1234"))
+							return
+						}
+					}()
+					go str.CancelRead(1234)
+					Eventually(done).Should(BeClosed())
+				}()
+			}
+			wg.Wait()
+			Expect(sess.CloseWithError(0, "")).To(Succeed())
+			numCanceled := atomic.LoadInt32(&counter)
+			fmt.Fprintf(GinkgoWriter, "canceled %d out of %d streams", numCanceled, numStreams)
+			Expect(numCanceled).ToNot(BeZero())
+			Eventually(serverCanceledCounterChan).Should(Receive())
+		})
 	})
 
 	Context("canceling the write side", func() {

internal/flowcontrol/stream_flow_controller.go

@@ -109,7 +109,10 @@ func (c *streamFlowController) AddBytesRead(n protocol.ByteCount) {
 }
 
 func (c *streamFlowController) Abandon() {
-	if unread := c.highestReceived - c.bytesRead; unread > 0 {
+	c.mutex.Lock()
+	unread := c.highestReceived - c.bytesRead
+	c.mutex.Unlock()
+	if unread > 0 {
 		c.connection.AddBytesRead(unread)
 	}
 }