router: configuration for async packet processing

doc/manuals/router.rst

@@ -125,6 +125,41 @@ considers the following options.
       If this is a relative path, it is interpreted as relative to the current working directory of the
       program (i.e. **not** relative to the location of this .toml configuration file).
 
+.. object:: router
+
+   .. option:: router.receive_buffer_size = <int> (Default: 0)
+
+      The receive buffer size in bytes. 0 means use system default.
+
+   .. option:: router.send_buffer_size = <int> (Default: 0)
+
+      The send buffer size in bytes. 0 means use system default.
+
+   .. option:: router.num_processors = <int> (Default: GOMAXPROCS)
+
+      Number of goroutines started for SCION packets processing.
+      These goroutines make the routing decision for the SCION packets by inspecting, validating and
+      updating the path information in the packet header. Packets are processed asynchronously from the
+      corresponding read/write operations on the individual interface sockets.
+
+      `Goroutines <https://en.wikipedia.org/wiki/Go_(programming_language)#Concurrency:_goroutines_and_channels>`_
+      are the Go pramming language's light-weight user-space concurrency primitives. Go's runtime schedules
+      goroutines on top of a fixed number of kernel threads. The number of kernel threads is controlled by
+      the ``GOMAXPROCS`` environment variable. See also the `go runtime documentation <https://pkg.go.dev/runtime#hdr-Environment_Variables>`_.
+
+      By default, the router uses ``GOMAXPROCS`` packet processor goroutines, i.e. exactly one goroutine for
+      each kernel thread created by the runtime.
+
+   .. option:: router.num_slow_processors = <int> (Default: 1)
+
+      Number of goroutines started for the slow-path processing. This feature will be implemented soon. Currently
+      this setting has no effect.
+
+   .. option:: router.batch_size = <int> (Default: 256)
+
+      The batch size used by the receiver and forwarder to
+      read or write from / to the network socket.
+
 .. _router-conf-topo:
 
 topology.json

private/underlay/conn/conn.go

@@ -214,7 +214,7 @@ func (cc *connUDPBase) initConnUDP(network string, laddr, raddr *net.UDPAddr, cf
 	}
 
 	// Set and confirm receive buffer size
-	{
+	if cfg.ReceiveBufferSize != 0 {
 		before, err := sockctrl.GetsockoptInt(c, syscall.SOL_SOCKET, syscall.SO_RCVBUF)
 		if err != nil {
 			return serrors.WrapStr("Error getting SO_RCVBUF socket option (before)", err,

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 := &router.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,55 @@ 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.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)
 }
 
 func (cfg *Config) InitDefaults() {
@@ -42,6 +93,7 @@ func (cfg *Config) InitDefaults() {
 		&cfg.Logging,
 		&cfg.Metrics,
 		&cfg.API,
+		&cfg.Router,
 	)
 }
 
@@ -52,6 +104,7 @@ func (cfg *Config) Validate() error {
 		&cfg.Logging,
 		&cfg.Metrics,
 		&cfg.API,
+		&cfg.Router,
 	)
 }
 
@@ -62,5 +115,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,38 @@
+// 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. 0 means use system default.
+# (default 0)
+receive_buffer_size = 0
+
+# The send buffer size in bytes. 0 means use system default.
+# (default 0)
+send_buffer_size = 0
+
+# The number of fast-path processors.
+# (default GOMAXPROCS)
+num_processors = 8
+
+# 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"
@@ -458,18 +457,19 @@ 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
 }
 
+type RunConfig struct {
+	NumProcessors         int
+	NumSlowPathProcessors int
+	BatchSize             int
+}
+
 func (d *DataPlane) Run(ctx context.Context, cfg *RunConfig) error {
 	d.mtx.Lock()
 	d.running = true
@@ -513,15 +513,6 @@ 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) {