From f4e42a892909863a7ec9897975555f9a5780ca19 Mon Sep 17 00:00:00 2001 From: Viktor Liu Date: Fri, 22 May 2026 17:32:07 +0200 Subject: [PATCH] Add tunnel batch-size knob to embed; rename to Performance --- client/embed/embed.go | 74 +++++++++++++---------------- proxy/cmd/proxy/cmd/debug.go | 33 +++---------- proxy/cmd/proxy/cmd/root.go | 36 +++++++------- proxy/internal/debug/client.go | 27 ++++------- proxy/internal/debug/handler.go | 33 +++++-------- proxy/internal/roundtrip/netbird.go | 2 + proxy/server.go | 5 ++ 7 files changed, 87 insertions(+), 123 deletions(-) diff --git a/client/embed/embed.go b/client/embed/embed.go index 129cf7c20..034ab7bb0 100644 --- a/client/embed/embed.go +++ b/client/embed/embed.go @@ -95,6 +95,26 @@ type Options struct { MTU *uint16 // DNSLabels defines additional DNS labels configured in the peer. DNSLabels []string + // Performance configures the tunnel's buffer pool cap and batch size. + Performance Performance +} + +// Performance configures the embedded client's tunnel memory/throughput knobs. +// +// These settings are process-global: any non-nil field also becomes the +// default for Clients constructed by later embed.New calls in the same +// process. Nil fields are ignored. +type Performance struct { + // PreallocatedBuffersPerPool caps the per-tunnel buffer pool. Zero + // leaves the pool unbounded. Lower values trade throughput for a + // tighter memory ceiling. May also be changed on a running Client via + // Client.SetPerformance, provided this field was nonzero at construction. + PreallocatedBuffersPerPool *uint32 + // MaxBatchSize overrides the number of packets the tunnel reads or + // writes per syscall, which also bounds eager buffer allocation per + // worker. Zero uses the platform default. Applied at construction + // only; ignored by Client.SetPerformance. + MaxBatchSize *uint32 } // validateCredentials checks that exactly one credential type is provided @@ -160,6 +180,13 @@ func New(opts Options) (*Client, error) { } } + if opts.Performance.PreallocatedBuffersPerPool != nil { + wgdevice.SetPreallocatedBuffersPerPool(*opts.Performance.PreallocatedBuffersPerPool) + } + if opts.Performance.MaxBatchSize != nil { + wgdevice.SetMaxBatchSizeOverride(*opts.Performance.MaxBatchSize) + } + var err error var parsedLabels domain.List if parsedLabels, err = domain.FromStringList(opts.DNSLabels); err != nil { @@ -474,19 +501,13 @@ func (c *Client) VerifySSHHostKey(peerAddress string, key []byte) error { return sshcommon.VerifyHostKey(storedKey, key, peerAddress) } -// WGTuning bundles runtime-adjustable WireGuard knobs exposed by the embed -// client. Nil fields are left unchanged; set a non-nil pointer to apply. -type WGTuning struct { - // PreallocatedBuffersPerPool caps each per-Device WaitPool. - // Zero means "unbounded" (no cap). Live-tunable only if the underlying - // Device was originally created with a nonzero cap. - PreallocatedBuffersPerPool *uint32 -} - -// SetWGTuning applies the given tuning to this client's live Device. -// Startup-only knobs (batch size) must be set via the package-level -// setters before Start. -func (c *Client) SetWGTuning(t WGTuning) error { +// SetPerformance retunes a running Client. Only PreallocatedBuffersPerPool +// takes effect, and only when it was nonzero at construction; +// MaxBatchSize is construction-only and is ignored here. +// +// Returns ErrClientNotStarted / ErrEngineNotStarted if the Client is not +// running yet. +func (c *Client) SetPerformance(t Performance) error { engine, err := c.getEngine() if err != nil { return err @@ -496,33 +517,6 @@ func (c *Client) SetWGTuning(t WGTuning) error { }) } -// SetWGDefaultPreallocatedBuffersPerPool sets the default WaitPool cap -// applied to Devices created after this call. Zero disables the cap. -// Existing Devices are unaffected; use Client.SetWGTuning for that. -func SetWGDefaultPreallocatedBuffersPerPool(n uint32) { - wgdevice.SetPreallocatedBuffersPerPool(n) -} - -// WGDefaultPreallocatedBuffersPerPool returns the current default WaitPool -// cap applied to newly-created Devices. -func WGDefaultPreallocatedBuffersPerPool() uint32 { - return wgdevice.PreallocatedBuffersPerPool -} - -// SetWGDefaultMaxBatchSize sets the default per-Device batch size applied -// to Devices created after this call. Zero means "use the bind+tun default" -// (NOT unlimited). Must be called before Start to take effect for a new -// Client. -func SetWGDefaultMaxBatchSize(n uint32) { - wgdevice.SetMaxBatchSizeOverride(n) -} - -// WGDefaultMaxBatchSize returns the current default batch-size override. -// Zero means "no override". -func WGDefaultMaxBatchSize() uint32 { - return wgdevice.MaxBatchSizeOverride -} - // StartCapture begins capturing packets on this client's tunnel device. // Only one capture can be active at a time; starting a new one stops the previous. // Call StopCapture (or CaptureSession.Stop) to end it. diff --git a/proxy/cmd/proxy/cmd/debug.go b/proxy/cmd/proxy/cmd/debug.go index 511251e04..360c7a516 100644 --- a/proxy/cmd/proxy/cmd/debug.go +++ b/proxy/cmd/proxy/cmd/debug.go @@ -109,24 +109,11 @@ var debugStopCmd = &cobra.Command{ SilenceUsage: true, } -var debugWGTuneCmd = &cobra.Command{ - Use: "wgtune", - Short: "Inspect and live-tune WireGuard pool settings", -} - -var debugWGTuneGetCmd = &cobra.Command{ - Use: "get", - Short: "Show pool cap and batch size defaults", - Args: cobra.NoArgs, - RunE: runDebugWGTuneGet, - SilenceUsage: true, -} - -var debugWGTuneSetCmd = &cobra.Command{ - Use: "set ", - Short: "Set the pool cap (new and live clients)", +var debugPerfCmd = &cobra.Command{ + Use: "perf ", + Short: "Live-retune the tunnel buffer pool cap on all running clients", Args: cobra.ExactArgs(1), - RunE: runDebugWGTuneSet, + RunE: runDebugPerfSet, SilenceUsage: true, } @@ -188,9 +175,7 @@ func init() { debugCmd.AddCommand(debugLogCmd) debugCmd.AddCommand(debugStartCmd) debugCmd.AddCommand(debugStopCmd) - debugWGTuneCmd.AddCommand(debugWGTuneGetCmd) - debugWGTuneCmd.AddCommand(debugWGTuneSetCmd) - debugCmd.AddCommand(debugWGTuneCmd) + debugCmd.AddCommand(debugPerfCmd) debugCmd.AddCommand(debugRuntimeCmd) debugCmd.AddCommand(debugCaptureCmd) @@ -253,16 +238,12 @@ func runDebugStop(cmd *cobra.Command, args []string) error { return getDebugClient(cmd).StopClient(cmd.Context(), args[0]) } -func runDebugWGTuneGet(cmd *cobra.Command, _ []string) error { - return getDebugClient(cmd).WGTuneGet(cmd.Context()) -} - -func runDebugWGTuneSet(cmd *cobra.Command, args []string) error { +func runDebugPerfSet(cmd *cobra.Command, args []string) error { n, err := strconv.ParseUint(args[0], 10, 32) if err != nil { return fmt.Errorf("invalid value %q: %w", args[0], err) } - return getDebugClient(cmd).WGTuneSet(cmd.Context(), uint32(n)) + return getDebugClient(cmd).PerfSet(cmd.Context(), uint32(n)) } func runDebugRuntime(cmd *cobra.Command, _ []string) error { diff --git a/proxy/cmd/proxy/cmd/root.go b/proxy/cmd/proxy/cmd/root.go index fc09ec492..5db506a78 100644 --- a/proxy/cmd/proxy/cmd/root.go +++ b/proxy/cmd/proxy/cmd/root.go @@ -22,13 +22,13 @@ import ( ) const ( - // envWGPreallocatedBuffers caps the per-Device WireGuard buffer pool - // size. Zero (unset) keeps the uncapped upstream default. - envWGPreallocatedBuffers = "NB_WG_PREALLOCATED_BUFFERS" - // envWGMaxBatchSize overrides the per-Device WireGuard batch size, - // which controls how many buffers each receive/TUN worker eagerly - // allocates. Zero (unset) keeps the bind+tun default. - envWGMaxBatchSize = "NB_WG_MAX_BATCH_SIZE" + // envPreallocatedBuffers caps the per-tunnel buffer pool. Zero (unset) + // keeps the upstream uncapped default. + envPreallocatedBuffers = "NB_PROXY_PREALLOCATED_BUFFERS" + // envMaxBatchSize overrides the per-tunnel batch size, which controls + // how many buffers each receive/TUN worker eagerly allocates. Zero + // (unset) keeps the platform default. + envMaxBatchSize = "NB_PROXY_MAX_BATCH_SIZE" ) const DefaultManagementURL = "https://api.netbird.io:443" @@ -157,23 +157,26 @@ func runServer(cmd *cobra.Command, args []string) error { logger.Infof("configured log level: %s", level) var wgPool, wgBatch uint64 - if raw := os.Getenv(envWGPreallocatedBuffers); raw != "" { + var perf embed.Performance + if raw := os.Getenv(envPreallocatedBuffers); raw != "" { n, err := strconv.ParseUint(raw, 10, 32) if err != nil { - return fmt.Errorf("invalid %s %q: %w", envWGPreallocatedBuffers, raw, err) + return fmt.Errorf("invalid %s %q: %w", envPreallocatedBuffers, raw, err) } wgPool = n - embed.SetWGDefaultPreallocatedBuffersPerPool(uint32(n)) - logger.Infof("wireguard preallocated buffers per pool: %d", n) + v := uint32(n) + perf.PreallocatedBuffersPerPool = &v + logger.Infof("tunnel preallocated buffers per pool: %d", n) } - if raw := os.Getenv(envWGMaxBatchSize); raw != "" { + if raw := os.Getenv(envMaxBatchSize); raw != "" { n, err := strconv.ParseUint(raw, 10, 32) if err != nil { - return fmt.Errorf("invalid %s %q: %w", envWGMaxBatchSize, raw, err) + return fmt.Errorf("invalid %s %q: %w", envMaxBatchSize, raw, err) } wgBatch = n - embed.SetWGDefaultMaxBatchSize(uint32(n)) - logger.Infof("wireguard max batch size override: %d", n) + v := uint32(n) + perf.MaxBatchSize = &v + logger.Infof("tunnel max batch size override: %d", n) } if wgPool > 0 { // Each bind recv goroutine (IPv4 + IPv6 + ICE relay) plus @@ -188,7 +191,7 @@ func runServer(cmd *cobra.Command, args []string) error { floor := batch * recvGoroutines if wgPool < floor { logger.Warnf("%s=%d is below the eager-allocation floor (~%d for batch=%d); startup may deadlock", - envWGPreallocatedBuffers, wgPool, floor, batch) + envPreallocatedBuffers, wgPool, floor, batch) } } @@ -231,6 +234,7 @@ func runServer(cmd *cobra.Command, args []string) error { CertLockMethod: nbacme.CertLockMethod(certLockMethod), WildcardCertDir: wildcardCertDir, WireguardPort: wgPort, + Performance: perf, ProxyProtocol: proxyProtocol, PreSharedKey: preSharedKey, SupportsCustomPorts: supportsCustomPorts, diff --git a/proxy/internal/debug/client.go b/proxy/internal/debug/client.go index 254a09065..aa87814f8 100644 --- a/proxy/internal/debug/client.go +++ b/proxy/internal/debug/client.go @@ -284,32 +284,21 @@ func (c *Client) printLogLevelResult(data map[string]any) { } } -// WGTuneGet fetches the current WireGuard pool cap. -func (c *Client) WGTuneGet(ctx context.Context) error { - return c.fetchAndPrint(ctx, "/debug/wgtune", c.printWGTuneGet) +// PerfSet live-retunes the tunnel buffer pool cap on all running embedded +// clients. Batch size is not live-tunable; configure it at proxy startup. +func (c *Client) PerfSet(ctx context.Context, value uint32) error { + path := fmt.Sprintf("/debug/perf?value=%d", value) + return c.fetchAndPrint(ctx, path, c.printPerfSet) } -// WGTuneSet updates the WireGuard pool cap on the global default and all live clients. -func (c *Client) WGTuneSet(ctx context.Context, value uint32) error { - path := fmt.Sprintf("/debug/wgtune?value=%d", value) - return c.fetchAndPrint(ctx, path, c.printWGTuneSet) -} - -func (c *Client) printWGTuneGet(data map[string]any) { - def, _ := data["default"].(float64) - batch, _ := data["batch_size"].(float64) - _, _ = fmt.Fprintf(c.out, "Default: %d\n", uint32(def)) - _, _ = fmt.Fprintf(c.out, "Batch size: %d (0 = unset)\n", uint32(batch)) -} - -func (c *Client) printWGTuneSet(data map[string]any) { +func (c *Client) printPerfSet(data map[string]any) { if errMsg, ok := data["error"].(string); ok && errMsg != "" { c.printError(data) return } - def, _ := data["default"].(float64) + val, _ := data["value"].(float64) applied, _ := data["applied"].(float64) - _, _ = fmt.Fprintf(c.out, "Default set to: %d\n", uint32(def)) + _, _ = fmt.Fprintf(c.out, "Pool cap set to: %d\n", uint32(val)) _, _ = fmt.Fprintf(c.out, "Applied to %d live clients\n", int(applied)) if failed, ok := data["failed"].(map[string]any); ok && len(failed) > 0 { _, _ = fmt.Fprintln(c.out, "Failed:") diff --git a/proxy/internal/debug/handler.go b/proxy/internal/debug/handler.go index 55308b61a..afdf5adf4 100644 --- a/proxy/internal/debug/handler.go +++ b/proxy/internal/debug/handler.go @@ -143,8 +143,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.handleListClients(w, r, wantJSON) case "/debug/health": h.handleHealth(w, r, wantJSON) - case "/debug/wgtune": - h.handleWGTune(w, r) + case "/debug/perf": + h.handlePerf(w, r) case "/debug/runtime": h.handleRuntime(w, r) default: @@ -650,33 +650,23 @@ func (h *Handler) handleClientStop(w http.ResponseWriter, r *http.Request, accou }) } -func (h *Handler) handleWGTune(w http.ResponseWriter, r *http.Request) { - values, ok := r.URL.Query()["value"] - if !ok { - h.writeJSON(w, map[string]any{ - "default": nbembed.WGDefaultPreallocatedBuffersPerPool(), - "batch_size": nbembed.WGDefaultMaxBatchSize(), - }) +func (h *Handler) handlePerf(w http.ResponseWriter, r *http.Request) { + raw := r.URL.Query().Get("value") + if raw == "" { + http.Error(w, "value parameter is required", http.StatusBadRequest) return } - if len(values) == 0 || values[0] == "" { - http.Error(w, "value parameter must not be empty", http.StatusBadRequest) - return - } - raw := values[0] - n, err := strconv.ParseUint(raw, 10, 32) if err != nil { http.Error(w, fmt.Sprintf("invalid value %q: %v", raw, err), http.StatusBadRequest) return } - nbembed.SetWGDefaultPreallocatedBuffersPerPool(uint32(n)) + capN := uint32(n) applied := 0 failed := map[string]string{} for accountID, client := range h.provider.ListClientsForStartup() { - capN := uint32(n) - if err := client.SetWGTuning(nbembed.WGTuning{PreallocatedBuffersPerPool: &capN}); err != nil { + if err := client.SetPerformance(nbembed.Performance{PreallocatedBuffersPerPool: &capN}); err != nil { failed[string(accountID)] = err.Error() continue } @@ -684,10 +674,9 @@ func (h *Handler) handleWGTune(w http.ResponseWriter, r *http.Request) { } resp := map[string]any{ - "success": true, - "default": uint32(n), - "batch_size": nbembed.WGDefaultMaxBatchSize(), - "applied": applied, + "success": true, + "value": capN, + "applied": applied, } if len(failed) > 0 { resp["failed"] = failed diff --git a/proxy/internal/roundtrip/netbird.go b/proxy/internal/roundtrip/netbird.go index e38e3dc4e..a261488ad 100644 --- a/proxy/internal/roundtrip/netbird.go +++ b/proxy/internal/roundtrip/netbird.go @@ -112,6 +112,7 @@ type ClientConfig struct { MgmtAddr string WGPort uint16 PreSharedKey string + Performance embed.Performance } type statusNotifier interface { @@ -267,6 +268,7 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account BlockInbound: true, WireguardPort: &wgPort, PreSharedKey: n.clientCfg.PreSharedKey, + Performance: n.clientCfg.Performance, }) if err != nil { return nil, fmt.Errorf("create netbird client: %w", err) diff --git a/proxy/server.go b/proxy/server.go index 6980e1df1..9b4614131 100644 --- a/proxy/server.go +++ b/proxy/server.go @@ -37,6 +37,7 @@ import ( "google.golang.org/grpc/keepalive" "google.golang.org/protobuf/types/known/timestamppb" + "github.com/netbirdio/netbird/client/embed" "github.com/netbirdio/netbird/proxy/internal/accesslog" "github.com/netbirdio/netbird/proxy/internal/acme" "github.com/netbirdio/netbird/proxy/internal/auth" @@ -162,6 +163,9 @@ type Server struct { // single-account deployments; multiple accounts will fail to bind // the same port. WireguardPort uint16 + // Performance configures the tunnel pool/batch sizes for every + // embedded client this proxy spawns. + Performance embed.Performance // ProxyProtocol enables PROXY protocol (v1/v2) on TCP listeners. // When enabled, the real client IP is extracted from the PROXY header // sent by upstream L4 proxies that support PROXY protocol. @@ -281,6 +285,7 @@ func (s *Server) ListenAndServe(ctx context.Context, addr string) (err error) { MgmtAddr: s.ManagementAddress, WGPort: s.WireguardPort, PreSharedKey: s.PreSharedKey, + Performance: s.Performance, }, s.Logger, s, s.mgmtClient) // Create health checker before the mapping worker so it can track