swarm/pss: changed comments, refactored helper func sendMessage

swarm/pss/forwarding_test.go

@@ -14,10 +14,33 @@ import (
 	whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
 )
 
+var testResMap map[pot.Address]int
+
+// this function substitutes the real send function, since
+// we only want to test the peer selection functionality
+func dummySendMsg(_ *Pss, sp *network.Peer, _ *PssMsg) bool {
+	a := pot.NewAddressFromBytes(sp.Address())
+	testResMap[a]++
+	return true
+}
+
+// setDummySendMsg replaces sendMessage function for testing purposes
+func setDummySendMsg() {
+	sendMessage = dummySendMsg
+}
+
+// resetSendMsgProduction resets sendMessage function to production version
+func resetSendMsgProduction() {
+	sendMessage = sendMessageProd
+}
+
 // the purpose of this test is to see that pss.forward() function correctly
 // selects the peers for message forwarding, depending on the message address
 // and kademlia constellation.
 func TestForwardBasic(t *testing.T) {
+	setDummySendMsg()
+	defer resetSendMsgProduction()
+
 	base := newBaseAddress() // 0xFFFFFF.......
 	var peerAddresses []pot.Address
 	var dst pot.Address
@@ -40,7 +63,6 @@ func TestForwardBasic(t *testing.T) {
 
 	const firstNearest = depth * 2 // first peer in the nearest neighbours' bin
 	nearestNeighbours := []int{firstNearest, firstNearest + 1, firstNearest + 2}
-	//fmt.Println(kad.String()) // print kademlia map for debugging, before any test starts
 
 	for i := 0; i < len(peerAddresses); i++ {
 		// send msg directly to the known peers (recipient address == peer address)
@@ -91,25 +113,20 @@ func TestForwardBasic(t *testing.T) {
 	testForwardMsg(900, t, ps, peerAddresses[19][:1], peerAddresses, all[16:])
 }
 
-// this function tests the forwarding of a single message. the recipient address (addr) is passed as param,
+// this function tests the forwarding of a single message. the recipient address is passed as param,
 // along with addreses of all peers, and indexes of those peers which are expected to receive the message.
-func testForwardMsg(testID int, t *testing.T, ps *Pss, addr []byte, addresses []pot.Address, expected []int) {
-	testResMap := make(map[pot.Address]int)
-	msg := newTestMsg(addr)
-	ps.forward(msg, func(p *Pss, sp *network.Peer, msg *PssMsg) bool {
-		// this function substitutes the real send function, since we only want to test the peer selection functionality
-		a := pot.NewAddressFromBytes(sp.Address())
-		testResMap[a]++
-		return true
-	})
+func testForwardMsg(testID int, t *testing.T, ps *Pss, recipientAddr []byte, peers []pot.Address, expected []int) {
+	testResMap = make(map[pot.Address]int)
+	msg := newTestMsg(recipientAddr)
+	ps.forward(msg)
 
 	// check test results
 	var fail bool
-	s := fmt.Sprintf("test id: %d, msg address: %x..., radius: %d", testID, addr[:len(addr)%4], 8*len(addr))
+	s := fmt.Sprintf("test id: %d, msg address: %x..., radius: %d", testID, recipientAddr[:len(recipientAddr)%4], 8*len(recipientAddr))
 
-	// false negatives
+	// false negatives (expected message didn't reach peer)
 	for _, i := range expected {
-		a := addresses[i]
+		a := peers[i]
 		received := testResMap[a]
 		if received != 1 {
 			s += fmt.Sprintf("\npeer number %d [%x...] received %d messages", i, a[:4], received)
@@ -118,13 +135,13 @@ func testForwardMsg(testID int, t *testing.T, ps *Pss, addr []byte, addresses []
 		testResMap[a] = 0
 	}
 
-	// false positives
+	// false positives (unexpected message reached peer)
 	for k, v := range testResMap {
 		if v != 0 {
 			// find the index of the false positive peer
 			var j int
-			for j = 0; j < len(addresses); j++ {
-				if addresses[j] == k {
+			for j = 0; j < len(peers); j++ {
+				if peers[j] == k {
 					break
 				}
 			}
@@ -156,7 +173,6 @@ func createPss(t *testing.T, kad *network.Kademlia) *Pss {
 }
 
 func newBaseAddress() pot.Address {
-	//base := network.RandomAddr().OAddr
 	base := make([]byte, 32)
 	for i := 0; i < len(base); i++ {
 		base[i] = 0xFF

swarm/pss/pss.go

@@ -225,7 +225,7 @@ func (p *Pss) Start(srv *p2p.Server) error {
 		for {
 			select {
 			case msg := <-p.outbox:
-				err := p.forward(msg, nil)
+				err := p.forward(msg)
 				if err != nil {
 					log.Error(err.Error())
 					metrics.GetOrRegisterCounter("pss.forward.err", nil).Inc(1)
@@ -886,8 +886,17 @@ func (p *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key []by
 	return nil
 }
 
+// sendMessage is a helper function that tries to send a message and returns true on success
+// It is set in the init function for usage in production, and optionally overridden in tests
+// for data validation.
+var sendMessage func(p *Pss, sp *network.Peer, msg *PssMsg) bool
+
+func init() {
+	sendMessage = sendMessageProd
+}
+
 // tries to send a message, returns true if successful
-func sendMessage(p *Pss, sp *network.Peer, msg *PssMsg) bool {
+func sendMessageProd(p *Pss, sp *network.Peer, msg *PssMsg) bool {
 	var isPssEnabled bool
 	info := sp.Info()
 	for _, capability := range info.Caps {
@@ -897,7 +906,7 @@ func sendMessage(p *Pss, sp *network.Peer, msg *PssMsg) bool {
 		}
 	}
 	if !isPssEnabled {
-		log.Trace("peer doesn't have matching pss capabilities, skipping", "peer", info.Name, "caps", info.Caps)
+		log.Error("peer doesn't have matching pss capabilities, skipping", "peer", info.Name, "caps", info.Caps)
 		return false
 	}
 
@@ -925,11 +934,7 @@ func sendMessage(p *Pss, sp *network.Peer, msg *PssMsg) bool {
 // are any; otherwise only to one peer, closest to the recipient address. In any case, if the message
 // forwarding fails, the node should try to forward it to the next best peer, until the message is
 // successfully forwarded to at least one peer.
-func (p *Pss) forward(msg *PssMsg, sendMsg func(p *Pss, sp *network.Peer, msg *PssMsg) bool) error {
-	if sendMsg == nil {
-		sendMsg = sendMessage
-	}
-
+func (p *Pss) forward(msg *PssMsg) error {
 	metrics.GetOrRegisterCounter("pss.forward", nil).Inc(1)
 	sent := 0 // number of successful sends
 	to := make([]byte, addressLength)
@@ -937,17 +942,19 @@ func (p *Pss) forward(msg *PssMsg, sendMsg func(p *Pss, sp *network.Peer, msg *P
 	neighbourhoodDepth := p.Kademlia.NeighbourhoodDepth()
 
 	// luminosity is the opposite of darkness. the more bytes are removed from the address, the higher is darkness,
-	// but the luminosity is less. here luminosity equals the number of bits present in the destination address.
+	// but the luminosity is less. here luminosity equals the number of bits given in the destination address.
 	luminosityRadius := len(msg.To) * 8
-	pof := pot.DefaultPof(neighbourhoodDepth) // pof function matching up to neighbourhoodDepth bits (pof <= neighbourhoodDepth)
+
+	// proximity order function matching up to neighbourhoodDepth bits (po <= neighbourhoodDepth)
+	pof := pot.DefaultPof(neighbourhoodDepth)
 	depth, _ := pof(to, p.BaseAddr(), 0)
 	if depth > luminosityRadius {
 		depth = luminosityRadius
 	}
 
-	// if measured from the recipient address (as opposed to the base address), then
-	// peers that fall in the same proximity bin will appear one bit closer (at least),
-	// under condition that these additional bits exist in the recipient address.
+	// if measured from the recipient address (as opposed to the base address), then peers
+	// that fall in the same proximity bin as recipient address will appear one bit closer
+	// (at least), under condition that these additional bits exist in the recipient address.
 	if depth < luminosityRadius && depth < neighbourhoodDepth {
 		depth++
 	}
@@ -956,10 +963,14 @@ func (p *Pss) forward(msg *PssMsg, sendMsg func(p *Pss, sp *network.Peer, msg *P
 		if po < depth && sent > 0 {
 			return false // stop iterating
 		}
-		if sendMsg(p, sp, msg) {
+		if sendMessage(p, sp, msg) {
 			sent++
+			if po == addressLength*8 {
+				// stop iterating if successfully sent to the exact recipient (perfect match of full address)
+				return false
+			}
 		}
-		return po < addressLength*8 // stop iterating in case of exact match of full address
+		return true
 	})
 
 	// if we failed to send to anyone, re-insert message in the send-queue

swarm/pss/pss_test.go

@@ -935,7 +935,7 @@ func TestPeerCapabilityMismatch(t *testing.T) {
 
 	// run the forward
 	// it is enough that it completes; trying to send to incapable peers would create segfault
-	ps.forward(pssmsg, nil)
+	ps.forward(pssmsg)
 
 }