From c870e2be26f18896700c873a7fee4b652025a1f1 Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 11 May 2017 18:07:10 -0700 Subject: [PATCH] WIP network rewrite --- p2p/protocols/protocol.go | 44 ++- p2p/simulations/adapters/inproc.go | 3 +- p2p/simulations/network.go | 1 + pot/address.go | 59 +++ pot/pot.go | 7 +- swarm/network/discovery.go | 64 ++-- swarm/network/discovery_test.go | 3 +- swarm/network/hive.go | 250 +++++++----- swarm/network/hive_test.go | 29 +- swarm/network/kademlia.go | 546 +++++++++++++-------------- swarm/network/kademlia_test.go | 36 +- swarm/network/protocol.go | 159 ++++---- swarm/network/protocol_test.go | 30 +- swarm/network/pss.go | 37 +- swarm/network/pss_test.go | 90 +++-- swarm/network/simulations/overlay.go | 56 ++- swarm/network/test_overlay.go | 371 +++++++++--------- swarm/swarm.go | 74 ++-- 18 files changed, 1024 insertions(+), 835 deletions(-) diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index cf86d8dd36..12fce6b93d 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -135,21 +135,25 @@ func (self *CodeMap) GetCode(msg interface{}) (uint64, bool) { return code, found } +// NewCodeMap construct the code to type map for the protocol func NewCodeMap(name string, version uint, maxMsgSize int) *CodeMap { return &CodeMap{ Name: name, Version: version, MaxMsgSize: maxMsgSize, messages: make(map[reflect.Type]uint64), + codes: make(map[uint64]reflect.Type), } } +// Length returns the current highes codepos + 1 func (self *CodeMap) Length() uint64 { return uint64(self.codepos) } +// Register defines a new series of codes starting on series, incrementing func (self *CodeMap) Register(series int, msgs ...interface{}) { - code := series + self.codepos = series for _, msg := range msgs { typ := reflect.TypeOf(msg) _, found := self.messages[typ] @@ -196,7 +200,7 @@ type Peer struct { rw p2p.MsgReadWriter // p2p.MsgReadWriter to send messages to and read messages from handlers map[reflect.Type][]func(interface{}) error // message type -> message handler callback(s) map Errc chan error - wErrc chan error // write error channel + ready chan bool // blocking send until handshake finishes } // NewPeer returns a new peer @@ -204,12 +208,14 @@ type Peer struct { // the first two arguments are comming the arguments passed to p2p.Protocol.Run function // the third argument is the CodeMap describing the protocol messages and options func NewPeer(p *p2p.Peer, ct *CodeMap, rw p2p.MsgReadWriter) *Peer { + ready := make(chan bool) + defer close(ready) return &Peer{ ct: ct, Peer: p, rw: rw, Errc: make(chan error), - wErrc: make(chan error), + ready: ready, handlers: make(map[reflect.Type][]func(interface{}) error), } } @@ -269,11 +275,17 @@ func (self *Peer) Drop(err error) { // this low level call will be wrapped by libraries providing routed or broadcast sends // but often just used to forward and push messages to directly connected peers func (self *Peer) Send(msg interface{}) error { + <-self.ready + return self.send(msg) +} + +func (self *Peer) send(msg interface{}) error { code, found := self.ct.GetCode(msg) if !found { return errorf(ErrInvalidMsgType, "%v", code) } log.Trace(fmt.Sprintf("=> msg #%d TO %v : %v", code, self.ID(), msg)) + return p2p.Send(self.rw, uint64(code), msg) } @@ -350,29 +362,25 @@ func (self *Peer) Handshake(hs interface{}, handshakeTimeout time.Duration) (rhs if !found { return nil, errorf(ErrHandshake, "unknown handshake message type: %v", typ) } - errc := make(chan error, 1) - go func() { - if err := self.Send(hs); err != nil { - errc <- errorf(ErrHandshake, "cannot send: %v", err) - } - }() - - hsc := make(chan interface{}) + self.ready = make(chan bool) + received := make(chan bool) + defer close(self.ready) go func() { + defer close(received) // receiving and validating remote handshake, expect code - rhs, err := self.handleIncoming() + rhs, err = self.handleIncoming() if err != nil { - errc <- errorf(ErrHandshake, "'%v': %v", self.ct.Name, err) - return + err = errorf(ErrHandshake, "'%v': %v", self.ct.Name, err) } - hsc <- rhs }() + if e := self.send(hs); e != nil { + return nil, errorf(ErrHandshake, "cannot send: %v", e) + } select { - case err = <-errc: - case rhs = <-hsc: + case <-received: case <-time.NewTimer(handshakeTimeout).C: - err = errorf(ErrHandshake, "timeout") + err = errorf(ErrHandshake, "timeout after %v", handshakeTimeout) } return rhs, err } diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 6f27607cf4..6a851fda29 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -211,7 +211,8 @@ func (self *SimNode) startRPC(service node.Service) error { self.lock.Lock() defer self.lock.Unlock() if self.client != nil { - return errors.New("RPC already started") + return nil + // return errors.New("RPC already started") } // add SimAdminAPI so that the network can call the diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 0adc696ca8..4254ff4ba0 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -275,6 +275,7 @@ func (self *Network) startWithSnapshot(id *adapters.NodeId, snapshot []byte) err } log.Trace(fmt.Sprintf("starting node %v: %v using %v", id, node.Up, self.nodeAdapter.Name())) if err := node.Start(snapshot); err != nil { + log.Warn(fmt.Sprintf("start up failed: %v", err)) return err } node.Up = true diff --git a/pot/address.go b/pot/address.go index 4e4a76d03e..e638a11c7d 100644 --- a/pot/address.go +++ b/pot/address.go @@ -287,3 +287,62 @@ func (self *BoolAddress) PO(val PotVal, pos int) (po int, eq bool) { } return po, true } + +type BytesAddress interface { + Bytes() []byte +} + +type bytesAddress struct { + bytes []byte + toBytes func(v AnyVal) []byte +} + +func NewBytesVal(v AnyVal, f func(v AnyVal) []byte) *bytesAddress { + if f == nil { + f = ToBytes + } + b := f(v) + return &bytesAddress{b, f} +} + +func ToBytes(v AnyVal) []byte { + b, ok := v.([]byte) + if !ok { + ba, ok := v.(BytesAddress) + if !ok { + panic(fmt.Sprintf("unsupported value type %T", v)) + } + b = ba.Bytes() + } + return b +} + +func (a *bytesAddress) String() string { + return fmt.Sprintf("%08b", a.bytes) +} +func (a *bytesAddress) Bytes() []byte { + return a.bytes +} + +func (a *bytesAddress) PO(val PotVal, i int) (int, bool) { + return proximityOrder(a.bytes, a.toBytes(val), i) +} + +func proximityOrder(one, other []byte, pos int) (int, bool) { + for i := pos / 8; i < len(one); i++ { + if one[i] == other[i] { + continue + } + oxo := one[i] ^ other[i] + start := 0 + if i == pos/8 { + start = pos % 8 + } + for j := start; j < 8; j++ { + if (uint8(oxo)>>uint8(7-j))&0x01 != 0 { + return i*8 + j, false + } + } + } + return len(one) * 8, true +} diff --git a/pot/pot.go b/pot/pot.go index bb1d4c936e..eef5037b4e 100644 --- a/pot/pot.go +++ b/pot/pot.go @@ -45,6 +45,8 @@ type PotVal interface { String() string } +type AnyVal interface{} + // Pot constructor. Requires value of type PotVal to pin // and po to point to a span in the PotVal key // The pinned item counts towards the size @@ -245,11 +247,12 @@ func remove(t *pot, val PotVal) (r *pot, po int, found bool) { // if f returns v' <> v then v' is inserted into the Pot // if v' == v the pot is not changed // it panics if v'.PO(k, 0) says v and k are not equal -func (t *Pot) Swap(val PotVal, f func(v PotVal) PotVal) (po int, found bool, change bool) { +func (t *Pot) Swap(val AnyVal, f func(v PotVal) PotVal) (po int, found bool, change bool) { t.lock.Lock() defer t.lock.Unlock() + ba := NewBytesVal(val, nil) var t0 *pot - t0, po, found, change = swap(t.pot, val, f) + t0, po, found, change = swap(t.pot, ba, f) if change { t.pot = t0 } diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go index 91ad122c6b..55a75ed858 100644 --- a/swarm/network/discovery.go +++ b/swarm/network/discovery.go @@ -16,7 +16,7 @@ var DiscoveryMsgs = []interface{}{ } type discPeer struct { - Peer + *bzzPeer overlay Overlay peers map[string]bool proxLimit uint8 // the proximity radius advertised by remote to subscribe to peers @@ -25,10 +25,10 @@ type discPeer struct { // discovery peer contructor // registers the handlers for discovery messages -func NewDiscovery(p Peer, o Overlay) *discPeer { +func NewDiscovery(p *bzzPeer, o Overlay) *discPeer { self := &discPeer{ overlay: o, - Peer: p, + bzzPeer: p, peers: make(map[string]bool), } self.seen(self) @@ -42,15 +42,15 @@ func NewDiscovery(p Peer, o Overlay) *discPeer { // NotifyPeer notifies the receiver remote end of a peer p or PO po. // callback for overlay driver -func (self *discPeer) NotifyPeer(p Peer, po uint8) error { +func (self *discPeer) NotifyPeer(p OverlayPeer, po uint8) error { log.Warn(fmt.Sprintf("peer %#v peers %v", p, self.peers)) if po < self.proxLimit || self.seen(p) { return nil } - log.Warn(fmt.Sprintf("notification about %x", p.OverlayAddr())) + log.Warn(fmt.Sprintf("notification about %x", p.Address())) resp := &peersMsg{ - Peers: []*peerAddr{&peerAddr{OAddr: p.OverlayAddr(), UAddr: p.UnderlayAddr()}}, // perhaps the PeerAddr interface is unnecessary generalization + Peers: []*bzzAddr{ToAddr(p)}, // perhaps the PeerAddr interface is unnecessary generalization } return self.Send(resp) } @@ -83,7 +83,7 @@ disconnected // used for communicating about known peers // relevant for bootstrapping connectivity and updating peersets type peersMsg struct { - Peers []*peerAddr + Peers []*bzzAddr } func (self peersMsg) String() string { @@ -113,14 +113,15 @@ func (self *discPeer) handleSubPeersMsg(msg interface{}) error { spm := msg.(*subPeersMsg) self.proxLimit = spm.ProxLimit if !self.sentPeers { - var peers []*peerAddr - self.overlay.EachLivePeer(self.OverlayAddr(), 255, func(p Peer, po int, isproxbin bool) bool { + var peers []*bzzAddr + self.overlay.EachConn(self.Over(), 255, func(p OverlayConn, po int, isproxbin bool) bool { if uint8(po) < self.proxLimit { return false } log.Warn(fmt.Sprintf("peer %#v proxlimit %v", p, self.proxLimit)) - self.seen(p.(*discPeer).Peer) - peers = append(peers, &peerAddr{p.OverlayAddr(), p.UnderlayAddr()}) + if !self.seen(p) { + peers = append(peers, ToAddr(p)) + } return true }) log.Warn(fmt.Sprintf("found initial %v peers not farther than %v", len(peers), self.proxLimit)) @@ -139,35 +140,37 @@ func (self *discPeer) handleSubPeersMsg(msg interface{}) error { // Register interface method func (self *discPeer) handlePeersMsg(msg interface{}) error { // register all addresses - var nas []PeerAddr - for _, na := range msg.(*peersMsg).Peers { - addr := PeerAddr(na) - nas = append(nas, addr) - self.seen(addr) - } - - if len(nas) == 0 { + as := msg.(*peersMsg).Peers + if len(as) == 0 { log.Debug(fmt.Sprintf("whoops, no peers in incoming peersMsg from %v", self)) return nil } - log.Debug(fmt.Sprintf("got peer addresses from %x, %v (%v)", self.OverlayAddr(), nas, len(nas))) - return self.overlay.Register(nas...) + + var c chan OverlayAddr + go func() { + for _, a := range as { + self.seen(a) + c <- a + } + }() + return self.overlay.Register(c) } // handleGetPeersMsg is called by the protocol when receiving a // peerset (for target address) request // peers suggestions are retrieved from the overlay topology driver -// using the EachLivePeer interface iterator method +// using the EachConn interface iterator method // peers sent are remembered throughout a session and not sent twice func (self *discPeer) handleGetPeersMsg(msg interface{}) error { - var peers []*peerAddr + var peers []*bzzAddr req := msg.(*getPeersMsg) i := 0 - self.overlay.EachLivePeer(self.OverlayAddr(), int(req.Order), func(n Peer, po int, isproxbin bool) bool { + self.overlay.EachConn(self.Over(), int(req.Order), func(p OverlayConn, po int, isproxbin bool) bool { i++ // only send peers we have not sent before in this session - if self.seen(n) { - peers = append(peers, &peerAddr{n.OverlayAddr(), n.UnderlayAddr()}) + a := ToAddr(p) + if self.seen(a) { + peers = append(peers, a) } return len(peers) < int(req.Max) }) @@ -189,9 +192,8 @@ func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) { } var i uint8 //var err error - k.EachLivePeer(nil, 255, func(n Peer, po int, isproxbin bool) bool { - log.Trace(fmt.Sprintf("%T sent to %v", req, n)) - if err := n.Send(req); err == nil { + k.EachConn(nil, 255, func(p OverlayConn, po int, isproxbin bool) bool { + if err := p.(Conn).Send(req); err == nil { i++ if i >= broadcastSize { return false @@ -202,8 +204,8 @@ func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) { log.Info(fmt.Sprintf("requesting bees of PO%03d from %v/%v (each max %v)", order, i, broadcastSize, maxPeers)) } -func (self *discPeer) seen(p PeerAddr) bool { - k := NodeId(p).NodeID.String() +func (self *discPeer) seen(p OverlayPeer) bool { + k := string(p.Address()) if self.peers[k] { return true } diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index d737b3e04f..22a61a04ea 100644 --- a/swarm/network/discovery_test.go +++ b/swarm/network/discovery_test.go @@ -18,7 +18,7 @@ func TestDiscovery(t *testing.T) { to := NewKademlia(addr.OAddr, NewKadParams()) ct := BzzCodeMap(DiscoveryMsgs...) - services := func(p Peer) error { + services := func(p *bzzPeer) error { dp := NewDiscovery(p, to) to.On(dp) log.Trace(fmt.Sprintf("kademlia on %v", p)) @@ -32,7 +32,6 @@ func TestDiscovery(t *testing.T) { defer s.Stop() s.runHandshakes() - // o := 0 s.TestExchanges(p2ptest.Exchange{ Label: "outgoing SubPeersMsg", Expects: []p2ptest.Expect{ diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 61642a1532..2ee8cd3b46 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -17,6 +17,7 @@ package network import ( + "encoding/json" "fmt" "sync" "time" @@ -39,38 +40,51 @@ and relay the peer request process to the Overlay module peer connections and disconnections are reported and registered to keep the nodetable uptodate */ + +// Overlay is the interface to Jaak ahd ka)a type Overlay interface { - Register(...PeerAddr) error + Register(chan OverlayAddr) error - On(Peer) - Off(Peer) + On(OverlayPeer) + Off(OverlayConn) - EachLivePeer([]byte, int, func(Peer, int, bool) bool) - EachPeer([]byte, int, func(PeerAddr, int) bool) + EachConn([]byte, int, func(OverlayConn, int, bool) bool) + EachAddr([]byte, int, func(OverlayAddr, int) bool) - SuggestPeer() (PeerAddr, int, bool) + SuggestPeer() (OverlayAddr, int, bool) String() string - GetAddr() PeerAddr + BaseAddr() []byte +} + +// ReadWriter interface to persist known peers, uses disk for real nodes +type ReadWriter interface { + ReadAll(string) ([]byte, error) + WriteAll(string, []byte) error } // Hive implements the PeerPool interface type Hive struct { - *HiveParams // settings - Overlay // the overlay topology driver - lock sync.Mutex - quit chan bool - toggle chan bool - more chan bool + *HiveParams // settings + Overlay // the overlay topology driver + RW ReadWriter // ReadWriter + + // bookkeeping + lock sync.Mutex + quit chan bool + toggle chan bool + more chan bool } +// HiveParams holds the config options to hive type HiveParams struct { - Discovery bool - PeersBroadcastSetSize uint8 - MaxPeersPerRequest uint8 - CallInterval uint + Discovery bool // if want discovery of not + PeersBroadcastSetSize uint8 // how many peers to use when relaying + MaxPeersPerRequest uint8 // max size for peer address batches + CallInterval uint // polling interval fir=== } +// NewHiveParams returns hive config with only the func NewHiveParams() *HiveParams { return &HiveParams{ Discovery: true, @@ -81,8 +95,8 @@ func NewHiveParams() *HiveParams { } // Hive constructor embeds both arguments -// HiveParams config parameters -// Overlay Topology Driver Interface +// HiveParams: config parameters +// Overlay: Topology Driver Interface func NewHive(params *HiveParams, overlay Overlay) *Hive { return &Hive{ HiveParams: params, @@ -91,11 +105,16 @@ func NewHive(params *HiveParams, overlay Overlay) *Hive { } // Start receives network info only at startup -// connectPeer is a function to connect to a peer based on its NodeID or enode URL +// server is used to connect to a peer based on its NodeID or enode URL // these are called on the p2p.Server which runs on the node // af() returns an arbitrary ticker channel -func (self *Hive) Start(server p2p.Server, af func() <-chan time.Time) error { - +// rw is a read writer for json configs +func (self *Hive) Start(server p2p.Server, af func() <-chan time.Time, rw ReadWriter) error { + if rw != nil { + if err := self.loadPeers(); err != nil { + return err + } + } self.toggle = make(chan bool) self.more = make(chan bool, 1) self.quit = make(chan bool) @@ -115,9 +134,9 @@ func (self *Hive) Start(server p2p.Server, af func() <-chan time.Time) error { if addr != nil { log.Info(fmt.Sprintf("========> connect to bee %v", addr)) - node, err := discover.ParseNode(NodeId(addr).NodeID.String()) + under, err := discover.ParseNode(string(addr.(Addr).Under())) if err == nil { - server.AddPeer(node) + server.AddPeer(under) } else { log.Error(fmt.Sprintf("===X====> connect to bee %v failed: invalid node URL: %v", addr, err)) } @@ -127,7 +146,7 @@ func (self *Hive) Start(server p2p.Server, af func() <-chan time.Time) error { want = want && self.Discovery if want { - go RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest) + RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest) } select { @@ -136,20 +155,98 @@ func (self *Hive) Start(server p2p.Server, af func() <-chan time.Time) error { case <-self.quit: return } - log.Info(fmt.Sprintf("%v", self)) + // log.Info(fmt.Sprintf("%v", self)) } }() return nil } +// Stop terminates the updateloop and saves the peers +func (self *Hive) Stop() { + if self.RW != nil { + self.savePeers() + } + // closing toggle channel quits the updateloop + close(self.quit) +} + +// default ticker, tickinterval is taken from KadParams.CallInterval func (self *Hive) ticker() <-chan time.Time { return time.NewTicker(time.Duration(self.CallInterval) * time.Millisecond).C } +// Add is called at the end of a successful protocol handshake +// to register a connected (live) peer +func (self *Hive) Add(p *bzzPeer) error { + defer self.wake() + dp := NewDiscovery(p, self.Overlay) + log.Debug(fmt.Sprintf("to add new bee %v", p)) + self.On(dp) + self.String() + log.Debug(fmt.Sprintf("%v", self)) + return nil +} + +// Remove called after peer is disconnected +func (self *Hive) Remove(p *bzzPeer) { + defer self.wake() + log.Debug(fmt.Sprintf("remove bee %v", p)) + self.Off(p) +} + +// NodeInfo function is used by the p2p.server RPC interface to display +// protocol specific node information +func (self *Hive) NodeInfo() interface{} { + return interface{}(self.String()) +} + +// PeerInfo function is used by the p2p.server RPC interface to display +// protocol specific information any connected peer referred to by their NodeID +func (self *Hive) PeerInfo(id discover.NodeID) interface{} { + self.lock.Lock() + defer self.lock.Unlock() + addr := NewAddrFromNodeId(adapters.NewNodeId(id[:])) + return interface{}(addr) +} + +// Healthy reports the health state of the kademlia connectivity +// +func (self *Hive) Healthy() bool { + // TODO: determine if we have enough peers to consider the network + // to be healthy + return true +} + +// wake triggers +func (self *Hive) wake() { + select { + case self.more <- true: + log.Trace("hive woken up") + case <-self.quit: + default: + log.Trace("hive already awake") + } +} + +// HexToBytes reads a hex string ontp +func HexToBytes(s string) []byte { + id := discover.MustHexID(s) + return id[:] +} + +// ToAddr returns the serialisable version of u +func ToAddr(pa OverlayPeer) *bzzAddr { + if addr, ok := pa.(*bzzAddr); ok { + return addr + } + return pa.(*bzzPeer).bzzAddr +} + // keepAlive is a forever loop // in its awake state it periodically triggers connection attempts // by writing to self.more until Kademlia Table is saturated // wake state is toggled by writing to self.toggle +// it goes to sleep mode if table is saturated // it restarts if the table becomes non-full again due to disconnections func (self *Hive) keepAlive(af func() <-chan time.Time) { log.Trace("keep alive loop started") @@ -173,63 +270,48 @@ func (self *Hive) keepAlive(af func() <-chan time.Time) { } } -// Add is called at the end of a successful protocol handshake -// to register a connected (live) peer -func (self *Hive) Add(p Peer) error { - defer self.wake() - dp := NewDiscovery(p, self.Overlay) - log.Debug(fmt.Sprintf("to add new bee %v", p)) - self.On(dp) - self.String() - log.Debug(fmt.Sprintf("%v", self)) +// loadPeers, savePeer implement persistence callback/ +func (self *Hive) loadPeers() error { + rw := self.RW + data, err := rw.ReadAll("peers") + if err != nil { + return err + } + if data == nil { + return nil + } + var as []*bzzAddr + if err := json.Unmarshal(data, &as); err != nil { + return err + } + + var c chan OverlayAddr + defer close(c) + go func() { + for _, a := range as { + c <- a + } + }() + return self.Overlay.Register(c) +} + +// savePeers, savePeer implement persistence callback/ +func (self *Hive) savePeers() error { + var peers []*bzzAddr + self.Overlay.EachAddr(nil, 256, func(pa OverlayAddr, i int) bool { + if pa == nil { + log.Warn(fmt.Sprintf("empty addr: %v", i)) + return true + } + peers = append(peers, ToAddr(pa)) + return true + }) + data, err := json.Marshal(peers) + if err != nil { + return fmt.Errorf("could not encode peers: %v", err) + } + if err := self.RW.WriteAll("peers", data); err != nil { + return fmt.Errorf("could not save peers: %v", err) + } return nil } - -// Remove called after peer is disconnected -func (self *Hive) Remove(p Peer) { - defer self.wake() - log.Debug(fmt.Sprintf("remove bee %v", p)) - self.Off(p) -} - -// NodeInfo function is used by the p2p.server RPC interface to display -// protocol specific node information -func (self *Hive) NodeInfo() interface{} { - return interface{}(self.String()) -} - -// PeerInfo function is used by the p2p.server RPC interface to display -// protocol specific information any connected peer referred to by their NodeID -func (self *Hive) PeerInfo(id discover.NodeID) interface{} { - self.lock.Lock() - defer self.lock.Unlock() - addr := NewPeerAddrFromNodeId(adapters.NewNodeId(id[:])) - return interface{}(addr) -} - -// Stop terminates the updateloop -func (self *Hive) Stop() { - // closing toggle channel quits the updateloop - close(self.quit) -} - -func (self *Hive) Healthy() bool { - // TODO: determine if we have enough peers to consider the network - // to be healthy - return true -} - -func (self *Hive) wake() { - select { - case self.more <- true: - log.Trace("hive woken up") - case <-self.quit: - default: - log.Trace("hive already awake") - } -} - -func HexToBytes(s string) []byte { - id := discover.MustHexID(s) - return id[:] -} diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index 54a8c9e3bf..aaa90e71a3 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -30,12 +30,13 @@ func (self *testConnect) connect(na string) error { func newHiveTester(t *testing.T, params *HiveParams) (*bzzTester, *Hive) { // setup - addr := RandomAddr() // tested peers peer address - to := NewTestOverlay(addr.OverlayAddr()) // overlay topology drive - pp := NewHive(params, to) // hive + addr := RandomAddr() // tested peers peer address + // to := NewTestOverlay(addr.Over()) // overlay topology drive + pp := NewHive(params, nil) // hive + // pp := NewHive(params, to) // hive ct := BzzCodeMap(DiscoveryMsgs...) // bzz protocol code map - services := func(p Peer) error { + services := func(p *bzzPeer) error { pp.Add(p) p.DisconnectHook(func(err error) { pp.Remove(p) @@ -53,14 +54,14 @@ func TestOverlayRegistration(t *testing.T) { defer s.Stop() id := s.Ids[0] - raddr := NewPeerAddrFromNodeId(id) + raddr := NewAddrFromNodeId(id) s.runHandshakes() // hive should have called the overlay - if pp.Overlay.(*testOverlay).posMap[string(raddr.OverlayAddr())] == nil { - t.Fatalf("Overlay#On not called on new peer") - } + // if pp.Overlay.(*testOverlay).posMap[string(raddr.Over())] == nil { + // t.Fatalf("Overlay#On not called on new peer") + // } } @@ -70,7 +71,7 @@ func TestRegisterAndConnect(t *testing.T) { defer s.Stop() id := s.Ids[0] - raddr := NewPeerAddrFromNodeId(id) + raddr := NewAddrFromNodeId(id) pp.Register(raddr) @@ -82,18 +83,18 @@ func TestRegisterAndConnect(t *testing.T) { }, ticker: make(chan time.Time), } - pp.Start(s, tc.ping) + pp.Start(s, tc.ping, nil) defer pp.Stop() tc.ticker <- time.Now() s.runHandshakes() - if pp.Overlay.(*testOverlay).posMap[string(raddr.OverlayAddr())] == nil { - t.Fatalf("Overlay#On not called on new peer") - } + // if pp.Overlay.(*testOverlay).posMap[string(raddr.Over())] == nil { + // t.Fatalf("Overlay#On not called on new peer") + // } // retrieve and broadcast - ord := order(raddr.OverlayAddr()) + ord := order(raddr.Over()) o := 0 if ord == 0 { o = 1 diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index 789575e804..72dc7c9ebf 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -46,24 +46,20 @@ a guaranteed constant maximum limit on the number of hops needed to reach one node from the other. */ -type KadDiscovery interface { - NotifyPeer(Peer, uint8) error - NotifyProx(uint8) error -} - +// KadParams holds the config params for Kademlia type KadParams struct { // adjustable parameters - MaxProxDisplay int - MinProxBinSize int - MinBinSize int - MaxBinSize int - RetryInterval int - RetryExponent int - MaxRetries int - PruneInterval int + MaxProxDisplay int // number of rows the table shows + MinProxBinSize int // nearest neighbour core minimum cardinality + MinBinSize int // minimum number of peers in a row + MaxBinSize int // maximum number of peers in a row before pruning + RetryInterval int // initial interval before a peer is first redialed + RetryExponent int // exponent to multiply retry intervals with + MaxRetries int // maximum number of redial attempts + PruneInterval int // interval between peer pruning cycles } -// NewKadParams() returns a params struct with default values +// NewKadParams returns a params struct with default values func NewKadParams() *KadParams { return &KadParams{ MaxProxDisplay: 8, @@ -79,141 +75,94 @@ func NewKadParams() *KadParams { // Kademlia is a table of live peers and a db of known peers type Kademlia struct { - addr PeerAddr // immutable baseaddress of the table - // addr *pot.HashAddress // immutable baseaddress of the table - *KadParams // Kademlia configuration parameters - addrs, peers *pot.Pot // pots container for peers - lastProxLimit uint8 // stores the last calculated proxlimit + *KadParams // Kademlia configuration parameters + base []byte // immutable baseaddress of the table + addrs *pot.Pot // pots container for known peer addresses + conns *pot.Pot // pots container for live peer connections + depth uint8 // stores the last calculated depth } -// NewKademlia(addr, params) creates a Kademlia table for base address addr +// NewKademlia creates a Kademlia table for base address addr // with parameters as in params // if params is nil, it uses default values func NewKademlia(addr []byte, params *KadParams) *Kademlia { if params == nil { params = NewKadParams() } - self := &Kademlia{ - addr: &peerAddr{OAddr: addr}, + return &Kademlia{ + base: addr, KadParams: params, addrs: pot.NewPot(nil, 0), - peers: pot.NewPot(nil, 0), + conns: pot.NewPot(nil, 0), } - return self } -// Prune implements a forever loop reacting to a ticker time channel given -// as the first argument -// the loop quits if the channel is closed -// it checks each kademlia bin and if the peer count is higher than -// the MaxBinSize parameter it drops the oldest n peers such that -// the bin is reduced to MinBinSize peers thus leaving slots to newly -// connecting peers -func (self *Kademlia) Prune(c <-chan time.Time) { - go func() { - for _ = range c { - log.Debug("pruning...") - total := 0 - self.peers.EachBin(self.addr, 0, func(po, size int, f func(func(pot.PotVal, int) bool) bool) bool { - extra := size - self.MinBinSize - if size > self.MaxBinSize { - n := 0 - f(func(v pot.PotVal, po int) bool { - p := v.(*KadPeer).Peer - if p != nil { - p.Drop(fmt.Errorf("bucket full")) - } - n++ - return n < extra - }) - total += extra - } - return true - }) - log.Debug(fmt.Sprintf("pruned %v peers", total)) - } - }() +type Notifier interface { + NotifyPeer(OverlayConn, uint8) error + NotifyDepth(uint8) error } -// KadPeer represents a Kademlia Peer and extends -// * PeerAddr interface (overlay and underlay addresses) -// * Peer interface (id, last seen, drop) -type KadPeer struct { - PeerAddr - Peer Peer +// OverlayPeer interface captures the common aspect of view of a peer from the Overlay +// topology driver +type OverlayPeer interface { + Address() []byte +} + +// OverlayConn represents a connected peer +type OverlayConn interface { + OverlayPeer + Drop(error) // call to indicate a peer should be expunged + Off() OverlayAddr // call to return a persitent OverlayAddr +} + +type OverlayAddr interface { + OverlayPeer + On(OverlayConn) OverlayConn // call to return the connected peer + Update(OverlayAddr) OverlayAddr // returns the updated version of the original +} + +// entry represents a Kademlia table entry (an extension of OverlayPeer) +// implements the pot.PotVal interface via BytesAddress, so entry can be +// used directly as a pot element +type entry struct { + pot.PotVal + OverlayPeer seenAt time.Time retries int } -func (self *KadPeer) String() string { - if self == nil { - return "" - } - return fmt.Sprintf("%x", self.OverlayAddr()) -} - -func (self *Kademlia) callable(val pot.PotVal) *KadPeer { - kp := val.(*KadPeer) - // not callable if peer is live or exceeded maxRetries - if kp.Peer != nil || kp.retries > self.MaxRetries { - log.Trace(fmt.Sprintf("peer %v (%T) not callable", kp, kp.Peer)) - return nil - } - // calculate the allowed number of retries based on time lapsed since last seen - timeAgo := time.Since(kp.seenAt) - var retries int - for delta := int(timeAgo) / self.RetryInterval; delta > 0; delta /= self.RetryExponent { - log.Trace(fmt.Sprintf("delta: %v", delta)) - retries++ - } - - // this is never called concurrently, so safe to increment - // peer can be retried again - if retries < kp.retries { - log.Trace(fmt.Sprintf("log time needed before retry %v, wait only warrants %v", kp.retries, retries)) - return nil - } - kp.retries++ - log.Trace(fmt.Sprintf("peer %v is callable", kp)) - - return kp -} - -// NewKadPeer creates a kademlia peer from a PeerAddr interface -func NewKadPeer(na PeerAddr) *KadPeer { - return &KadPeer{ - PeerAddr: na, - seenAt: time.Now(), +// newEntry creates a kademlia peer from an OverlayPeer interface +func newEntry(p OverlayPeer) *entry { + return &entry{ + PotVal: pot.NewBytesVal(p, nil), + OverlayPeer: p, + seenAt: time.Now(), } } -// retrieve the base address -// which is the overlayaddress used by peers to reach us -func (self *Kademlia) GetAddr() PeerAddr { - return self.addr +func (self *entry) String() string { + return fmt.Sprintf("%x", self.Address()) } -// Register(nas) enters each PeerAddr as kademlia peers into the -// database of known peers -func (self *Kademlia) Register(nas ...PeerAddr) error { - label := fmt.Sprintf("%x", RandomAddr().OverlayAddr()) +// Register enters each OverlayAddr as kademlia peer record into the +// database of known peer addresses +func (self *Kademlia) Register(peers chan OverlayAddr) error { + if len(peers) == 0 { + return fmt.Errorf("empty peers list") + } np := pot.NewPot(nil, 0) - for _, na := range nas { - if bytes.Equal(na.OverlayAddr(), self.addr.OverlayAddr()) { - log.Warn(fmt.Sprintf("[%06s] add peers: %x is self.. skipped ", label, self.addr.OverlayAddr())) - continue + defer func() { self.addrs.Merge(np) }() + for p := range peers { + // error if self received, peer should know better + if bytes.Equal(p.Address(), self.base) { + return fmt.Errorf("add peers: %x is self", self.base) } - p := NewKadPeer(na) - np, _, _ = pot.Add(np, pot.PotVal(p)) + np, _, _ = pot.Add(np, pot.PotVal(newEntry(p))) } - oldpeers := pot.NewPot(nil, 0) - oldpeers.Merge(self.addrs) - self.addrs.Merge(np) + // TODO: remove this check m := make(map[string]bool) self.addrs.Each(func(val pot.PotVal, i int) bool { _, found := m[val.String()] - // TODO: remove this check - // log.Debug(fmt.Sprintf("-> %v %v", val, i)) if found { panic("duplicate found") } @@ -223,148 +172,32 @@ func (self *Kademlia) Register(nas ...PeerAddr) error { return nil } -// On(p) inserts the peer as a kademlia peer into the live peers -func (self *Kademlia) On(p Peer) { - kp := NewKadPeer(p) - kp.Peer = p - self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal { - // if not found live - if v == nil { - // switch the offline peer - self.addrs.Swap(kp, func(v pot.PotVal) pot.PotVal { - return pot.PotVal(kp) - }) - // insert new peer - return pot.PotVal(kp) - } - // found among live peers, do nothing - return v - }) - prox := self.proxLimit() - - vp, ok := kp.Peer.(KadDiscovery) - if !ok { - // log.Trace(fmt.Sprintf("not discovery peer %T", kp)) - return - } - go vp.NotifyProx(uint8(prox)) - f := func(val pot.PotVal, po int) { - dp := val.(*KadPeer).Peer.(KadDiscovery) - log.Debug(fmt.Sprintf("peer %v notified of %v (%v)", dp, kp, po)) - dp.NotifyPeer(kp.Peer, uint8(po)) - if uint8(prox) != self.lastProxLimit { - self.lastProxLimit = uint8(prox) - dp.NotifyProx(uint8(prox)) - } - log.Debug("peer notified") - } - self.peers.EachNeighbourAsync(kp, 1024, 255, f, false) -} - -// Off removes a peer from among live peers -func (self *Kademlia) Off(p Peer) { - kp := NewKadPeer(p) - self.addrs.Swap(kp, func(v pot.PotVal) pot.PotVal { - if v != nil { - self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal { - return nil - }) - } - return nil - }) -} - -type ByteAddr struct { - key []byte -} - -// EachLivePeer(base, po, f) is an iterator applying f to each live peer -// that has proximity order po or less as measured from the base -// if base is nil, kademlia base address is used -func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int, bool) bool) { - var p pot.PotVal - if base == nil { - p = pot.PotVal(self.addr) - } else { - p = pot.PotVal(&peerAddr{OAddr: base}) - } - self.peers.EachNeighbour(p, func(val pot.PotVal, po int) bool { - if po > o { - return true - } - isproxbin := false - if l, _ := p.PO(val, 0); l >= self.proxLimit() { - isproxbin = true - } - return f(val.(*KadPeer).Peer, po, isproxbin) - }) -} - -// EachPeer(base, po, f) is an iterator applying f to each known peer -// that has proximity order po or less as measured from the base -// if base is nil, kademlia base address is used -func (self *Kademlia) EachPeer(base []byte, o int, f func(PeerAddr, int) bool) { - var p pot.PotVal - if base == nil { - p = pot.PotVal(self.addr) - } else { - p = pot.NewHashAddressFromBytes(base) - } - self.addrs.EachNeighbour(p, func(val pot.PotVal, po int) bool { - if po > o { - return true - } - return f(val.(*KadPeer).Peer, po) - }) -} - -// proxLimit() returns the proximity order that defines the distance of -// the nearest neighbour set with cardinality >= MinProxBinSize -// if there is altogether less than MinProxBinSize peers it returns 0 -func (self *Kademlia) proxLimit() int { - if self.peers.Size() < self.MinProxBinSize { - return 0 - } - var proxLimit int - var size int - f := func(v pot.PotVal, i int) bool { - size++ - proxLimit = i - return size < self.MinProxBinSize - } - self.peers.EachNeighbour(pot.PotVal(self.addr), f) - return proxLimit -} - // SuggestPeer returns a known peer for the lowest proximity bin for the -// lowest bincount below proxLimit +// lowest bincount below depth // naturally if there is an empty row it returns a peer for that // -func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) { +func (self *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) { minsize := self.MinBinSize - proxLimit := self.proxLimit() + depth := self.Depth() // if there is a callable neighbour within the current proxBin, connect // this makes sure nearest neighbour set is fully connected - log.Trace(fmt.Sprintf("candidate prox peer checking above PO %v", proxLimit)) + log.Trace(fmt.Sprintf("candidate prox peer checking above PO %v", depth)) var ppo int - self.addrs.EachNeighbour(self.addr, func(val pot.PotVal, po int) bool { - r := self.callable(val) - if r == nil { - return po >= proxLimit - } - p = r + ba := pot.NewBytesVal(self.base, nil) + self.addrs.EachNeighbour(ba, func(val pot.PotVal, po int) bool { + a = self.callable(val) ppo = po - return false + return a != nil && po >= depth }) - if p != nil { - log.Trace(fmt.Sprintf("candidate prox peer found: %v (%v), %v", p, ppo, p)) - return p, 0, false + if a != nil { + log.Trace(fmt.Sprintf("candidate prox peer found: %v (%v)", a, ppo)) + return a, 0, false } - log.Trace(fmt.Sprintf("no candidate prox peers to connect to (ProxLimit: %v, minProxSize: %v)", proxLimit, self.MinProxBinSize)) + log.Trace(fmt.Sprintf("no candidate prox peers to connect to (Depth: %v, minProxSize: %v)", depth, self.MinProxBinSize)) var bpo []int prev := -1 - self.peers.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + self.conns.EachBin(pot.NewBytesVal(self.base, nil), 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { log.Trace(fmt.Sprintf("check PO%02d: ", po)) prev++ if po > prev { @@ -375,7 +208,7 @@ func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) { minsize = size bpo = append(bpo, po) } - return size > 0 && po < proxLimit + return size > 0 && po < depth }) // all buckets are full // minsize == self.MinBinSize @@ -387,28 +220,165 @@ func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) { // try to select a candidate peer for i := len(bpo) - 1; i >= 0; i-- { // find the first callable peer - self.addrs.EachBin(self.addr, bpo[i], func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + self.addrs.EachBin(ba, bpo[i], func(po, size int, f func(func(pot.PotVal, int) bool) bool) bool { // for each bin we find callable candidate peers - f(func(val pot.PotVal, i int) bool { - r := self.callable(val) - log.Trace(fmt.Sprintf("check PO%02d: ", po)) - if r == nil { - return i < proxLimit - } - p = r - return false + log.Trace(fmt.Sprintf("check PO%02d: ", po)) + f(func(val pot.PotVal, j int) bool { + a = self.callable(val) + return a != nil && po < depth }) return false }) // found a candidate - if p != nil { + if a != nil { break } // cannot find a candidate, ask for more for this proximity bin specifically o = bpo[i] want = true } - return p, o, want + return a, o, want +} + +// On inserts the peer as a kademlia peer into the live peers +func (self *Kademlia) On(p OverlayPeer) { + e := newEntry(p) + self.conns.Swap(p, func(v pot.PotVal) pot.PotVal { + // if not found live + if v == nil { + // insert new online peer into addrs + self.addrs.Swap(p, func(v pot.PotVal) pot.PotVal { + return e + }) + // insert new online peer into conns + return e + } + // found among live peers, do nothing + return v + }) + + np, ok := p.(Notifier) + if !ok { + return + } + + depth := uint8(self.Depth()) + if depth != self.depth { + self.depth = depth + } else { + depth = 0 + } + + go np.NotifyDepth(depth) + f := func(val pot.PotVal, po int) { + dp := val.(Notifier) + dp.NotifyPeer(p.(OverlayConn), uint8(po)) + log.Trace(fmt.Sprintf("peer %v notified of %v (%v)", dp, p, po)) + if depth > 0 { + dp.NotifyDepth(depth) + log.Trace("peer %v notified of new depth limit %v", dp, depth) + } + } + self.conns.EachNeighbourAsync(e, 1024, 255, f, false) +} + +// Off removes a peer from among live peers +func (self *Kademlia) Off(p OverlayConn) { + self.addrs.Swap(p, func(v pot.PotVal) pot.PotVal { + // v cannot be nil, must check otherwise we overwrite entry + if v == nil { + panic(fmt.Sprintf("connected peer not found %v", p)) + } + self.conns.Swap(p, func(v pot.PotVal) pot.PotVal { + // v cannot nil, but no need to check + return nil + }) + return newEntry(p) + }) +} + +// EachConn is an iterator with args (base, po, f) applies f to each live peer +// that has proximity order po or less as measured from the base +// if base is nil, kademlia base address is used +func (self *Kademlia) EachConn(base []byte, o int, f func(OverlayConn, int, bool) bool) { + if len(base) == 0 { + base = self.base + } + p := pot.NewBytesVal(base, nil) + self.conns.EachNeighbour(p, func(val pot.PotVal, po int) bool { + if po > o { + return true + } + isproxbin := false + if l, _ := p.PO(val, 0); l >= self.Depth() { + isproxbin = true + } + return f(val.(OverlayConn), po, isproxbin) + }) +} + +// EachAddr(base, po, f) is an iterator applying f to each known peer +// that has proximity order po or less as measured from the base +// if base is nil, kademlia base address is used +func (self *Kademlia) EachAddr(base []byte, o int, f func(OverlayAddr, int) bool) { + if len(base) == 0 { + base = self.base + } + p := pot.NewBytesVal(base, nil) + self.addrs.EachNeighbour(p, func(val pot.PotVal, po int) bool { + if po > o { + return true + } + return f(val.(OverlayAddr), po) + }) +} + +// Depth returns the proximity order that defines the distance of +// the nearest neighbour set with cardinality >= MinProxBinSize +// if there is altogether less than MinProxBinSize peers it returns 0 +func (self *Kademlia) Depth() (depth int) { + if self.conns.Size() < self.MinProxBinSize { + return 0 + } + var size int + f := func(v pot.PotVal, i int) bool { + size++ + depth = i + return size < self.MinProxBinSize + } + self.conns.EachNeighbour(pot.NewBytesVal(self.base, nil), f) + return depth +} + +func (self *Kademlia) callable(val pot.PotVal) OverlayAddr { + e := val.(*entry) + // not callable if peer is live or exceeded maxRetries + if _, live := val.(OverlayConn); live || e.retries > self.MaxRetries { + log.Trace(fmt.Sprintf("peer %v (%T) not callable", e, e.OverlayPeer)) + return nil + } + // calculate the allowed number of retries based on time lapsed since last seen + timeAgo := time.Since(e.seenAt) + var retries int + for delta := int(timeAgo) / self.RetryInterval; delta > 0; delta /= self.RetryExponent { + log.Trace(fmt.Sprintf("delta: %v", delta)) + retries++ + } + + // this is never called concurrently, so safe to increment + // peer can be retried again + if retries < e.retries { + log.Trace(fmt.Sprintf("log time needed before retry %v, wait only warrants %v", e.retries, retries)) + return nil + } + e.retries++ + log.Trace(fmt.Sprintf("peer %v is callable", e)) + + return val.(OverlayAddr) +} + +func (self *Kademlia) BaseAddr() []byte { + return self.base } // kademlia table + kaddb table displayed with ascii @@ -417,16 +387,16 @@ func (self *Kademlia) String() string { var rows []string rows = append(rows, "=========================================================================") - rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), fmt.Sprintf("%x", self.addr.OverlayAddr()[:3]))) - rows = append(rows, fmt.Sprintf("population: %d (%d), MinProxBinSize: %d, MinBinSize: %d, MaxBinSize: %d", self.peers.Size(), self.addrs.Size(), self.MinProxBinSize, self.MinBinSize, self.MaxBinSize)) + rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), self)) + rows = append(rows, fmt.Sprintf("population: %d (%d), MinProxBinSize: %d, MinBinSize: %d, MaxBinSize: %d", self.conns.Size(), self.addrs.Size(), self.MinProxBinSize, self.MinBinSize, self.MaxBinSize)) liverows := make([]string, self.MaxProxDisplay) peersrows := make([]string, self.MaxProxDisplay) - var proxLimit int + var depth int prev := -1 - var proxLimitSet bool - rest := self.peers.Size() - self.peers.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + var depthSet bool + rest := self.conns.Size() + self.conns.EachBin(pot.NewBytesVal(self.base, nil), 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { var rowlen int if po >= self.MaxProxDisplay { po = self.MaxProxDisplay - 1 @@ -434,13 +404,13 @@ func (self *Kademlia) String() string { row := []string{fmt.Sprintf("%2d", size)} rest -= size f(func(val pot.PotVal, vpo int) bool { - row = append(row, val.(*KadPeer).String()[:6]) + row = append(row, val.(*entry).String()[:6]) rowlen++ return rowlen < 4 }) - if !proxLimitSet && (po > prev+1 || rest < self.MinProxBinSize) { - proxLimitSet = true - proxLimit = prev + 1 + if !depthSet && (po > prev+1 || rest < self.MinProxBinSize) { + depthSet = true + depth = prev + 1 } for ; rowlen <= 5; rowlen++ { row = append(row, " ") @@ -450,7 +420,7 @@ func (self *Kademlia) String() string { return true }) - self.addrs.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + self.addrs.EachBin(pot.NewBytesVal(self.base, nil), 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { var rowlen int if po >= self.MaxProxDisplay { po = self.MaxProxDisplay - 1 @@ -461,8 +431,7 @@ func (self *Kademlia) String() string { row := []string{fmt.Sprintf("%2d", size)} // we are displaying live peers too f(func(val pot.PotVal, vpo int) bool { - kp := val.(*KadPeer) - row = append(row, kp.String()[:6]) + row = append(row, val.(*entry).String()[:6]) rowlen++ return rowlen < 4 }) @@ -471,7 +440,7 @@ func (self *Kademlia) String() string { }) for i := 0; i < self.MaxProxDisplay; i++ { - if i == proxLimit { + if i == depth { rows = append(rows, fmt.Sprintf("============ PROX LIMIT: %d ==========================================", i)) } left := liverows[i] @@ -487,3 +456,32 @@ func (self *Kademlia) String() string { rows = append(rows, "=========================================================================") return "\n" + strings.Join(rows, "\n") } + +// Prune implements a forever loop reacting to a ticker time channel given +// as the first argument +// the loop quits if the channel is closed +// it checks each kademlia bin and if the peer count is higher than +// the MaxBinSize parameter it drops the oldest n peers such that +// the bin is reduced to MinBinSize peers thus leaving slots to newly +// connecting peers +func (self *Kademlia) Prune(c <-chan time.Time) { + go func() { + for range c { + total := 0 + self.conns.EachBin(nil, 0, func(po, size int, f func(func(pot.PotVal, int) bool) bool) bool { + extra := size - self.MinBinSize + if size > self.MaxBinSize { + n := 0 + f(func(v pot.PotVal, po int) bool { + v.(OverlayConn).Drop(fmt.Errorf("bucket full")) + n++ + return n < extra + }) + total += extra + } + return true + }) + log.Debug(fmt.Sprintf("pruned %v peers", total)) + } + }() +} diff --git a/swarm/network/kademlia_test.go b/swarm/network/kademlia_test.go index 81d5184bbf..40e1cbac97 100644 --- a/swarm/network/kademlia_test.go +++ b/swarm/network/kademlia_test.go @@ -21,13 +21,12 @@ import ( "testing" "time" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/pot" ) -func testKadPeerAddr(s string) *peerAddr { +func testKadPeerAddr(s string) *bzzAddr { a := pot.NewHashAddress(s).Bytes() - return &peerAddr{OAddr: a, UAddr: a} + return &bzzAddr{OAddr: a, UAddr: a} } type testDropPeer struct { @@ -70,7 +69,7 @@ func (self *testDiscPeer) NotifyProx(po uint8) error { return nil } -func (self *testDiscPeer) NotifyPeer(p Peer, po uint8) error { +func (self *testDiscPeer) NotifyPeer(p OverlayPeer, po uint8) error { key := overlayStr(self) key += overlayStr(p) self.lock.Lock() @@ -102,15 +101,15 @@ func newTestKademlia(b string) *testKademlia { } func (k *testKademlia) newTestKadPeer(s string) Peer { - dp := &testDropPeer{&bzzPeer{peerAddr: testKadPeerAddr(s)}, k.dropc} + dp := &testDropPeer{&bzzPeer{bzzAddr: testKadPeerAddr(s)}, k.dropc} if k.Discovery { return Peer(&testDiscPeer{dp, k.lock, k.notifications}) } return Peer(dp) } -func overlayStr(a PeerAddr) string { - log.Error(fmt.Sprintf("PeerAddr: %v (%T)", a, a)) +func overlayStr(a OverlayPeer) string { + // log.Error(fmt.Sprintf("PeerAddr: %v (%T)", a, a)) // if a == (*KadPeer)(nil) || a == (*testDiscPeer)(nil) || a == (*bzzPeer)(nil) || a == nil { // return "" // } @@ -126,14 +125,15 @@ func overlayStr(a PeerAddr) string { // return "" // } // return pot.NewHashAddressFromBytes(p.OverlayAddr()).Bin()[:6] - if a == nil { - return "" - } - k, ok := a.(*KadPeer) - if ok && k.Peer != nil { - return pot.ToBin(a.(*KadPeer).Peer.OverlayAddr())[:6] - } - return pot.ToBin(a.OverlayAddr())[:6] + // if a == nil { + // return "" + // } + // k, ok := a.(*KadPeer) + // if ok && k.Peer != nil { + // return pot.ToBin(a.(*KadPeer).Peer.Over())[:6] + // } + // return pot.ToBin(a.Over())[:6] + return pot.ToBin(a.Address()) } func (k *testKademlia) On(ons ...string) *testKademlia { @@ -146,16 +146,16 @@ func (k *testKademlia) On(ons ...string) *testKademlia { func (k *testKademlia) Off(offs ...string) *testKademlia { for _, s := range offs { - k.Kademlia.Off(k.newTestKadPeer(s)) + k.Kademlia.Off(k.newTestKadPeer(s).(OverlayConn)) } return k } func (k *testKademlia) Register(regs ...string) *testKademlia { - var ps []PeerAddr + var ps []Addr for _, s := range regs { - ps = append(ps, PeerAddr(testKadPeerAddr(s))) + ps = append(ps, Addr(testKadPeerAddr(s))) } k.Kademlia.Register(ps...) return k diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 66963215e4..523031db17 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -26,7 +26,6 @@ import ( "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" - "github.com/ethereum/go-ethereum/pot" ) const ( @@ -36,51 +35,68 @@ const ( ProtocolMaxMsgSize = 10 * 1024 * 1024 ) -// bzz is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer) -type bzzPeer struct { - *protocols.Peer - localAddr *peerAddr - *peerAddr // remote address - lastActive time.Time +// the Addr interface that peerPool needs +type Addr interface { + OverlayPeer + Over() []byte + Under() []byte + String() string } +// Peer interface represents an live peer connection +type Peer interface { + Addr // the address of a peer + Conn // the live connection (protocols.Peer) + LastActive() time.Time // last time active +} + +// Conn interface represents an live peer connection +type Conn interface { + ID() discover.NodeID // the key that uniquely identifies the Node for the peerPool + Handshake(interface{}, time.Duration) (interface{}, error) // can send messages + Send(interface{}) error // can send messages + Drop(error) // disconnect this peer + Register(interface{}, func(interface{}) error) uint64 // register message-handler callbacks + DisconnectHook(func(error)) // register message-handler callbacks + Run() error // the run function to run a protocol +} + +// bzzPeer is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer) +// implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer +type bzzPeer struct { + Conn // represents the connection for online peers + localAddr *bzzAddr // local Peers address + *bzzAddr // remote address -> implements Addr interface = protocols.Peer + lastActive time.Time // time is updated whenever mutexes are releasing +} + +// Off returns the overlay peer record for offline persistance +func (self *bzzPeer) Off() OverlayAddr { + return self.bzzAddr +} + +// LastActive returns the time the peer was last active func (self *bzzPeer) LastActive() time.Time { return self.lastActive } -// implemented by peerAddr -type PeerAddr interface { - OverlayAddr() []byte - UnderlayAddr() []byte - PO(pot.PotVal, int) (int, bool) - String() string -} - -// the Peer interface that peerPool needs -type Peer interface { - PeerAddr - // String() string // pretty printable the Node - ID() discover.NodeID // the key that uniquely identifies the Node for the peerPool - Send(interface{}) error // can send messages - Drop(error) // disconnect this peer - Register(interface{}, func(interface{}) error) uint64 // register message-handler callbacks - DisconnectHook(func(error)) -} - +// BzzCodeMap compiles the message codes and message types bzz wire protocol. +// note each call to Register can start a new series (initial code is arg1) +// the initial offset for a series is arbitrary (to ensure u) func BzzCodeMap(msgs ...interface{}) *protocols.CodeMap { ct := protocols.NewCodeMap(ProtocolName, Version, ProtocolMaxMsgSize) - ct.Register(&bzzHandshake{}) - ct.Register(msgs...) + ct.Register(0, &bzzHandshake{}) + ct.Register(1, msgs...) return ct } -// Bzz is the protocol constructor +// NewBzz is the protocol constructor // returns p2p.Protocol that is to be offered by the node.Service -func Bzz(oAddr, uAddr []byte, ct *protocols.CodeMap, services func(Peer) error, peerInfo func(id discover.NodeID) interface{}, nodeInfo func() interface{}) *p2p.Protocol { +func NewBzz(over, under []byte, ct *protocols.CodeMap, services func(*bzzPeer) error, peerInfo func(id discover.NodeID) interface{}, nodeInfo func() interface{}) *p2p.Protocol { run := func(p *protocols.Peer) error { bee := &bzzPeer{ - Peer: p, - localAddr: &peerAddr{oAddr, uAddr}, + Conn: p, + localAddr: &bzzAddr{over, under}, } // protocol handshake and its validation // sets remote peer address @@ -115,57 +131,43 @@ func Bzz(oAddr, uAddr []byte, ct *protocols.CodeMap, services func(Peer) error, type bzzHandshake struct { Version uint64 NetworkId uint64 - Addr *peerAddr + Addr *bzzAddr } func (self *bzzHandshake) String() string { return fmt.Sprintf("Handshake: Version: %v, NetworkId: %v, Addr: %v", self.Version, self.NetworkId, self.Addr) } -// peerAddr implements the PeerAddress interface -type peerAddr struct { +// bzzAddr implements the PeerAddr interface +type bzzAddr struct { OAddr []byte UAddr []byte } -func (self *peerAddr) OverlayAddr() []byte { +// implements OverlayPeer interface to be used in pot package +func (self *bzzAddr) Address() []byte { return self.OAddr } -func (self *peerAddr) UnderlayAddr() []byte { +func (self *bzzAddr) Over() []byte { + return self.OAddr +} + +func (self *bzzAddr) Under() []byte { return self.UAddr } -func (self *peerAddr) PO(val pot.PotVal, pos int) (int, bool) { - kp := val.(PeerAddr) - one := kp.OverlayAddr() - other := self.OAddr - for i := pos / 8; i < len(one); i++ { - if one[i] == other[i] { - continue - } - oxo := one[i] ^ other[i] - start := 0 - if i == pos/8 { - start = pos % 8 - } - for j := start; j < 8; j++ { - if (uint8(oxo)>>uint8(7-j))&0x01 != 0 { - return i*8 + j, false - } - } - } - return len(one) * 8, true - // var ha *pot.HashAddress - // var left, right string - // if ok { - // ha = kp.HashAddress - // } else { - // ha = val.(*pot.HashAddress) - // } +func (self *bzzAddr) On(p OverlayConn) OverlayConn { + bp := p.(*bzzPeer) + bp.bzzAddr = self + return bp } -func (self *peerAddr) String() string { +func (self *bzzAddr) Update(a OverlayAddr) OverlayAddr { + return &bzzAddr{self.OAddr, a.(Addr).Under()} +} + +func (self *bzzAddr) String() string { return fmt.Sprintf("%x <%x>", self.OAddr, self.UAddr) } @@ -180,22 +182,20 @@ func (self *bzzPeer) bzzHandshake() error { Addr: self.localAddr, } - hs, err := self.Handshake(lhs) + hs, err := self.Handshake(lhs, time.Second) if err != nil { log.Error(fmt.Sprintf("handshake failed: %v", err)) return err } rhs := hs.(*bzzHandshake) - self.peerAddr = rhs.Addr + self.bzzAddr = rhs.Addr err = checkBzzHandshake(rhs) if err != nil { - log.Error(fmt.Sprintf("handshake between %v and %v failed: %v", self.localAddr, self.peerAddr, err)) + log.Error(fmt.Sprintf("handshake between %v and %v failed: %v", self.localAddr, self.bzzAddr, err)) return err } - return nil - } // checkBzzHandshake checks for the validity and compatibility of the remote handshake @@ -213,7 +213,7 @@ func checkBzzHandshake(rhs *bzzHandshake) error { } // RandomAddr is a utility method generating an address from a public key -func RandomAddr() *peerAddr { +func RandomAddr() *bzzAddr { key, err := crypto.GenerateKey() if err != nil { panic("unable to generate key") @@ -221,23 +221,20 @@ func RandomAddr() *peerAddr { pubkey := crypto.FromECDSAPub(&key.PublicKey) var id discover.NodeID copy(id[:], pubkey[1:]) - return &peerAddr{ + return &bzzAddr{ OAddr: crypto.Keccak256(pubkey[1:]), UAddr: id[:], } } -// NodeId transforms the underlay address to an adapters.NodeId -func NodeId(addr PeerAddr) *adapters.NodeId { - return adapters.NewNodeId(addr.UnderlayAddr()) +// NewNodeIdFromAddr transforms the underlay address to an adapters.NodeId +func NewNodeIdFromAddr(addr Addr) *adapters.NodeId { + return adapters.NewNodeId(addr.Under()) } -// NewPeerAddrFromNodeId constucts a peerAddr from an adapters.NodeId +// NewAddrFromNodeId constucts a bzzAddr from an adapters.NodeId // the overlay address is derived as the hash of the nodeId -func NewPeerAddrFromNodeId(n *adapters.NodeId) *peerAddr { +func NewAddrFromNodeId(n *adapters.NodeId) *bzzAddr { id := n.NodeID - return &peerAddr{ - OAddr: crypto.Keccak256(id[:]), - UAddr: id[:], - } + return &bzzAddr{crypto.Keccak256(id[:]), id[:]} } diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index cf750b614d..cc4372a763 100644 --- a/swarm/network/protocol_test.go +++ b/swarm/network/protocol_test.go @@ -34,28 +34,28 @@ func bzzHandshakeExchange(lhs, rhs *bzzHandshake, id *adapters.NodeId) []p2ptest } } -func newBzzBaseTester(t *testing.T, n int, addr *peerAddr, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { +func newBzzBaseTester(t *testing.T, n int, addr *bzzAddr, ct *protocols.CodeMap, services func(*bzzPeer) error) *bzzTester { if ct == nil { ct = BzzCodeMap() } cs := make(map[string]chan bool) - srv := func(p Peer) error { + srv := func(p *bzzPeer) error { defer close(cs[p.ID().String()]) return services(p) } - protocall := Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, srv, nil, nil).Run + protocall := NewBzz(addr.Over(), addr.Under(), ct, srv, nil, nil).Run - s := p2ptest.NewProtocolTester(t, NodeId(addr), n, protocall) + s := p2ptest.NewProtocolTester(t, NewNodeIdFromAddr(addr), n, protocall) for _, id := range s.Ids { cs[id.NodeID.String()] = make(chan bool) } return &bzzTester{ - addr: addr, + addr: addr.Address(), ProtocolTester: s, cs: cs, } @@ -63,13 +63,13 @@ func newBzzBaseTester(t *testing.T, n int, addr *peerAddr, ct *protocols.CodeMap type bzzTester struct { *p2ptest.ProtocolTester - addr *peerAddr + addr []byte cs map[string]chan bool } -func newBzzTester(t *testing.T, n int, addr *peerAddr, pp *p2ptest.TestPeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { +func newBzzTester(t *testing.T, n int, addr *bzzAddr, pp *p2ptest.TestPeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { - extraservices := func(p Peer) error { + extraservices := func(p *bzzPeer) error { pp.Add(p) p.DisconnectHook(func(err error) { pp.Remove(p) @@ -88,7 +88,7 @@ func newBzzTester(t *testing.T, n int, addr *peerAddr, pp *p2ptest.TestPeerPool, // should test handshakes in one exchange? parallelisation func (s *bzzTester) testHandshake(lhs, rhs *bzzHandshake, disconnects ...*p2ptest.Disconnect) { var peers []*adapters.NodeId - id := NodeId(rhs.Addr) + id := NewNodeIdFromAddr(rhs.Addr) if len(disconnects) > 0 { for _, d := range disconnects { peers = append(peers, d.Peer) @@ -106,13 +106,13 @@ func (s *bzzTester) runHandshakes(ids ...*adapters.NodeId) { ids = s.Ids } for _, id := range ids { - s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewPeerAddrFromNodeId(id))) + s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewAddrFromNodeId(id))) <-s.cs[id.NodeID.String()] } } -func correctBzzHandshake(addr *peerAddr) *bzzHandshake { +func correctBzzHandshake(addr *bzzAddr) *bzzHandshake { return &bzzHandshake{0, 322, addr} } @@ -125,7 +125,7 @@ func TestBzzHandshakeNetworkIdMismatch(t *testing.T) { id := s.Ids[0] s.testHandshake( correctBzzHandshake(addr), - &bzzHandshake{0, 321, NewPeerAddrFromNodeId(id)}, + &bzzHandshake{0, 321, NewAddrFromNodeId(id)}, &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")}, ) } @@ -139,7 +139,7 @@ func TestBzzHandshakeVersionMismatch(t *testing.T) { id := s.Ids[0] s.testHandshake( correctBzzHandshake(addr), - &bzzHandshake{1, 322, NewPeerAddrFromNodeId(id)}, + &bzzHandshake{1, 322, NewAddrFromNodeId(id)}, &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("version mismatch 1 (!= 0)")}, ) } @@ -153,7 +153,7 @@ func TestBzzHandshakeSuccess(t *testing.T) { id := s.Ids[0] s.testHandshake( correctBzzHandshake(addr), - &bzzHandshake{0, 322, NewPeerAddrFromNodeId(id)}, + &bzzHandshake{0, 322, NewAddrFromNodeId(id)}, ) } @@ -215,7 +215,7 @@ func TestBzzPeerPoolNotAdd(t *testing.T) { defer s.Stop() id := s.Ids[0] - s.testHandshake(correctBzzHandshake(addr), &bzzHandshake{0, 321, NewPeerAddrFromNodeId(id)}, &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")}) + s.testHandshake(correctBzzHandshake(addr), &bzzHandshake{0, 321, NewAddrFromNodeId(id)}, &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")}) if pp.Has(id) { t.Fatalf("peer %v incorrectly added: %v", id, pp) } diff --git a/swarm/network/pss.go b/swarm/network/pss.go index 3bd5e8d3ee..fa1265982a 100644 --- a/swarm/network/pss.go +++ b/swarm/network/pss.go @@ -80,7 +80,7 @@ type pssEnvelope struct { TTL uint16 Payload []byte SenderOAddr []byte - SenderUAddr []byte + // SenderUAddr []byte } // Pre-Whisper placeholder @@ -128,6 +128,7 @@ type Pss struct { fwdcache map[pssDigest]pssCacheEntry // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg cachettl time.Duration // how long to keep messages in fwdcache hasher func(string) storage.Hasher // hasher to digest message to cache + baseAddr []byte lock sync.Mutex } @@ -135,7 +136,7 @@ func (self *Pss) hashMsg(msg *PssMsg) pssDigest { hasher := self.hasher("SHA3")() hasher.Reset() hasher.Write(msg.GetRecipient()) - hasher.Write(msg.Payload.SenderUAddr) + // hasher.Write(msg.Payload.SenderUAddr) hasher.Write(msg.Payload.SenderOAddr) hasher.Write(msg.Payload.Topic[:]) hasher.Write(msg.Payload.Payload) @@ -147,6 +148,7 @@ func (self *Pss) hashMsg(msg *PssMsg) pssDigest { // // TODO error check overlay integrity func NewPss(k Overlay, params *PssParams) *Pss { + baseAddr := k.BaseAddr() return &Pss{ Overlay: k, //peerPool: make(map[pot.Address]map[PssTopic]*PssReadWriter, PssPeerCapacity), @@ -156,6 +158,7 @@ func NewPss(k Overlay, params *PssParams) *Pss { fwdcache: make(map[pssDigest]pssCacheEntry), cachettl: params.Cachettl, hasher: storage.MakeHashFunc, + baseAddr: baseAddr, } } @@ -301,11 +304,11 @@ func (self *Pss) alertSubscribers(topic *PssTopic, msg []byte) error { func (self *Pss) Send(to []byte, topic PssTopic, msg []byte) error { pssenv := pssEnvelope{ - SenderOAddr: self.Overlay.GetAddr().OverlayAddr(), - SenderUAddr: self.Overlay.GetAddr().UnderlayAddr(), - Topic: topic, - TTL: DefaultTTL, - Payload: msg, + SenderOAddr: self.baseAddr, + // SenderUAddr: self.baseAddr.Under(), + Topic: topic, + TTL: DefaultTTL, + Payload: msg, } pssmsg := &PssMsg{ @@ -328,7 +331,7 @@ func (self *Pss) Forward(msg *PssMsg) error { digest := self.hashMsg(msg) if self.checkFwdCache(nil, digest) { - log.Trace(fmt.Sprintf("pss relay block-cache match: FROM %x TO %x", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()))) + log.Trace(fmt.Sprintf("pss relay block-cache match: FROM %x TO %x", common.ByteLabel(self.baseAddr), common.ByteLabel(msg.GetRecipient()))) //return errorBlockByCache return nil } @@ -339,22 +342,22 @@ func (self *Pss) Forward(msg *PssMsg) error { // send with kademlia // find the closest peer to the recipient and attempt to send - self.Overlay.EachLivePeer(msg.GetRecipient(), 256, func(p Peer, po int, isproxbin bool) bool { - if self.checkFwdCache(p.OverlayAddr(), digest) { - log.Warn(fmt.Sprintf("BOUNCE DEFER PSS-relay FROM %x TO %x THRU %x:", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr()))) + self.Overlay.EachConn(msg.GetRecipient(), 256, func(p OverlayConn, po int, isproxbin bool) bool { + if self.checkFwdCache(p.Address(), digest) { + log.Warn(fmt.Sprintf("BOUNCE DEFER PSS-relay FROM %x TO %x THRU %x:", common.ByteLabel(self.baseAddr), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.Address()))) return true } - log.Warn(fmt.Sprintf("Attempting PSS-relay FROM %x TO %x THRU %x", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr()))) - err := p.Send(msg) + log.Warn(fmt.Sprintf("Attempting PSS-relay FROM %x TO %x THRU %x", common.ByteLabel(self.baseAddr), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.Address()))) + err := p.(Peer).Send(msg) if err != nil { - log.Warn(fmt.Sprintf("FAILED PSS-relay FROM %x TO %x THRU %x: %v", common.ByteLabel(self.Overlay.GetAddr().OverlayAddr()), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.OverlayAddr()), err)) + log.Warn(fmt.Sprintf("FAILED PSS-relay FROM %x TO %x THRU %x: %v", common.ByteLabel(self.baseAddr), common.ByteLabel(msg.GetRecipient()), common.ByteLabel(p.Address()), err)) return true } sent++ - if bytes.Equal(msg.GetRecipient(), p.OverlayAddr()) || !isproxbin { + if bytes.Equal(msg.GetRecipient(), p.Address()) || !isproxbin { return false } - log.Trace(fmt.Sprintf("%x is in proxbin, so we continue sending", common.ByteLabel(p.OverlayAddr()))) + log.Trace(fmt.Sprintf("%x is in proxbin, so we continue sending", common.ByteLabel(p.Address()))) return true }) if sent == 0 { @@ -469,7 +472,7 @@ func (self *PssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) erro } func (self *Pss) IsSelfRecipient(msg *PssMsg) bool { - if bytes.Equal(msg.GetRecipient(), self.Overlay.GetAddr().OverlayAddr()) { + if bytes.Equal(msg.GetRecipient(), self.baseAddr) { return true } return false diff --git a/swarm/network/pss_test.go b/swarm/network/pss_test.go index c0e385c4a7..68b3f2abfa 100644 --- a/swarm/network/pss_test.go +++ b/swarm/network/pss_test.go @@ -59,19 +59,15 @@ type pssTestNode struct { apifunc func() []rpc.API } -func (n *pssTestNode) Add(peer Peer) error { +func (n *pssTestNode) Add(peer *bzzPeer) error { err := n.Hive.Add(peer) time.Sleep(time.Millisecond * 250) n.triggerCheck() return err } -func (n *pssTestNode) Remove(peer Peer) { - n.Hive.Remove(peer) -} - func (n *pssTestNode) hiveKeepAlive() <-chan time.Time { - return time.Tick(time.Second * 10) + return time.Tick(time.Millisecond * 300) } func (n *pssTestNode) triggerCheck() { @@ -79,7 +75,7 @@ func (n *pssTestNode) triggerCheck() { } func (n *pssTestNode) OverlayAddr() []byte { - return n.Pss.Overlay.GetAddr().OverlayAddr() + return n.Pss.Overlay.BaseAddr() } func (n *pssTestNode) UnderlayAddr() []byte { @@ -112,7 +108,7 @@ func newPssTestService(t *testing.T, handlefunc func(interface{}) error, testnod } func (self *pssTestService) Start(server p2p.Server) error { - return self.node.Hive.Start(server, self.node.hiveKeepAlive) + return self.node.Hive.Start(server, self.node.hiveKeepAlive, nil) } func (self *pssTestService) Stop() error { @@ -122,12 +118,12 @@ func (self *pssTestService) Stop() error { func (self *pssTestService) Protocols() []p2p.Protocol { ct := BzzCodeMap() + ct.Register(0, &PssMsg{}) for _, m := range DiscoveryMsgs { - ct.Register(m) + ct.Register(1, m) } - ct.Register(&PssMsg{}) - srv := func(p Peer) error { + srv := func(p *bzzPeer) error { p.Register(&PssMsg{}, self.msgFunc) self.node.Add(p) p.DisconnectHook(func(err error) { @@ -136,7 +132,7 @@ func (self *pssTestService) Protocols() []p2p.Protocol { return nil } - proto := Bzz(self.node.OverlayAddr(), self.node.UnderlayAddr(), ct, srv, nil, nil) + proto := NewBzz(self.node.OverlayAddr(), self.node.UnderlayAddr(), ct, srv, nil, nil) return []p2p.Protocol{*proto} } @@ -167,9 +163,9 @@ func TestPssCache(t *testing.T) { Payload: pssEnvelope{ TTL: 0, SenderOAddr: oaddr, - SenderUAddr: uaddr, - Topic: topic, - Payload: data, + // SenderUAddr: uaddr, + Topic: topic, + Payload: data, }, } msg.SetRecipient(to) @@ -177,10 +173,10 @@ func TestPssCache(t *testing.T) { msgtwo := &PssMsg{ Payload: pssEnvelope{ TTL: 0, - SenderOAddr: uaddr, - SenderUAddr: oaddr, - Topic: topic, - Payload: data, + SenderOAddr: oaddr, + // SenderUAddr: oaddr, + Topic: topic, + Payload: data, }, } msgtwo.SetRecipient(to) @@ -197,16 +193,16 @@ func TestPssCache(t *testing.T) { } // check the sender cache - err = ps.addFwdCacheSender(fwdaddr.OverlayAddr(), digest) + err = ps.addFwdCacheSender(fwdaddr.Over(), digest) if err != nil { t.Fatalf("write to pss sender cache failed: %v", err) } - if !ps.checkFwdCache(fwdaddr.OverlayAddr(), digest) { + if !ps.checkFwdCache(fwdaddr.Over(), digest) { t.Fatalf("message %v should have SENDER record in cache but checkCache returned false", msg) } - if ps.checkFwdCache(fwdaddr.OverlayAddr(), digesttwo) { + if ps.checkFwdCache(fwdaddr.Over(), digesttwo) { t.Fatalf("message %v should NOT have SENDER record in cache but checkCache returned true", msgtwo) } @@ -229,12 +225,12 @@ func TestPssCache(t *testing.T) { t.Fatalf("message %v should have expired from cache but checkCache returned true", msg) } - err = ps.AddToCache(fwdaddr.OverlayAddr(), msgtwo) + err = ps.AddToCache(fwdaddr.Over(), msgtwo) if err != nil { t.Fatalf("public accessor cache write failed: %v", err) } - if !ps.checkFwdCache(fwdaddr.OverlayAddr(), digesttwo) { + if !ps.checkFwdCache(fwdaddr.Over(), digesttwo) { t.Fatalf("message %v should have SENDER record in cache but checkCache returned false", msgtwo) } } @@ -243,7 +239,7 @@ func TestPssRegisterHandler(t *testing.T) { var topic PssTopic var err error addr := RandomAddr() - ps := makePss(addr.UnderlayAddr()) + ps := makePss(addr.Under()) topic, _ = MakeTopic(protocolName, protocolVersion) err = ps.Register(topic, func(msg []byte, p *p2p.Peer, sender []byte) error { return nil }) @@ -305,7 +301,8 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in expectnodesids := []*adapters.NodeId{} // the nodes to expect on (needed by checker) expectnodesresults := make(map[*adapters.NodeId][]int) // which messages expect actually got - vct := protocols.NewCodeMap(protocolName, protocolVersion, 65535, &pssTestPayload{}) + vct := protocols.NewCodeMap(protocolName, protocolVersion, ProtocolMaxMsgSize) + vct.Register(0, &pssTestPayload{}) topic, _ := MakeTopic(protocolName, protocolVersion) trigger := make(chan *adapters.NodeId) @@ -471,8 +468,8 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in for i := 0; i < len(sends); i += 2 { t.Logf("Pss #%d: oaddr %x -> %x (uaddr %x -> %x)", i/2+1, - common.ByteLabel(nodes[fullnodes[sends[i]]].Pss.GetAddr().OverlayAddr()), - common.ByteLabel(nodes[fullnodes[sends[i+1]]].Pss.GetAddr().OverlayAddr()), + common.ByteLabel(nodes[fullnodes[sends[i]]].Pss.BaseAddr()), + common.ByteLabel(nodes[fullnodes[sends[i+1]]].Pss.BaseAddr()), common.ByteLabel(fullnodes[sends[i]].Bytes()), common.ByteLabel(fullnodes[sends[i+1]].Bytes())) } @@ -484,15 +481,15 @@ func testPssFullRandom(t *testing.T, numsends int, numnodes int, numfullnodes in fails++ } } - t.Logf("Node oaddr %x (uaddr %x) was sent %d msgs, of which %d failed", common.ByteLabel(nodes[id].Pss.GetAddr().OverlayAddr()), common.ByteLabel(id.Bytes()), len(results), fails) + t.Logf("Node oaddr %x (uaddr %x) was sent %d msgs, of which %d failed", common.ByteLabel(nodes[id].Pss.BaseAddr()), common.ByteLabel(id.Bytes()), len(results), fails) totalfails += fails } t.Logf("Total sent: %d, total fail: %d (%.2f%%)", len(sends)/2, totalfails, (float32(totalfails)/float32(len(sends)/2+1))*100) for _, node := range nodes { - logstring := fmt.Sprintf("Node oaddr %x kademlia: ", common.ByteLabel(node.Pss.Overlay.GetAddr().OverlayAddr())) - node.Pss.Overlay.EachLivePeer(nil, 256, func(p Peer, po int, isprox bool) bool { - logstring += fmt.Sprintf("%x ", common.ByteLabel(p.OverlayAddr())) + logstring := fmt.Sprintf("Node oaddr %x kademlia: ", common.ByteLabel(node.Pss.Overlay.BaseAddr())) + node.Pss.Overlay.EachConn(nil, 256, func(p Peer, po int, isprox bool) bool { + logstring += fmt.Sprintf("%x ", common.ByteLabel(p.Over())) return true }) t.Log(logstring) @@ -511,7 +508,8 @@ func TestPssFullLinearEcho(t *testing.T) { var firstpssnode *adapters.NodeId var secondpssnode *adapters.NodeId - vct := protocols.NewCodeMap(protocolName, protocolVersion, 65535, &pssTestPayload{}) + vct := protocols.NewCodeMap(protocolName, protocolVersion, ProtocolMaxMsgSize) + vct.Register(0, &pssTestPayload{}) topic, _ := MakeTopic(protocolName, protocolVersion) fullnodes := []*adapters.NodeId{} @@ -559,9 +557,8 @@ func TestPssFullLinearEcho(t *testing.T) { node, ok := nodes[id] if !ok { return false, fmt.Errorf("unknown node: %s (%v)", id, node) - } else { - log.Trace(fmt.Sprintf("sim check ok node %v", id)) } + log.Trace(fmt.Sprintf("sim check ok node %v", id)) return true, nil } @@ -589,7 +586,7 @@ func TestPssFullLinearEcho(t *testing.T) { // first find a node that we're connected to for firstpssnode == nonode { log.Debug(fmt.Sprintf("Waiting for pss relaypeer for %x close to %x ...", common.ByteLabel(nodes[fullnodes[0]].OverlayAddr()), common.ByteLabel(nodes[ids[1]].OverlayAddr()))) - nodes[fullnodes[0]].Pss.Overlay.EachLivePeer(nodes[fullnodes[1]].OverlayAddr(), 256, func(p Peer, po int, isprox bool) bool { + nodes[fullnodes[0]].Pss.Overlay.EachConn(nodes[fullnodes[1]].OverlayAddr(), 256, func(p Peer, po int, isprox bool) bool { for _, id := range ids { if id.NodeID == p.ID() { firstpssnode = id @@ -609,7 +606,7 @@ func TestPssFullLinearEcho(t *testing.T) { // then find the node it's connected to for secondpssnode == nonode { log.Debug(fmt.Sprintf("PSS kademlia: Waiting for recipientpeer for %x close to %x ...", common.ByteLabel(nodes[firstpssnode].OverlayAddr()), common.ByteLabel(nodes[fullnodes[1]].OverlayAddr()))) - nodes[firstpssnode].Pss.Overlay.EachLivePeer(nodes[fullnodes[1]].OverlayAddr(), 256, func(p Peer, po int, isprox bool) bool { + nodes[firstpssnode].Pss.Overlay.Eachc(nodes[fullnodes[1]].OverlayAddr(), 256, func(p Peer, po int, isprox bool) bool { for _, id := range ids { if id.NodeID == p.ID() && id.NodeID != fullnodes[0].NodeID { secondpssnode = id @@ -692,7 +689,8 @@ func TestPssFullWS(t *testing.T) { var firstpssnode, secondpssnode *adapters.NodeId fullnodes := []*adapters.NodeId{} - vct := protocols.NewCodeMap(protocolName, protocolVersion, 65535, &pssTestPayload{}) + vct := protocols.NewCodeMap(protocolName, protocolVersion, ProtocolMaxMsgSize) + vct.Register(0, &pssTestPayload{}) topic, _ := MakeTopic(pingTopicName, pingTopicVersion) trigger := make(chan *adapters.NodeId) @@ -788,7 +786,7 @@ func TestPssFullWS(t *testing.T) { // then find the node it's connected to for secondpssnode == nonode { log.Debug(fmt.Sprintf("PSS kademlia: Waiting for recipientpeer for %x close to %x ...", common.ByteLabel(nodes[firstpssnode].OverlayAddr()), common.ByteLabel(nodes[fullnodes[1]].OverlayAddr()))) - nodes[firstpssnode].Pss.Overlay.EachLivePeer(nodes[fullnodes[1]].OverlayAddr(), 256, func(p Peer, po int, isprox bool) bool { + nodes[firstpssnode].Pss.Overlay.EachConn(nodes[fullnodes[1]].OverlayAddr(), 256, func(p Peer, po int, isprox bool) bool { for _, id := range ids { if id.NodeID == p.ID() && id.NodeID != fullnodes[0].NodeID { secondpssnode = id @@ -855,7 +853,7 @@ func TestPssFullWS(t *testing.T) { action = func(ctx context.Context) error { go func() { clientrecv.EthSubscribe(ctx, ch, "newMsg", topic) - clientsend.Call(nil, "eth_sendRaw", nodes[secondpssnode].Pss.Overlay.GetAddr().OverlayAddr(), topic, []byte("ping")) + clientsend.Call(nil, "eth_sendRaw", nodes[secondpssnode].Pss.Overlay.BaseAddr(), topic, []byte("ping")) trigger <- secondpssnode }() return nil @@ -935,7 +933,7 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigge if testpeers[id] != nil { handlefunc = makePssHandleProtocol(psss[id]) - log.Trace(fmt.Sprintf("Making full protocol id %x addr %x (testpeers %p)", common.ByteLabel(id.Bytes()), common.ByteLabel(addr.OverlayAddr()), testpeers)) + log.Trace(fmt.Sprintf("Making full protocol id %x addr %x (testpeers %p)", common.ByteLabel(id.Bytes()), common.ByteLabel(addr.Over()), testpeers)) } else { handlefunc = makePssHandleForward(psss[id]) } @@ -965,7 +963,7 @@ func newPssSimulationTester(t *testing.T, numnodes int, numfullnodes int, trigge } for i, conf := range configs { addr := NewPeerAddrFromNodeId(conf.Id) - psss[conf.Id] = makePss(addr.OverlayAddr()) + psss[conf.Id] = makePss(addr.Over()) if i < numfullnodes { tp := &pssTestPeer{ Peer: &protocols.Peer{ @@ -1018,7 +1016,7 @@ func makeCustomProtocol(name string, version int, ct *protocols.CodeMap, testpee return protocols.NewProtocol(name, uint(version), run, ct, nil, nil) } -func makeFakeMsg(ps *Pss, ct *protocols.CodeMap, topic PssTopic, senderaddr PeerAddr, content string) PssMsg { +func makeFakeMsg(ps *Pss, ct *protocols.CodeMap, topic PssTopic, senderaddr Addr, content string) PssMsg { data := pssTestPayload{} code, found := ct.GetCode(&data) if !found { @@ -1033,8 +1031,8 @@ func makeFakeMsg(ps *Pss, ct *protocols.CodeMap, topic PssTopic, senderaddr Peer } pssenv := pssEnvelope{ - SenderOAddr: senderaddr.OverlayAddr(), - SenderUAddr: senderaddr.UnderlayAddr(), + SenderOAddr: senderaddr.Over(), + SenderUAddr: senderaddr.Under(), Topic: topic, TTL: DefaultTTL, Payload: rlpbundle, @@ -1042,7 +1040,7 @@ func makeFakeMsg(ps *Pss, ct *protocols.CodeMap, topic PssTopic, senderaddr Peer pssmsg := PssMsg{ Payload: pssenv, } - pssmsg.SetRecipient(ps.Overlay.GetAddr().OverlayAddr()) + pssmsg.SetRecipient(ps.Overlay.BaseAddr()) return pssmsg } diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index 198de66394..1fe613676b 100644 --- a/swarm/network/simulations/overlay.go +++ b/swarm/network/simulations/overlay.go @@ -24,10 +24,31 @@ import ( // SimNode is the adapter used by Swarm simulations. type SimNode struct { + id *adapters.NodeId + rw network.ReadWriter hive *network.Hive protocol *p2p.Protocol } +type simReadWriter struct { + m map[string][]byte +} + +func (self *simReadWriter) ReadAll(s string) ([]byte, error) { + return self.m[s], nil +} + +func (self *simReadWriter) WriteAll(s string, data []byte) error { + self.m[s] = data + return nil +} + +func NewSimReadWriter() *simReadWriter { + return &simReadWriter{ + make(map[string][]byte), + } +} + func (s *SimNode) Protocols() []p2p.Protocol { return []p2p.Protocol{*s.protocol} } @@ -44,19 +65,12 @@ func af() <-chan time.Time { // Start() starts up the hive // makes SimNode implement node.Service func (self *SimNode) Start(server p2p.Server) error { - return self.hive.Start(server, af) + self.init() + return self.hive.Start(server, af, self.rw) } -// Stop() shuts down the hive -// makes SimNode implement node.Service -func (self *SimNode) Stop() error { - self.hive.Stop() - return nil -} - -// NewSimNode creates adapters for nodes in the simulation. -func NewSimNode(id *adapters.NodeId, snapshot []byte) node.Service { - addr := network.NewPeerAddrFromNodeId(id) +func (self *SimNode) init() { + addr := network.NewPeerAddrFromNodeId(self.id) kp := network.NewKadParams() kp.MinProxBinSize = 2 @@ -83,11 +97,25 @@ func NewSimNode(id *adapters.NodeId, snapshot []byte) node.Service { ct := network.BzzCodeMap(network.DiscoveryMsgs...) // bzz protocol code map nodeInfo := func() interface{} { return pp.String() } + self.hive = pp + self.protocol = network.Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, services, nil, nodeInfo) +} - return &SimNode{ - hive: pp, - protocol: network.Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, services, nil, nodeInfo), +// Stop() shuts down the hive +// makes SimNode implement node.Service +func (self *SimNode) Stop() error { + self.hive.Stop() + return nil +} + +// NewSimNode creates adapters for nodes in the simulation. +func NewSimNode(id *adapters.NodeId, snapshot []byte) node.Service { + s := &SimNode{ + id: id, + rw: NewSimReadWriter(), } + s.init() + return s } func createMockers() map[string]*simulations.MockerConfig { diff --git a/swarm/network/test_overlay.go b/swarm/network/test_overlay.go index de839bcbab..ebcbbc3108 100644 --- a/swarm/network/test_overlay.go +++ b/swarm/network/test_overlay.go @@ -1,182 +1,193 @@ package network -import ( - "fmt" - "strings" - "sync" - - "github.com/ethereum/go-ethereum/log" -) - -const orders = 8 - -type testOverlay struct { - mu sync.Mutex - addr []byte - pos [][]*testPeerAddr - posMap map[string]*testPeerAddr -} - -type testPeerAddr struct { - PeerAddr - Peer Peer -} - -func (self *testOverlay) Register(nas ...PeerAddr) error { - self.mu.Lock() - defer self.mu.Unlock() - return self.register(nas...) -} - -func (self *testOverlay) GetAddr() PeerAddr { - return &peerAddr{ - OAddr: self.addr, - UAddr: []byte{}, - } -} - -func (self *testOverlay) register(nas ...PeerAddr) error { - for _, na := range nas { - tna := &testPeerAddr{PeerAddr: na} - addr := na.OverlayAddr() - if self.posMap[string(addr)] != nil { - continue - } - self.posMap[string(addr)] = tna - o := order(addr) - log.Trace(fmt.Sprintf("PO: %v, orders: %v", o, orders)) - self.pos[o] = append(self.pos[o], tna) - } - return nil -} - -func order(addr []byte) int { - return int(addr[0]) / 32 -} - -func (self *testOverlay) On(n Peer) { - self.mu.Lock() - defer self.mu.Unlock() - addr := n.OverlayAddr() - na := self.posMap[string(addr)] - if na == nil { - self.register(n) - na = self.posMap[string(addr)] - } else if na.Peer != nil { - return - } - log.Trace(fmt.Sprintf("Online: %x", addr[:4])) - na.Peer = n - return -} - -func (self *testOverlay) Off(n Peer) { - self.mu.Lock() - defer self.mu.Unlock() - addr := n.OverlayAddr() - na := self.posMap[string(addr)] - if na == nil { - return - } - delete(self.posMap, string(addr)) - na.Peer = nil -} - -// caller must hold the lock -func (self *testOverlay) on(po []*testPeerAddr) (nodes []Peer) { - for _, na := range po { - if na.Peer != nil { - nodes = append(nodes, na.Peer) - } - } - return nodes -} - -// caller must hold the lock -func (self *testOverlay) off(po []*testPeerAddr) (nas []PeerAddr) { - for _, na := range po { - if na.Peer == (*bzzPeer)(nil) { - nas = append(nas, PeerAddr(na)) - } - } - return nas -} - -func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer, int, bool) bool) { - if base == nil { - base = self.addr - } - for i := o; i < len(self.pos); i++ { - for _, na := range self.pos[i] { - if na.Peer != nil { - if !f(na.Peer, o, false) { - return - } - } - } - } -} - -func (self *testOverlay) EachPeer(base []byte, o int, f func(PeerAddr, int) bool) { - if base == nil { - base = self.addr - } - for i := o; i < len(self.pos); i++ { - for _, na := range self.pos[i] { - if !f(na, i) { - return - } - } - } -} - -func (self *testOverlay) SuggestPeer() (PeerAddr, int, bool) { - self.mu.Lock() - defer self.mu.Unlock() - for i, po := range self.pos { - ons := self.on(po) - if len(ons) < 2 { - offs := self.off(po) - if len(offs) > 0 { - log.Trace(fmt.Sprintf("node %v is off", offs[0])) - return offs[0], i, true - } - } - } - return nil, 0, true -} - -func (self *testOverlay) String() string { - self.mu.Lock() - defer self.mu.Unlock() - var t []string - var ons, offs int - var ns []Peer - var nas []PeerAddr - for o, po := range self.pos { - var row []string - ns = self.on(po) - nas = self.off(po) - ons = len(ns) - for _, n := range ns { - addr := n.OverlayAddr() - row = append(row, fmt.Sprintf("%x", addr[:4])) - } - row = append(row, "|") - offs = len(nas) - for _, na := range nas { - addr := na.OverlayAddr() - row = append(row, fmt.Sprintf("%x", addr[:4])) - } - t = append(t, fmt.Sprintf("%v: (%v/%v) %v", o, ons, offs, strings.Join(row, " "))) - } - return strings.Join(t, "\n") -} - -func NewTestOverlay(addr []byte) *testOverlay { - return &testOverlay{ - addr: addr, - posMap: make(map[string]*testPeerAddr), - pos: make([][]*testPeerAddr, orders), - } -} +// +// import ( +// "fmt" +// "strings" +// "sync" +// +// "github.com/ethereum/go-ethereum/log" +// ) +// +// const orders = 8 +// +// type testOverlay struct { +// mu sync.Mutex +// addr []byte +// pos [][]OverlayAddr +// posMap map[string]OverlayAddr +// } +// +// type testPeerAddr struct { +// Addr +// Peer +// } +// +// func (self *testPeerAddr) Address() []byte { +// return nil +// } +// +// func (self *testPeerAddr) Update(a OverlayAddr) OverlayAddr { +// return self +// } +// +// func (self *testPeerAddr) On(p OverlayConn) OverlayConn { +// return self +// } +// +// func (self *testPeerAddr) Off() OverlayAddr { +// return self +// } +// +// func (self *testOverlay) Register(peers chan OverlayAddr) error { +// self.mu.Lock() +// defer self.mu.Unlock() +// var nas []OverlayAddr +// for a := range peers { +// nas = append(nas, a) +// } +// return self.register(nas...) +// } +// +// func (self *testOverlay) BaseAddr() []byte { +// return nil +// } +// +// func (self *testOverlay) register(nas ...OverlayAddr) error { +// for _, na := range nas { +// addr := na.Address() +// if self.posMap[string(addr)] != nil { +// continue +// } +// self.posMap[string(addr)] = na +// o := order(addr) +// log.Trace(fmt.Sprintf("PO: %v, orders: %v", o, orders)) +// self.pos[o] = append(self.pos[o], na) +// } +// return nil +// } +// +// func order(addr []byte) int { +// return int(addr[0]) / 32 +// } +// +// func (self *testOverlay) On(n OverlayConn) { +// self.mu.Lock() +// defer self.mu.Unlock() +// addr := n.Address() +// na := self.posMap[string(addr)] +// if na == nil { +// self.register(n) +// na = self.posMap[string(addr)] +// } else if na.Peer != nil { +// return +// } +// log.Trace(fmt.Sprintf("Online: %x", addr[:4])) +// na.Peer = n +// return +// } +// +// func (self *testOverlay) Off(n OverlayConn) { +// self.mu.Lock() +// defer self.mu.Unlock() +// addr := n.Over() +// na := self.posMap[string(addr)] +// if na == nil { +// return +// } +// delete(self.posMap, string(addr)) +// na.Peer = nil +// } +// +// // caller must hold the lock +// func (self *testOverlay) on(po []*testPeerAddr) (nodes []OverlayConn) { +// for _, na := range po { +// if na.Peer != nil { +// nodes = append(nodes, na) +// } +// } +// return nodes +// } +// +// // caller must hold the lock +// func (self *testOverlay) off(po []*testPeerAddr) (nas []OverlayAddr) { +// for _, na := range po { +// if na.Peer == (*bzzPeer)(nil) { +// nas = append(nas, Addr(na)) +// } +// } +// return nas +// } +// +// func (self *testOverlay) EachConn(base []byte, o int, f func(OverlayConn, int, bool) bool) { +// for i := o; i < len(self.pos); i++ { +// for _, na := range self.pos[i] { +// if na.Peer != nil { +// if !f(na, o, false) { +// return +// } +// } +// } +// } +// } +// +// func (self *testOverlay) EachAddr(base []byte, o int, f func(OverlayAddr, int) bool) { +// for i := o; i < len(self.pos); i++ { +// for _, na := range self.pos[i] { +// if !f(na, i) { +// return +// } +// } +// } +// } +// +// func (self *testOverlay) SuggestPeer() (OverlayAddr, int, bool) { +// self.mu.Lock() +// defer self.mu.Unlock() +// for i, po := range self.pos { +// ons := self.on(po) +// if len(ons) < 2 { +// offs := self.off(po) +// if len(offs) > 0 { +// log.Trace(fmt.Sprintf("node %v is off", offs[0])) +// return offs[0], i, true +// } +// } +// } +// return nil, 0, true +// } +// +// func (self *testOverlay) String() string { +// self.mu.Lock() +// defer self.mu.Unlock() +// var t []string +// var ons, offs int +// var ns []Peer +// var nas []Addr +// for o, po := range self.pos { +// var row []string +// ns = self.on(po) +// nas = self.off(po) +// ons = len(ns) +// for _, n := range ns { +// addr := n.Over() +// row = append(row, fmt.Sprintf("%x", addr[:4])) +// } +// row = append(row, "|") +// offs = len(nas) +// for _, na := range nas { +// addr := na.Over() +// row = append(row, fmt.Sprintf("%x", addr[:4])) +// } +// t = append(t, fmt.Sprintf("%v: (%v/%v) %v", o, ons, offs, strings.Join(row, " "))) +// } +// return strings.Join(t, "\n") +// } +// +// func NewTestOverlay(addr []byte) *testOverlay { +// return &testOverlay{ +// addr: addr, +// posMap: make(map[string]*testPeerAddr), +// pos: make([][]*testPeerAddr, orders), +// } +// } diff --git a/swarm/swarm.go b/swarm/swarm.go index c900f23b61..a8b6353dd9 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -32,7 +32,6 @@ 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/simulations/adapters" "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/api" httpapi "github.com/ethereum/go-ethereum/swarm/api/http" @@ -182,9 +181,10 @@ func (self *Swarm) Start(net p2p.Server) error { func() <-chan time.Time { return time.NewTicker(time.Second).C }, + nil, ) - log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.GetAddr())) + log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.BaseAddr())) if self.pssEnabled { pssparams := network.NewPssParams() @@ -238,48 +238,46 @@ func (self *Swarm) Stop() error { // implements the node.Service interface func (self *Swarm) Protocols() []p2p.Protocol { ct := network.BzzCodeMap() - for _, m := range network.DiscoveryMsgs { - ct.Register(m) - } if self.pssEnabled { - ct.Register(&network.PssMsg{}) + ct.Register(1, &network.PssMsg{}) } + ct.Register(2, network.DiscoveryMsgs...) - srv := func(p network.Peer) error { - if self.pssEnabled { - p.Register(&network.PssMsg{}, func(msg interface{}) error { - pssmsg := msg.(*network.PssMsg) - - if self.pss.IsSelfRecipient(pssmsg) { - log.Trace("pss for us, yay! ... let's process!") - env := pssmsg.Payload - umsg := env.Payload - f := self.pss.GetHandler(env.Topic) - if f == nil { - return fmt.Errorf("No registered handler for topic '%s'", env.Topic) - } - nid := adapters.NewNodeId(env.SenderUAddr) - p := p2p.NewPeer(nid.NodeID, fmt.Sprintf("%x", common.ByteLabel(nid.Bytes())), []p2p.Cap{}) - return f(umsg, p, env.SenderOAddr) - } else { - log.Trace("pss was for someone else :'( ... forwarding") - return self.pss.Forward(pssmsg) - } - return nil - }) - } - self.hive.Add(p) - p.DisconnectHook(func(err error) { - self.hive.Remove(p) - }) - return nil - } + // srv := func(p network.Peer) error { + // if self.pssEnabled { + // p.Register(&network.PssMsg{}, func(msg interface{}) error { + // pssmsg := msg.(*network.PssMsg) + // + // if self.pss.IsSelfRecipient(pssmsg) { + // log.Trace("pss for us, yay! ... let's process!") + // env := pssmsg.Payload + // umsg := env.Payload + // f := self.pss.GetHandler(env.Topic) + // if f == nil { + // return fmt.Errorf("No registered handler for topic '%s'", env.Topic) + // } + // nid := adapters.NewNodeId(env.SenderUAddr) + // p := p2p.NewPeer(nid.NodeID, fmt.Sprintf("%x", common.ByteLabel(nid.Bytes())), []p2p.Cap{}) + // return f(umsg, p, env.SenderOAddr) + // } else { + // log.Trace("pss was for someone else :'( ... forwarding") + // return self.pss.Forward(pssmsg) + // } + // return nil + // }) + // } + // self.hive.Add(p) + // p.DisconnectHook(func(err error) { + // self.hive.Remove(p) + // }) + // return nil + // } proto := network.Bzz( - self.hive.Overlay.GetAddr().OverlayAddr(), - self.hive.Overlay.GetAddr().UnderlayAddr(), + self.hive.Overlay.GetAddr().Over(), + self.hive.Overlay.GetAddr().Under(), ct, - srv, + nil, nil, nil, )