Merge pull request #47 from dmcgowan/fix-expiration-race

Fix race in idle expiration
moby · Mar 2, 2015 · e9bf991 · e9bf991
2 parents 29e1da2 + 1fdaf41
commit e9bf991
Show file tree

Hide file tree

Showing 3 changed files with 95 additions and 5 deletions.
diff --git a/connection.go b/connection.go
@@ -72,11 +72,17 @@ Loop:
 				timer.Reset(i.timeout)
 			}
 		case <-expired:
-			for _, stream := range i.conn.streams {
-				stream.Reset()
-			}
-			i.conn.Close()
-			break Loop
+			i.conn.streamCond.L.Lock()
+			streams := i.conn.streams
+			i.conn.streams = make(map[spdy.StreamId]*Stream)
+			i.conn.streamCond.Broadcast()
+			i.conn.streamCond.L.Unlock()
+			go func() {
+				for _, stream := range streams {
+					stream.resetStream()
+				}
+				i.conn.Close()
+			}()
 		case <-i.conn.closeChan:
 			if timer != nil {
 				timer.Stop()

diff --git a/spdy_test.go b/spdy_test.go
@@ -396,6 +396,49 @@ func TestCloseNotification(t *testing.T) {
 	}
 }
 
+func TestIdleShutdownRace(t *testing.T) {
+	var wg sync.WaitGroup
+	server, listen, serverErr := runServer(&wg)
+	if serverErr != nil {
+		t.Fatalf("Error initializing server: %s", serverErr)
+	}
+
+	conn, dialErr := net.Dial("tcp", listen)
+	if dialErr != nil {
+		t.Fatalf("Error dialing server: %s", dialErr)
+	}
+
+	spdyConn, spdyErr := NewConnection(conn, false)
+	if spdyErr != nil {
+		t.Fatalf("Error creating spdy connection: %s", spdyErr)
+	}
+	go spdyConn.Serve(NoOpStreamHandler)
+
+	authenticated = true
+	stream, err := spdyConn.CreateStream(http.Header{}, nil, false)
+	if err != nil {
+		t.Fatalf("Error creating stream: %v", err)
+	}
+
+	spdyConn.SetIdleTimeout(5 * time.Millisecond)
+	go func() {
+		time.Sleep(5 * time.Millisecond)
+		stream.Reset()
+	}()
+
+	select {
+	case <-spdyConn.CloseChan():
+	case <-time.After(20 * time.Millisecond):
+		t.Fatal("Timed out waiting for idle connection closure")
+	}
+
+	closeErr := server.Close()
+	if closeErr != nil {
+		t.Fatalf("Error shutting down server: %s", closeErr)
+	}
+	wg.Wait()
+}
+
 func TestIdleNoTimeoutSet(t *testing.T) {
 	var wg sync.WaitGroup
 	server, listen, serverErr := runServer(&wg)
@@ -558,6 +601,44 @@ Loop:
 	wg.Wait()
 }
 
+func TestIdleRace(t *testing.T) {
+	var wg sync.WaitGroup
+	server, listen, serverErr := runServer(&wg)
+	if serverErr != nil {
+		t.Fatalf("Error initializing server: %s", serverErr)
+	}
+
+	conn, dialErr := net.Dial("tcp", listen)
+	if dialErr != nil {
+		t.Fatalf("Error dialing server: %s", dialErr)
+	}
+
+	spdyConn, spdyErr := NewConnection(conn, false)
+	if spdyErr != nil {
+		t.Fatalf("Error creating spdy connection: %s", spdyErr)
+	}
+	go spdyConn.Serve(NoOpStreamHandler)
+
+	spdyConn.SetIdleTimeout(10 * time.Millisecond)
+
+	authenticated = true
+
+	for i := 0; i < 10; i++ {
+		_, err := spdyConn.CreateStream(http.Header{}, nil, false)
+		if err != nil {
+			t.Fatalf("Error creating stream: %v", err)
+		}
+	}
+
+	<-spdyConn.CloseChan()
+
+	closeErr := server.Close()
+	if closeErr != nil {
+		t.Fatalf("Error shutting down server: %s", closeErr)
+	}
+	wg.Wait()
+}
+
 func TestHalfClosedIdleTimeout(t *testing.T) {
 	listener, listenErr := net.Listen("tcp", "localhost:0")
 	if listenErr != nil {

diff --git a/stream.go b/stream.go
@@ -168,7 +168,10 @@ func (s *Stream) Close() error {
 // Reset sends a reset frame, putting the stream into the fully closed state.
 func (s *Stream) Reset() error {
 	s.conn.removeStream(s)
+	return s.resetStream()
+}
 
+func (s *Stream) resetStream() error {
 	s.finishLock.Lock()
 	if s.finished {
 		s.finishLock.Unlock()