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

// This integration test exercises the behavior of heartbeatconn
// where the server is slow at handling the data received over the connection.
// Note that the server is still sending heartbeats to the client, it's just the
// data handling (usually I/O in case of zrepl endpoint.Receiver) that is slow.
//
// In commit 082335df5d85e1b0b9faa35ff182c71886142d3e and earlier, heartbeatconn would fail
// this benchmark with a writev I/O timeout (here the ss(8) output at the time of failure)
//
//  ESTAB        33369    0                                        127.0.0.1:12345                                                127.0.0.1:57282                    users:(("heartbeatconn_i",pid=25953,fd=5))
//	 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
//	 --
//	 ESTAB        0        3956805                                  127.0.0.1:57282                                                127.0.0.1:12345                    users:(("heartbeatconn_i",pid=26100,fd=3))
//		  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
//	 panic: writev tcp 127.0.0.1:57282->127.0.0.1:12345: i/o timeout
//
// The assumed reason for those writev timeouts is the following:
// - Sporadic server stalls (sever data handling, usually I/O) cause TCP exponential backoff on the client for client->server
// - Go runtime unblocks after the deadline expires, resultin gin writev I/O timeout
// - That is, even though the client observed heartbeats from the server
// -> TCP doesn't assume symmetric connection behavior, but our implementation does.
//
// The fix contained in the commit this message was committed with resets the deadline whenever
// a heartbeat is received from the server.
//
//
// How to run this integration test:
//
//
//  Terminal 1:
//     $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go  -mode server -addr 127.0.0.1:12345
//     rpc/dataconn/heartbeatconn: send heartbeat
//     rpc/dataconn/heartbeatconn: send heartbeat
//     ...
//
//  Terminal 2:
//     $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go  -mode  client -addr 127.0.0.1:12345
//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
//     rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
//     rpc/dataconn/heartbeatconn: send heartbeat
//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
//     rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
//     ...
//
//  You should observe
package main

import (
	"bytes"
	"flag"
	"fmt"
	"io"
	"log"
	"math/rand"
	"net"
	"os"
	"os/exec"
	"path"
	"time"

	"github.com/zrepl/zrepl/util/devnoop"

	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
)

func orDie(err error) {
	if err != nil {
		grepfield := path.Base(os.Args[0])[:10]
		fmt.Fprintf(os.Stderr, "grepping for %s\n", grepfield)
		sh := fmt.Sprintf("ss -ntpi | grep -A1 %s", grepfield)
		cmd := exec.Command("bash", "-c", sh)
		o, _ := cmd.CombinedOutput()
		buf := bytes.NewBuffer(o)
		_, _ = io.Copy(os.Stderr, buf)
		panic(err)
	}
}

var mode string
var addr string

func main() {

	flag.StringVar(&mode, "mode", "", "server|client")
	flag.StringVar(&addr, "addr", "INVALID", "")
	flag.Parse()

	modemap := map[string]func(){
		"server": server,
		"client": client,
	}
	modemap[mode]()

}

func server() {
	ln, err := net.Listen("tcp", addr)
	orDie(err)
	l := ln.(*net.TCPListener)
	for {
		c, err := l.AcceptTCP()
		if err != nil {
			log.Printf("accept err: %s", err)
			continue
		}
		hc := heartbeatconn.Wrap(c, 5*time.Second, 10*time.Second)

		for {
			f, err := hc.ReadFrame()
			orDie(err)
			// _, err = buf.Write(f.Buffer.Bytes())
			// orDie(err)
			sleep := time.Duration(rand.NormFloat64()*500) * time.Millisecond
			time.Sleep(sleep)
			f.Buffer.Free()
		}

	}

}

func client() {
	c, err := net.Dial("tcp", addr)
	orDie(err)
	hc := heartbeatconn.Wrap(c.(*net.TCPConn), 5*time.Second, 10*time.Second)

	// follow API requirements to always ReadFrame
	go func() {
		for {
			f, err := hc.ReadFrame()
			orDie(err)
			f.Buffer.Free()
		}
	}()

	dn := devnoop.Get()
	var buf [1 << 10]byte
	for {
		n, err := dn.Read(buf[:])
		orDie(err)
		err = hc.WriteFrame(buf[:n], 23)
		orDie(err)
	}
}