router: configuration for async packet processing

router/BUILD.bazel

@@ -29,6 +29,7 @@ go_library(
         "//private/topology:go_default_library",
         "//private/underlay/conn:go_default_library",
         "//router/bfd:go_default_library",
+        "//router/config:go_default_library",
         "//router/control:go_default_library",
         "@com_github_google_gopacket//:go_default_library",
         "@com_github_google_gopacket//layers:go_default_library",
@@ -62,6 +63,7 @@ go_test(
         "//pkg/slayers/path/scion:go_default_library",
         "//private/topology:go_default_library",
         "//private/underlay/conn:go_default_library",
+        "//router/config:go_default_library",
         "//router/control:go_default_library",
         "//router/mock_router:go_default_library",
         "@com_github_golang_mock//gomock:go_default_library",

router/cmd/router/main.go

@@ -60,6 +60,8 @@ func realMain(ctx context.Context) error {
 		DataPlane: router.DataPlane{
 			Metrics: metrics,
 		},
+		ReceiveBufferSize: globalCfg.Router.ReceiveBufferSize,
+		SendBufferSize:    globalCfg.Router.SendBufferSize,
 	}
 	iaCtx := &control.IACtx{
 		Config: controlConfig,
@@ -121,8 +123,11 @@ func realMain(ctx context.Context) error {
 	})
 	g.Go(func() error {
 		defer log.HandlePanic()
-		runConfig := &router.RunConfig{}
-		runConfig.LoadDefaults()
+		runConfig := &config.RunConfig{
+			NumProcessors:         globalCfg.Router.NumProcessors,
+			NumSlowPathProcessors: globalCfg.Router.NumSlowPathProcessors,
+			BatchSize:             globalCfg.Router.BatchSize,
+		}
 		if err := dp.DataPlane.Run(errCtx, runConfig); err != nil {
 			return serrors.WrapStr("running dataplane", err)
 		}

router/config/BUILD.bazel

@@ -2,11 +2,15 @@ load("//tools/lint:go.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["config.go"],
+    srcs = [
+        "config.go",
+        "sample.go",
+    ],
     importpath = "github.com/scionproto/scion/router/config",
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/log:go_default_library",
+        "//pkg/private/serrors:go_default_library",
         "//private/config:go_default_library",
         "//private/env:go_default_library",
         "//private/mgmtapi:go_default_library",

router/config/config.go

@@ -18,8 +18,10 @@ package config
 
 import (
 	"io"
+	"runtime"
 
 	"github.com/scionproto/scion/pkg/log"
+	"github.com/scionproto/scion/pkg/private/serrors"
 	"github.com/scionproto/scion/private/config"
 	"github.com/scionproto/scion/private/env"
 	api "github.com/scionproto/scion/private/mgmtapi"
@@ -33,6 +35,64 @@ type Config struct {
 	Logging  log.Config   `toml:"log,omitempty"`
 	Metrics  env.Metrics  `toml:"metrics,omitempty"`
 	API      api.Config   `toml:"api,omitempty"`
+	Router   RouterConfig `toml:"router,omitempty"`
+}
+
+type RouterConfig struct {
+	ReceiveBufferSize     int `toml:"receive_buffer_size,omitempty"`
+	SendBufferSize        int `toml:"send_buffer_size,omitempty"`
+	NumProcessors         int `toml:"num_processors,omitempty"`
+	NumSlowPathProcessors int `toml:"num_slow_processors,omitempty"`
+	BatchSize             int `toml:"batch_size,omitempty"`
+}
+
+func (cfg *RouterConfig) ConfigName() string {
+	return "router"
+}
+
+func (cfg *RouterConfig) Validate() error {
+	if cfg.ReceiveBufferSize < 0 {
+		return serrors.New("Provided router config is invalid. ReceiveBufferSize < 0")
+	}
+	if cfg.SendBufferSize < 0 {
+		return serrors.New("Provided router config is invalid. SendBufferSize < 0")
+	}
+	if cfg.BatchSize < 1 {
+		return serrors.New("Provided router config is invalid. BatchSize < 1")
+	}
+	if cfg.NumProcessors < 0 {
+		return serrors.New("Provided router config is invalid. NumProcessors < 0")
+	}
+	if cfg.NumSlowPathProcessors < 1 {
+		return serrors.New("Provided router config is invalid. NumSlowPathProcessors < 1")
+	}
+
+	return nil
+}
+
+func (cfg *RouterConfig) InitDefaults() {
+	if cfg.ReceiveBufferSize == 0 {
+		cfg.ReceiveBufferSize = 1 << 20
+	}
+	if cfg.NumProcessors == 0 {
+		cfg.NumProcessors = runtime.GOMAXPROCS(0)
+	}
+	if cfg.NumSlowPathProcessors == 0 {
+		cfg.NumSlowPathProcessors = 1
+	}
+	if cfg.BatchSize == 0 {
+		cfg.BatchSize = 256
+	}
+}
+
+func (cfg *RouterConfig) Sample(dst io.Writer, path config.Path, ctx config.CtxMap) {
+	config.WriteString(dst, routerConfigSample)
+}
+
+type RunConfig struct {
+	NumProcessors         int
+	NumSlowPathProcessors int
+	BatchSize             int
 }
 
 func (cfg *Config) InitDefaults() {
@@ -42,6 +102,7 @@ func (cfg *Config) InitDefaults() {
 		&cfg.Logging,
 		&cfg.Metrics,
 		&cfg.API,
+		&cfg.Router,
 	)
 }
 
@@ -52,6 +113,7 @@ func (cfg *Config) Validate() error {
 		&cfg.Logging,
 		&cfg.Metrics,
 		&cfg.API,
+		&cfg.Router,
 	)
 }
 
@@ -62,5 +124,6 @@ func (cfg *Config) Sample(dst io.Writer, path config.Path, _ config.CtxMap) {
 		&cfg.Logging,
 		&cfg.Metrics,
 		&cfg.API,
+		&cfg.Router,
 	)
 }

router/config/sample.go

@@ -0,0 +1,34 @@
+// Copyright 2023 ETH Zurich
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+const routerConfigSample = `
+# The receive buffer size in bytes.
+# (default 1048576)
+receive_buffer_size = 1048576
+
+# The send buffer size in bytes. 0 means use system default.
+# (default 0)
+send_buffer_size = 0
+
+# The number of slow path processors.
+# (default 1)
+num_slow_processors = 1
+
+# The batch size used by the receiver and forwarder to
+# read or write from / to the network socket.
+# (default 256)
+batch_size = 256
+`

router/connector.go

@@ -27,9 +27,6 @@ import (
 	"github.com/scionproto/scion/router/control"
 )
 
-// receiveBufferSize is the size of receive buffers used by the router.
-const receiveBufferSize = 1 << 20
-
 // Connector implements the Dataplane API of the router control process. It sets
 // up connections for the DataPlane.
 type Connector struct {
@@ -40,6 +37,9 @@ type Connector struct {
 	internalInterfaces []control.InternalInterface
 	externalInterfaces map[uint16]control.ExternalInterface
 	siblingInterfaces  map[uint16]control.SiblingInterface
+
+	ReceiveBufferSize int
+	SendBufferSize    int
 }
 
 var errMultiIA = serrors.New("different IA not allowed")
@@ -65,7 +65,7 @@ func (c *Connector) AddInternalInterface(ia addr.IA, local net.UDPAddr) error {
 		return serrors.WithCtx(errMultiIA, "current", c.ia, "new", ia)
 	}
 	connection, err := conn.New(&local, nil,
-		&conn.Config{ReceiveBufferSize: receiveBufferSize})
+		&conn.Config{ReceiveBufferSize: c.ReceiveBufferSize, SendBufferSize: c.SendBufferSize})
 	if err != nil {
 		return err
 	}
@@ -130,7 +130,7 @@ func (c *Connector) AddExternalInterface(localIfID common.IFIDType, link control
 	}
 
 	connection, err := conn.New(link.Local.Addr, link.Remote.Addr,
-		&conn.Config{ReceiveBufferSize: receiveBufferSize})
+		&conn.Config{ReceiveBufferSize: c.ReceiveBufferSize, SendBufferSize: c.SendBufferSize})
 	if err != nil {
 		return err
 	}

router/dataplane.go

@@ -27,7 +27,6 @@ import (
 	"math/big"
 	"net"
 	"net/netip"
-	"runtime"
 	"strconv"
 	"sync"
 	"time"
@@ -54,6 +53,7 @@ import (
 	"github.com/scionproto/scion/private/topology"
 	underlayconn "github.com/scionproto/scion/private/underlay/conn"
 	"github.com/scionproto/scion/router/bfd"
+	"github.com/scionproto/scion/router/config"
 	"github.com/scionproto/scion/router/control"
 )
 
@@ -458,19 +458,14 @@ func (d *DataPlane) AddNextHopBFD(ifID uint16, src, dst *net.UDPAddr, cfg contro
 	return d.addBFDController(ifID, s, cfg, m)
 }
 
-type RunConfig struct {
-	NumProcessors int
-	BatchSize     int
-}
-
 func max(a int, b int) int {
 	if a > b {
 		return a
 	}
 	return b
 }
 
-func (d *DataPlane) Run(ctx context.Context, cfg *RunConfig) error {
+func (d *DataPlane) Run(ctx context.Context, cfg *config.RunConfig) error {
 	d.mtx.Lock()
 	d.running = true
 	d.initMetrics()
@@ -513,18 +508,9 @@ func (d *DataPlane) Run(ctx context.Context, cfg *RunConfig) error {
 	return nil
 }
 
-// loadDefaults sets the default configuration for the number of
-// processors and the batch size
-func (r *RunConfig) LoadDefaults() {
-	// TODO(rohrerj) move this logic to configuration in configuration PR
-	r.NumProcessors = runtime.GOMAXPROCS(0)
-	r.BatchSize = 256
-
-}
-
 // initializePacketPool calculates the size of the packet pool based on the
 // current dataplane settings and allocates all the buffers
-func (d *DataPlane) initPacketPool(cfg *RunConfig, processorQueueSize int) {
+func (d *DataPlane) initPacketPool(cfg *config.RunConfig, processorQueueSize int) {
 	poolSize := len(d.interfaces)*cfg.BatchSize +
 		cfg.NumProcessors*(processorQueueSize+1) +
 		len(d.interfaces)*(2*cfg.BatchSize)
@@ -537,7 +523,7 @@ func (d *DataPlane) initPacketPool(cfg *RunConfig, processorQueueSize int) {
 }
 
 // initializes the processing routines and forwarders queues
-func initQueues(cfg *RunConfig, interfaces map[uint16]BatchConn,
+func initQueues(cfg *config.RunConfig, interfaces map[uint16]BatchConn,
 	processorQueueSize int) ([]chan packet, map[uint16]chan packet) {
 
 	procQs := make([]chan packet, cfg.NumProcessors)
@@ -563,7 +549,7 @@ type packet struct {
 	rawPacket []byte
 }
 
-func (d *DataPlane) runReceiver(ifID uint16, conn BatchConn, cfg *RunConfig,
+func (d *DataPlane) runReceiver(ifID uint16, conn BatchConn, cfg *config.RunConfig,
 	procQs []chan packet) {
 
 	log.Debug("Run receiver for", "interface", ifID)
@@ -704,7 +690,7 @@ func (d *DataPlane) runProcessor(id int, q <-chan packet,
 }
 
 func (d *DataPlane) runForwarder(ifID uint16, conn BatchConn,
-	cfg *RunConfig, c <-chan packet) {
+	cfg *config.RunConfig, c <-chan packet) {
 
 	log.Debug("Initialize forwarder for", "interface", ifID)
 	writeMsgs := make(underlayconn.Messages, cfg.BatchSize)

router/dataplane_internal_test.go

@@ -38,6 +38,7 @@ import (
 	"github.com/scionproto/scion/pkg/slayers/path"
 	"github.com/scionproto/scion/pkg/slayers/path/scion"
 	underlayconn "github.com/scionproto/scion/private/underlay/conn"
+	"github.com/scionproto/scion/router/config"
 	"github.com/scionproto/scion/router/mock_router"
 )
 
@@ -75,7 +76,7 @@ func TestReceiver(t *testing.T) {
 
 	_ = dp.AddInternalInterface(mInternal, net.IP{})
 
-	runConfig := &RunConfig{
+	runConfig := &config.RunConfig{
 		NumProcessors: 1,
 		BatchSize:     64,
 	}
@@ -169,7 +170,7 @@ func TestForwarder(t *testing.T) {
 		return ret
 	}
 	dp := prepareDP(ctrl)
-	runConfig := &RunConfig{
+	runConfig := &config.RunConfig{
 		NumProcessors: 20,
 		BatchSize:     64,
 	}

router/dataplane_test.go

@@ -46,6 +46,7 @@ import (
 	"github.com/scionproto/scion/private/topology"
 	underlayconn "github.com/scionproto/scion/private/underlay/conn"
 	"github.com/scionproto/scion/router"
+	"github.com/scionproto/scion/router/config"
 	"github.com/scionproto/scion/router/control"
 	"github.com/scionproto/scion/router/mock_router"
 )
@@ -538,7 +539,7 @@ func TestDataPlaneRun(t *testing.T) {
 		name, tc := name, tc
 		t.Run(name, func(t *testing.T) {
 			t.Parallel()
-			runConfig := &router.RunConfig{
+			runConfig := &config.RunConfig{
 				NumProcessors: 8,
 				BatchSize:     256,
 			}

tools/topology/go.py

@@ -97,6 +97,12 @@ def _build_br_conf(self, topo_id, ia, base, name, v):
             'features': translate_features(self.args.features),
             'api': {
                 'addr': prom_addr(v['internal_addr'], DEFAULT_BR_PROM_PORT+700)
+            },
+            'router': {
+                'receive_buffer_size': 1 << 20,
+                'send_buffer_size': 0,
+                'num_slow_processors': 1,
+                'batch_size': 256
             }
         }
         return raw_entry