Collecting network vars and functions in networks structure

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()
+		nets.RLock()
-		defer networksMutex.RUnlock()
+		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
 }

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) {
+func (nets *Networks) sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *protocol.PacketRaw, remote *net.UDPAddr, receiverPortInternal uint16, traversePeer *PeerInfo, sequenceData interface{}) (err error) {
-	networksMutex.RLock()
+	nets.RLock()
-	defer networksMutex.RUnlock()
+	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

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) {
+func (nets *Networks) networkChangeInterfaceRemove(iface string, addresses []net.Addr) {
-	networksMutex.RLock()
+	nets.RLock()
-	defer networksMutex.RUnlock()
+	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]
+			networksNew := nets.networks6[:n]
-			if n < len(networks6)-1 {
+			if n < len(nets.networks6)-1 {
-				networksNew = append(networksNew, networks6[n+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]
+			networksNew := nets.networks4[:n]
-			if n < len(networks4)-1 {
+			if n < len(nets.networks4)-1 {
-				networksNew = append(networksNew, networks4[n+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) {
+func (nets *Networks) networkChangeIPRemove(iface net.Interface, address net.Addr) {
-	networksMutex.RLock()
+	nets.RLock()
-	defer networksMutex.RUnlock()
+	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]
+			networksNew := nets.networks6[:n]
-			if n < len(networks6)-1 {
+			if n < len(nets.networks6)-1 {
-				networksNew = append(networksNew, networks6[n+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]
+			networksNew := nets.networks4[:n]
-			if n < len(networks4)-1 {
+			if n < len(nets.networks4)-1 {
-				networksNew = append(networksNew, networks4[n+1:]...)
+				networksNew = append(networksNew, nets.networks4[n+1:]...)
 			}
-			networks4 = networksNew
+			nets.networks4 = networksNew
 		}
 	}
 }

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}
 	}
 }
 

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}
 	}
 }
 

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)
+		nets.networks4 = append(nets.networks4, network)
-		networksMutex.Unlock()
+		nets.Unlock()
 		network.BroadcastIPv4()
 	} else {
-		networks6 = append(networks6, network)
+		nets.networks6 = append(nets.networks6, network)
-		networksMutex.Unlock()
+		nets.Unlock()
 		network.MulticastIPv6Join()
 	}
 

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()
 	}
 }

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) {
+func (nets *Networks) packetWorker() {
-	for packet := range packets {
+	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<<ActionSequenceLast) > 0); !valid {
+				if valid, rtt := nets.Sequences.ValidateSequence(raw, response.Actions&(1<<ActionSequenceLast) > 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

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
+	sync.RWMutex
-	//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.
-	//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()

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()
 }