diff --git a/swarm/pss/forwarding_test.go b/swarm/pss/forwarding_test.go new file mode 100644 index 0000000000..5706185e11 --- /dev/null +++ b/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 +} diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index ffdca8960a..f7fd0785c4 100644 --- a/swarm/pss/pss.go +++ b/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 diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 32404aaaf9..3aeac2e5ea 100644 --- a/swarm/pss/pss_test.go +++ b/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) }