possibly reduce multicast cpu usage even more

This commit is contained in:
Arceliar 2020-05-02 17:23:20 -05:00
parent 127b7e311c
commit 02e1cb180d
5 changed files with 32 additions and 21 deletions

View File

@ -36,6 +36,8 @@ type Multicast struct {
isOpen bool isOpen bool
monitor *time.Timer monitor *time.Timer
platformhandler *time.Timer platformhandler *time.Timer
_interfaces map[string]net.Interface
_interfaceAddrs map[string][]net.Addr
} }
type multicastInterface struct { type multicastInterface struct {
@ -100,8 +102,8 @@ func (m *Multicast) _start() error {
m.isOpen = true m.isOpen = true
go m.listen() go m.listen()
m.Act(m, m.multicastStarted) m.Act(nil, m._multicastStarted)
m.Act(m, m.monitorInterfaceChanges) m.Act(nil, m._monitorInterfaceChanges)
return nil return nil
} }
@ -145,7 +147,7 @@ func (m *Multicast) _stop() error {
// and then signals the various module goroutines to reconfigure themselves if // and then signals the various module goroutines to reconfigure themselves if
// needed. // needed.
func (m *Multicast) UpdateConfig(config *config.NodeConfig) { func (m *Multicast) UpdateConfig(config *config.NodeConfig) {
m.Act(m, func() { m._updateConfig(config) }) m.Act(nil, func() { m._updateConfig(config) })
} }
func (m *Multicast) _updateConfig(config *config.NodeConfig) { func (m *Multicast) _updateConfig(config *config.NodeConfig) {
@ -167,17 +169,14 @@ func (m *Multicast) _updateConfig(config *config.NodeConfig) {
m.log.Debugln("Reloaded multicast configuration successfully") m.log.Debugln("Reloaded multicast configuration successfully")
} }
func (m *Multicast) monitorInterfaceChanges() { func (m *Multicast) _monitorInterfaceChanges() {
interfaces := m.Interfaces() m._updateInterfaces() // update interfaces and interfaceAddrs
// Look for interfaces we don't know about yet. // Look for interfaces we don't know about yet.
for name, intf := range interfaces { for name, intf := range m._interfaces {
if _, ok := m.listeners[name]; !ok { if _, ok := m.listeners[name]; !ok {
// Look up interface addresses. // Look up interface addresses.
addrs, err := intf.Addrs() addrs := m._interfaceAddrs[intf.Name]
if err != nil {
continue
}
// Find the first link-local address. // Find the first link-local address.
for _, addr := range addrs { for _, addr := range addrs {
addrIP, _, _ := net.ParseCIDR(addr.String()) addrIP, _, _ := net.ParseCIDR(addr.String())
@ -198,7 +197,7 @@ func (m *Multicast) monitorInterfaceChanges() {
stop: make(chan interface{}), stop: make(chan interface{}),
zone: name, zone: name,
} }
multicastInterface.Act(multicastInterface, multicastInterface.announce) multicastInterface.Act(m, multicastInterface._announce)
m.listeners[name] = multicastInterface m.listeners[name] = multicastInterface
m.log.Debugln("Started multicasting on", name) m.log.Debugln("Started multicasting on", name)
break break
@ -207,7 +206,7 @@ func (m *Multicast) monitorInterfaceChanges() {
} }
// Look for interfaces we knew about but are no longer there. // Look for interfaces we knew about but are no longer there.
for name, intf := range m.listeners { for name, intf := range m.listeners {
if _, ok := interfaces[name]; !ok { if _, ok := m._interfaces[name]; !ok {
// This is a disappeared interface. Stop the announcer. // This is a disappeared interface. Stop the announcer.
close(intf.stop) close(intf.stop)
delete(m.listeners, name) delete(m.listeners, name)
@ -216,11 +215,11 @@ func (m *Multicast) monitorInterfaceChanges() {
} }
// Queue the next check. // Queue the next check.
m.monitor = time.AfterFunc(time.Second, func() { m.monitor = time.AfterFunc(time.Second, func() {
m.Act(m, m.monitorInterfaceChanges) m.Act(nil, m._monitorInterfaceChanges)
}) })
} }
func (m *multicastInterface) announce() { func (m *multicastInterface) _announce() {
// Check if the multicast interface has been stopped. This will happen // Check if the multicast interface has been stopped. This will happen
// if it disappears from the system or goes down. // if it disappears from the system or goes down.
select { select {
@ -240,7 +239,7 @@ func (m *multicastInterface) announce() {
m.interval += time.Second m.interval += time.Second
} }
m.timer = time.AfterFunc(m.interval, func() { m.timer = time.AfterFunc(m.interval, func() {
m.Act(m, m.announce) m.Act(nil, m._announce)
}) })
} }
@ -248,6 +247,14 @@ func (m *multicastInterface) announce() {
// expected that UpdateInterfaces has been called at least once before calling // expected that UpdateInterfaces has been called at least once before calling
// this method. // this method.
func (m *Multicast) Interfaces() map[string]net.Interface { func (m *Multicast) Interfaces() map[string]net.Interface {
var interfaces map[string]net.Interface
phony.Block(m, func() {
interfaces = m._interfaces
})
return interfaces
}
func (m *Multicast) _updateInterfaces() {
interfaces := make(map[string]net.Interface) interfaces := make(map[string]net.Interface)
// Get interface expressions from config // Get interface expressions from config
current := m.config.GetCurrent() current := m.config.GetCurrent()
@ -258,6 +265,7 @@ func (m *Multicast) Interfaces() map[string]net.Interface {
panic(err) panic(err)
} }
// Work out which interfaces to announce on // Work out which interfaces to announce on
interfaceAddrs := make(map[string][]net.Addr)
for _, iface := range allifaces { for _, iface := range allifaces {
if iface.Flags&net.FlagUp == 0 { if iface.Flags&net.FlagUp == 0 {
// Ignore interfaces that are down // Ignore interfaces that are down
@ -293,10 +301,12 @@ func (m *Multicast) Interfaces() map[string]net.Interface {
// Does the interface match the regular expression? Store it if so // Does the interface match the regular expression? Store it if so
if e.MatchString(iface.Name) { if e.MatchString(iface.Name) {
interfaces[iface.Name] = iface interfaces[iface.Name] = iface
interfaceAddrs[iface.Name] = addrs
} }
} }
} }
return interfaces m._interfaces = interfaces
m._interfaceAddrs = interfaceAddrs
} }
func (m *Multicast) listen() { func (m *Multicast) listen() {
@ -333,6 +343,7 @@ func (m *Multicast) listen() {
if addr.IP.String() != from.IP.String() { if addr.IP.String() != from.IP.String() {
continue continue
} }
// Note that m.Interfaces would block if it was being run by the actor itself
if _, ok := m.Interfaces()[from.Zone]; ok { if _, ok := m.Interfaces()[from.Zone]; ok {
addr.Zone = "" addr.Zone = ""
if err := m.core.CallPeer("tcp://"+addr.String(), from.Zone); err != nil { if err := m.core.CallPeer("tcp://"+addr.String(), from.Zone); err != nil {

View File

@ -31,7 +31,7 @@ import (
var awdlGoroutineStarted bool var awdlGoroutineStarted bool
func (m *Multicast) multicastStarted() { func (m *Multicast) _multicastStarted() {
C.StopAWDLBrowsing() C.StopAWDLBrowsing()
for intf := range m.Interfaces() { for intf := range m.Interfaces() {
if intf == "awdl0" { if intf == "awdl0" {
@ -40,7 +40,7 @@ func (m *Multicast) multicastStarted() {
} }
} }
m.platformhandler = time.AfterFunc(time.Minute, func() { m.platformhandler = time.AfterFunc(time.Minute, func() {
m.Act(m, m.multicastStarted) m.Act(m, m._multicastStarted)
}) })
} }

View File

@ -4,7 +4,7 @@ package multicast
import "syscall" import "syscall"
func (m *Multicast) multicastStarted() { func (m *Multicast) _multicastStarted() {
} }

View File

@ -5,7 +5,7 @@ package multicast
import "syscall" import "syscall"
import "golang.org/x/sys/unix" import "golang.org/x/sys/unix"
func (m *Multicast) multicastStarted() { func (m *Multicast) _multicastStarted() {
} }

View File

@ -5,7 +5,7 @@ package multicast
import "syscall" import "syscall"
import "golang.org/x/sys/windows" import "golang.org/x/sys/windows"
func (m *Multicast) multicastStarted() { func (m *Multicast) _multicastStarted() {
} }