From 4b3dd9103dcf4009fbb6b088fbc44fc0db760128 Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Thu, 2 Jul 2026 20:42:43 +0200 Subject: [PATCH] [client] Fix slow wg operations (#6633) * [iface] Drop redundant device dump in kernel configure() wgctrl.ConfigureDevice already returns an error when the interface is missing, so the preceding wg.Device() existence check is redundant. That check dumps the entire device (all peers) on every configure() call, making it O(peers) per call and turning bulk peer insertion into O(peers^2): inserting N peers one by one re-parsed the whole growing peer list N times. Removing it keeps each peer write constant-time regardless of how many peers are already configured. * [iface] Cache WireGuard stats to collapse per-peer device dumps Each peer runs a WGWatcher that polls GetStats(), and every call dumps the whole device, so with N peers the watchers perform O(N) full dumps per poll cycle (O(N^2) work) while each keeps only its own peer's entry. Wrap the kernel and userspace configurer GetStats() in a short-TTL cache with singleflight: the staggered per-peer calls share a single device dump per window and concurrent misses collapse into one dump. The kernel and userspace WireGuard APIs have no per-peer stats query (a get always returns the whole device), so a shared cached snapshot avoids the repeated full dumps. * Ignore .claude directory --- .gitignore | 1 + client/iface/configurer/kernel_unix.go | 23 +++---- client/iface/configurer/stats_cache.go | 52 +++++++++++++++ client/iface/configurer/stats_cache_test.go | 70 +++++++++++++++++++++ client/iface/configurer/usp.go | 10 ++- 5 files changed, 144 insertions(+), 12 deletions(-) create mode 100644 client/iface/configurer/stats_cache.go create mode 100644 client/iface/configurer/stats_cache_test.go diff --git a/.gitignore b/.gitignore index 783fe77f3..305f3cb50 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +.claude .idea .run *.iml diff --git a/client/iface/configurer/kernel_unix.go b/client/iface/configurer/kernel_unix.go index a29fe181a..da69c2a35 100644 --- a/client/iface/configurer/kernel_unix.go +++ b/client/iface/configurer/kernel_unix.go @@ -17,12 +17,15 @@ import ( type KernelConfigurer struct { deviceName string + statsCache *statsCache } func NewKernelConfigurer(deviceName string) *KernelConfigurer { - return &KernelConfigurer{ + c := &KernelConfigurer{ deviceName: deviceName, } + c.statsCache = newStatsCache(statsCacheTTL, c.fetchStats) + return c } func (c *KernelConfigurer) ConfigureInterface(privateKey string, port int) error { @@ -246,12 +249,6 @@ func (c *KernelConfigurer) configure(config wgtypes.Config) error { } }() - // validate if device with name exists - _, err = wg.Device(c.deviceName) - if err != nil { - return err - } - return wg.ConfigureDevice(c.deviceName, config) } @@ -300,6 +297,14 @@ func (c *KernelConfigurer) FullStats() (*Stats, error) { } func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) { + return c.statsCache.get() +} + +func (c *KernelConfigurer) LastActivities() map[string]monotime.Time { + return nil +} + +func (c *KernelConfigurer) fetchStats() (map[string]WGStats, error) { stats := make(map[string]WGStats) wg, err := wgctrl.New() if err != nil { @@ -326,7 +331,3 @@ func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) { } return stats, nil } - -func (c *KernelConfigurer) LastActivities() map[string]monotime.Time { - return nil -} diff --git a/client/iface/configurer/stats_cache.go b/client/iface/configurer/stats_cache.go new file mode 100644 index 000000000..71a4e88fc --- /dev/null +++ b/client/iface/configurer/stats_cache.go @@ -0,0 +1,52 @@ +package configurer + +import ( + "sync" + "time" + + "golang.org/x/sync/singleflight" +) + +const statsCacheTTL = 1 * time.Second + +type statsCache struct { + ttl time.Duration + fetch func() (map[string]WGStats, error) + + mu sync.RWMutex + value map[string]WGStats + expireAt time.Time + + sf singleflight.Group +} + +func newStatsCache(ttl time.Duration, fetch func() (map[string]WGStats, error)) *statsCache { + return &statsCache{ttl: ttl, fetch: fetch} +} + +func (c *statsCache) get() (map[string]WGStats, error) { + c.mu.RLock() + if c.value != nil && time.Now().Before(c.expireAt) { + value := c.value + c.mu.RUnlock() + return value, nil + } + c.mu.RUnlock() + + value, err, _ := c.sf.Do("stats", func() (interface{}, error) { + res, err := c.fetch() + if err != nil { + return nil, err + } + + c.mu.Lock() + c.value = res + c.expireAt = time.Now().Add(c.ttl) + c.mu.Unlock() + return res, nil + }) + if err != nil { + return nil, err + } + return value.(map[string]WGStats), nil +} diff --git a/client/iface/configurer/stats_cache_test.go b/client/iface/configurer/stats_cache_test.go new file mode 100644 index 000000000..bcee5cd52 --- /dev/null +++ b/client/iface/configurer/stats_cache_test.go @@ -0,0 +1,70 @@ +package configurer + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestStatsCache_CachesWithinTTL(t *testing.T) { + var calls atomic.Int64 + c := newStatsCache(50*time.Millisecond, func() (map[string]WGStats, error) { + calls.Add(1) + return map[string]WGStats{"p": {}}, nil + }) + + for i := 0; i < 10; i++ { + _, err := c.get() + require.NoError(t, err) + } + require.Equal(t, int64(1), calls.Load(), "within TTL only one underlying fetch") + + time.Sleep(60 * time.Millisecond) + _, err := c.get() + require.NoError(t, err) + require.Equal(t, int64(2), calls.Load(), "after TTL expiry a fresh fetch happens") +} + +func TestStatsCache_SingleFlight(t *testing.T) { + var calls atomic.Int64 + release := make(chan struct{}) + c := newStatsCache(time.Minute, func() (map[string]WGStats, error) { + calls.Add(1) + <-release + return map[string]WGStats{}, nil + }) + + const n = 50 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + _, _ = c.get() + }() + } + time.Sleep(20 * time.Millisecond) + close(release) + wg.Wait() + + require.Equal(t, int64(1), calls.Load(), "concurrent misses collapse into one fetch") +} + +func TestStatsCache_ErrorNotCached(t *testing.T) { + var calls atomic.Int64 + wantErr := errors.New("dump failed") + c := newStatsCache(time.Minute, func() (map[string]WGStats, error) { + calls.Add(1) + return nil, wantErr + }) + + _, err := c.get() + require.ErrorIs(t, err, wantErr) + _, err = c.get() + require.ErrorIs(t, err, wantErr) + require.Equal(t, int64(2), calls.Load(), "errors are not cached; each call retries") +} diff --git a/client/iface/configurer/usp.go b/client/iface/configurer/usp.go index 9b070aab8..0a25c55bc 100644 --- a/client/iface/configurer/usp.go +++ b/client/iface/configurer/usp.go @@ -40,6 +40,7 @@ type WGUSPConfigurer struct { device *device.Device deviceName string activityRecorder *bind.ActivityRecorder + statsCache *statsCache uapiListener net.Listener } @@ -50,16 +51,19 @@ func NewUSPConfigurer(device *device.Device, deviceName string, activityRecorder deviceName: deviceName, activityRecorder: activityRecorder, } + wgCfg.statsCache = newStatsCache(statsCacheTTL, wgCfg.fetchStats) wgCfg.startUAPI() return wgCfg } func NewUSPConfigurerNoUAPI(device *device.Device, deviceName string, activityRecorder *bind.ActivityRecorder) *WGUSPConfigurer { - return &WGUSPConfigurer{ + wgCfg := &WGUSPConfigurer{ device: device, deviceName: deviceName, activityRecorder: activityRecorder, } + wgCfg.statsCache = newStatsCache(statsCacheTTL, wgCfg.fetchStats) + return wgCfg } func (c *WGUSPConfigurer) ConfigureInterface(privateKey string, port int) error { @@ -348,6 +352,10 @@ func (t *WGUSPConfigurer) Close() { } func (t *WGUSPConfigurer) GetStats() (map[string]WGStats, error) { + return t.statsCache.get() +} + +func (t *WGUSPConfigurer) fetchStats() (map[string]WGStats, error) { ipc, err := t.device.IpcGet() if err != nil { return nil, fmt.Errorf("ipc get: %w", err)