Migrate to new multicast server with more robust interface selection

Author: g3force

cmd/ssl-game-controller/main.go

@@ -19,6 +19,7 @@ var trackerAddress = flag.String("trackerAddress", "", "The address (ip+port) fr
 var publishAddress = flag.String("publishAddress", "", "The address (ip+port) to which referee command should be sent")
 var timeAcquisitionMode = flag.String("timeAcquisitionMode", "", "The time acquisitionMode to use (system, ci, vision)")
 var skipInterfaces = flag.String("skipInterfaces", "", "Comma separated list of interface names to ignore when receiving multicast packets")
+var verbose = flag.Bool("verbose", false, "Verbose output")
 
 const configFileName = "config/ssl-game-controller.yaml"
 
@@ -54,6 +55,7 @@ func setupGameController() {
 	}
 
 	gameController := gc.NewGameController(cfg)
+	gameController.SetVerbose(*verbose)
 	gameController.Start()
 
 	sigs := make(chan os.Signal, 1)

internal/app/gc/gc.go

@@ -49,13 +49,18 @@ func NewGameController(cfg config.Controller) (c *GameController) {
 	c.ciServer = ci.NewServer(cfg.Server.Ci.Address)
 	c.visionReceiver = vision.NewReceiver(cfg.Network.VisionAddress)
 	c.visionReceiver.GeometryCallback = c.gcEngine.ProcessGeometry
-	c.visionReceiver.MulticastReceiver.SkipInterfaces = cfg.Network.SkipInterfaces
+	c.visionReceiver.MulticastServer.SkipInterfaces = cfg.Network.SkipInterfaces
 	c.trackerReceiver = tracker.NewReceiver(cfg.Network.TrackerAddress)
 	c.trackerReceiver.Callback = c.gcEngine.ProcessTrackerFrame
-	c.trackerReceiver.MulticastReceiver.SkipInterfaces = cfg.Network.SkipInterfaces
+	c.trackerReceiver.MulticastServer.SkipInterfaces = cfg.Network.SkipInterfaces
 	return
 }
 
+func (c *GameController) SetVerbose(verbose bool) {
+	c.visionReceiver.MulticastServer.Verbose = verbose
+	c.trackerReceiver.MulticastServer.Verbose = verbose
+}
+
 // Start starts all go routines
 func (c *GameController) Start() {
 

internal/app/tracker/trackerReceiver.go

@@ -8,29 +8,29 @@ import (
 )
 
 type Receiver struct {
-	address           string
-	Callback          func(*TrackerWrapperPacket)
-	mutex             sync.Mutex
-	MulticastReceiver *sslnet.MulticastReceiver
+	address         string
+	Callback        func(*TrackerWrapperPacket)
+	mutex           sync.Mutex
+	MulticastServer *sslnet.MulticastServer
 }
 
 // NewReceiver creates a new receiver
 func NewReceiver(address string) (v *Receiver) {
 	v = new(Receiver)
 	v.address = address
 	v.Callback = func(*TrackerWrapperPacket) {}
-	v.MulticastReceiver = sslnet.NewMulticastReceiver(v.consumeData)
+	v.MulticastServer = sslnet.NewMulticastServer(v.consumeData)
 	return
 }
 
 // Start starts the receiver
 func (v *Receiver) Start() {
-	v.MulticastReceiver.Start(v.address)
+	v.MulticastServer.Start(v.address)
 }
 
 // Stop stops the receiver
 func (v *Receiver) Stop() {
-	v.MulticastReceiver.Stop()
+	v.MulticastServer.Stop()
 }
 
 func (v *Receiver) consumeData(data []byte) {

internal/app/vision/visionReceiver.go

@@ -15,7 +15,7 @@ type Receiver struct {
 	GeometryCallback  func(*SSL_GeometryData)
 	latestTimestamp   time.Time
 	mutex             sync.Mutex
-	MulticastReceiver *sslnet.MulticastReceiver
+	MulticastServer   *sslnet.MulticastServer
 }
 
 // NewReceiver creates a new receiver
@@ -24,18 +24,18 @@ func NewReceiver(address string) (v *Receiver) {
 	v.address = address
 	v.DetectionCallback = func(*SSL_DetectionFrame) {}
 	v.GeometryCallback = func(data *SSL_GeometryData) {}
-	v.MulticastReceiver = sslnet.NewMulticastReceiver(v.consumeData)
+	v.MulticastServer = sslnet.NewMulticastServer(v.consumeData)
 	return
 }
 
 // Start starts the receiver
 func (v *Receiver) Start() {
-	v.MulticastReceiver.Start(v.address)
+	v.MulticastServer.Start(v.address)
 }
 
 // Stop stops the receiver
 func (v *Receiver) Stop() {
-	v.MulticastReceiver.Stop()
+	v.MulticastServer.Stop()
 }
 
 func (v *Receiver) consumeData(data []byte) {

pkg/sslnet/multicast_receiver.go

@@ -0,0 +1,138 @@
+package sslnet
+
+import (
+	"log"
+	"net"
+	"sync"
+	"time"
+)
+
+const maxDatagramSize = 8192
+
+type MulticastServer struct {
+	connection     *net.UDPConn
+	running        bool
+	consumer       func([]byte)
+	mutex          sync.Mutex
+	SkipInterfaces []string
+	Verbose        bool
+}
+
+func NewMulticastServer(consumer func([]byte)) (r *MulticastServer) {
+	r = new(MulticastServer)
+	r.consumer = consumer
+	return
+}
+
+func (r *MulticastServer) Start(multicastAddress string) {
+	r.running = true
+	go r.receive(multicastAddress)
+}
+
+func (r *MulticastServer) Stop() {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	r.running = false
+	if err := r.connection.Close(); err != nil {
+		log.Println("Could not close connection: ", err)
+	}
+}
+
+func (r *MulticastServer) receive(multicastAddress string) {
+	var currentIfiIdx = 0
+	for r.isRunning() {
+		ifis := r.interfaces()
+		currentIfiIdx = currentIfiIdx % len(ifis)
+		ifi := ifis[currentIfiIdx]
+		r.receiveOnInterface(multicastAddress, ifi)
+		currentIfiIdx++
+		if currentIfiIdx >= len(ifis) {
+			// cycled though all interfaces once, make a short break to avoid producing endless log messages
+			time.Sleep(1 * time.Second)
+		}
+	}
+}
+
+func (r *MulticastServer) isRunning() bool {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	return r.running
+}
+
+func (r *MulticastServer) interfaces() (interfaces []net.Interface) {
+	interfaces = []net.Interface{}
+	ifis, err := net.Interfaces()
+	if err != nil {
+		log.Println("Could not get available interfaces: ", err)
+	}
+	for _, ifi := range ifis {
+		if ifi.Flags&net.FlagMulticast == 0 || // No multicast support
+			r.skipInterface(ifi.Name) {
+			continue
+		}
+		interfaces = append(interfaces, ifi)
+	}
+	return
+}
+
+func (r *MulticastServer) skipInterface(ifiName string) bool {
+	for _, skipIfi := range r.SkipInterfaces {
+		if skipIfi == ifiName {
+			return true
+		}
+	}
+	return false
+}
+
+func (r *MulticastServer) receiveOnInterface(multicastAddress string, ifi net.Interface) {
+	addr, err := net.ResolveUDPAddr("udp", multicastAddress)
+	if err != nil {
+		log.Printf("Could resolve multicast address %v: %v", multicastAddress, err)
+		return
+	}
+
+	r.connection, err = net.ListenMulticastUDP("udp", &ifi, addr)
+	if err != nil {
+		log.Printf("Could not listen at %v: %v", multicastAddress, err)
+		return
+	}
+
+	if err := r.connection.SetReadBuffer(maxDatagramSize); err != nil {
+		log.Println("Could not set read buffer: ", err)
+	}
+
+	if r.Verbose {
+		log.Printf("Listening on %s (%s)", multicastAddress, ifi.Name)
+	}
+
+	first := true
+	data := make([]byte, maxDatagramSize)
+	for {
+		if err := r.connection.SetDeadline(time.Now().Add(300 * time.Millisecond)); err != nil {
+			log.Println("Could not set deadline on connection: ", err)
+		}
+		n, _, err := r.connection.ReadFromUDP(data)
+		if err != nil {
+			if r.Verbose {
+				log.Println("ReadFromUDP failed:", err)
+			}
+			break
+		}
+
+		if first && r.Verbose {
+			log.Printf("Got first data packets from %s (%s)", multicastAddress, ifi.Name)
+			first = false
+		}
+
+		r.consumer(data[:n])
+	}
+
+	if r.Verbose {
+		log.Printf("Stop listening on %s (%s)", multicastAddress, ifi.Name)
+	}
+
+	if err := r.connection.Close(); err != nil {
+		log.Println("Could not close listener: ", err)
+	}
+	return
+}