From d1ee02fb29c9f44bb52e7287a925901d74724e1a Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Tue, 6 Oct 2026 02:00:34 +0300 Subject: [PATCH 1/2] daemon: much less CPU under send bursts Profiling two local daemons under 32- and 256-way bursts of `pilotctl send-message` showed the sender spending about 85% of its CPU in the Go scheduler, and 73% of its allocations on `info` replies. - Dials no longer poll their connection every 10 ms. SYN-ACK already signalled DialCh; RST and shutdown now do too, and the ticker is a 250 ms backstop. - Concurrent info requests share one reply for 200 ms. Every send asks for info, and the reply serialises every connection, so N concurrent sends cost N builds over a table of N rows. Requests arriving during a build wait for it instead of starting their own (this also stops a burst from issuing one registry lookup per request when the network-membership cache expires). - The daemon caps GOMAXPROCS at 4 unless the variable is set. Sender CPU per message: 4.5 -> 1.4 ms at 32 concurrent, 24 -> 2.6 ms at 256. Most of it comes from the first two changes (3.5 ms at 256 with 16 threads); the thread cap takes the rest. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 14 ++++ cmd/daemon/main.go | 12 +++ pkg/daemon/daemon.go | 4 +- pkg/daemon/ipc.go | 47 +++++++++-- pkg/daemon/zz_burst_efficiency_test.go | 111 +++++++++++++++++++++++++ 5 files changed, 180 insertions(+), 8 deletions(-) create mode 100644 pkg/daemon/zz_burst_efficiency_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ffbbd60..8d516981 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.4 ms at 32 concurrent sends and from 24 ms to 2.6 ms at 256, and its + memory after a burst is 9–13 MB lower. Three 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. + - The daemon caps itself at 4 OS threads unless `GOMAXPROCS` is set. With + one thread per core most of its CPU in a burst went to waking and parking + idle threads. - **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/cmd/daemon/main.go b/cmd/daemon/main.go index 2e50df24..e6c8efdc 100644 --- a/cmd/daemon/main.go +++ b/cmd/daemon/main.go @@ -13,6 +13,7 @@ import ( "os" "os/signal" "path/filepath" + goruntime "runtime" "strconv" "strings" "syscall" @@ -47,7 +48,18 @@ import ( var version = "dev" +// daemonMaxProcs caps the daemon's OS threads unless GOMAXPROCS is set. +// The daemon's work is many short waits on sockets and timers. With one +// thread per core (16 on a laptop, 64 on the service-agent VM) a burst of +// connections spends most of its CPU in the Go scheduler waking and parking +// idle threads: measured 85% of the daemon's CPU in a 256-way send burst, +// and 25–40% less CPU per message at 2 threads than at 16. +const daemonMaxProcs = 4 + func main() { + if os.Getenv("GOMAXPROCS") == "" && goruntime.GOMAXPROCS(0) > daemonMaxProcs { + goruntime.GOMAXPROCS(daemonMaxProcs) + } configPath := flag.String("config", "", "path to config file (JSON)") securityProfile := flag.String("security-profile", envString("PILOT_SECURITY_PROFILE", securityProfileCompatible), "locked security profile: compatible or enterprise") registryDefault := defaultRegistryAddr 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) + } +} From 08cc04d0b30f0541f66f26c3de795c3e684d8578 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 8 Oct 2026 11:49:23 +0300 Subject: [PATCH 2/2] daemon: no hard thread cap; GOMAXPROCS stays the operator's choice The cap of 4 OS threads applied to every install unless GOMAXPROCS was set, including large hosts and the hosted variant. Keep the dial wake-up and the shared info reply; the changelog quotes this PR's measurements at 16 threads, which is what ships. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 +++++----- cmd/daemon/main.go | 12 ------------ 2 files changed, 5 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d516981..f2677560 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -306,8 +306,8 @@ Detailed per-release notes are on the (`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.4 ms at 32 concurrent sends and from 24 ms to 2.6 ms at 256, and its - memory after a burst is 9–13 MB lower. Three changes: + 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. @@ -315,9 +315,9 @@ Detailed per-release notes are on the `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. - - The daemon caps itself at 4 OS threads unless `GOMAXPROCS` is set. With - one thread per core most of its CPU in a burst went to waking and parking - idle threads. + + 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/cmd/daemon/main.go b/cmd/daemon/main.go index e6c8efdc..2e50df24 100644 --- a/cmd/daemon/main.go +++ b/cmd/daemon/main.go @@ -13,7 +13,6 @@ import ( "os" "os/signal" "path/filepath" - goruntime "runtime" "strconv" "strings" "syscall" @@ -48,18 +47,7 @@ import ( var version = "dev" -// daemonMaxProcs caps the daemon's OS threads unless GOMAXPROCS is set. -// The daemon's work is many short waits on sockets and timers. With one -// thread per core (16 on a laptop, 64 on the service-agent VM) a burst of -// connections spends most of its CPU in the Go scheduler waking and parking -// idle threads: measured 85% of the daemon's CPU in a 256-way send burst, -// and 25–40% less CPU per message at 2 threads than at 16. -const daemonMaxProcs = 4 - func main() { - if os.Getenv("GOMAXPROCS") == "" && goruntime.GOMAXPROCS(0) > daemonMaxProcs { - goruntime.GOMAXPROCS(daemonMaxProcs) - } configPath := flag.String("config", "", "path to config file (JSON)") securityProfile := flag.String("security-profile", envString("PILOT_SECURITY_PROFILE", securityProfileCompatible), "locked security profile: compatible or enterprise") registryDefault := defaultRegistryAddr