From b7286a99093b566a2e282291ab13a40b6ef1315b Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 6 Sep 2018 13:13:31 +0200 Subject: [PATCH] swarm, swarm/network/stream, swarm/netork/simulations,, swarm/pss: adapt to new Kad API --- .../simulations/discovery/discovery_test.go | 4 +- swarm/network/stream/delivery.go | 12 ++--- swarm/network/stream/snapshot_sync_test.go | 9 +--- swarm/network/stream/stream.go | 18 ++++--- swarm/network_test.go | 3 +- swarm/pss/pss.go | 43 +++++++---------- swarm/pss/pss_test.go | 47 ++++++------------- swarm/swarm.go | 2 +- 8 files changed, 53 insertions(+), 85 deletions(-) diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index acf3479e52..913d6d8373 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -556,8 +556,8 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) { kp.MinProxBinSize = testMinProxBinSize if ctx.Config.Reachable != nil { - kp.Reachable = func(o network.OverlayAddr) bool { - return ctx.Config.Reachable(o.(*network.BzzAddr).ID()) + kp.Reachable = func(o *network.BzzAddr) bool { + return ctx.Config.Reachable(o.ID()) } } kad := network.NewKademlia(addr.Over(), kp) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 36040339d3..6273525355 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -47,15 +47,15 @@ var ( type Delivery struct { db *storage.DBAPI - overlay network.Overlay + kad *network.Kademlia receiveC chan *ChunkDeliveryMsg getPeer func(discover.NodeID) *Peer } -func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery { +func NewDelivery(kad *network.Kademlia, db *storage.DBAPI) *Delivery { d := &Delivery{ db: db, - overlay: overlay, + kad: kad, receiveC: make(chan *ChunkDeliveryMsg, deliveryCap), } @@ -172,7 +172,7 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req * t := time.NewTimer(10 * time.Minute) defer t.Stop() - log.Debug("waiting delivery", "peer", sp.ID(), "hash", req.Addr, "node", common.Bytes2Hex(d.overlay.BaseAddr()), "created", created) + log.Debug("waiting delivery", "peer", sp.ID(), "hash", req.Addr, "node", common.Bytes2Hex(d.kad.BaseAddr()), "created", created) start := time.Now() select { case <-chunk.ReqC: @@ -269,8 +269,8 @@ func (d *Delivery) RequestFromPeers(ctx context.Context, hash []byte, skipCheck var err error requestFromPeersCount.Inc(1) - d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool { - spId := p.(network.Peer).ID() + d.kad.EachConn(hash, 255, func(p *network.Peer, po int, nn bool) bool { + spId := p.ID() for _, p := range peersToSkip { if p == spId { log.Trace("Delivery.RequestFromPeers: skip peer", "peer", spId) diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 4e1ab09fcf..313019d6a1 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -457,15 +457,10 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error { //returns the number of subscriptions requested func startSyncing(r *Registry, conf *synctestConfig) (int, error) { var err error - - kad, ok := r.delivery.overlay.(*network.Kademlia) - if !ok { - return 0, fmt.Errorf("Not a Kademlia!") - } - + kad := r.delivery.kad subCnt := 0 //iterate over each bin and solicit needed subscription to bins - kad.EachBin(r.addr.Over(), pof, 0, func(conn network.OverlayConn, po int) bool { + kad.EachBin(r.addr.Over(), pof, 0, func(conn *network.Peer, po int) bool { //identify begin and start index of the bin(s) we want to subscribe to histRange := &Range{} diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index cd0580a0c0..21620c501d 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -130,7 +130,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i // wait for kademlia table to be healthy time.Sleep(options.SyncUpdateDelay) - kad := streamer.delivery.overlay.(*network.Kademlia) + kad := streamer.delivery.kad depthC := latestIntC(kad.NeighbourhoodDepthC()) addressBookSizeC := latestIntC(kad.AddrCountC()) @@ -398,9 +398,7 @@ func (r *Registry) Run(p *network.BzzPeer) error { // and they are no longer required after iteration, request to Quit // them will be send to appropriate peers. func (r *Registry) updateSyncing() { - // if overlay in not Kademlia, panic - kad := r.delivery.overlay.(*network.Kademlia) - + kad := r.delivery.kad // map of all SYNC streams for all peers // used at the and of the function to remove servers // that are not needed anymore @@ -421,8 +419,7 @@ func (r *Registry) updateSyncing() { r.peersMu.RUnlock() // request subscriptions for all nodes and bins - kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(conn network.OverlayConn, bin int) bool { - p := conn.(network.Peer) + kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(p *network.Peer, bin int) bool { log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin)) // bin is always less then 256 and it is safe to convert it to type uint8 @@ -461,10 +458,11 @@ func (r *Registry) updateSyncing() { func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { peer := protocols.NewPeer(p, rw, Spec) - bzzPeer := network.NewBzzTestPeer(peer, r.addr) - r.delivery.overlay.On(bzzPeer) - defer r.delivery.overlay.Off(bzzPeer) - return r.Run(bzzPeer) + bp := network.NewBzzTestPeer(peer, r.addr) + np := network.NewPeer(bp, r.delivery.kad) + r.delivery.kad.On(np) + defer r.delivery.kad.Off(np) + return r.Run(bp) } // HandleMsg is the message handler that delegates incoming messages diff --git a/swarm/network_test.go b/swarm/network_test.go index 176c635d82..9bc6143d84 100644 --- a/swarm/network_test.go +++ b/swarm/network_test.go @@ -34,7 +34,6 @@ import ( "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/swarm/api" - "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/simulation" "github.com/ethereum/go-ethereum/swarm/storage" colorable "github.com/mattn/go-colorable" @@ -293,7 +292,7 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa if err != nil { return nil, cleanup, err } - bucket.Store(simulation.BucketKeyKademlia, swarm.bzz.Hive.Overlay.(*network.Kademlia)) + bucket.Store(simulation.BucketKeyKademlia, swarm.bzz.Hive.Kademlia) log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr())) return swarm, cleanup, nil }, diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 8459211ddb..14285f7f1a 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -110,10 +110,10 @@ func (params *PssParams) WithPrivateKey(privatekey *ecdsa.PrivateKey) *PssParams // // Implements node.Service type Pss struct { - network.Overlay // we can get the overlayaddress from this - privateKey *ecdsa.PrivateKey // pss can have it's own independent key - w *whisper.Whisper // key and encryption backend - auxAPIs []rpc.API // builtins (handshake, test) can add APIs + *network.Kademlia // we can get the Kademliaaddress from this + privateKey *ecdsa.PrivateKey // pss can have it's own independent key + w *whisper.Whisper // key and encryption backend + auxAPIs []rpc.API // builtins (handshake, test) can add APIs // sending and forwarding fwdPool map[string]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer @@ -151,9 +151,9 @@ func (p *Pss) String() string { // Creates a new Pss instance. // -// In addition to params, it takes a swarm network overlay +// In addition to params, it takes a swarm network Kademlia // and a FileStore storage for message cache storage. -func NewPss(k network.Overlay, params *PssParams) (*Pss, error) { +func NewPss(k *network.Kademlia, params *PssParams) (*Pss, error) { if params.privateKey == nil { return nil, errors.New("missing private key for pss") } @@ -162,7 +162,7 @@ func NewPss(k network.Overlay, params *PssParams) (*Pss, error) { Version: pssVersion, } ps := &Pss{ - Overlay: k, + Kademlia: k, privateKey: params.privateKey, w: whisper.New(&whisper.DefaultConfig), quitC: make(chan struct{}), @@ -290,9 +290,9 @@ func (p *Pss) addAPI(api rpc.API) { p.auxAPIs = append(p.auxAPIs, api) } -// Returns the swarm overlay address of the pss node +// Returns the swarm Kademlia address of the pss node func (p *Pss) BaseAddr() []byte { - return p.Overlay.BaseAddr() + return p.Kademlia.BaseAddr() } // Returns the pss node's public key @@ -356,11 +356,11 @@ func (p *Pss) handlePssMsg(ctx context.Context, msg interface{}) error { } if int64(pssmsg.Expire) < time.Now().Unix() { metrics.GetOrRegisterCounter("pss.expire", nil).Inc(1) - log.Warn("pss filtered expired message", "from", common.ToHex(p.Overlay.BaseAddr()), "to", common.ToHex(pssmsg.To)) + log.Warn("pss filtered expired message", "from", common.ToHex(p.Kademlia.BaseAddr()), "to", common.ToHex(pssmsg.To)) return nil } if p.checkFwdCache(pssmsg) { - log.Trace("pss relay block-cache match (process)", "from", common.ToHex(p.Overlay.BaseAddr()), "to", (common.ToHex(pssmsg.To))) + log.Trace("pss relay block-cache match (process)", "from", common.ToHex(p.Kademlia.BaseAddr()), "to", (common.ToHex(pssmsg.To))) return nil } p.addFwdCache(pssmsg) @@ -442,12 +442,12 @@ func (p *Pss) executeHandlers(topic Topic, payload []byte, from *PssAddress, asy // will return false if using partial address func (p *Pss) isSelfRecipient(msg *PssMsg) bool { - return bytes.Equal(msg.To, p.Overlay.BaseAddr()) + return bytes.Equal(msg.To, p.Kademlia.BaseAddr()) } -// test match of leftmost bytes in given message to node's overlay address +// test match of leftmost bytes in given message to node's Kademlia address func (p *Pss) isSelfPossibleRecipient(msg *PssMsg) bool { - local := p.Overlay.BaseAddr() + local := p.Kademlia.BaseAddr() return bytes.Equal(msg.To[:], local[:len(msg.To)]) } @@ -816,14 +816,7 @@ func (p *Pss) forward(msg *PssMsg) error { // send with kademlia // find the closest peer to the recipient and attempt to send sent := 0 - p.Overlay.EachConn(to, 256, func(op network.OverlayConn, po int, isproxbin bool) bool { - // we need p2p.protocols.Peer.Send - // cast and resolve - sp, ok := op.(senderPeer) - if !ok { - log.Crit("Pss cannot use kademlia peer type") - return false - } + p.Kademlia.EachConn(to, 256, func(sp *network.Peer, po int, isproxbin bool) bool { info := sp.Info() // check if the peer is running pss @@ -840,7 +833,7 @@ func (p *Pss) forward(msg *PssMsg) error { } // get the protocol peer from the forwarding peer cache - sendMsg := fmt.Sprintf("MSG TO %x FROM %x VIA %x", to, p.BaseAddr(), op.Address()) + sendMsg := fmt.Sprintf("MSG TO %x FROM %x VIA %x", to, p.BaseAddr(), sp.Address()) p.fwdPoolMu.RLock() pp := p.fwdPool[sp.Info().ID] p.fwdPoolMu.RUnlock() @@ -859,11 +852,11 @@ func (p *Pss) forward(msg *PssMsg) error { // - if the peer is end recipient but the full address has not been disclosed // - if the peer address matches the partial address fully // - if the peer is in proxbin - if len(msg.To) < addressLength && bytes.Equal(msg.To, op.Address()[:len(msg.To)]) { + if len(msg.To) < addressLength && bytes.Equal(msg.To, sp.Address()[:len(msg.To)]) { log.Trace(fmt.Sprintf("Pss keep forwarding: Partial address + full partial match")) return true } else if isproxbin { - log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ToHex(op.Address()))) + log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ToHex(sp.Address()))) return true } // at this point we stop forwarding, and the state is as follows: diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 41b03db286..6ba04cb5d1 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -556,23 +556,6 @@ OUTER: } } -type pssTestPeer struct { - *protocols.Peer - addr []byte -} - -func (t *pssTestPeer) Address() []byte { - return t.addr -} - -func (t *pssTestPeer) Update(addr network.OverlayAddr) network.OverlayAddr { - return addr -} - -func (t *pssTestPeer) Off() network.OverlayAddr { - return &pssTestPeer{} -} - // forwarding should skip peers that do not have matching pss capabilities func TestMismatch(t *testing.T) { @@ -582,7 +565,7 @@ func TestMismatch(t *testing.T) { t.Fatal(err) } - // initialize overlay + // initialize kad baseaddr := network.RandomAddr() kad := network.NewKademlia((baseaddr).Over(), network.NewKadParams()) rw := &p2p.MsgPipeRW{} @@ -594,10 +577,10 @@ func TestMismatch(t *testing.T) { Version: 0, } nid, _ := discover.HexID("0x01") - wrongpsspeer := &pssTestPeer{ - Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(wrongpssaddr.Over()), []p2p.Cap{wrongpsscap}), rw, nil), - addr: wrongpssaddr.Over(), - } + wrongpsspeer := network.NewPeer(&network.BzzPeer{ + Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(wrongpssaddr.Over()), []p2p.Cap{wrongpsscap}), rw, nil), + BzzAddr: &network.BzzAddr{OAddr: wrongpssaddr.Over(), UAddr: nil}, + }, kad) // one peer doesn't even have pss (boo!) nopssaddr := network.RandomAddr() @@ -606,16 +589,16 @@ func TestMismatch(t *testing.T) { Version: 1, } nid, _ = discover.HexID("0x02") - nopsspeer := &pssTestPeer{ - Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(nopssaddr.Over()), []p2p.Cap{nopsscap}), rw, nil), - addr: nopssaddr.Over(), - } + nopsspeer := network.NewPeer(&network.BzzPeer{ + Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(nopssaddr.Over()), []p2p.Cap{nopsscap}), rw, nil), + BzzAddr: &network.BzzAddr{OAddr: nopssaddr.Over(), UAddr: nil}, + }, kad) // add peers to kademlia and activate them // it's safe so don't check errors - kad.Register([]network.OverlayAddr{wrongpsspeer}) + kad.Register(wrongpsspeer.BzzAddr) kad.On(wrongpsspeer) - kad.Register([]network.OverlayAddr{nopsspeer}) + kad.Register(nopsspeer.BzzAddr) kad.On(nopsspeer) // create pss @@ -1636,17 +1619,17 @@ func newServices(allowRaw bool) adapters.Services { } } -func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *PssParams) *Pss { +func newTestPss(privkey *ecdsa.PrivateKey, kad *network.Kademlia, ppextra *PssParams) *Pss { var nid discover.NodeID copy(nid[:], crypto.FromECDSAPub(&privkey.PublicKey)) addr := network.NewAddrFromNodeID(nid) // set up routing if kademlia is not passed to us - if overlay == nil { + if kad == nil { kp := network.NewKadParams() kp.MinProxBinSize = 3 - overlay = network.NewKademlia(addr.Over(), kp) + kad = network.NewKademlia(addr.Over(), kp) } // create pss @@ -1654,7 +1637,7 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss if ppextra != nil { pp.SymKeyCacheCapacity = ppextra.SymKeyCacheCapacity } - ps, err := NewPss(overlay, pp) + ps, err := NewPss(kad, pp) if err != nil { return nil } diff --git a/swarm/swarm.go b/swarm/swarm.go index 736cd37de8..baf71b9623 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -356,7 +356,7 @@ func (self *Swarm) Start(srv *p2p.Server) error { log.Error("bzz failed", "err", err) return err } - log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.Overlay.BaseAddr())) + log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.BaseAddr())) if self.ps != nil { self.ps.Start(srv)