Harden internode DeadlineConn (#20631)

Since DeadlineConn would send deadline updates directly upstream,
it would race with Read/Write operations. The stdlib will perform a read, 
but do an async SetReadDeadLine(unix(1)) to cancel the Read in 
`abortPendingRead`. In this case, the Read may override the 
deadline intended to cancel the read.

Stop updating deadlines if a deadline in the past is seen and when Close is called. 
A mutex now protects all upstream deadline calls to avoid races. 

This should fix the short-term buildup of...

```
365 @ 0x44112e 0x4756b9 0x475699 0x483525 0x732286 0x737407 0x73816b 0x479601
#	0x475698	sync.runtime_notifyListWait+0x138		runtime/sema.go:569
#	0x483524	sync.(*Cond).Wait+0x84				sync/cond.go:70
#	0x732285	net/http.(*connReader).abortPendingRead+0xa5	net/http/server.go:729
#	0x737406	net/http.(*response).finishRequest+0x86		net/http/server.go:1676
#	0x73816a	net/http.(*conn).serve+0x62a			net/http/server.go:2050
```

AFAICT Only affects internode calls that create a connection (non-grid).
This commit is contained in:
Klaus Post
2024-11-11 09:15:17 -08:00
committed by GitHub
parent 8ce101c174
commit 55f5c18fd9
2 changed files with 195 additions and 16 deletions

View File

@@ -19,6 +19,7 @@ package deadlineconn
import (
"bufio"
"fmt"
"io"
"net"
"sync"
@@ -115,3 +116,77 @@ func TestBuffConnReadTimeout(t *testing.T) {
wg.Wait()
}
// Test deadlineconn handles read timeout properly by reading two messages beyond deadline.
func TestBuffConnReadCheckTimeout(t *testing.T) {
l, err := net.Listen("tcp", "localhost:0")
if err != nil {
t.Fatalf("unable to create listener. %v", err)
}
defer l.Close()
serverAddr := l.Addr().String()
tcpListener, ok := l.(*net.TCPListener)
if !ok {
t.Fatalf("failed to assert to net.TCPListener")
}
var cerr error
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
tcpConn, terr := tcpListener.AcceptTCP()
if terr != nil {
cerr = fmt.Errorf("failed to accept new connection. %v", terr)
return
}
deadlineconn := New(tcpConn)
deadlineconn.WithReadDeadline(time.Second)
deadlineconn.WithWriteDeadline(time.Second)
defer deadlineconn.Close()
// Read a line
b := make([]byte, 12)
_, terr = deadlineconn.Read(b)
if terr != nil {
cerr = fmt.Errorf("failed to read from client. %v", terr)
return
}
received := string(b)
if received != "message one\n" {
cerr = fmt.Errorf(`server: expected: "message one\n", got: %v`, received)
return
}
// Set a deadline in the past to indicate we want the next read to fail.
// Ensure we don't override it on read.
deadlineconn.SetReadDeadline(time.Unix(1, 0))
// Be sure to exceed update interval
time.Sleep(updateInterval * 2)
_, terr = deadlineconn.Read(b)
if terr == nil {
cerr = fmt.Errorf("could read from client, expected error, got %v", terr)
return
}
}()
c, err := net.Dial("tcp", serverAddr)
if err != nil {
t.Fatalf("unable to connect to server. %v", err)
}
defer c.Close()
_, err = io.WriteString(c, "message one\n")
if err != nil {
t.Fatalf("failed to write to server. %v", err)
}
_, _ = io.WriteString(c, "message two\n")
wg.Wait()
if cerr != nil {
t.Fatal(cerr)
}
}