diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ffbbd60..f2677560 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -304,6 +304,20 @@ Detailed per-release notes are on the CLI 7.5 ms of CPU and 2.9 MB of allocations, more than the rest of a send. The second request happened on every send to an agent in the trusted list (`list-agents`, `pilot-mom`) and to any public peer not yet trusted. +- **The daemon uses far less CPU under bursts of sends.** Measured on two + local daemons, the sending side's CPU per message fell from 4.5 ms to + 1.6 ms at 32 concurrent sends and from 24 ms to 3.5 ms at 256. Two + changes: + - A dial no longer polls its connection every 10 ms. SYN-ACK, RST and + shutdown wake the dialer directly; the poll is a 250 ms backstop. With + hundreds of dials in flight the poll alone kept the Go scheduler busy. + - Concurrent `info` requests share one reply for 200 ms. Every + `pilotctl send-message` asks for `info`, and the reply lists every + connection, so a burst of N sends built N replies over a table that grew + with N: 73% of everything the daemon allocated during a 256-way burst. + + Hosts with many cores can cut scheduler overhead further by setting + `GOMAXPROCS` (for example 4) in the daemon's environment. - **The tunnel socket asks the kernel for 4 MB buffers** in each direction instead of the default (about 200 KB on Linux). Every tunnel shares the one socket, and several streams sending at once overflowed it. The kernel caps diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 431c692d..a520f5ac 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -287,7 +287,7 @@ const ( DialMaxRetries = 7 // total attempts (direct + relay). 3 direct + 4 relay. With DialInitialRTO=250ms exponential-backoff capped at DialMaxRTO=8s, the relay phase is ~7.75s — covers cold-start handshake (key_exchange + flushPending + SYN/SYN-ACK round trip) for typical peers while keeping bad dials from blocking longer than the user's --timeout. The probe-and-adapt machinery (see srttHistory below) will let us shorten this for peers we've successfully dialed before. DialInitialRTO = 250 * time.Millisecond // initial SYN retransmission timeout. Lowered from 1s — modern relay RTT is <200ms; waiting a full second before assuming loss makes cold dials feel like a stall. Three direct retries with exponential backoff (250→500→1000) still cover up to 1.75s of jitter before flipping to relay; that's plenty for an unhealthy direct path while letting the common case (peer is reachable, single retry needed) feel snappy. DialMaxRTO = 8 * time.Second // max backoff for SYN retransmission - DialCheckInterval = 10 * time.Millisecond // backstop poll for state changes during dial (conn.DialCh is the fast path) + DialCheckInterval = 250 * time.Millisecond // backstop poll for state changes during dial; every transition that ends a dial (SYN-ACK, RST, shutdown) signals conn.DialCh RetxCheckInterval = 100 * time.Millisecond // retransmission check ticker MaxRetxAttempts = 8 // abandon connection after this many retransmissions HeartbeatReregThresh = 3 // heartbeat failures before re-registration @@ -1730,6 +1730,7 @@ func (d *Daemon) doStop() { conn.Mu.Lock() conn.State = StateClosed conn.Mu.Unlock() + conn.signalDial() conn.CloseRecvBuf() d.ports.RemoveConnection(conn.ID) } @@ -3630,6 +3631,7 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { conn.Mu.Lock() conn.State = StateClosed conn.Mu.Unlock() + conn.signalDial() // a refused dial ends now, not at the next backstop tick conn.CloseRecvBuf() d.ports.RemoveConnection(conn.ID) d.publishEvent("conn.rst", map[string]interface{}{ diff --git a/pkg/daemon/ipc.go b/pkg/daemon/ipc.go index 0a10a978..a8ee589f 100644 --- a/pkg/daemon/ipc.go +++ b/pkg/daemon/ipc.go @@ -542,8 +542,22 @@ type IPCServer struct { clients map[*ipcConn]bool closeOnce sync.Once done chan struct{} + + // infoMu guards the last info reply. Every `pilotctl send-message` + // asks for info, and the reply serialises every peer and connection + // row, so a burst of N concurrent sends cost N full builds over a + // table that itself grows with N. Requests that arrive while a reply + // younger than infoReplyTTL exists share it; requests that arrive + // during a build wait for it instead of starting their own. + infoMu sync.Mutex + infoAt time.Time + infoData []byte } +// infoReplyTTL is how long one info reply is reused. A var so tests that +// assert on a value changed a moment earlier can turn reuse off. +var infoReplyTTL = 200 * time.Millisecond + func NewIPCServer(socketPath string, d *Daemon) *IPCServer { return &IPCServer{ socketPath: socketPath, @@ -1246,6 +1260,31 @@ type ipcInfoConn struct { } func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) { + s.infoMu.Lock() + if infoReplyTTL > 0 && s.infoData != nil && time.Since(s.infoAt) < infoReplyTTL { + data := s.infoData + s.infoMu.Unlock() + if err := conn.writeReply(CmdInfoOK, reqID, data); err != nil { + slog.Debug("IPC info reply failed", "err", err) + } + return + } + data, err := s.buildInfoReply() + if err == nil { + s.infoData, s.infoAt = data, time.Now() + } + s.infoMu.Unlock() + if err != nil { + s.sendError(conn, reqID, fmt.Sprintf("info marshal: %v", err)) + return + } + if err := conn.writeReply(CmdInfoOK, reqID, data); err != nil { + slog.Debug("IPC info reply failed", "err", err) + } +} + +// buildInfoReply assembles the JSON body of an info reply. +func (s *IPCServer) buildInfoReply() ([]byte, error) { info := s.daemon.Info() // Peer and connection rows are typed structs, not maps: `info` is called @@ -1348,13 +1387,7 @@ func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) { data, err = marshalRowsWithinBudget(body, "conn_list", conns, ipcReplyBudget) } } - if err != nil { - s.sendError(conn, reqID, fmt.Sprintf("info marshal: %v", err)) - return - } - if err := conn.writeReply(CmdInfoOK, reqID, data); err != nil { - slog.Debug("IPC info reply failed", "err", err) - } + return data, err } func (s *IPCServer) handleHealth(conn *ipcConn, reqID uint64) { diff --git a/pkg/daemon/zz_burst_efficiency_test.go b/pkg/daemon/zz_burst_efficiency_test.go new file mode 100644 index 00000000..5ffa4083 --- /dev/null +++ b/pkg/daemon/zz_burst_efficiency_test.go @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "encoding/json" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// infoHostname runs one info request and returns the hostname in the reply. +func infoHostname(t *testing.T, s *IPCServer) string { + t.Helper() + ic, client := newIPCTestConn(t) + reply := runHandler(t, client, func() { s.handleInfo(ic, 0) }) + if reply[0] != CmdInfoOK { + t.Fatalf("opcode = 0x%02X, want CmdInfoOK", reply[0]) + } + var info map[string]interface{} + if err := json.Unmarshal(reply[1:], &info); err != nil { + t.Fatalf("info json: %v", err) + } + h, _ := info["hostname"].(string) + return h +} + +// A burst of info requests shares one reply for infoReplyTTL; a request +// after that sees current state again. Not parallel: it reads the +// package-level TTL. +func TestInfoReplyIsSharedWithinTTLThenRebuilt(t *testing.T) { + client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} { + return map[string]interface{}{"networks": []interface{}{float64(0)}} + }) + defer cleanup() + d, s := newSimpleHandlerDaemon(t, client) + + d.config.Hostname = "before" + if got := infoHostname(t, s); got != "before" { + t.Fatalf("first reply hostname = %q, want before", got) + } + + d.config.Hostname = "after" + if got := infoHostname(t, s); got != "before" { + t.Fatalf("reply inside the TTL was rebuilt (hostname %q); concurrent senders would each rebuild it", got) + } + + s.infoMu.Lock() + s.infoAt = time.Now().Add(-2 * infoReplyTTL) + s.infoMu.Unlock() + if got := infoHostname(t, s); got != "after" { + t.Fatalf("reply after the TTL still stale (hostname %q)", got) + } +} + +// With reuse turned off every request is built fresh. +func TestInfoReplyReuseCanBeTurnedOff(t *testing.T) { + client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} { + return map[string]interface{}{"networks": []interface{}{float64(0)}} + }) + defer cleanup() + d, s := newSimpleHandlerDaemon(t, client) + + prev := infoReplyTTL + infoReplyTTL = 0 + t.Cleanup(func() { infoReplyTTL = prev }) + + d.config.Hostname = "one" + _ = infoHostname(t, s) + d.config.Hostname = "two" + if got := infoHostname(t, s); got != "two" { + t.Fatalf("hostname = %q, want two with reuse off", got) + } +} + +// A RST for a connection still in SYN_SENT must wake the dialer at once. +// The dial loop's ticker is only a 250 ms backstop now; before, a refused +// dial was noticed by a 10 ms poll that ran for every dialing connection. +func TestResetWakesAWaitingDial(t *testing.T) { + t.Parallel() + d, _ := newPacketDaemon(t, nil) + + remote := protocol.Addr{Network: 0, Node: 77} + conn := d.ports.NewConnection(41000, remote, protocol.PortDataExchange) + conn.Mu.Lock() + conn.State = StateSynSent + conn.Mu.Unlock() + + d.handleStreamPacket(&protocol.Packet{ + Version: protocol.Version, + Protocol: protocol.ProtoStream, + Flags: protocol.FlagRST, + Src: remote, + Dst: protocol.Addr{Network: 0, Node: 42}, + SrcPort: protocol.PortDataExchange, + DstPort: 41000, + }) + + select { + case <-conn.DialCh: + default: + t.Fatal("RST did not signal the dialer; it would wait for the backstop tick") + } + conn.Mu.Lock() + st := conn.State + conn.Mu.Unlock() + if st != StateClosed { + t.Fatalf("state = %v, want CLOSED", st) + } +}