Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

router: configuration for async packet processing #4374

Merged
merged 7 commits into from
Aug 4, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions doc/manuals/router.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion private/underlay/conn/conn.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
9 changes: 7 additions & 2 deletions router/cmd/router/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
}
Expand Down
6 changes: 5 additions & 1 deletion router/config/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
54 changes: 54 additions & 0 deletions router/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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() {
Expand All @@ -42,6 +93,7 @@ func (cfg *Config) InitDefaults() {
&cfg.Logging,
&cfg.Metrics,
&cfg.API,
&cfg.Router,
)
}

Expand All @@ -52,6 +104,7 @@ func (cfg *Config) Validate() error {
&cfg.Logging,
&cfg.Metrics,
&cfg.API,
&cfg.Router,
)
}

Expand All @@ -62,5 +115,6 @@ func (cfg *Config) Sample(dst io.Writer, path config.Path, _ config.CtxMap) {
&cfg.Logging,
&cfg.Metrics,
&cfg.API,
&cfg.Router,
)
}
38 changes: 38 additions & 0 deletions router/config/sample.go
Original file line number Diff line number Diff line change
@@ -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
`
10 changes: 5 additions & 5 deletions router/connector.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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")
Expand All @@ -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
}
Expand Down Expand Up @@ -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
}
Expand Down
21 changes: 6 additions & 15 deletions router/dataplane.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@ import (
"math/big"
"net"
"net/netip"
"runtime"
"strconv"
"sync"
"time"
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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) {
Expand Down
Loading