Refactor sequence code into SequenceManager

Connection.go

@@ -426,7 +426,7 @@ func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *protocol.Packet
 		}
 
 		if sequenceData != nil {
-			packet.Sequence = ArbitrarySequence(receiverPublicKey, sequenceData).SequenceNumber
+			packet.Sequence = networks.Sequences.ArbitrarySequence(receiverPublicKey, sequenceData).SequenceNumber
 		}
 		err = (&Connection{Network: network, Address: remote, PortInternal: receiverPortInternal, traversePeer: traversePeer}).send(packet, receiverPublicKey, isFirstPacket)
 		isFirstPacket = false

Message Send.go

@@ -15,7 +15,7 @@ import (
 
 // pingConnection sends a ping to the target peer via the specified connection
 func (peer *PeerInfo) pingConnection(connection *Connection) {
-	raw := &protocol.PacketRaw{Command: protocol.CommandPing, Sequence: NewSequence(peer.PublicKey, &peer.messageSequence, nil).SequenceNumber}
+	raw := &protocol.PacketRaw{Command: protocol.CommandPing, Sequence: networks.Sequences.NewSequence(peer.PublicKey, &peer.messageSequence, nil).SequenceNumber}
 	Filters.MessageOutPing(peer, raw, connection)
 
 	err := peer.sendConnection(raw, connection)
@@ -37,7 +37,7 @@ func (peer *PeerInfo) sendAnnouncement(sendUA, findSelf bool, findPeer []KeyHash
 	packets := EncodeAnnouncement(sendUA, findSelf, findPeer, findValue, files, FeatureSupport(), blockchainHeight, blockchainVersion)
 
 	for _, packet := range packets {
-		raw := &protocol.PacketRaw{Command: protocol.CommandAnnouncement, Payload: packet, Sequence: NewSequence(peer.PublicKey, &peer.messageSequence, sequenceData).SequenceNumber}
+		raw := &protocol.PacketRaw{Command: protocol.CommandAnnouncement, Payload: packet, Sequence: networks.Sequences.NewSequence(peer.PublicKey, &peer.messageSequence, sequenceData).SequenceNumber}
 		Filters.MessageOutAnnouncement(peer.PublicKey, peer, raw, findSelf, findPeer, findValue, files)
 		peer.send(raw)
 	}

Message Sequence.go

@@ -25,13 +25,16 @@ import (
 	"github.com/btcsuite/btcd/btcec"
 )
 
-// SequenceReplyTimeout is the round-trip timeout for message sequences.
-var SequenceReplyTimeout = 20
+// SequenceManager stores all message sequence numbers that are valid at the moment
+type SequenceManager struct {
+	ReplyTimeout int // The round-trip timeout for message sequences.
 
-// sequences stores all sequence numbers that are valid at the moment. The value represents the time the sequence number was used.
-// Key = Peer ID + Sequence Number
-var sequences map[string]*SequenceExpiry
-var sequencesMutex sync.Mutex
+	// sequences is the list of sequence numbers that are valid at the moment. The value represents the time the sequence number.
+	// Key = Peer ID + Sequence Number
+	sequences map[string]*SequenceExpiry
+
+	sync.Mutex // synchronized access to the sequences
+}
 
 // SequenceExpiry contains the decoded sequence information of a message.
 type SequenceExpiry struct {
@@ -42,65 +45,73 @@ type SequenceExpiry struct {
 	Data           interface{} // Optional high-level data associated with the sequence
 }
 
-func initMessageSequence() {
-	sequences = make(map[string]*SequenceExpiry)
+// NewSequenceManager creates a new sequence manager. The ReplyTimeout is in seconds. The expiration function is started immediately.
+func NewSequenceManager(ReplyTimeout int) (manager *SequenceManager) {
+	manager = &SequenceManager{
+		ReplyTimeout: ReplyTimeout,
+		sequences:    make(map[string]*SequenceExpiry),
+	}
 
-	// auto-delete worker to remove expired sequences
-	go func() {
-		for {
-			time.Sleep(time.Duration(SequenceReplyTimeout) * time.Second)
-			now := time.Now()
+	go manager.autoDeleteExpired()
 
-			sequencesMutex.Lock()
-			for key, sequence := range sequences {
-				if sequence.expires.Before(now) {
-					delete(sequences, key)
-				}
+	return
+}
+
+// autoDeleteExpired deletes all sequences that are expired.
+func (manager *SequenceManager) autoDeleteExpired() {
+	for {
+		time.Sleep(time.Duration(manager.ReplyTimeout) * time.Second)
+		now := time.Now()
+
+		manager.Lock()
+		for key, sequence := range manager.sequences {
+			if sequence.expires.Before(now) {
+				delete(manager.sequences, key)
 			}
-			sequencesMutex.Unlock()
 		}
-	}()
+		manager.Unlock()
+	}
 }
 
 // NewSequence returns a new sequence and registers it. messageSequence must point to the variable holding the continuous next sequence number.
 // Use only for Announcement and Ping messages.
-func NewSequence(publicKey *btcec.PublicKey, messageSequence *uint32, data interface{}) (info *SequenceExpiry) {
+func (manager *SequenceManager) NewSequence(publicKey *btcec.PublicKey, messageSequence *uint32, data interface{}) (info *SequenceExpiry) {
 	info = &SequenceExpiry{
 		SequenceNumber: atomic.AddUint32(messageSequence, 1),
 		created:        time.Now(),
-		expires:        time.Now().Add(time.Duration(SequenceReplyTimeout) * time.Second),
+		expires:        time.Now().Add(time.Duration(manager.ReplyTimeout) * time.Second),
 		Data:           data,
 	}
 
 	// Add the sequence to the list. Sequences are unique enough that collisions are unlikely and negligible.
 	key := string(publicKey.SerializeCompressed()) + strconv.FormatUint(uint64(info.SequenceNumber), 10)
-	sequencesMutex.Lock()
-	sequences[key] = info
-	sequencesMutex.Unlock()
+	manager.Lock()
+	manager.sequences[key] = info
+	manager.Unlock()
 
 	return
 }
 
 // ArbitrarySequence returns an arbitrary sequence to be used for uncontacted peers
-func ArbitrarySequence(publicKey *btcec.PublicKey, data interface{}) (info *SequenceExpiry) {
+func (manager *SequenceManager) ArbitrarySequence(publicKey *btcec.PublicKey, data interface{}) (info *SequenceExpiry) {
 	info = &SequenceExpiry{
 		SequenceNumber: rand.Uint32(),
 		created:        time.Now(),
-		expires:        time.Now().Add(time.Duration(SequenceReplyTimeout) * time.Second),
+		expires:        time.Now().Add(time.Duration(manager.ReplyTimeout) * time.Second),
 		Data:           data,
 	}
 
 	// Add the sequence to the list. Sequences are unique enough that collisions are unlikely and negligible.
 	key := string(publicKey.SerializeCompressed()) + strconv.FormatUint(uint64(info.SequenceNumber), 10)
-	sequencesMutex.Lock()
-	sequences[key] = info
-	sequencesMutex.Unlock()
+	manager.Lock()
+	manager.sequences[key] = info
+	manager.Unlock()
 
 	return
 }
 
 // ValidateSequence validates the sequence number of an incoming message. It will set raw.sequence if valid.
-func ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Duration) {
+func (manager *SequenceManager) ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Duration) {
 	// Only Response and Pong
 	if raw.Command != protocol.CommandResponse && raw.Command != protocol.CommandPong {
 		return true, rtt
@@ -108,11 +119,11 @@ func ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Du
 
 	key := string(raw.SenderPublicKey.SerializeCompressed()) + strconv.FormatUint(uint64(raw.Sequence), 10)
 
-	sequencesMutex.Lock()
-	defer sequencesMutex.Unlock()
+	manager.Lock()
+	defer manager.Unlock()
 
 	// lookup the sequence
-	sequence, ok := sequences[key]
+	sequence, ok := manager.sequences[key]
 	if !ok {
 		return false, rtt
 	}
@@ -127,10 +138,10 @@ func ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Du
 
 	// invalidate the sequence immediately?
 	if invalidate {
-		delete(sequences, key)
+		delete(manager.sequences, key)
 	} else if raw.Command == protocol.CommandResponse {
 		// Special case CommandResponse: Extend validity in case there are follow-up responses by half of the round-trip time since they will be sent one-way.
-		sequence.expires = time.Now().Add(time.Duration(SequenceReplyTimeout) * time.Second / 2)
+		sequence.expires = time.Now().Add(time.Duration(manager.ReplyTimeout) * time.Second / 2)
 	}
 
 	raw.SequenceInfo = sequence
@@ -139,7 +150,7 @@ func ValidateSequence(raw *MessageRaw, invalidate bool) (valid bool, rtt time.Du
 }
 
 // InvalidateSequence invalidates the sequence number.
-func InvalidateSequence(raw *MessageRaw) {
+func (manager *SequenceManager) InvalidateSequence(raw *MessageRaw) {
 	// Only Response
 	if raw.Command != protocol.CommandResponse {
 		return
@@ -147,7 +158,7 @@ func InvalidateSequence(raw *MessageRaw) {
 
 	key := string(raw.SenderPublicKey.SerializeCompressed()) + strconv.FormatUint(uint64(raw.Sequence), 10)
 
-	sequencesMutex.Lock()
-	delete(sequences, key)
-	sequencesMutex.Unlock()
+	manager.Lock()
+	delete(manager.sequences, key)
+	manager.Unlock()
 }

Network.go

@@ -195,7 +195,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 := ValidateSequence(raw, response.Actions&(1<<ActionSequenceLast) > 0); !valid {
+				if valid, rtt := networks.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 +237,7 @@ func packetWorker(packets <-chan networkWire) {
 
 		case protocol.CommandPong: // Ping
 			// Validate sequence number which prevents unsolicited responses.
-			if valid, rtt := ValidateSequence(raw, true); !valid {
+			if valid, rtt := networks.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 {

Networks.go

@@ -0,0 +1,34 @@
+/*
+File Name:  Networks.go
+Copyright:  2021 Peernet s.r.o.
+Author:     Peter Kleissner
+*/
+
+package core
+
+// Networks is the collection of all connected networks
+type Networks struct {
+	// networks is a list of all connected networks
+	//networks4, networks6 []*Network
+
+	// Mutex for both network lists. Higher granularity currently not needed.
+	//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
+
+	Sequences *SequenceManager
+}
+
+//  ReplyTimeout is the round-trip timeout for message sequences.
+const ReplyTimeout = 20
+
+var networks *Networks
+
+func initMessageSequence() {
+	networks = &Networks{}
+
+	networks.Sequences = NewSequenceManager(ReplyTimeout)
+}