Skip to content

Commit d6304f4

Browse files
problameproblame
committed
rpc/dataconn: fix I/O timeout on variable receive rate
refs #162
1 parent 082335d commit d6304f4

4 files changed

Lines changed: 162 additions & 4 deletions

File tree

‎rpc/dataconn/frameconn/frameconn.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -222,6 +222,10 @@ func (c *Conn) writeFrame(payload []byte, frameType uint32) error {
222222
return nil
223223
}
224224

225+
func (c *Conn) ResetWriteTimeout() error {
226+
return c.nc.RenewWriteDeadline()
227+
}
228+
225229
func (c *Conn) Shutdown(deadline time.Time) error {
226230
// TCP connection teardown is a bit wonky if we are in a situation
227231
// where there is still data in flight (DIF) to our side:

‎rpc/dataconn/heartbeatconn/heartbeatconn.go‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,17 @@ func (c *Conn) readFrameFiltered() (frameconn.Frame, error) {
123123
return frameconn.Frame{}, fmt.Errorf("unknown frame type %x", f.Header.Type)
124124
}
125125
// drop heartbeat frame
126-
debug("received heartbeat")
126+
debug("received heartbeat, resetting write timeout")
127+
// the peer's heartbeat proves to us that the peer is still live
128+
// => trust the peer at this point (DoS risks are ignored ATM)
129+
// => we assume that the connection is symmetric duplex, i.e., if receiving works for us,
130+
// sending works for us, too.
131+
// So, let's grant the peer another write timeout.
132+
err = c.fc.ResetWriteTimeout()
133+
debug("renew frameconn write timeout returned errT=%T err=%s", err, err)
134+
if err != nil {
135+
return frameconn.Frame{}, err
136+
}
127137
continue
128138
}
129139
}

‎rpc/dataconn/heartbeatconn/integration_test_variablereceiverate/heartbeatconn_integration_variablereceiverate.go‎

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
// This integration test exercises the behavior of heartbeatconn
2+
// where the server is slow at handling the data received over the connection.
3+
// Note that the server is still sending heartbeats to the client, it's just the
4+
// data handling (usually I/O in case of zrepl endpoint.Receiver) that is slow.
5+
//
6+
// In commit 082335df5d85e1b0b9faa35ff182c71886142d3e and earlier, heartbeatconn would fail
7+
// this benchmark with a writev I/O timeout (here the ss(8) output at the time of failure)
8+
//
9+
// ESTAB 33369 0 127.0.0.1:12345 127.0.0.1:57282 users:(("heartbeatconn_i",pid=25953,fd=5))
10+
// cubic wscale:7,7 rto:203 rtt:2.992/5.849 ato:162 mss:32768 pmtu:65535 rcvmss:32741 advmss:65483 cwnd:10 bytes_sent:48 bytes_acked:48 bytes_received:195401 segs_out:44 segs_in:57 data_segs_out:6 data_segs_in:34 send 876.1Mbps lastsnd:125 lastrcv:9390 lastack:125 pacing_rate 1752.0Mbps delivery_rate 6393.8Mbps delivered:7 app_limited busy:42ms rcv_rtt:1 rcv_space:65483 rcv_ssthresh:65483 minrtt:0.029
11+
// --
12+
// ESTAB 0 3956805 127.0.0.1:57282 127.0.0.1:12345 users:(("heartbeatconn_i",pid=26100,fd=3))
13+
// cubic wscale:7,7 rto:211 backoff:5 rtt:10.38/16.937 ato:40 mss:32768 pmtu:65535 rcvmss:536 advmss:65483 cwnd:10 bytes_sent:195401 bytes_acked:195402 bytes_received:48 segs_out:57 segs_in:45 data_segs_out:34 data_segs_in:6 send 252.5Mbps lastsnd:9390 lastrcv:125 lastack:125 pacing_rate 505.1Mbps delivery_rate 1971.0Mbps delivered:35 busy:30127ms rwnd_limited:30086ms(99.9%) rcv_space:65495 rcv_ssthresh:65495 notsent:3956805 minrtt:0.007
14+
// panic: writev tcp 127.0.0.1:57282->127.0.0.1:12345: i/o timeout
15+
//
16+
// The assumed reason for those writev timeouts is the following:
17+
// - Sporadic server stalls (sever data handling, usually I/O) cause TCP exponential backoff on the client for client->server
18+
// - Go runtime unblocks after the deadline expires, resultin gin writev I/O timeout
19+
// - That is, even though the client observed heartbeats from the server
20+
// -> TCP doesn't assume symmetric connection behavior, but our implementation does.
21+
//
22+
// The fix contained in the commit this message was committed with resets the deadline whenever
23+
// a heartbeat is received from the server.
24+
//
25+
//
26+
// How to run this integration test:
27+
//
28+
//
29+
// Terminal 1:
30+
// $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go -mode server -addr 127.0.0.1:12345
31+
// rpc/dataconn/heartbeatconn: send heartbeat
32+
// rpc/dataconn/heartbeatconn: send heartbeat
33+
// ...
34+
//
35+
// Terminal 2:
36+
// $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go -mode client -addr 127.0.0.1:12345
37+
// rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
38+
// rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
39+
// rpc/dataconn/heartbeatconn: send heartbeat
40+
// rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
41+
// rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
42+
// rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
43+
// ...
44+
//
45+
// You should observe
46+
package main
47+
48+
import (
49+
"bytes"
50+
"flag"
51+
"fmt"
52+
"io"
53+
"log"
54+
"math/rand"
55+
"net"
56+
"os"
57+
"os/exec"
58+
"path"
59+
"time"
60+
61+
"github.com/zrepl/zrepl/util/devnoop"
62+
63+
"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
64+
)
65+
66+
func orDie(err error) {
67+
if err != nil {
68+
grepfield := path.Base(os.Args[0])[:10]
69+
fmt.Fprintf(os.Stderr, "grepping for %s\n", grepfield)
70+
sh := fmt.Sprintf("ss -ntpi | grep -A1 %s", grepfield)
71+
cmd := exec.Command("bash", "-c", sh)
72+
o, _ := cmd.CombinedOutput()
73+
buf := bytes.NewBuffer(o)
74+
_, _ = io.Copy(os.Stderr, buf)
75+
panic(err)
76+
}
77+
}
78+
79+
var mode string
80+
var addr string
81+
82+
func main() {
83+
84+
flag.StringVar(&mode, "mode", "", "server|client")
85+
flag.StringVar(&addr, "addr", "INVALID", "")
86+
flag.Parse()
87+
88+
modemap := map[string]func(){
89+
"server": server,
90+
"client": client,
91+
}
92+
modemap[mode]()
93+
94+
}
95+
96+
func server() {
97+
ln, err := net.Listen("tcp", addr)
98+
orDie(err)
99+
l := ln.(*net.TCPListener)
100+
for {
101+
c, err := l.AcceptTCP()
102+
if err != nil {
103+
log.Printf("accept err: %s", err)
104+
continue
105+
}
106+
hc := heartbeatconn.Wrap(c, 5*time.Second, 10*time.Second)
107+
108+
for {
109+
f, err := hc.ReadFrame()
110+
orDie(err)
111+
// _, err = buf.Write(f.Buffer.Bytes())
112+
// orDie(err)
113+
sleep := time.Duration(rand.NormFloat64()*500) * time.Millisecond
114+
time.Sleep(sleep)
115+
f.Buffer.Free()
116+
}
117+
118+
}
119+
120+
}
121+
122+
func client() {
123+
c, err := net.Dial("tcp", addr)
124+
orDie(err)
125+
hc := heartbeatconn.Wrap(c.(*net.TCPConn), 5*time.Second, 10*time.Second)
126+
127+
// follow API requirements to always ReadFrame
128+
go func() {
129+
for {
130+
f, err := hc.ReadFrame()
131+
orDie(err)
132+
f.Buffer.Free()
133+
}
134+
}()
135+
136+
dn := devnoop.Get()
137+
var buf [1 << 10]byte
138+
for {
139+
n, err := dn.Read(buf[:])
140+
orDie(err)
141+
err = hc.WriteFrame(buf[:n], 23)
142+
orDie(err)
143+
}
144+
}

‎rpc/dataconn/timeoutconn/timeoutconn.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ func (c *Conn) renewReadDeadline() error {
8080
return c.SetReadDeadline(time.Now().Add(c.idleTimeout))
8181
}
8282

83-
func (c *Conn) renewWriteDeadline() error {
83+
func (c *Conn) RenewWriteDeadline() error {
8484
if atomic.LoadInt32(&c.renewDeadlinesDisabled) != 0 {
8585
return nil
8686
}
@@ -107,7 +107,7 @@ restart:
107107
func (c Conn) Write(p []byte) (n int, err error) {
108108
n = 0
109109
restart:
110-
if err := c.renewWriteDeadline(); err != nil {
110+
if err := c.RenewWriteDeadline(); err != nil {
111111
return n, err
112112
}
113113
var nCurWrite int
@@ -127,7 +127,7 @@ restart:
127127
func (c Conn) WritevFull(bufs net.Buffers) (n int64, err error) {
128128
n = 0
129129
restart:
130-
if err := c.renewWriteDeadline(); err != nil {
130+
if err := c.RenewWriteDeadline(); err != nil {
131131
return n, err
132132
}
133133
var nCurWrite int64

0 commit comments

Comments
 (0)