pkg/rpctype: prepare for not using for target communication
Remove things that are only needed for target VM communication:
conditional compression, timeout scaling, traffic stats.
To minimize diffs when we switch target VM communication to flatrpc.
diff --git a/pkg/instance/instance.go b/pkg/instance/instance.go
index 3e59209..d47c093 100644
--- a/pkg/instance/instance.go
+++ b/pkg/instance/instance.go
@@ -456,10 +456,9 @@
}
type OptionalFuzzerArgs struct {
- Slowdown int
- SandboxArg int
- PprofPort int
- NetCompression bool
+ Slowdown int
+ SandboxArg int
+ PprofPort int
}
type FuzzerCmdArgs struct {
@@ -496,7 +495,6 @@
{Name: "slowdown", Value: fmt.Sprint(args.Optional.Slowdown)},
{Name: "sandbox_arg", Value: fmt.Sprint(args.Optional.SandboxArg)},
{Name: "pprof_port", Value: fmt.Sprint(args.Optional.PprofPort)},
- {Name: "net_compression", Value: fmt.Sprint(args.Optional.NetCompression)},
}
optionalArg = " " + tool.OptionalFlags(flags)
}
diff --git a/pkg/rpctype/rpc.go b/pkg/rpctype/rpc.go
index d195e99..f6b8d1e 100644
--- a/pkg/rpctype/rpc.go
+++ b/pkg/rpctype/rpc.go
@@ -13,18 +13,14 @@
"time"
"github.com/google/syzkaller/pkg/log"
- "github.com/google/syzkaller/pkg/stats"
)
type RPCServer struct {
- ln net.Listener
- s *rpc.Server
- useCompression bool
- statSent *stats.Val
- statRecv *stats.Val
+ ln net.Listener
+ s *rpc.Server
}
-func NewRPCServer(addr, name string, receiver interface{}, useCompression bool) (*RPCServer, error) {
+func NewRPCServer(addr, name string, receiver interface{}) (*RPCServer, error) {
ln, err := net.Listen("tcp", addr)
if err != nil {
return nil, fmt.Errorf("failed to listen on %v: %w", addr, err)
@@ -34,13 +30,8 @@
return nil, err
}
serv := &RPCServer{
- ln: ln,
- s: s,
- useCompression: useCompression,
- statSent: stats.Create("go rpc sent", "Uncompressed outbound RPC traffic",
- stats.Graph("traffic"), stats.Rate{}, stats.FormatMB),
- statRecv: stats.Create("go rpc recv", "Uncompressed inbound RPC traffic",
- stats.Graph("traffic"), stats.Rate{}, stats.FormatMB),
+ ln: ln,
+ s: s,
}
return serv, nil
}
@@ -53,7 +44,7 @@
continue
}
setupKeepAlive(conn, time.Minute)
- go serv.s.ServeConn(maybeFlateConn(newCountedConn(serv, conn), serv.useCompression))
+ go serv.s.ServeConn(newFlateConn(conn))
}
}
@@ -62,51 +53,35 @@
}
type RPCClient struct {
- conn net.Conn
- c *rpc.Client
- timeScale time.Duration
- useTimeouts bool
- useCompression bool
+ conn net.Conn
+ c *rpc.Client
}
-func Dial(addr string, timeScale time.Duration) (net.Conn, error) {
- if timeScale <= 0 {
- return nil, fmt.Errorf("bad rpc time scale %v", timeScale)
- }
+func NewRPCClient(addr string) (*RPCClient, error) {
var conn net.Conn
var err error
if addr == "stdin" {
// This is used by vm/gvisor which passes us a unix socket connection in stdin.
- return net.FileConn(os.Stdin)
+ // TODO: remove this once we switch to flatrpc for target communication.
+ conn, err = net.FileConn(os.Stdin)
+ } else {
+ conn, err = net.DialTimeout("tcp", addr, 3*time.Minute)
}
- if conn, err = net.DialTimeout("tcp", addr, time.Minute*timeScale); err != nil {
- return nil, err
- }
- setupKeepAlive(conn, time.Minute*timeScale)
- return conn, nil
-}
-
-func NewRPCClient(addr string, timeScale time.Duration, useTimeouts, useCompression bool) (*RPCClient, error) {
- conn, err := Dial(addr, timeScale)
if err != nil {
return nil, err
}
+ setupKeepAlive(conn, time.Minute)
cli := &RPCClient{
- conn: conn,
- c: rpc.NewClient(maybeFlateConn(conn, useCompression)),
- timeScale: timeScale,
- useTimeouts: useTimeouts,
- useCompression: useCompression,
+ conn: conn,
+ c: rpc.NewClient(newFlateConn(conn)),
}
return cli, nil
}
func (cli *RPCClient) Call(method string, args, reply interface{}) error {
- if cli.useTimeouts {
- // Note: SetDeadline is not implemented on fuchsia, so don't fail on error.
- cli.conn.SetDeadline(time.Now().Add(3 * time.Minute * cli.timeScale))
- defer cli.conn.SetDeadline(time.Time{})
- }
+ // Note: SetDeadline is not implemented on fuchsia, so don't fail on error.
+ cli.conn.SetDeadline(time.Now().Add(10 * time.Minute))
+ defer cli.conn.SetDeadline(time.Time{})
return cli.c.Call(method, args, reply)
}
@@ -130,10 +105,7 @@
c io.Closer
}
-func maybeFlateConn(conn io.ReadWriteCloser, useCompression bool) io.ReadWriteCloser {
- if !useCompression {
- return conn
- }
+func newFlateConn(conn io.ReadWriteCloser) io.ReadWriteCloser {
w, err := flate.NewWriter(conn, 9)
if err != nil {
panic(err)
@@ -173,29 +145,3 @@
}
return err0
}
-
-// countedConn wraps net.Conn to record the transferred bytes.
-type countedConn struct {
- io.ReadWriteCloser
- server *RPCServer
-}
-
-func newCountedConn(server *RPCServer,
- conn io.ReadWriteCloser) io.ReadWriteCloser {
- return &countedConn{
- ReadWriteCloser: conn,
- server: server,
- }
-}
-
-func (cc countedConn) Read(p []byte) (n int, err error) {
- n, err = cc.ReadWriteCloser.Read(p)
- cc.server.statRecv.Add(n)
- return
-}
-
-func (cc countedConn) Write(b []byte) (n int, err error) {
- n, err = cc.ReadWriteCloser.Write(b)
- cc.server.statSent.Add(n)
- return
-}
diff --git a/syz-fuzzer/fuzzer.go b/syz-fuzzer/fuzzer.go
index c350f2c..bdccf9d 100644
--- a/syz-fuzzer/fuzzer.go
+++ b/syz-fuzzer/fuzzer.go
@@ -72,14 +72,13 @@
debug.SetGCPercent(50)
var (
- flagName = flag.String("name", "test", "unique name for manager")
- flagOS = flag.String("os", runtime.GOOS, "target OS")
- flagArch = flag.String("arch", runtime.GOARCH, "target arch")
- flagManager = flag.String("manager", "", "manager rpc address")
- flagProcs = flag.Int("procs", 1, "number of parallel test processes")
- flagTest = flag.Bool("test", false, "enable image testing mode") // used by syz-ci
- flagPprofPort = flag.Int("pprof_port", 0, "HTTP port for the pprof endpoint (disabled if 0)")
- flagNetCompression = flag.Bool("net_compression", false, "use network compression for RPC calls")
+ flagName = flag.String("name", "test", "unique name for manager")
+ flagOS = flag.String("os", runtime.GOOS, "target OS")
+ flagArch = flag.String("arch", runtime.GOARCH, "target arch")
+ flagManager = flag.String("manager", "", "manager rpc address")
+ flagProcs = flag.Int("procs", 1, "number of parallel test processes")
+ flagTest = flag.Bool("test", false, "enable image testing mode") // used by syz-ci
+ flagPprofPort = flag.Int("pprof_port", 0, "HTTP port for the pprof endpoint (disabled if 0)")
)
defer tool.Init()()
log.Logf(0, "fuzzer started")
@@ -123,7 +122,7 @@
}
log.Logf(0, "dialing manager at %v", *flagManager)
- manager, err := rpctype.NewRPCClient(*flagManager, timeouts.Scale, false, *flagNetCompression)
+ manager, err := rpctype.NewRPCClient(*flagManager)
if err != nil {
log.SyzFatalf("failed to create an RPC client: %v ", err)
}
diff --git a/syz-fuzzer/testing.go b/syz-fuzzer/testing.go
index a3f1b88..bd2eb33 100644
--- a/syz-fuzzer/testing.go
+++ b/syz-fuzzer/testing.go
@@ -6,6 +6,7 @@
import (
"fmt"
"io"
+ "net"
"strings"
"time"
@@ -30,7 +31,8 @@
func testImage(hostAddr string, args *checkArgs) {
log.Logf(0, "connecting to host at %v", hostAddr)
- conn, err := rpctype.Dial(hostAddr, args.ipcConfig.Timeouts.Scale)
+ timeout := time.Minute * args.ipcConfig.Timeouts.Scale
+ conn, err := net.DialTimeout("tcp", hostAddr, timeout)
if err != nil {
log.SyzFatalf("failed to connect to host: %v", err)
}
diff --git a/syz-hub/hub.go b/syz-hub/hub.go
index f7e7e3e..187d4a1 100644
--- a/syz-hub/hub.go
+++ b/syz-hub/hub.go
@@ -61,7 +61,7 @@
hub.initHTTP(cfg.HTTP)
- s, err := rpctype.NewRPCServer(cfg.RPC, "Hub", hub, true)
+ s, err := rpctype.NewRPCServer(cfg.RPC, "Hub", hub)
if err != nil {
log.Fatalf("failed to create rpc server: %v", err)
}
diff --git a/syz-manager/hub.go b/syz-manager/hub.go
index 8de9f1f..dd0cbbc 100644
--- a/syz-manager/hub.go
+++ b/syz-manager/hub.go
@@ -129,7 +129,7 @@
if err != nil {
return nil, err
}
- hub, err := rpctype.NewRPCClient(hc.cfg.HubAddr, 1, true, true)
+ hub, err := rpctype.NewRPCClient(hc.cfg.HubAddr)
if err != nil {
return nil, err
}
@@ -162,7 +162,7 @@
if err != nil {
return nil, err
}
- hub, err = rpctype.NewRPCClient(hc.cfg.HubAddr, 1, true, true)
+ hub, err = rpctype.NewRPCClient(hc.cfg.HubAddr)
if err != nil {
return nil, err
}
diff --git a/syz-manager/manager.go b/syz-manager/manager.go
index 469ac38..f5437f8 100644
--- a/syz-manager/manager.go
+++ b/syz-manager/manager.go
@@ -65,7 +65,6 @@
vmStop chan bool
checkFeatures *host.Features
fresh bool
- netCompression bool
expertMode bool
nextInstanceID atomic.Uint64
@@ -177,7 +176,6 @@
memoryLeakFrames: make(map[string]bool),
dataRaceFrames: make(map[string]bool),
fresh: true,
- netCompression: vm.UseNetCompression(cfg.Type),
vmStop: make(chan bool),
externalReproQueue: make(chan *Crash, 10),
needMoreRepros: make(chan chan bool),
@@ -811,10 +809,9 @@
Debug: *flagDebug,
Test: false,
Optional: &instance.OptionalFuzzerArgs{
- Slowdown: mgr.cfg.Timeouts.Slowdown,
- SandboxArg: mgr.cfg.SandboxArg,
- PprofPort: inst.PprofPort(),
- NetCompression: mgr.netCompression,
+ Slowdown: mgr.cfg.Timeouts.Slowdown,
+ SandboxArg: mgr.cfg.SandboxArg,
+ PprofPort: inst.PprofPort(),
},
}
cmd := instance.FuzzerCmd(args)
diff --git a/syz-manager/rpc.go b/syz-manager/rpc.go
index 062d19e..3d452fd 100644
--- a/syz-manager/rpc.go
+++ b/syz-manager/rpc.go
@@ -128,7 +128,7 @@
"End-to-end fuzzer RPC Exchange call latency (us)", stats.Distribution{}),
statCoverFiltered: stats.Create("filtered coverage", "", stats.NoGraph),
}
- s, err := rpctype.NewRPCServer(mgr.cfg.RPC, "Manager", serv, mgr.netCompression)
+ s, err := rpctype.NewRPCServer(mgr.cfg.RPC, "Manager", serv)
if err != nil {
return nil, err
}
diff --git a/syz-runner/runner.go b/syz-runner/runner.go
index 3b37039..329b340 100644
--- a/syz-runner/runner.go
+++ b/syz-runner/runner.go
@@ -45,8 +45,7 @@
log.Fatalf("failed to create default ipc config: %v", err)
}
- timeouts := config.Timeouts
- vrf, err := rpctype.NewRPCClient(*flagAddr, timeouts.Scale, true, true)
+ vrf, err := rpctype.NewRPCClient(*flagAddr)
if err != nil {
log.Fatalf("failed to connect to verifier : %v", err)
}
diff --git a/syz-verifier/rpcserver.go b/syz-verifier/rpcserver.go
index 8c0e6b7..f80fb7a 100644
--- a/syz-verifier/rpcserver.go
+++ b/syz-verifier/rpcserver.go
@@ -33,7 +33,7 @@
notChecked: len(vrf.pools),
}
- s, err := rpctype.NewRPCServer(vrf.addr, "Verifier", srv, true)
+ s, err := rpctype.NewRPCServer(vrf.addr, "Verifier", srv)
if err != nil {
return nil, err
}
diff --git a/tools/syz-hubtool/hubtool.go b/tools/syz-hubtool/hubtool.go
index a331780..dc59ef2 100644
--- a/tools/syz-hubtool/hubtool.go
+++ b/tools/syz-hubtool/hubtool.go
@@ -57,7 +57,7 @@
return
}
log.Printf("connecting to hub at %v...", *flagHubAddress)
- conn, err := rpctype.NewRPCClient(*flagHubAddress, 1, true, true)
+ conn, err := rpctype.NewRPCClient(*flagHubAddress)
if err != nil {
log.Fatalf("failed to connect to hub: %v", err)
}
diff --git a/tools/syz-runtest/runtest.go b/tools/syz-runtest/runtest.go
index 1de8f6a..586e4c2 100644
--- a/tools/syz-runtest/runtest.go
+++ b/tools/syz-runtest/runtest.go
@@ -69,7 +69,7 @@
}
mgr.checkFiles, mgr.checkProgs = mgr.checker.StartCheck()
mgr.needCheckResults = len(mgr.checkProgs)
- s, err := rpctype.NewRPCServer(cfg.RPC, "Manager", mgr, false)
+ s, err := rpctype.NewRPCServer(cfg.RPC, "Manager", mgr)
if err != nil {
log.Fatalf("failed to create rpc server: %v", err)
}
diff --git a/vm/adb/adb.go b/vm/adb/adb.go
index de1bcd2..bd98634 100644
--- a/vm/adb/adb.go
+++ b/vm/adb/adb.go
@@ -26,7 +26,7 @@
)
func init() {
- vmimpl.Register("adb", ctor, false, true)
+ vmimpl.Register("adb", ctor, false)
}
type Device struct {
diff --git a/vm/bhyve/bhyve.go b/vm/bhyve/bhyve.go
index 3e82cbc..c3b53e0 100644
--- a/vm/bhyve/bhyve.go
+++ b/vm/bhyve/bhyve.go
@@ -21,7 +21,7 @@
)
func init() {
- vmimpl.Register("bhyve", ctor, true, false)
+ vmimpl.Register("bhyve", ctor, true)
}
type Config struct {
diff --git a/vm/cuttlefish/cuttlefish.go b/vm/cuttlefish/cuttlefish.go
index d17bd78..94610ed 100644
--- a/vm/cuttlefish/cuttlefish.go
+++ b/vm/cuttlefish/cuttlefish.go
@@ -28,7 +28,7 @@
)
func init() {
- vmimpl.Register("cuttlefish", ctor, true, true)
+ vmimpl.Register("cuttlefish", ctor, true)
}
type Pool struct {
diff --git a/vm/gce/gce.go b/vm/gce/gce.go
index 6a292d4..16f3f99 100644
--- a/vm/gce/gce.go
+++ b/vm/gce/gce.go
@@ -35,7 +35,7 @@
)
func init() {
- vmimpl.Register("gce", ctor, true, true)
+ vmimpl.Register("gce", ctor, true)
}
type Config struct {
diff --git a/vm/gvisor/gvisor.go b/vm/gvisor/gvisor.go
index f8194ab..45c076b 100644
--- a/vm/gvisor/gvisor.go
+++ b/vm/gvisor/gvisor.go
@@ -25,7 +25,7 @@
)
func init() {
- vmimpl.Register("gvisor", ctor, true, false)
+ vmimpl.Register("gvisor", ctor, true)
}
type Config struct {
diff --git a/vm/isolated/isolated.go b/vm/isolated/isolated.go
index 3383e91..51a995b 100755
--- a/vm/isolated/isolated.go
+++ b/vm/isolated/isolated.go
@@ -23,7 +23,7 @@
const pstoreConsoleFile = "/sys/fs/pstore/console-ramoops-0"
func init() {
- vmimpl.Register("isolated", ctor, false, true)
+ vmimpl.Register("isolated", ctor, false)
}
type Config struct {
diff --git a/vm/proxyapp/init.go b/vm/proxyapp/init.go
index 9c96e48..467187f 100644
--- a/vm/proxyapp/init.go
+++ b/vm/proxyapp/init.go
@@ -32,7 +32,7 @@
func(env *vmimpl.Env) (vmimpl.Pool, error) {
return ctor(makeDefaultParams(), env)
},
- false, true)
+ false)
}
// Package configuration VARs are mostly needed for tests.
diff --git a/vm/qemu/qemu.go b/vm/qemu/qemu.go
index 2dd016a..6f5cb4f 100644
--- a/vm/qemu/qemu.go
+++ b/vm/qemu/qemu.go
@@ -27,7 +27,7 @@
func init() {
var _ vmimpl.Infoer = (*instance)(nil)
- vmimpl.Register("qemu", ctor, true, false)
+ vmimpl.Register("qemu", ctor, true)
}
type Config struct {
diff --git a/vm/starnix/starnix.go b/vm/starnix/starnix.go
index 2f5a087..51f2f67 100644
--- a/vm/starnix/starnix.go
+++ b/vm/starnix/starnix.go
@@ -22,7 +22,7 @@
func init() {
var _ vmimpl.Infoer = (*instance)(nil)
- vmimpl.Register("starnix", ctor, true, false)
+ vmimpl.Register("starnix", ctor, true)
}
type Config struct {
diff --git a/vm/vm.go b/vm/vm.go
index 351163f..4887ef6 100644
--- a/vm/vm.go
+++ b/vm/vm.go
@@ -92,12 +92,6 @@
return vmimpl.Types[vmType(typ)].Overcommit
}
-// UseNetCompression says if it's beneficial to use network compression for this VM type.
-// Local VMs (qemu) generally don't benefit from compression, while remote machines may benefit.
-func UseNetCompression(typ string) bool {
- return vmimpl.Types[vmType(typ)].NetCompression
-}
-
// Create creates a VM pool that can be used to create individual VMs.
func Create(cfg *mgrconfig.Config, debug bool) (*Pool, error) {
typ, ok := vmimpl.Types[vmType(cfg.Type)]
diff --git a/vm/vm_test.go b/vm/vm_test.go
index 6acc003..184f613 100644
--- a/vm/vm_test.go
+++ b/vm/vm_test.go
@@ -80,7 +80,7 @@
ctor := func(env *vmimpl.Env) (vmimpl.Pool, error) {
return &testPool{}, nil
}
- vmimpl.Register("test", ctor, false, false)
+ vmimpl.Register("test", ctor, false)
}
type Test struct {
diff --git a/vm/vmimpl/vmimpl.go b/vm/vmimpl/vmimpl.go
index bd565c8..a9afdc1 100644
--- a/vm/vmimpl/vmimpl.go
+++ b/vm/vmimpl/vmimpl.go
@@ -129,18 +129,16 @@
}
// Register registers a new VM type within the package.
-func Register(typ string, ctor ctorFunc, allowsOvercommit, netCompression bool) {
+func Register(typ string, ctor ctorFunc, allowsOvercommit bool) {
Types[typ] = Type{
- Ctor: ctor,
- Overcommit: allowsOvercommit,
- NetCompression: netCompression,
+ Ctor: ctor,
+ Overcommit: allowsOvercommit,
}
}
type Type struct {
- Ctor ctorFunc
- Overcommit bool
- NetCompression bool
+ Ctor ctorFunc
+ Overcommit bool
}
type ctorFunc func(env *Env) (Pool, error)
diff --git a/vm/vmm/vmm.go b/vm/vmm/vmm.go
index 67f12f4..71dfd39 100644
--- a/vm/vmm/vmm.go
+++ b/vm/vmm/vmm.go
@@ -25,7 +25,7 @@
var vmctlStatusRegex = regexp.MustCompile(`^\s+([0-9]+)\b.*\brunning`)
func init() {
- vmimpl.Register("vmm", ctor, true, false)
+ vmimpl.Register("vmm", ctor, true)
}
type Config struct {
diff --git a/vm/vmware/vmware.go b/vm/vmware/vmware.go
index 6bccbaa..4c05bac 100644
--- a/vm/vmware/vmware.go
+++ b/vm/vmware/vmware.go
@@ -22,7 +22,7 @@
)
func init() {
- vmimpl.Register("vmware", ctor, false, false)
+ vmimpl.Register("vmware", ctor, false)
}
type Config struct {
