@@ -1359,7 +1359,7 @@ func TestSession_sendMsg_Timeout(t *testing.T) {
13591359
13601360 hdr := encode (typePing , flagACK , 0 , 0 )
13611361 for {
1362- err := client .sendMsg (hdr , nil , nil )
1362+ err := client .sendMsg (hdr , nil , nil , true )
13631363 if err == nil {
13641364 continue
13651365 } else if err == ErrConnectionWriteTimeout {
@@ -1382,14 +1382,14 @@ func TestWindowOverflow(t *testing.T) {
13821382 defer server .Close ()
13831383
13841384 hdr1 := encode (typeData , flagSYN , i , 0 )
1385- _ = client .sendMsg (hdr1 , nil , nil )
1385+ _ = client .sendMsg (hdr1 , nil , nil , true )
13861386 s , err := server .AcceptStream ()
13871387 if err != nil {
13881388 t .Fatal (err )
13891389 }
13901390 msg := make ([]byte , client .config .MaxStreamWindowSize * 2 )
13911391 hdr2 := encode (typeData , 0 , i , uint32 (len (msg )))
1392- _ = client .sendMsg (hdr2 , msg , nil )
1392+ _ = client .sendMsg (hdr2 , msg , nil , true )
13931393 _ , err = io .ReadAll (s )
13941394 if err == nil {
13951395 t .Fatal ("expected to read no data" )
@@ -1874,3 +1874,108 @@ func TestErrorCodeErrorIsErrStreamReset(t *testing.T) {
18741874 ge := & GoAwayError {}
18751875 require .True (t , errors .Is (ge , ErrStreamReset ))
18761876}
1877+
1878+ func testTCPConn (t * testing.T ) (net.Conn , net.Conn ) {
1879+ listener , err := net .Listen ("tcp" , "127.0.0.1:0" )
1880+ if err != nil {
1881+ t .Fatal (err )
1882+ }
1883+ defer listener .Close ()
1884+
1885+ // Channel to receive the server-side connection
1886+ serverConnCh := make (chan net.Conn , 1 )
1887+ errCh := make (chan error , 1 )
1888+
1889+ // Accept connection in goroutine
1890+ go func () {
1891+ conn , err := listener .Accept ()
1892+ if err != nil {
1893+ errCh <- err
1894+ return
1895+ }
1896+ serverConnCh <- conn
1897+ }()
1898+
1899+ // Connect to the listener
1900+ clientConn , err := net .Dial ("tcp" , listener .Addr ().String ())
1901+ if err != nil {
1902+ t .Fatal (err )
1903+ }
1904+
1905+ // Wait for server connection or error
1906+ select {
1907+ case serverConn := <- serverConnCh :
1908+ return clientConn , serverConn
1909+ case err := <- errCh :
1910+ clientConn .Close ()
1911+ t .Fatal (err )
1912+ return nil , nil
1913+ case <- time .After (time .Second ):
1914+ clientConn .Close ()
1915+ t .Fatal ("timeout waiting for connection" )
1916+ return nil , nil
1917+ }
1918+ }
1919+
1920+ // TestSessionCloseDeadlock is a regression test for a deadlock on closing connections.
1921+ // See: https://github.com/libp2p/go-yamux/issues/129 for details
1922+ func TestSessionCloseDeadlock (t * testing.T ) {
1923+ const n = 10
1924+ const numWrites = 1000
1925+ var closeWG sync.WaitGroup
1926+ closeWG .Add (numWrites * n )
1927+ for i := 0 ; i < n ; i ++ {
1928+ go func () {
1929+ conn1 , conn2 := testTCPConn (t )
1930+ conf := DefaultConfig ()
1931+ conf .LogOutput = io .Discard
1932+ client , err := Client (conn1 , conf , nil )
1933+ require .NoError (t , err )
1934+ defer client .Close ()
1935+
1936+ conf = DefaultConfig ()
1937+ conf .LogOutput = io .Discard
1938+ server , err := Server (conn2 , conf , nil )
1939+ require .NoError (t , err )
1940+ defer server .Close ()
1941+
1942+ // Create a stream from client
1943+ stream , err := client .OpenStream (context .Background ())
1944+ require .NoError (t , err )
1945+ defer stream .Close ()
1946+
1947+ // Accept the stream on server side
1948+ serverStream , err := server .AcceptStream ()
1949+ require .NoError (t , err )
1950+ defer serverStream .Close ()
1951+
1952+ // Send an incomplete dataframe to the server to ensure that the recvloop is reading the rest
1953+ // of the message.
1954+ buf := make ([]byte , 1 << 10 )
1955+ hdr := encode (typeData , serverStream .sendFlags (), serverStream .id , uint32 (len (buf ))+ 1000 )
1956+ err = server .sendMsg (hdr , buf , nil , true )
1957+ require .NoError (t , err )
1958+
1959+ time .Sleep (1 * time .Second ) // Let the read loop block on reading rest of the data frame
1960+
1961+ // Make many writes so that these writes takes up all the buffer space in the channel
1962+ // and the recvLoop deadlocks because it's unable to send the goAwayProtoErr
1963+ var writeWG sync.WaitGroup
1964+ writeWG .Add (numWrites )
1965+ for i := 0 ; i < numWrites ; i ++ {
1966+ go func () {
1967+ defer closeWG .Done ()
1968+ buf := make ([]byte , 1 ) // make small writes so that we queue up the channel
1969+ writeWG .Done () // This is used to trigger concurrent streamWrite with CloseWrite.
1970+ stream .Write (buf )
1971+ }()
1972+ }
1973+ // Wait for the client to exit
1974+ writeWG .Wait ()
1975+ // Close the Write half of the client conn so that the send loop exits first.
1976+ err = conn1 .(* net.TCPConn ).CloseWrite ()
1977+ require .NoError (t , err )
1978+ }()
1979+ }
1980+ closeWG .Wait ()
1981+ }
0 commit comments