Merge pull request #216 from neilalexander/switchoptions

Add SwitchOptions and MaxTotalQueueSize

src/yggdrasil/admin.go

@@ -638,7 +638,7 @@ func (a *admin) getData_getSwitchQueues() admin_nodeInfo {
 			{"queues_size", switchTable.queues.size},
 			{"highest_queues_count", switchTable.queues.maxbufs},
 			{"highest_queues_size", switchTable.queues.maxsize},
-			{"maximum_queues_size", switch_buffer_maxSize},
+			{"maximum_queues_size", switchTable.queueTotalMaxSize},
 		}
 	}
 	a.core.switchTable.doAdmin(getSwitchQueues)

src/yggdrasil/config/config.go

@@ -18,6 +18,7 @@ type NodeConfig struct {
 	IfMTU                       int                 `comment:"Maximux Transmission Unit (MTU) size for your local TUN/TAP interface.\nDefault is the largest supported size for your platform. The lowest\npossible value is 1280."`
 	SessionFirewall             SessionFirewall     `comment:"The session firewall controls who can send/receive network traffic\nto/from. This is useful if you want to protect this node without\nresorting to using a real firewall. This does not affect traffic\nbeing routed via this node to somewhere else. Rules are prioritised as\nfollows: blacklist, whitelist, always allow outgoing, direct, remote."`
 	TunnelRouting               TunnelRouting       `comment:"Allow tunneling non-Yggdrasil traffic over Yggdrasil. This effectively\nallows you to use Yggdrasil to route to, or to bridge other networks,\nsimilar to a VPN tunnel. Tunnelling works between any two nodes and\ndoes not require them to be directly peered."`
+	SwitchOptions               SwitchOptions       `comment:"Advanced options for tuning the switch. Normally you will not need\nto edit these options."`
 	//Net                         NetConfig `comment:"Extended options for connecting to peers over other networks."`
 }
 
@@ -45,3 +46,8 @@ type TunnelRouting struct {
 	IPv4Destinations map[string]string `comment:"IPv4 CIDR subnets, mapped to the EncryptionPublicKey to which they\nshould be routed, e.g. { \"a.b.c.d/e\": \"boxpubkey\", ... }"`
 	IPv4Sources      []string          `comment:"IPv4 source subnets which are allowed to be tunnelled. Unlike for\nIPv6, this option is required for bridging IPv4 traffic. Only\ntraffic with a source matching these subnets will be tunnelled."`
 }
+
+// SwitchOptions contains tuning options for the switch
+type SwitchOptions struct {
+	MaxTotalQueueSize uint64 `comment:"Maximum size of all switch queues combined (in bytes)."`
+}

src/yggdrasil/core.go

@@ -100,6 +100,10 @@ func (c *Core) Start(nc *config.NodeConfig, log *log.Logger) error {
 		return err
 	}
 
+	if nc.SwitchOptions.MaxTotalQueueSize >= SwitchQueueTotalMinSize {
+		c.switchTable.queueTotalMaxSize = nc.SwitchOptions.MaxTotalQueueSize
+	}
+
 	if err := c.switchTable.start(); err != nil {
 		c.log.Println("Failed to start switch")
 		return err

src/yggdrasil/switch.go

@@ -156,21 +156,25 @@ type switchData struct {
 
 // All the information stored by the switch.
 type switchTable struct {
-	core     *Core
+	core              *Core
-	key      sigPubKey           // Our own key
+	key               sigPubKey           // Our own key
-	time     time.Time           // Time when locator.tstamp was last updated
+	time              time.Time           // Time when locator.tstamp was last updated
-	drop     map[sigPubKey]int64 // Tstamp associated with a dropped root
+	drop              map[sigPubKey]int64 // Tstamp associated with a dropped root
-	mutex    sync.RWMutex        // Lock for reads/writes of switchData
+	mutex             sync.RWMutex        // Lock for reads/writes of switchData
-	parent   switchPort          // Port of whatever peer is our parent, or self if we're root
+	parent            switchPort          // Port of whatever peer is our parent, or self if we're root
-	data     switchData          //
+	data              switchData          //
-	updater  atomic.Value        // *sync.Once
+	updater           atomic.Value        // *sync.Once
-	table    atomic.Value        // lookupTable
+	table             atomic.Value        // lookupTable
-	packetIn chan []byte         // Incoming packets for the worker to handle
+	packetIn          chan []byte         // Incoming packets for the worker to handle
-	idleIn   chan switchPort     // Incoming idle notifications from peer links
+	idleIn            chan switchPort     // Incoming idle notifications from peer links
-	admin    chan func()         // Pass a lambda for the admin socket to query stuff
+	admin             chan func()         // Pass a lambda for the admin socket to query stuff
-	queues   switch_buffers      // Queues - not atomic so ONLY use through admin chan
+	queues            switch_buffers      // Queues - not atomic so ONLY use through admin chan
+	queueTotalMaxSize uint64              // Maximum combined size of queues
 }
 
+// Minimum allowed total size of switch queues.
+const SwitchQueueTotalMinSize = 4 * 1024 * 1024
+
 // Initializes the switchTable struct.
 func (t *switchTable) init(core *Core, key sigPubKey) {
 	now := time.Now()
@@ -185,6 +189,7 @@ func (t *switchTable) init(core *Core, key sigPubKey) {
 	t.packetIn = make(chan []byte, 1024)
 	t.idleIn = make(chan switchPort, 1024)
 	t.admin = make(chan func())
+	t.queueTotalMaxSize = SwitchQueueTotalMinSize
 }
 
 // Safely gets a copy of this node's locator.
@@ -620,8 +625,6 @@ type switch_packetInfo struct {
 	time  time.Time // Timestamp of when the packet arrived
 }
 
-const switch_buffer_maxSize = 4 * 1048576 // Maximum 4 MB
-
 // Used to keep track of buffered packets
 type switch_buffer struct {
 	packets []switch_packetInfo // Currently buffered packets, which may be dropped if it grows too large
@@ -629,10 +632,11 @@ type switch_buffer struct {
 }
 
 type switch_buffers struct {
-	bufs    map[string]switch_buffer // Buffers indexed by StreamID
+	switchTable *switchTable
-	size    uint64                   // Total size of all buffers, in bytes
+	bufs        map[string]switch_buffer // Buffers indexed by StreamID
-	maxbufs int
+	size        uint64                   // Total size of all buffers, in bytes
-	maxsize uint64
+	maxbufs     int
+	maxsize     uint64
 }
 
 func (b *switch_buffers) cleanup(t *switchTable) {
@@ -649,7 +653,7 @@ func (b *switch_buffers) cleanup(t *switchTable) {
 		}
 	}
 
-	for b.size > switch_buffer_maxSize {
+	for b.size > b.switchTable.queueTotalMaxSize {
 		// Drop a random queue
 		target := rand.Uint64() % b.size
 		var size uint64 // running total
@@ -719,6 +723,7 @@ func (t *switchTable) handleIdle(port switchPort) bool {
 
 // The switch worker does routing lookups and sends packets to where they need to be
 func (t *switchTable) doWorker() {
+	t.queues.switchTable = t
 	t.queues.bufs = make(map[string]switch_buffer) // Packets per PacketStreamID (string)
 	idle := make(map[switchPort]struct{})          // this is to deduplicate things
 	for {

yggdrasil.go

@@ -69,6 +69,7 @@ func generateConfig(isAutoconf bool) *nodeConfig {
 	cfg.SessionFirewall.Enable = false
 	cfg.SessionFirewall.AllowFromDirect = true
 	cfg.SessionFirewall.AllowFromRemote = true
+	cfg.SwitchOptions.MaxTotalQueueSize = yggdrasil.SwitchQueueTotalMinSize
 
 	return &cfg
 }

yggdrasilctl.go

@@ -216,8 +216,8 @@ func main() {
 				fmt.Printf("Highest queue size: %d bytes\n", uint(highestqueuesize))
 			}
 			if m, ok := v["maximum_queues_size"].(float64); ok {
-				fmt.Printf("Maximum queue size: %d bytes\n", uint(maximumqueuesize))
 				maximumqueuesize = m
+				fmt.Printf("Maximum queue size: %d bytes\n", uint(maximumqueuesize))
 			}
 			if queues, ok := v["queues"].([]interface{}); ok {
 				if len(queues) != 0 {