Skip to content

Commit

Permalink
Improve network interface handling for multicast send and receive
Browse files Browse the repository at this point in the history
  • Loading branch information
g3force committed Nov 3, 2024
1 parent 6ac4fd9 commit f13d823
Show file tree
Hide file tree
Showing 2 changed files with 59 additions and 10 deletions.
17 changes: 15 additions & 2 deletions internal/app/publish/publisher.go
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,19 @@ func (p *Publisher) disconnect() {
p.conns = []*net.UDPConn{}
}

func (p *Publisher) disconnectConn(conn *net.UDPConn) {
if err := conn.Close(); err != nil {
log.Println("Failed to close referee connection:", err)
}

for i, c := range p.conns {
if c == conn {
p.conns = append(p.conns[:i], p.conns[i+1:]...)
break
}
}
}

func (p *Publisher) SendMessage(refereeMsg *state.Referee) {
if len(p.conns) == 0 && !p.connect() {
return
Expand All @@ -88,8 +101,8 @@ func (p *Publisher) SendMessage(refereeMsg *state.Referee) {
for _, conn := range p.conns {
_, err = conn.Write(bytes)
if err != nil {
log.Println("Could not write referee message:", err)
p.disconnect()
log.Printf("Could not write referee message on %v: %v", conn, err)
p.disconnectConn(conn)
}
}
}
Expand Down
52 changes: 44 additions & 8 deletions pkg/sslnet/multicast_server.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package sslnet
import (
"log"
"net"
"net/netip"
"sync"
"time"
)
Expand Down Expand Up @@ -40,11 +41,20 @@ func (r *MulticastServer) Stop() {

func (r *MulticastServer) receive(multicastAddress string) {
var currentIfiIdx = 0
var badIfis []string
for r.isRunning() {
ifis := r.interfaces()
ifis := r.interfaces(badIfis)
if len(ifis) == 0 {
log.Println("No valid interfaces found to listen on")
return
}
currentIfiIdx = currentIfiIdx % len(ifis)
ifi := ifis[currentIfiIdx]
r.receiveOnInterface(multicastAddress, ifi)
badIfi := r.receiveOnInterface(multicastAddress, ifi)
if badIfi {
badIfis = append(badIfis, ifi.Name)
}

currentIfiIdx++
if currentIfiIdx >= len(ifis) {
// cycled though all interfaces once, make a short break to avoid producing endless log messages
Expand All @@ -59,41 +69,65 @@ func (r *MulticastServer) isRunning() bool {
return r.running
}

func (r *MulticastServer) interfaces() (interfaces []net.Interface) {
func (r *MulticastServer) interfaces(ignoreIfis []string) (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) {
ifi.Flags&net.FlagUp == 0 || // Not up
ifi.Flags&net.FlagRunning == 0 || // Not running
r.skipInterface(r.SkipInterfaces, ifi.Name) ||
r.skipInterface(ignoreIfis, ifi.Name) ||
isInvalidInterface(&ifi) {
continue
}
interfaces = append(interfaces, ifi)
}
return
}

func (r *MulticastServer) skipInterface(ifiName string) bool {
for _, skipIfi := range r.SkipInterfaces {
func isInvalidInterface(ifi *net.Interface) bool {
addrs, err := ifi.Addrs()
if err != nil {
log.Printf("Could not get addresses of interface %v: %v", ifi, err)
return false
}
for _, addr := range addrs {
ipNet := addr.(*net.IPNet)
ip := netip.MustParseAddr(ipNet.IP.String())
if ip.Is4() {
return false
}
}
return true
}

func (r *MulticastServer) skipInterface(ifis []string, ifiName string) bool {
for _, skipIfi := range ifis {
if skipIfi == ifiName {
return true
}
}
return false
}

func (r *MulticastServer) receiveOnInterface(multicastAddress string, ifi net.Interface) {
func (r *MulticastServer) receiveOnInterface(multicastAddress string, ifi net.Interface) (badIfi bool) {
badIfi = false

addr, err := net.ResolveUDPAddr("udp", multicastAddress)
if err != nil {
log.Printf("Could resolve multicast address %v: %v", multicastAddress, err)
badIfi = true
return
}

r.connection, err = net.ListenMulticastUDP("udp", &ifi, addr)
if err != nil {
log.Printf("Could not listen at %v: %v", multicastAddress, err)
log.Printf("Could not listen at %v (%v): %v", multicastAddress, ifi.Name, err)
badIfi = true
return
}

Expand All @@ -109,6 +143,7 @@ func (r *MulticastServer) receiveOnInterface(multicastAddress string, ifi net.In
data := make([]byte, maxDatagramSize)
for {
if err := r.connection.SetDeadline(time.Now().Add(300 * time.Millisecond)); err != nil {
badIfi = true
log.Println("Could not set deadline on connection: ", err)
}
n, _, err := r.connection.ReadFromUDP(data)
Expand All @@ -132,6 +167,7 @@ func (r *MulticastServer) receiveOnInterface(multicastAddress string, ifi net.In
}

if err := r.connection.Close(); err != nil {
badIfi = true
log.Println("Could not close listener: ", err)
}
return
Expand Down

0 comments on commit f13d823

Please sign in to comment.