swarm/pss: forwarding test added, bugs fixed

swarm/pss/forwarding_test.go

@@ -0,0 +1,183 @@
+package pss
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/swarm/network"
+	"github.com/ethereum/go-ethereum/swarm/pot"
+	whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
+)
+
+func TestForwardBasic(t *testing.T) {
+	base := newBaseAddress() // 0xFFFFFF.......
+	var peerAddresses []pot.Address
+	var dst pot.Address
+	const depth = 9
+	for i := 0; i <= depth; i++ {
+		a := pot.RandomAddressAt(base, i)
+		peerAddresses = append(peerAddresses, a)
+		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)
+
+	kad := network.NewKademlia(base[:], network.NewKadParams())
+	ps := createPss(t, kad)
+	addPeers(kad, peerAddresses)
+
+	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)
+		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})
+	}
+
+	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)
+	}
+
+	// send msg with proximity order higher than the last nearest neighbour
+	dst = pot.RandomAddressAt(base, 29)
+	testForwardMsg(400, t, ps, dst[:], 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})
+		}
+		testForwardMsg(550+i, t, ps, peerAddresses[i][:part], peerAddresses, []int{po * 2, po*2 + 1})
+	}
+
+	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)
+
+	// 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}
+	testForwardMsg(800, t, ps, []byte{}, peerAddresses, all)
+	testForwardMsg(900, t, ps, peerAddresses[19][:1], peerAddresses, all[16:])
+}
+
+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
+	})
+
+	// 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))
+
+	// false negatives
+	for _, i := range expected {
+		a := addresses[i]
+		received := testResMap[a]
+		if received != 1 {
+			s += fmt.Sprintf("\npeer number %d [%x...] received %d messages", i, a[:4], received)
+			fail = true
+		}
+		testResMap[a] = 0
+	}
+
+	// false positives
+	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 {
+					break
+				}
+			}
+			s += fmt.Sprintf("\npeer number %d [%x...] received %d messages", j, k[:4], v)
+			fail = true
+		}
+	}
+
+	if fail {
+		t.Fatal(s)
+	}
+}
+
+func addPeers(kad *network.Kademlia, addresses []pot.Address) {
+	for _, a := range addresses {
+		p := newTestDiscoveryPeer(a, kad)
+		kad.On(p)
+	}
+}
+
+func createPss(t *testing.T, kad *network.Kademlia) *Pss {
+	privKey, err := crypto.GenerateKey()
+	pssp := NewPssParams().WithPrivateKey(privKey)
+	ps, err := NewPss(kad, pssp)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+	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{})
+	pp := protocols.NewPeer(p, rw, &protocols.Spec{})
+	bp := &network.BzzPeer{
+		Peer: pp,
+		BzzAddr: &network.BzzAddr{
+			OAddr: addr.Bytes(),
+			UAddr: []byte(fmt.Sprintf("%x", addr[:])),
+		},
+	}
+	return network.NewPeer(bp, kad)
+}
+
+func newTestMsg(addr []byte) *PssMsg {
+	msg := newPssMsg(&msgParams{})
+	msg.To = addr[:]
+	msg.Expire = uint32(time.Now().Add(time.Second * 60).Unix())
+	msg.Payload = &whisper.Envelope{
+		Topic: [4]byte{},
+		Data:  []byte("i have nothing to hide"),
+	}
+	return msg
+}

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)
+				err := p.forward(msg, nil)
 				if err != nil {
 					log.Error(err.Error())
 					metrics.GetOrRegisterCounter("pss.forward.err", nil).Inc(1)
@@ -887,7 +887,7 @@ func (p *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key []by
 }
 
 // tries to send a message, returns true if successful
-func (p *Pss) trySend(sp *network.Peer, msg *PssMsg) bool {
+func trySendMsg(p *Pss, sp *network.Peer, msg *PssMsg) bool {
 	var isPssEnabled bool
 	info := sp.Info()
 	for _, capability := range info.Caps {
@@ -915,16 +915,21 @@ func (p *Pss) trySend(sp *network.Peer, msg *PssMsg) bool {
 	return err == nil
 }
 
-// Forwards a pss message to the peer(s) closest to the to recipient address in the PssMsg struct
-// The recipient address can be of any length, and the byte slice will be matched to the MSB slice
-// of the peer address of the equivalent length.
+// Forwards a pss message to the peer(s) based on recipient address according to the algorithm
+// described below. The recipient address can be of any length, and the byte slice will be matched
+// to the MSB slice of the peer address of the equivalent length.
+//
 // If the recipient address (or partial address) is within the neighbourhood depth of the forwarding
 // node, then it will be forwarded to all the nearest neighbours of the forwarding node. In case of
 // partial address, it should be forwarded to all the peers matching the partial address, if there
 // 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) error {
+func (p *Pss) forward(msg *PssMsg, trySend func(p *Pss, sp *network.Peer, msg *PssMsg) bool) error {
+	if trySend == nil {
+		trySend = trySendMsg
+	}
+
 	metrics.GetOrRegisterCounter("pss.forward", nil).Inc(1)
 	sent := 0 // number of successful sends
 	to := make([]byte, addressLength)
@@ -940,14 +945,21 @@ func (p *Pss) forward(msg *PssMsg) error {
 		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 depth < luminosityRadius && depth < neighbourhoodDepth {
+		depth++
+	}
+
 	p.Kademlia.EachConn(to, addressLength*8, func(sp *network.Peer, po int, _ bool) bool {
 		if po < depth && sent > 0 {
 			return false // stop iterating
 		}
-		if p.trySend(sp, msg) {
+		if trySend(p, sp, msg) {
 			sent++
 		}
-		return true // continue
+		return po < addressLength*8 // stop iterating in case of exact match of full address
 	})
 
 	// 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)
+	ps.forward(pssmsg, nil)
 
 }