From 5d93a749819f19530f984ce97f29c947da5a1e12 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Mon, 18 Oct 2021 02:12:36 +0200 Subject: [PATCH] Collecting network vars and functions in networks structure --- Bootstrap.go | 12 +++--- Connection.go | 14 +++---- Network Detection.go | 78 +++++++++++++++++++-------------------- Network IPv4 Broadcast.go | 4 +- Network IPv6 Multicast.go | 4 +- Network Init.go | 26 ++++++------- Network UPnP.go | 4 +- Network.go | 41 ++++++++------------ Networks.go | 14 +++++-- Peernet.go | 6 +-- 10 files changed, 98 insertions(+), 105 deletions(-) diff --git a/Bootstrap.go b/Bootstrap.go index 8a3bcbe..6230592 100644 --- a/Bootstrap.go +++ b/Bootstrap.go @@ -165,18 +165,18 @@ func bootstrap() { Filters.LogError("bootstrap", "unable to connect to at least 2 root peers, aborting\n") } -func autoMulticastBroadcast() { +func (nets *Networks) autoMulticastBroadcast() { sendMulticastBroadcast := func() { - networksMutex.RLock() - defer networksMutex.RUnlock() + nets.RLock() + defer nets.RUnlock() - for _, network := range networks6 { + for _, network := range nets.networks6 { if err := network.MulticastIPv6Send(); err != nil { Filters.LogError("autoMulticastBroadcast", "multicast from network address '%s': %v\n", network.address.IP.String(), err.Error()) } } - for _, network := range networks4 { + for _, network := range nets.networks4 { if err := network.BroadcastIPv4Send(); err != nil { Filters.LogError("autoMulticastBroadcast", "broadcast from network address '%s': %v\n", network.address.IP.String(), err.Error()) } @@ -216,7 +216,7 @@ func contactArbitraryPeer(publicKey *btcec.PublicKey, address *net.UDPAddr, rece Filters.MessageOutAnnouncement(publicKey, nil, raw, findSelf, nil, nil, nil) - sendAllNetworks(publicKey, raw, address, receiverPortInternal, nil, &bootstrapFindSelf{}) + networks.sendAllNetworks(publicKey, raw, address, receiverPortInternal, nil, &bootstrapFindSelf{}) return true } diff --git a/Connection.go b/Connection.go index c95743c..92d51c8 100644 --- a/Connection.go +++ b/Connection.go @@ -351,7 +351,7 @@ func (c *Connection) send(packet *protocol.PacketRaw, receiverPublicKey *btcec.P func (peer *PeerInfo) send(packet *protocol.PacketRaw) (err error) { if peer.isVirtual { // special case for peers that were not contacted before for _, address := range peer.targetAddresses { - sendAllNetworks(peer.PublicKey, packet, &net.UDPAddr{IP: address.IP, Port: int(address.Port)}, address.PortInternal, peer.traversePeer, nil) + networks.sendAllNetworks(peer.PublicKey, packet, &net.UDPAddr{IP: address.IP, Port: int(address.Port)}, address.PortInternal, peer.traversePeer, nil) } return } @@ -407,13 +407,13 @@ func (peer *PeerInfo) sendConnection(packet *protocol.PacketRaw, connection *Con // sendAllNetworks sends a raw packet via all networks. It assigns a new sequence for each sent packet. // receiverPortInternal is important for NAT detection and sending the traverse message. -func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *protocol.PacketRaw, remote *net.UDPAddr, receiverPortInternal uint16, traversePeer *PeerInfo, sequenceData interface{}) (err error) { - networksMutex.RLock() - defer networksMutex.RUnlock() +func (nets *Networks) sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *protocol.PacketRaw, remote *net.UDPAddr, receiverPortInternal uint16, traversePeer *PeerInfo, sequenceData interface{}) (err error) { + nets.RLock() + defer nets.RUnlock() - networksTarget := networks4 + networksTarget := nets.networks4 if IsIPv6(remote.IP.To16()) { - networksTarget = networks6 + networksTarget = nets.networks6 } successCount := 0 @@ -426,7 +426,7 @@ func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *protocol.Packet } if sequenceData != nil { - packet.Sequence = networks.Sequences.ArbitrarySequence(receiverPublicKey, sequenceData).SequenceNumber + packet.Sequence = nets.Sequences.ArbitrarySequence(receiverPublicKey, sequenceData).SequenceNumber } err = (&Connection{Network: network, Address: remote, PortInternal: receiverPortInternal, traversePeer: traversePeer}).send(packet, receiverPublicKey, isFirstPacket) isFirstPacket = false diff --git a/Network Detection.go b/Network Detection.go index 29877dd..fe5c09f 100644 --- a/Network Detection.go +++ b/Network Detection.go @@ -93,7 +93,7 @@ func IsNetworkErrorFatal(err error) bool { const changeMonitorFrequency = 10 // networkChangeMonitor() monitors for network changes to act accordingly -func networkChangeMonitor() { +func (nets *Networks) networkChangeMonitor() { // If manual IPs are entered, no need for monitoring for any network changes. if len(config.Listen) > 0 { return @@ -119,9 +119,9 @@ func networkChangeMonitor() { ifacesNew[iface.Name] = addressesNew // was the interface added? - addressesExist, ok := networks.ipListen.ifacesExist[iface.Name] + addressesExist, ok := nets.ipListen.ifacesExist[iface.Name] if !ok { - networkChangeInterfaceNew(iface, addressesNew) + nets.networkChangeInterfaceNew(iface, addressesNew) } else { // new IPs added for this interface? for _, addr := range addressesNew { @@ -134,7 +134,7 @@ func networkChangeMonitor() { } if !exists { - networkChangeIPNew(iface, addr) + nets.networkChangeIPNew(iface, addr) } } @@ -149,30 +149,30 @@ func networkChangeMonitor() { } if removed { - networkChangeIPRemove(iface, exist) + nets.networkChangeIPRemove(iface, exist) } } } } // was an existing interface removed? - for ifaceExist, addressesExist := range networks.ipListen.ifacesExist { + for ifaceExist, addressesExist := range nets.ipListen.ifacesExist { if _, ok := ifacesNew[ifaceExist]; !ok { - networkChangeInterfaceRemove(ifaceExist, addressesExist) + nets.networkChangeInterfaceRemove(ifaceExist, addressesExist) } } - networks.ipListen.ifacesExist = ifacesNew + nets.ipListen.ifacesExist = ifacesNew } } // networkChangeInterfaceNew is called when a new interface is detected -func networkChangeInterfaceNew(iface net.Interface, addresses []net.Addr) { +func (nets *Networks) networkChangeInterfaceNew(iface net.Interface, addresses []net.Addr) { Filters.LogError("networkChangeInterfaceNew", "new interface '%s' (%d IPs)\n", iface.Name, len(addresses)) - networks := networks.InterfaceStart(iface, addresses) + networksNew := nets.InterfaceStart(iface, addresses) - for _, network := range networks { + for _, network := range networksNew { go network.upnpAuto() } @@ -180,44 +180,44 @@ func networkChangeInterfaceNew(iface net.Interface, addresses []net.Addr) { } // networkChangeInterfaceRemove is called when an existing interface is removed -func networkChangeInterfaceRemove(iface string, addresses []net.Addr) { - networksMutex.RLock() - defer networksMutex.RUnlock() +func (nets *Networks) networkChangeInterfaceRemove(iface string, addresses []net.Addr) { + nets.RLock() + defer nets.RUnlock() Filters.LogError("networkChangeInterfaceRemove", "removing interface '%s' (%d IPs)\n", iface, len(addresses)) - for n, network := range networks6 { + for n, network := range nets.networks6 { if network.iface != nil && network.iface.Name == iface { network.Terminate() // remove from list - networksNew := networks6[:n] - if n < len(networks6)-1 { - networksNew = append(networksNew, networks6[n+1:]...) + networksNew := nets.networks6[:n] + if n < len(nets.networks6)-1 { + networksNew = append(networksNew, nets.networks6[n+1:]...) } - networks6 = networksNew + nets.networks6 = networksNew } } - for n, network := range networks4 { + for n, network := range nets.networks4 { if network.iface != nil && network.iface.Name == iface { network.Terminate() // remove from list - networksNew := networks4[:n] - if n < len(networks4)-1 { - networksNew = append(networksNew, networks4[n+1:]...) + networksNew := nets.networks4[:n] + if n < len(nets.networks4)-1 { + networksNew = append(networksNew, nets.networks4[n+1:]...) } - networks4 = networksNew + nets.networks4 = networksNew } } } // networkChangeIPNew is called when an existing interface lists a new IP -func networkChangeIPNew(iface net.Interface, address net.Addr) { +func (nets *Networks) networkChangeIPNew(iface net.Interface, address net.Addr) { Filters.LogError("networkChangeIPNew", "new interface '%s' IP %s\n", iface.Name, address.String()) - networksNew := networks.InterfaceStart(iface, []net.Addr{address}) + networksNew := nets.InterfaceStart(iface, []net.Addr{address}) for _, network := range networksNew { go network.upnpAuto() @@ -227,35 +227,35 @@ func networkChangeIPNew(iface net.Interface, address net.Addr) { } // networkChangeIPRemove is called when an existing interface removes an IP -func networkChangeIPRemove(iface net.Interface, address net.Addr) { - networksMutex.RLock() - defer networksMutex.RUnlock() +func (nets *Networks) networkChangeIPRemove(iface net.Interface, address net.Addr) { + nets.RLock() + defer nets.RUnlock() Filters.LogError("networkChangeIPRemove", "remove interface '%s' IP %s\n", iface.Name, address.String()) - for n, network := range networks6 { + for n, network := range nets.networks6 { if network.address.IP.Equal(address.(*net.IPNet).IP) { network.Terminate() // remove from list - networksNew := networks6[:n] - if n < len(networks6)-1 { - networksNew = append(networksNew, networks6[n+1:]...) + networksNew := nets.networks6[:n] + if n < len(nets.networks6)-1 { + networksNew = append(networksNew, nets.networks6[n+1:]...) } - networks6 = networksNew + nets.networks6 = networksNew } } - for n, network := range networks4 { + for n, network := range nets.networks4 { if network.address.IP.Equal(address.(*net.IPNet).IP) { network.Terminate() // remove from list - networksNew := networks4[:n] - if n < len(networks4)-1 { - networksNew = append(networksNew, networks4[n+1:]...) + networksNew := nets.networks4[:n] + if n < len(nets.networks4)-1 { + networksNew = append(networksNew, nets.networks4[n+1:]...) } - networks4 = networksNew + nets.networks4 = networksNew } } } diff --git a/Network IPv4 Broadcast.go b/Network IPv4 Broadcast.go index 029a471..2b3366d 100644 --- a/Network IPv4 Broadcast.go +++ b/Network IPv4 Broadcast.go @@ -68,7 +68,7 @@ func (network *Network) BroadcastIPv4Listen() { continue } - if networks.ipListen.IsAddressSelf(sender.(*net.UDPAddr)) { + if network.networkGroup.ipListen.IsAddressSelf(sender.(*net.UDPAddr)) { continue } @@ -85,7 +85,7 @@ func (network *Network) BroadcastIPv4Listen() { } // send the packet to a channel which is processed by multiple workers. - rawPacketsIncoming <- networkWire{network: network, sender: sender.(*net.UDPAddr), raw: buffer[:length], receiverPublicKey: ipv4BroadcastPublicKey, unicast: false} + network.networkGroup.rawPacketsIncoming <- networkWire{network: network, sender: sender.(*net.UDPAddr), raw: buffer[:length], receiverPublicKey: ipv4BroadcastPublicKey, unicast: false} } } diff --git a/Network IPv6 Multicast.go b/Network IPv6 Multicast.go index 7ba982f..6113db2 100644 --- a/Network IPv6 Multicast.go +++ b/Network IPv6 Multicast.go @@ -114,7 +114,7 @@ func (network *Network) MulticastIPv6Listen() { } // skip incoming packets that were looped back - if networks.ipListen.IsAddressSelf(sender.(*net.UDPAddr)) { + if network.networkGroup.ipListen.IsAddressSelf(sender.(*net.UDPAddr)) { continue } @@ -131,7 +131,7 @@ func (network *Network) MulticastIPv6Listen() { } // send the packet to a channel which is processed by multiple workers. - rawPacketsIncoming <- networkWire{network: network, sender: sender.(*net.UDPAddr), raw: buffer[:length], receiverPublicKey: ipv6MulticastPublicKey, unicast: false} + network.networkGroup.rawPacketsIncoming <- networkWire{network: network, sender: sender.(*net.UDPAddr), raw: buffer[:length], receiverPublicKey: ipv6MulticastPublicKey, unicast: false} } } diff --git a/Network Init.go b/Network Init.go index 751b072..0872b46 100644 --- a/Network Init.go +++ b/Network Init.go @@ -30,20 +30,16 @@ type networkWire struct { unicast bool // True if the message was sent via unicast. False if sent via IPv4 broadcast or IPv6 multicast. } -var ( - rawPacketsIncoming chan networkWire // channel for processing incoming decoded packets by workers -) - // initNetwork sets up the network configuration and starts listening. func initNetwork() { - rawPacketsIncoming = make(chan networkWire, 1000) // buffer up to 1000 UDP packets before they get buffered by the OS network stack and eventually dropped - rand.Seed(time.Now().UnixNano()) // we are not using "crypto/rand" for speed tradeoff + rand.Seed(time.Now().UnixNano()) // we are not using "crypto/rand" for speed tradeoff + // start listen workers if config.ListenWorkers == 0 { config.ListenWorkers = 2 } for n := 0; n < config.ListenWorkers; n++ { - go packetWorker(rawPacketsIncoming) + go networks.packetWorker() } // check if user specified where to listen @@ -110,7 +106,7 @@ func (nets *Networks) InterfaceStart(iface net.Interface, addresses []net.Addr) continue } - networkNew, err := networks.PrepareListen(net1.IP.String(), 0) + networkNew, err := nets.PrepareListen(net1.IP.String(), 0) if err != nil { // Do not log common errors: @@ -124,7 +120,7 @@ func (nets *Networks) InterfaceStart(iface net.Interface, addresses []net.Addr) continue } - networks.ipListen.Add(networkNew.address) + nets.ipListen.Add(networkNew.address) Filters.LogError("networks.InterfaceStart", "listen on network '%s' UDP %s\n", iface.Name, networkNew.address.String()) @@ -141,7 +137,7 @@ func (nets *Networks) PrepareListen(ipA string, port int) (network *Network, err return nil, errors.New("invalid input IP") } - network = new(Network) + network = &Network{networkGroup: nets} network.terminateSignal = make(chan interface{}) // get the network interface that belongs to the IP @@ -157,16 +153,16 @@ func (nets *Networks) PrepareListen(ipA string, port int) (network *Network, err return nil, err } - networksMutex.Lock() + nets.Lock() // Success - port is open. Add to the list and start accepting incoming messages. if IsIPv4(ip) { - networks4 = append(networks4, network) - networksMutex.Unlock() + nets.networks4 = append(nets.networks4, network) + nets.Unlock() network.BroadcastIPv4() } else { - networks6 = append(networks6, network) - networksMutex.Unlock() + nets.networks6 = append(nets.networks6, network) + nets.Unlock() network.MulticastIPv6Join() } diff --git a/Network UPnP.go b/Network UPnP.go index 41b48d1..1cbfbae 100644 --- a/Network UPnP.go +++ b/Network UPnP.go @@ -20,7 +20,7 @@ import ( var upnpListInterfaces map[string]struct{} var upnpMutex sync.RWMutex -func startUPnP() { +func (nets *Networks) startUPnP() { upnpListInterfaces = make(map[string]struct{}) for _, cidr := range []string{ @@ -40,7 +40,7 @@ func startUPnP() { return } - for _, network := range networks4 { + for _, network := range nets.networks4 { go network.upnpAuto() } } diff --git a/Network.go b/Network.go index 75ae6af..ea69895 100644 --- a/Network.go +++ b/Network.go @@ -34,18 +34,9 @@ type Network struct { isTerminated bool // If true, the network was signaled for termination terminateSignal chan interface{} // gets closed on termination signal, can be used in select via "case _ = <- network.terminateSignal:" sync.RWMutex // for sychronized closing + networkGroup *Networks // Pointer to the pool of networks that this is part of } -// networks is a list of all connected networks -var networks4, networks6 []*Network - -// single mutex for both network lists. Higher granularity currently not needed. -var networksMutex sync.RWMutex - -// countListenX is the number of networks listened to, excluding link-local only listeners. This number might be different than len(networksN). -// This is useful to determine if there are any IPv4 or IPv6 listeners for potential external connections. This can be used to determine IPv4_LISTEN and IPv6_LISTEN. -var countListen4, countListen6 int64 - // Default ports to use. This may be randomized in the future to prevent fingerprinting (and subsequent blocking) by corporate and ISP firewalls. const defaultPort = 'p' // 112 @@ -106,9 +97,9 @@ const maxPacketSize = 65536 func (network *Network) Listen() { if !network.address.IP.IsLinkLocalUnicast() { if IsIPv4(network.address.IP) { - atomic.AddInt64(&countListen4, 1) + atomic.AddInt64(&network.networkGroup.countListen4, 1) } else { - atomic.AddInt64(&countListen6, 1) + atomic.AddInt64(&network.networkGroup.countListen6, 1) } } @@ -135,13 +126,13 @@ func (network *Network) Listen() { } // send the packet to a channel which is processed by multiple workers. - rawPacketsIncoming <- networkWire{network: network, sender: sender, raw: buffer[:length], receiverPublicKey: peerPublicKey, unicast: true} + network.networkGroup.rawPacketsIncoming <- networkWire{network: network, sender: sender, raw: buffer[:length], receiverPublicKey: peerPublicKey, unicast: true} } } // packetWorker handles incoming packets. -func packetWorker(packets <-chan networkWire) { - for packet := range packets { +func (nets *Networks) packetWorker() { + for packet := range nets.rawPacketsIncoming { decoded, senderPublicKey, err := protocol.PacketDecrypt(packet.raw, packet.receiverPublicKey) if err != nil { //Filters.LogError("packetWorker", "decrypting packet from '%s': %s\n", packet.sender.String(), err.Error()) // Only log for debug purposes. @@ -195,7 +186,7 @@ func packetWorker(packets <-chan networkWire) { case protocol.CommandResponse: // Response if response, _ := msgDecodeResponse(raw); response != nil { // Validate sequence number which prevents unsolicited responses. - if valid, rtt := networks.Sequences.ValidateSequence(raw, response.Actions&(1< 0); !valid { + if valid, rtt := nets.Sequences.ValidateSequence(raw, response.Actions&(1< 0); !valid { //Filters.LogError("packetWorker", "message with invalid sequence %d command %d from %s\n", raw.Sequence, raw.Command, raw.connection.Address.String()) // Only log for debug purposes. continue } else if rtt > 0 { @@ -237,7 +228,7 @@ func packetWorker(packets <-chan networkWire) { case protocol.CommandPong: // Ping // Validate sequence number which prevents unsolicited responses. - if valid, rtt := networks.Sequences.ValidateSequence(raw, true); !valid { + if valid, rtt := nets.Sequences.ValidateSequence(raw, true); !valid { //Filters.LogError("packetWorker", "message with invalid sequence %d command %d from %s\n", raw.Sequence, raw.Command, raw.connection.Address.String()) // Only log for debug purposes. continue } else if rtt > 0 { @@ -271,12 +262,12 @@ func packetWorker(packets <-chan networkWire) { } // GetNetworks returns the list of connected networks -func GetNetworks(networkType int) (networks []*Network) { +func GetNetworks(networkType int) (networksConnected []*Network) { switch networkType { case 4: - return networks4 + return networks.networks4 case 6: - return networks6 + return networks.networks6 } return nil } @@ -305,9 +296,9 @@ func (network *Network) Terminate() { if !network.address.IP.IsLinkLocalUnicast() { if IsIPv4(network.address.IP) { - atomic.AddInt64(&countListen4, -1) + atomic.AddInt64(&network.networkGroup.countListen4, -1) } else { - atomic.AddInt64(&countListen6, -1) + atomic.AddInt64(&network.networkGroup.countListen6, -1) } } @@ -316,7 +307,7 @@ func (network *Network) Terminate() { close(network.terminateSignal) // safety guaranteed via lock network.socket.Close() // Will stop the listener from blocking on network.socket.ReadFromUDP - networks.ipListen.Remove(network.address) + network.networkGroup.ipListen.Remove(network.address) } // SelfReportedPorts returns the internal and external ports as self-reported by the peer to others. @@ -341,10 +332,10 @@ func (network *Network) SelfReportedPorts() (portI, portE uint16) { // FeatureSupport returns supported features by this peer func FeatureSupport() (feature byte) { - if countListen4 > 0 { + if networks.countListen4 > 0 { feature |= 1 << FeatureIPv4Listen } - if countListen6 > 0 { + if networks.countListen6 > 0 { feature |= 1 << FeatureIPv6Listen } return feature diff --git a/Networks.go b/Networks.go index defe316..09d83a4 100644 --- a/Networks.go +++ b/Networks.go @@ -6,18 +6,22 @@ Author: Peter Kleissner package core +import "sync" + // Networks is the collection of all connected networks type Networks struct { // networks is a list of all connected networks - //networks4, networks6 []*Network + networks4, networks6 []*Network // Mutex for both network lists. Higher granularity currently not needed. - //sync.RWMutex - //networksMutex sync.RWMutex + sync.RWMutex // countListenX is the number of networks listened to, excluding link-local only listeners. This number might be different than len(networksN). // This is useful to determine if there are any IPv4 or IPv6 listeners for potential external connections. This can be used to determine IPv4_LISTEN and IPv6_LISTEN. - //countListen4, countListen6 int64 + countListen4, countListen6 int64 + + // channel for processing incoming decoded packets by workers, across all networks + rawPacketsIncoming chan networkWire // Sequences keeps track of all message sequence number, regardless of the network connection. Sequences *SequenceManager @@ -34,6 +38,8 @@ var networks *Networks func initMessageSequence() { networks = &Networks{} + networks.rawPacketsIncoming = make(chan networkWire, 1000) // buffer up to 1000 UDP packets before they get buffered by the OS network stack and eventually dropped + networks.Sequences = NewSequenceManager(ReplyTimeout) networks.ipListen = NewIPList() diff --git a/Peernet.go b/Peernet.go index 6470eab..14865bd 100644 --- a/Peernet.go +++ b/Peernet.go @@ -25,9 +25,9 @@ func Init() { func Connect() { go bootstrapKademlia() go bootstrap() - go autoMulticastBroadcast() + go networks.autoMulticastBroadcast() go autoPingAll() - go networkChangeMonitor() + go networks.networkChangeMonitor() + go networks.startUPnP() go autoBucketRefresh() - go startUPnP() }