swarm/pss: forwarding function refactoring

swarm/pss/forwarding_test.go

@@ -14,95 +14,122 @@ 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) {
-	base := newBaseAddress() // 0xFFFFFF.......
+	setDummySendMsg()
+	defer resetSendMsgProduction()
+
+	baseAddrBytes := make([]byte, 32)
+	for i := 0; i < len(baseAddrBytes); i++ {
+		baseAddrBytes[i] = 0xFF
+	}
+	base := pot.NewAddressFromBytes(baseAddrBytes)
 	var peerAddresses []pot.Address
-	var dst pot.Address
-	const depth = 9
+	var a pot.Address
+	const depth = 10
 	for i := 0; i <= depth; i++ {
-		a := pot.RandomAddressAt(base, i)
-		peerAddresses = append(peerAddresses, a)
+		// add one peer for each proximity order
 		a = pot.RandomAddressAt(base, i)
 		peerAddresses = append(peerAddresses, a)
 	}
 
-	// skip one level, add one peer at one level below
-	a := pot.RandomAddressAt(base, depth+2)
-	peerAddresses = append(peerAddresses, a)
+	// add one peer to the "depth" level, then skip one level, add one peer at one level below.
+	// as a result, we will have an edge case of three peers in nearest neighbours' bin.
+	peerAddresses = append(peerAddresses, pot.RandomAddressAt(base, depth))
+	peerAddresses = append(peerAddresses, pot.RandomAddressAt(base, depth+2))
 
 	kad := network.NewKademlia(base[:], network.NewKadParams())
 	ps := createPss(t, kad)
 	addPeers(kad, peerAddresses)
 
-	const firstNearest = depth * 2 // first peer in the nearest neighbours' bin
+	const firstNearest = depth // shallowest 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)
 		testForwardMsg(100+i, t, ps, peerAddresses[i][:], peerAddresses, []int{i})
 	}
 
 	for i := 0; i < firstNearest; i++ {
-		// send random messages with different proximity orders
-		po := i / 2
-		dst := pot.RandomAddressAt(base, po)
-		testForwardMsg(200+i, t, ps, dst[:], peerAddresses, []int{po * 2, po*2 + 1})
+		// send random messages with proximity orders, corresponding to PO of each bin
+		a = pot.RandomAddressAt(base, i)
+		testForwardMsg(200+i, t, ps, a[:], peerAddresses, []int{i})
 	}
 
 	for i := firstNearest; i < len(peerAddresses); i++ {
 		// recipient address falls into the nearest neighbours' bin
-		dst := pot.RandomAddressAt(base, i)
-		testForwardMsg(300+i, t, ps, dst[:], peerAddresses, nearestNeighbours)
+		a = pot.RandomAddressAt(base, i)
+		testForwardMsg(300+i, t, ps, a[:], peerAddresses, nearestNeighbours)
 	}
 
-	// send msg with proximity order higher than the last nearest neighbour
-	dst = pot.RandomAddressAt(base, 29)
-	testForwardMsg(400, t, ps, dst[:], peerAddresses, nearestNeighbours)
+	// send msg with proximity order much deeper than the deepest nearest neighbour
+	a = pot.RandomAddressAt(base, 77)
+	testForwardMsg(400, t, ps, a[:], peerAddresses, nearestNeighbours)
 
 	// test with partial addresses
 	const part = 12
 
 	for i := 0; i < firstNearest; i++ {
 		// send messages with partial address falling into different proximity orders
-		po := i / 2
-		if po%8 != 0 {
-			testForwardMsg(500+i, t, ps, peerAddresses[i][:po], peerAddresses, []int{po * 2, po*2 + 1})
+		if i%8 != 0 {
+			testForwardMsg(500+i, t, ps, peerAddresses[i][:i], peerAddresses, []int{i})
 		}
-		testForwardMsg(550+i, t, ps, peerAddresses[i][:part], peerAddresses, []int{po * 2, po*2 + 1})
+		testForwardMsg(550+i, t, ps, peerAddresses[i][:part], peerAddresses, []int{i})
 	}
 
 	for i := firstNearest; i < len(peerAddresses); i++ {
 		// partial address falls into the nearest neighbours' bin
 		testForwardMsg(600+i, t, ps, peerAddresses[i][:part], peerAddresses, nearestNeighbours)
 	}
 
-	// partial address with proximity order higher than the last nearest neighbour
-	dst = pot.RandomAddressAt(base, part)
-	testForwardMsg(700, t, ps, dst[:part], peerAddresses, nearestNeighbours)
+	// partial address with proximity order deeper than any of the nearest neighbour
+	a = pot.RandomAddressAt(base, part)
+	testForwardMsg(700, t, ps, a[:part], peerAddresses, nearestNeighbours)
 
 	// special cases where partial address matches a large group of peers
-	all := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20}
+	all := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12}
 	testForwardMsg(800, t, ps, []byte{}, peerAddresses, all)
-	testForwardMsg(900, t, ps, peerAddresses[19][:1], peerAddresses, all[16:])
+
+	// luminous radius of one byte (8 bits)
+	testForwardMsg(900, t, ps, baseAddrBytes[:1], peerAddresses, all[8:])
 }
 
-func testForwardMsg(num 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 {
-		a := pot.NewAddressFromBytes(sp.Address())
-		testResMap[a]++
-		return true
-	})
+// this function tests the forwarding of a single message. the recipient address is passed as param,
+// along with addresses of all peers, and indices of those peers which are expected to receive the message.
+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", num, 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)
@@ -111,13 +138,13 @@ func testForwardMsg(num int, t *testing.T, ps *Pss, addr []byte, addresses []pot
 		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
 				}
 			}
@@ -148,15 +175,6 @@ func createPss(t *testing.T, kad *network.Kademlia) *Pss {
 	return ps
 }
 
-func newBaseAddress() pot.Address {
-	//base := network.RandomAddr().OAddr
-	base := make([]byte, 32)
-	for i := 0; i < len(base); i++ {
-		base[i] = 0xFF
-	}
-	return pot.NewAddressFromBytes(base)
-}
-
 func newTestDiscoveryPeer(addr pot.Address, kad *network.Kademlia) *network.Peer {
 	rw := &p2p.MsgPipeRW{}
 	p := p2p.NewPeer(enode.ID{}, "test", []p2p.Cap{})

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 trySendMsg(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 trySendMsg(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,41 +934,45 @@ func trySendMsg(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, trySend func(p *Pss, sp *network.Peer, msg *PssMsg) bool) error {
-	if trySend == nil {
-		trySend = trySendMsg
-	}
-
+func (p *Pss) forward(msg *PssMsg) error {
 	metrics.GetOrRegisterCounter("pss.forward", nil).Inc(1)
 	sent := 0 // number of successful sends
 	to := make([]byte, addressLength)
 	copy(to[:len(msg.To)], msg.To)
 	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)
-	depth, _ := pof(to, p.BaseAddr(), 0)
-	if depth > luminosityRadius {
-		depth = luminosityRadius
+
+	// proximity order function matching up to neighbourhoodDepth bits (po <= neighbourhoodDepth)
+	pof := pot.DefaultPof(neighbourhoodDepth)
+
+	// soft threshold for msg broadcast
+	broadcastThreshold, _ := pof(to, p.BaseAddr(), 0)
+	if broadcastThreshold > luminosityRadius {
+		broadcastThreshold = 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 depth < luminosityRadius && depth < neighbourhoodDepth {
-		depth++
+	// if measured from the recipient address as opposed to the base address (see Kademlia.EachConn
+	// call below), then peers that fall in the same proximity bin as recipient address will appear
+	// [at least] one bit closer, but only if these additional bits are given in the recipient address.
+	if broadcastThreshold < luminosityRadius && broadcastThreshold < neighbourhoodDepth {
+		broadcastThreshold++
 	}
 
 	p.Kademlia.EachConn(to, addressLength*8, func(sp *network.Peer, po int, _ bool) bool {
-		if po < depth && sent > 0 {
+		if po < broadcastThreshold && sent > 0 {
 			return false // stop iterating
 		}
-		if trySend(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)
 
 }