Merge pull request #700 from Arceliar/multicast

Multicast

src/multicast/multicast.go

@@ -15,57 +15,45 @@ import (
 	"golang.org/x/net/ipv6"
 )
 
-const (
-	// GroupAddr contains the multicast group and port used for multicast packets.
-	GroupAddr = "[ff02::114]:9001"
-)
-
 // Multicast represents the multicast advertisement and discovery mechanism used
 // by Yggdrasil to find peers on the same subnet. When a beacon is received on a
 // configured multicast interface, Yggdrasil will attempt to peer with that node
 // automatically.
 type Multicast struct {
 	phony.Inbox
-	core            *yggdrasil.Core
-	config          *config.NodeState
-	log             *log.Logger
-	sock            *ipv6.PacketConn
-	groupAddr       *net.UDPAddr
-	listeners       map[string]*multicastInterface
-	listenPort      uint16
-	isOpen          bool
-	monitor         *time.Timer
-	platformhandler *time.Timer
-	_interfaces     map[string]net.Interface
-	_interfaceAddrs map[string]addrInfo
+	core        *yggdrasil.Core
+	config      *config.NodeState
+	log         *log.Logger
+	sock        *ipv6.PacketConn
+	groupAddr   string
+	listeners   map[string]*listenerInfo
+	listenPort  uint16
+	isOpen      bool
+	_interfaces map[string]interfaceInfo
 }
 
-type addrInfo struct {
+type interfaceInfo struct {
+	iface net.Interface
 	addrs []net.Addr
-	time  time.Time
 }
 
-type multicastInterface struct {
-	phony.Inbox
-	sock     *ipv6.PacketConn
-	destAddr net.UDPAddr
+type listenerInfo struct {
 	listener *yggdrasil.TcpListener
-	zone     string
-	timer    *time.Timer
+	time     time.Time
 	interval time.Duration
-	stop     chan interface{}
 }
 
 // Init prepares the multicast interface for use.
-func (m *Multicast) Init(core *yggdrasil.Core, state *config.NodeState, log *log.Logger, options interface{}) (err error) {
+func (m *Multicast) Init(core *yggdrasil.Core, state *config.NodeState, log *log.Logger, options interface{}) error {
 	m.core = core
 	m.config = state
 	m.log = log
-	m.listeners = make(map[string]*multicastInterface)
+	m.listeners = make(map[string]*listenerInfo)
+	m._interfaces = make(map[string]interfaceInfo)
 	current := m.config.GetCurrent()
 	m.listenPort = current.LinkLocalTCPPort
-	m.groupAddr, err = net.ResolveUDPAddr("udp6", GroupAddr)
-	return
+	m.groupAddr = "[ff02::114]:9001"
+	return nil
 }
 
 // Start starts the multicast interface. This launches goroutines which will
@@ -88,7 +76,7 @@ func (m *Multicast) _start() error {
 		return nil
 	}
 	m.log.Infoln("Starting multicast module")
-	addr, err := net.ResolveUDPAddr("udp", GroupAddr)
+	addr, err := net.ResolveUDPAddr("udp", m.groupAddr)
 	if err != nil {
 		return err
 	}
@@ -108,7 +96,7 @@ func (m *Multicast) _start() error {
 	m.isOpen = true
 	go m.listen()
 	m.Act(nil, m._multicastStarted)
-	m.Act(nil, m._monitorInterfaceChanges)
+	m.Act(nil, m._announce)
 
 	return nil
 }
@@ -135,13 +123,6 @@ func (m *Multicast) Stop() error {
 func (m *Multicast) _stop() error {
 	m.log.Infoln("Stopping multicast module")
 	m.isOpen = false
-	for name := range m.listeners {
-		close(m.listeners[name].stop)
-		delete(m.listeners, name)
-	}
-	if m.platformhandler != nil {
-		m.platformhandler.Stop()
-	}
 	if m.sock != nil {
 		m.sock.Close()
 	}
@@ -174,92 +155,35 @@ func (m *Multicast) _updateConfig(config *config.NodeConfig) {
 	m.log.Debugln("Reloaded multicast configuration successfully")
 }
 
-func (m *Multicast) _monitorInterfaceChanges() {
-	m._updateInterfaces() // update interfaces and interfaceAddrs
-
-	// Look for interfaces we don't know about yet.
-	for name, intf := range m._interfaces {
-		if _, ok := m.listeners[name]; !ok {
-			// Look up interface addresses.
-			addrs := m._interfaceAddrs[intf.Name].addrs
-			// Find the first link-local address.
-			for _, addr := range addrs {
-				addrIP, _, _ := net.ParseCIDR(addr.String())
-				// Join the multicast group.
-				m.sock.JoinGroup(&intf, m.groupAddr)
-				// Construct a listener on this address.
-				listenaddr := fmt.Sprintf("[%s%%%s]:%d", addrIP, intf.Name, m.listenPort)
-				listener, err := m.core.ListenTCP(listenaddr)
-				if err != nil {
-					m.log.Warnln("Not multicasting on", name, "due to error:", err)
-					continue
-				}
-				// This is a new interface. Start an announcer for it.
-				multicastInterface := &multicastInterface{
-					sock:     m.sock,
-					destAddr: *m.groupAddr,
-					listener: listener,
-					stop:     make(chan interface{}),
-					zone:     name,
-				}
-				multicastInterface.Act(m, multicastInterface._announce)
-				m.listeners[name] = multicastInterface
-				m.log.Debugln("Started multicasting on", name)
-				break
-			}
+func (m *Multicast) _updateInterfaces() {
+	interfaces := make(map[string]interfaceInfo)
+	intfs := m.getAllowedInterfaces()
+	for _, intf := range intfs {
+		addrs, err := intf.Addrs()
+		if err != nil {
+			m.log.Warnf("Failed up get addresses for interface %s: %s", intf.Name, err)
+			continue
+		}
+		interfaces[intf.Name] = interfaceInfo{
+			iface: intf,
+			addrs: addrs,
 		}
 	}
-	// Look for interfaces we knew about but are no longer there.
-	for name, intf := range m.listeners {
-		if _, ok := m._interfaces[name]; !ok {
-			// This is a disappeared interface. Stop the announcer.
-			close(intf.stop)
-			delete(m.listeners, name)
-			m.log.Debugln("Stopped multicasting on", name)
-		}
-	}
-	// Queue the next check.
-	m.monitor = time.AfterFunc(time.Second, func() {
-		m.Act(nil, m._monitorInterfaceChanges)
-	})
+	m._interfaces = interfaces
 }
 
-func (m *multicastInterface) _announce() {
-	// Check if the multicast interface has been stopped. This will happen
-	// if it disappears from the system or goes down.
-	select {
-	case <-m.stop:
-		return
-	default:
-	}
-	// Send the beacon.
-	lladdr := m.listener.Listener.Addr().String()
-	if a, err := net.ResolveTCPAddr("tcp6", lladdr); err == nil {
-		a.Zone = ""
-		msg := []byte(a.String())
-		m.sock.WriteTo(msg, nil, &m.destAddr)
-	}
-	// Queue the next beacon.
-	if m.interval.Seconds() < 15 {
-		m.interval += time.Second
-	}
-	m.timer = time.AfterFunc(m.interval, func() {
-		m.Act(nil, m._announce)
-	})
-}
-
-// GetInterfaces returns the currently known/enabled multicast interfaces. It is
-// expected that UpdateInterfaces has been called at least once before calling
-// this method.
 func (m *Multicast) Interfaces() map[string]net.Interface {
-	var interfaces map[string]net.Interface
+	interfaces := make(map[string]net.Interface)
 	phony.Block(m, func() {
-		interfaces = m._interfaces
+		for _, info := range m._interfaces {
+			interfaces[info.iface.Name] = info.iface
+		}
 	})
 	return interfaces
 }
 
-func (m *Multicast) _updateInterfaces() {
+// getAllowedInterfaces returns the currently known/enabled multicast interfaces.
+func (m *Multicast) getAllowedInterfaces() map[string]net.Interface {
 	interfaces := make(map[string]net.Interface)
 	// Get interface expressions from config
 	current := m.config.GetCurrent()
@@ -270,7 +194,6 @@ func (m *Multicast) _updateInterfaces() {
 		panic(err)
 	}
 	// Work out which interfaces to announce on
-	interfaceAddrs := make(map[string]addrInfo)
 	for _, iface := range allifaces {
 		if iface.Flags&net.FlagUp == 0 {
 			// Ignore interfaces that are down
@@ -284,26 +207,6 @@ func (m *Multicast) _updateInterfaces() {
 			// Ignore point-to-point interfaces
 			continue
 		}
-		var aInfo addrInfo
-		var isIn bool
-		if aInfo, isIn = m._interfaceAddrs[iface.Name]; isIn && time.Since(aInfo.time) < time.Minute {
-			// don't call iface.Addrs, it's unlikely things have changed
-		} else {
-			aInfo.addrs, _ = iface.Addrs()
-			aInfo.time = time.Now()
-		}
-		lladdrs := aInfo.addrs[:0]
-		for _, addr := range aInfo.addrs {
-			addrIP, _, _ := net.ParseCIDR(addr.String())
-			if addrIP.To4() == nil && addrIP.IsLinkLocalUnicast() {
-				lladdrs = append(lladdrs, addr)
-			}
-		}
-		aInfo.addrs = lladdrs
-		if len(lladdrs) == 0 {
-			// Ignore interfaces without link-local addresses
-			continue
-		}
 		for _, expr := range exprs {
 			// Compile each regular expression
 			e, err := regexp.Compile(expr)
@@ -313,16 +216,132 @@ func (m *Multicast) _updateInterfaces() {
 			// Does the interface match the regular expression? Store it if so
 			if e.MatchString(iface.Name) {
 				interfaces[iface.Name] = iface
-				interfaceAddrs[iface.Name] = aInfo
 			}
 		}
 	}
-	m._interfaces = interfaces
-	m._interfaceAddrs = interfaceAddrs
+	return interfaces
+}
+
+func (m *Multicast) _announce() {
+	if !m.isOpen {
+		return
+	}
+	m._updateInterfaces()
+	groupAddr, err := net.ResolveUDPAddr("udp6", m.groupAddr)
+	if err != nil {
+		panic(err)
+	}
+	destAddr, err := net.ResolveUDPAddr("udp6", m.groupAddr)
+	if err != nil {
+		panic(err)
+	}
+	// There might be interfaces that we configured listeners for but are no
+	// longer up - if that's the case then we should stop the listeners
+	for name, info := range m.listeners {
+		// Prepare our stop function!
+		stop := func() {
+			info.listener.Stop()
+			delete(m.listeners, name)
+			m.log.Debugln("No longer multicasting on", name)
+		}
+		// If the interface is no longer visible on the system then stop the
+		// listener, as another one will be started further down
+		if _, ok := m._interfaces[name]; !ok {
+			stop()
+			continue
+		}
+		// It's possible that the link-local listener address has changed so if
+		// that is the case then we should clean up the interface listener
+		found := false
+		listenaddr, err := net.ResolveTCPAddr("tcp6", info.listener.Listener.Addr().String())
+		if err != nil {
+			stop()
+			continue
+		}
+		// Find the interface that matches the listener
+		if intf, err := net.InterfaceByName(name); err == nil {
+			if addrs, err := intf.Addrs(); err == nil {
+				// Loop through the addresses attached to that listener and see if any
+				// of them match the current address of the listener
+				for _, addr := range addrs {
+					if ip, _, err := net.ParseCIDR(addr.String()); err == nil {
+						// Does the interface address match our listener address?
+						if ip.Equal(listenaddr.IP) {
+							found = true
+							break
+						}
+					}
+				}
+			}
+		}
+		// If the address has not been found on the adapter then we should stop
+		// and clean up the TCP listener. A new one will be created below if a
+		// suitable link-local address is found
+		if !found {
+			stop()
+		}
+	}
+	// Now that we have a list of valid interfaces from the operating system,
+	// we can start checking if we can send multicasts on them
+	for _, info := range m._interfaces {
+		iface := info.iface
+		for _, addr := range info.addrs {
+			addrIP, _, _ := net.ParseCIDR(addr.String())
+			// Ignore IPv4 addresses
+			if addrIP.To4() != nil {
+				continue
+			}
+			// Ignore non-link-local addresses
+			if !addrIP.IsLinkLocalUnicast() {
+				continue
+			}
+			// Join the multicast group
+			m.sock.JoinGroup(&iface, groupAddr)
+			// Try and see if we already have a TCP listener for this interface
+			var info *listenerInfo
+			if nfo, ok := m.listeners[iface.Name]; !ok || nfo.listener.Listener == nil {
+				// No listener was found - let's create one
+				listenaddr := fmt.Sprintf("[%s%%%s]:%d", addrIP, iface.Name, m.listenPort)
+				if li, err := m.core.ListenTCP(listenaddr); err == nil {
+					m.log.Debugln("Started multicasting on", iface.Name)
+					// Store the listener so that we can stop it later if needed
+					info = &listenerInfo{listener: li, time: time.Now()}
+					m.listeners[iface.Name] = info
+				} else {
+					m.log.Warnln("Not multicasting on", iface.Name, "due to error:", err)
+				}
+			} else {
+				// An existing listener was found
+				info = m.listeners[iface.Name]
+			}
+			// Make sure nothing above failed for some reason
+			if info == nil {
+				continue
+			}
+			if time.Since(info.time) < info.interval {
+				continue
+			}
+			// Get the listener details and construct the multicast beacon
+			lladdr := info.listener.Listener.Addr().String()
+			if a, err := net.ResolveTCPAddr("tcp6", lladdr); err == nil {
+				a.Zone = ""
+				destAddr.Zone = iface.Name
+				msg := []byte(a.String())
+				m.sock.WriteTo(msg, nil, destAddr)
+			}
+			if info.interval.Seconds() < 15 {
+				info.interval += time.Second
+			}
+			break
+		}
+	}
+	time.AfterFunc(time.Second, func() {
+		m.Act(nil, m._announce)
+	})
 }
 
 func (m *Multicast) listen() {
-	groupAddr, err := net.ResolveUDPAddr("udp6", GroupAddr)
+	groupAddr, err := net.ResolveUDPAddr("udp6", m.groupAddr)
 	if err != nil {
 		panic(err)
 	}
@@ -355,8 +374,11 @@ func (m *Multicast) listen() {
 		if addr.IP.String() != from.IP.String() {
 			continue
 		}
-		// Note that m.Interfaces would block if it was being run by the actor itself
-		if _, ok := m.Interfaces()[from.Zone]; ok {
+		var interfaces map[string]interfaceInfo
+		phony.Block(m, func() {
+			interfaces = m._interfaces
+		})
+		if _, ok := interfaces[from.Zone]; ok {
 			addr.Zone = ""
 			if err := m.core.CallPeer("tcp://"+addr.String(), from.Zone); err != nil {
 				m.log.Debugln("Call from multicast failed:", err)

src/multicast/multicast_darwin.go

@@ -32,6 +32,9 @@ import (
 var awdlGoroutineStarted bool
 
 func (m *Multicast) _multicastStarted() {
+	if !m.isOpen {
+		return
+	}
 	C.StopAWDLBrowsing()
 	for intf := range m._interfaces {
 		if intf == "awdl0" {
@@ -39,7 +42,7 @@ func (m *Multicast) _multicastStarted() {
 			break
 		}
 	}
-	m.platformhandler = time.AfterFunc(time.Minute, func() {
+	time.AfterFunc(time.Minute, func() {
 		m.Act(nil, m._multicastStarted)
 	})
 }