Add tunnel batch-size knob to embed; rename to Performance

2026-07-20 15:31:30 +02:00 · 2026-05-22 17:32:07 +02:00
parent 8f7fd69813
commit f4e42a8929
7 changed files with 87 additions and 123 deletions
--- 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.
--- 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 <pool-cap>",
-	Short:        "Set the pool cap (new and live clients)",
+var debugPerfCmd = &cobra.Command{
+	Use:          "perf <pool-cap>",
+	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 {
--- 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,
--- 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:")
--- 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
--- 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)
--- 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