From a924c8572e521e849b96eda373767b8d97bbf2e4 Mon Sep 17 00:00:00 2001 From: nolash Date: Thu, 18 May 2017 06:11:59 +0200 Subject: [PATCH] swarm/pss, swarm/network, p2p/simulations: pss prototest fix --- p2p/simulations/adapters/state.go | 20 ++++++++++++++++++++ swarm/network/protocol.go | 8 ++++---- swarm/pss/common.go | 24 ++++++++++++++++++++++-- swarm/pss/pss.go | 25 +++++++++++++++---------- swarm/pss/pss_test.go | 31 ++++++++++++++++++++++++------- 5 files changed, 85 insertions(+), 23 deletions(-) create mode 100644 p2p/simulations/adapters/state.go diff --git a/p2p/simulations/adapters/state.go b/p2p/simulations/adapters/state.go new file mode 100644 index 0000000000..2d7882c86b --- /dev/null +++ b/p2p/simulations/adapters/state.go @@ -0,0 +1,20 @@ +package adapters + +type SimStateStore struct { + m map[string][]byte +} + +func (self *SimStateStore) Load(s string) ([]byte, error) { + return self.m[s], nil +} + +func (self *SimStateStore) Save(s string, data []byte) error { + self.m[s] = data + return nil +} + +func NewSimStateStore() *SimStateStore { + return &SimStateStore{ + make(map[string][]byte), + } +} diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 3901eadc2c..7753a69b1c 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -347,15 +347,15 @@ func NewNodeIdFromAddr(addr Addr) *adapters.NodeId { // NewAddrFromNodeId constucts a bzzAddr from an adapters.NodeId // the overlay address is derived as the hash of the nodeId -func NewAddrFromNodeId(n *adapters.NodeId) *bzzAddr { +func NewAddrFromNodeId(n *adapters.NodeId) Addr { id := n.NodeID return &bzzAddr{ - OAddr: ToOverlayAddr(n), + OAddr: ToOverlayAddr(n.Bytes()), UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()), } } // ToOverlayAddr creates an overlayaddress from NodeID -func ToOverlayAddr(id *adapters.NodeId) []byte { - return crypto.Keccak256(id.Bytes()) +func ToOverlayAddr(id []byte) []byte { + return crypto.Keccak256(id) } diff --git a/swarm/pss/common.go b/swarm/pss/common.go index ea0a9bf90a..c3f03698ec 100644 --- a/swarm/pss/common.go +++ b/swarm/pss/common.go @@ -68,7 +68,7 @@ func newTestPss(addr []byte) *Pss { return ps } -func newPssPingMsg(ps *Pss, spec *protocols.Spec, topic PssTopic, senderaddr []byte) PssMsg { +func newPssPingMsg(ps *Pss, to []byte, spec *protocols.Spec, topic PssTopic, senderaddr []byte) PssMsg { data := pssPingMsg{ Created: time.Now(), } @@ -83,7 +83,7 @@ func newPssPingMsg(ps *Pss, spec *protocols.Spec, topic PssTopic, senderaddr []b } pssmsg := PssMsg{ - To: ps.Overlay.BaseAddr(), + To: to, Payload: NewPssEnvelope(senderaddr, topic, rlpbundle), } @@ -103,3 +103,23 @@ func newPssPingProtocol(handler func (interface{}) error) *p2p.Protocol { }, } } + +type testPssPeer struct { + *protocols.Peer + addr []byte +} + +func (self *testPssPeer) Address() []byte { + return self.addr +} + +func (self *testPssPeer) Off() network.OverlayAddr { + return self +} + +func (self *testPssPeer) Drop(err error) { +} + +func (self *testPssPeer) Update(o network.OverlayAddr) network.OverlayAddr { + return self +} diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 26ac098cae..36873fcee0 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -37,6 +37,11 @@ var ( errorForwardToSelf = errors.New("forward to self") ) +type senderPeer interface { + Address() []byte + Send(interface{}) error +} + // Defines params for Pss type PssParams struct { Cachettl time.Duration @@ -99,7 +104,7 @@ func (msg *PssMsg) serialize() []byte { } -var pssTransportProtocol = &protocols.Spec{ +var pssSpec = &protocols.Spec{ Name: "pss", Version: 1, MaxMsgSize: 10 * 1024 * 1024, @@ -188,16 +193,16 @@ func (self *Pss) Stop() error { func (self *Pss) Protocols() []p2p.Protocol { return []p2p.Protocol{ p2p.Protocol{ - Name: pssTransportProtocol.Name, - Version: pssTransportProtocol.Version, - Length: pssTransportProtocol.Length(), + Name: pssSpec.Name, + Version: pssSpec.Version, + Length: pssSpec.Length(), Run: self.Run, }, } } func (self *Pss) Run(p *p2p.Peer, rw p2p.MsgReadWriter) error { - pp := protocols.NewPeer(p, rw, pssTransportProtocol) + pp := protocols.NewPeer(p, rw, pssSpec) return pp.Run(self.handlePssMsg) } @@ -371,13 +376,13 @@ func (self *Pss) Forward(msg *PssMsg) error { // send with kademlia // find the closest peer to the recipient and attempt to send self.Overlay.EachConn(msg.To, 256, func(op network.OverlayConn, po int, isproxbin bool) bool { - p, ok := op.(network.Peer) + p, ok := op.(senderPeer) if !ok { return true } addr := self.Overlay.BaseAddr() - sendMsg := fmt.Sprintf("%x: msg to %x via %x", common.ByteLabel(addr), common.ByteLabel(msg.To), common.ByteLabel(p.Over())) - if self.checkFwdCache(p.Over(), digest) { + sendMsg := fmt.Sprintf("%x: msg to %x via %x", common.ByteLabel(addr), common.ByteLabel(msg.To), common.ByteLabel(p.Address())) + if self.checkFwdCache(p.Address(), digest) { log.Info(fmt.Sprintf("%v: peer already forwarded to", sendMsg)) return true } @@ -389,10 +394,10 @@ func (self *Pss) Forward(msg *PssMsg) error { log.Trace(fmt.Sprintf("%v: successfully forwarded", sendMsg)) sent++ // if equality holds, p is always the first peer given in the iterator - if bytes.Equal(msg.To, p.Over()) || !isproxbin { + if bytes.Equal(msg.To, p.Address()) || !isproxbin { return false } - log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(p.Over()))) + log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(p.Address()))) return true }) diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 1a02a70067..d9849d042a 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -13,6 +13,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" @@ -131,7 +132,7 @@ func TestPssRegisterHandler(t *testing.T) { ps := newTestPss(addr.OAddr) from := network.RandomAddr() payload := []byte("payload") - topic := NewTopic(pssTransportProtocol.Name, int(pssTransportProtocol.Version)) + topic := NewTopic(pssSpec.Name, int(pssSpec.Version)) wrongtopic := NewTopic("foo", 42) checkMsg := func(msg []byte, p *p2p.Peer, sender []byte) error { if !bytes.Equal(from.OAddr, sender) { @@ -172,23 +173,36 @@ func TestPssRegisterHandler(t *testing.T) { } func TestPssSimpleLinear(t *testing.T) { + var err error nodeconfig := adapters.RandomNodeConfig() addr := network.NewAddrFromNodeId(nodeconfig.Id) _ = p2ptest.NewTestPeerPool() - ps := newTestPss(addr.OAddr) + ps := newTestPss(addr.Over()) ping := &pssPing{ quitC: make(chan struct{}), } - err := RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler)) + err = RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler)) if err != nil { t.Fatalf("Failed to register virtual protocol in pss: %v", err) } - pt := p2ptest.NewProtocolTester(t, nodeconfig.Id, 2, ps.Run) + run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { + id := p.ID() + bp := &testPssPeer{ + Peer: protocols.NewPeer(p, rw, pssSpec), + addr: network.ToOverlayAddr(id[:]), + } + ps.Overlay.On(bp) + defer ps.Overlay.Off(bp) + log.Debug(fmt.Sprintf("%v", ps.Overlay)) + return bp.Run(ps.handlePssMsg) + } - msg := newPssPingMsg(ps, pssPingProtocol, pssPingTopic, []byte{1, 2, 3}) + pt := p2ptest.NewProtocolTester(t, nodeconfig.Id, 2, run) + + msg := newPssPingMsg(ps, network.ToOverlayAddr(pt.Ids[0].Bytes()), pssPingProtocol, pssPingTopic, []byte{1, 2, 3}) exchange := p2ptest.Exchange{ Expects: []p2ptest.Expect{ @@ -207,7 +221,10 @@ func TestPssSimpleLinear(t *testing.T) { }, } - pt.TestExchanges(exchange) + err = pt.TestExchanges(exchange) + if err != nil { + t.Fatalf("exchange failed %v", err) + } } func TestPssFullRandom10_5_5(t *testing.T) { @@ -367,7 +384,7 @@ func newServices() adapters.Services { kadparams.MaxRetries = 1000 kadparams.RetryExponent = 2 kadparams.RetryInterval = 1000000 - kademlia := network.NewKademlia(addr.OAddr, kadparams) + kademlia := network.NewKademlia(addr.Over(), kadparams) config := &network.BzzConfig{ OverlayAddr: addr.Over(),