diff --git a/pot/address.go b/pot/address.go index e638a11c7d..05f2f82167 100644 --- a/pot/address.go +++ b/pot/address.go @@ -289,7 +289,7 @@ func (self *BoolAddress) PO(val PotVal, pos int) (po int, eq bool) { } type BytesAddress interface { - Bytes() []byte + Address() []byte } type bytesAddress struct { @@ -312,7 +312,7 @@ func ToBytes(v AnyVal) []byte { if !ok { panic(fmt.Sprintf("unsupported value type %T", v)) } - b = ba.Bytes() + b = ba.Address() } return b } @@ -320,7 +320,7 @@ func ToBytes(v AnyVal) []byte { func (a *bytesAddress) String() string { return fmt.Sprintf("%08b", a.bytes) } -func (a *bytesAddress) Bytes() []byte { +func (a *bytesAddress) Address() []byte { return a.bytes } diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go index b1830118d7..80e8e27c70 100644 --- a/swarm/network/discovery.go +++ b/swarm/network/discovery.go @@ -12,12 +12,11 @@ type discPeer struct { *bzzPeer overlay Overlay peers map[string]bool - proxLimit uint8 // the proximity radius advertised by remote to subscribe to peers + depth uint8 // the proximity radius advertised by remote to subscribe to peers sentPeers bool // set to true when the peer is first notifed of peers close to them } -// discovery peer contructor -// registers the handlers for discovery messages +// NewDiscovery discovery peer contructor func NewDiscovery(p *bzzPeer, o Overlay) *discPeer { self := &discPeer{ overlay: o, @@ -47,25 +46,24 @@ func (self *discPeer) HandleMsg(msg interface{}) error { // NotifyPeer notifies the receiver remote end of a peer p or PO po. // callback for overlay driver -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) { +func (self *discPeer) NotifyPeer(a OverlayAddr, po uint8) error { + if po < self.depth || self.seen(a) { return nil } - log.Warn(fmt.Sprintf("notification about %x", p.Address())) + log.Warn(fmt.Sprintf("notification about %x", a.Address())) resp := &peersMsg{ - Peers: []*bzzAddr{ToAddr(p)}, // perhaps the PeerAddr interface is unnecessary generalization + Peers: []*bzzAddr{ToAddr(a)}, // perhaps the PeerAddr interface is unnecessary generalization } return self.Send(resp) } -// NotifyProx sends a subPeers Msg to the receiver notifying them about +// NotifyDepth sends a subPeers Msg to the receiver notifying them about // a change in the prox limit (radius of the set including the nearest X peers // or first empty row) // callback for overlay driver -func (self *discPeer) NotifyProx(po uint8) error { - return self.Send(&subPeersMsg{ProxLimit: po}) +func (self *discPeer) NotifyDepth(po uint8) error { + return self.Send(&subPeersMsg{Depth: po}) } /* @@ -107,28 +105,28 @@ func (self getPeersMsg) String() string { // subPeers msg is communicating the depth/sharpness/focus of the overlay table of a peer type subPeersMsg struct { - ProxLimit uint8 + Depth uint8 } func (self subPeersMsg) String() string { - return fmt.Sprintf("%T: request peers > PO%02d. ", self, self.ProxLimit) + return fmt.Sprintf("%T: request peers > PO%02d. ", self, self.Depth) } func (self *discPeer) handleSubPeersMsg(msg *subPeersMsg) error { - self.proxLimit = msg.ProxLimit + self.depth = msg.Depth if !self.sentPeers { var peers []*bzzAddr self.overlay.EachConn(self.Over(), 255, func(p OverlayConn, po int, isproxbin bool) bool { - if uint8(po) < self.proxLimit { + if uint8(po) < self.depth { return false } - log.Warn(fmt.Sprintf("peer %#v proxlimit %v", p, self.proxLimit)) + log.Warn(fmt.Sprintf("peer %#v depth %v", p, self.depth)) 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)) + log.Warn(fmt.Sprintf("found initial %v peers not farther than %v", len(peers), self.depth)) if len(peers) > 0 { if err := self.Send(&peersMsg{Peers: peers}); err != nil { return err diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index dfab1eef5a..91a280e0df 100644 --- a/swarm/network/discovery_test.go +++ b/swarm/network/discovery_test.go @@ -33,7 +33,7 @@ func TestDiscovery(t *testing.T) { Expects: []p2ptest.Expect{ p2ptest.Expect{ Code: 3, - Msg: &subPeersMsg{ProxLimit: 0}, + Msg: &subPeersMsg{Depth: 0}, Peer: s.ProtocolTester.Ids[0], }, }, diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 8a9c3d950a..dfa356214b 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -45,7 +45,7 @@ to keep the nodetable uptodate type Overlay interface { Register(chan OverlayAddr) error - On(OverlayPeer) + On(OverlayConn) Off(OverlayConn) EachConn([]byte, int, func(OverlayConn, int, bool) bool) @@ -57,21 +57,6 @@ type Overlay interface { BaseAddr() []byte } -// Hive implements the PeerPool interface -type Hive struct { - *HiveParams // settings - Overlay // the overlay topology driver - store Store - - // bookkeeping - lock sync.Mutex - quit chan bool - toggle chan bool - more chan bool - - newTicker func() hiveTicker -} - // HiveParams holds the config options to hive type HiveParams struct { Discovery bool // if want discovery of not @@ -90,6 +75,21 @@ func NewHiveParams() *HiveParams { } } +// Hive implements the PeerPool interface +type Hive struct { + *HiveParams // settings + Overlay // the overlay topology driver + store Store + + // bookkeeping + lock sync.Mutex + quit chan bool + toggle chan bool + more chan bool + + newTicker func() hiveTicker +} + // Hive constructor embeds both arguments // HiveParams: config parameters // Overlay: Topology Driver Interface @@ -167,36 +167,27 @@ func (self *Hive) Stop() { close(self.quit) } -func (self *Hive) Run(peer *bzzPeer) error { - discPeer := NewDiscovery(peer, self) - self.On(discPeer) - defer self.Off(discPeer) - return peer.Run(discPeer.HandleMsg) -} - -// 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() +func (self *Hive) Run(p *bzzPeer) error { 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 + self.wake() + defer self.wake() + defer self.Off(dp) + return p.Run(dp.HandleMsg) } // 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) -} +// 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()) + return self.String() } // PeerInfo function is used by the p2p.server RPC interface to display diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index 6eaa48c48e..8cd11b13a6 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -67,11 +67,6 @@ func TestRegisterAndConnect(t *testing.T) { pp.Start(s.Server) defer pp.Stop() tc.ticker <- time.Now() - - // if pp.Overlay.(*testOverlay).posMap[string(raddr.Over())] == nil { - // t.Fatalf("Overlay#On not called on new peer") - // } - // retrieve and broadcast ord := raddr.Over()[0] / 32 o := 0 diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index e553b29e6d..e3d4604597 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -98,7 +98,7 @@ func NewKademlia(addr []byte, params *KadParams) *Kademlia { } type Notifier interface { - NotifyPeer(OverlayConn, uint8) error + NotifyPeer(OverlayAddr, uint8) error NotifyDepth(uint8) error } @@ -117,7 +117,6 @@ type OverlayConn interface { type OverlayAddr interface { OverlayPeer - On(OverlayConn) OverlayConn // call to return the connected peer Update(OverlayAddr) OverlayAddr // returns the updated version of the original } @@ -140,18 +139,24 @@ func newEntry(p OverlayPeer) *entry { } } +func (self *entry) addr() OverlayAddr { + a, _ := self.OverlayPeer.(OverlayAddr) + return a +} + +func (self *entry) conn() OverlayConn { + c, _ := self.OverlayPeer.(OverlayConn) + return c +} + func (self *entry) String() string { - return fmt.Sprintf("%x", self.Address()) + return fmt.Sprintf("%x", self.OverlayPeer.Address()) } // 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) - 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) { @@ -159,6 +164,9 @@ func (self *Kademlia) Register(peers chan OverlayAddr) error { } np, _, _ = pot.Add(np, pot.PotVal(newEntry(p))) } + com := self.addrs.Merge(np) + log.Trace(fmt.Sprintf("merged %v peers, %v known", np.Size(), com)) + // TODO: remove this check m := make(map[string]bool) self.addrs.Each(func(val pot.PotVal, i int) bool { @@ -186,14 +194,15 @@ func (self *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) { ba := pot.NewBytesVal(self.base, nil) self.addrs.EachNeighbour(ba, func(val pot.PotVal, po int) bool { a = self.callable(val) + log.Trace(fmt.Sprintf("candidate prox peer at %x: %v (%v). a == nil is %v", val.(*entry).Address(), a, po, a == nil)) ppo = po - return a != nil && po >= depth + return a == nil && po >= depth }) 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 (Depth: %v, minProxSize: %v)", depth, self.MinProxBinSize)) + log.Trace(fmt.Sprintf("no candidate prox peers to connect to (Depth: %v, minProxSize: %v) %#v", depth, self.MinProxBinSize, a)) var bpo []int prev := -1 @@ -225,7 +234,7 @@ func (self *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) { 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 a == nil && po < depth }) return false }) @@ -241,7 +250,7 @@ func (self *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) { } // On inserts the peer as a kademlia peer into the live peers -func (self *Kademlia) On(p OverlayPeer) { +func (self *Kademlia) On(p OverlayConn) { e := newEntry(p) self.conns.Swap(p, func(v pot.PotVal) pot.PotVal { // if not found live @@ -257,10 +266,12 @@ func (self *Kademlia) On(p OverlayPeer) { return v }) + log.Trace(fmt.Sprintf("Notifier:%#v", p)) np, ok := p.(Notifier) if !ok { return } + log.Trace(fmt.Sprintf("notify:%v", p)) depth := uint8(self.Depth()) if depth != self.depth { @@ -271,12 +282,12 @@ func (self *Kademlia) On(p OverlayPeer) { go np.NotifyDepth(depth) f := func(val pot.PotVal, po int) { - dp := val.(Notifier) - dp.NotifyPeer(p.(OverlayConn), uint8(po)) + dp := val.(*entry).OverlayPeer.(Notifier) + dp.NotifyPeer(p.Off(), 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) + log.Trace("peer %v notified of new depth %v", dp, depth) } } self.conns.EachNeighbourAsync(e, 1024, 255, f, false) @@ -290,10 +301,10 @@ func (self *Kademlia) Off(p OverlayConn) { 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 + // v cannot be nil, but no need to check return nil }) - return newEntry(p) + return newEntry(p.Off()) }) } @@ -313,7 +324,7 @@ func (self *Kademlia) EachConn(base []byte, o int, f func(OverlayConn, int, bool if l, _ := p.PO(val, 0); l >= self.Depth() { isproxbin = true } - return f(val.(OverlayConn), po, isproxbin) + return f(val.(*entry).conn(), po, isproxbin) }) } @@ -329,7 +340,7 @@ func (self *Kademlia) EachAddr(base []byte, o int, f func(OverlayAddr, int) bool if po > o { return true } - return f(val.(OverlayAddr), po) + return f(val.(*entry).addr(), po) }) } @@ -353,7 +364,7 @@ func (self *Kademlia) Depth() (depth int) { 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 { + if e.conn() != nil || e.retries > self.MaxRetries { log.Trace(fmt.Sprintf("peer %v (%T) not callable", e, e.OverlayPeer)) return nil } @@ -368,26 +379,27 @@ func (self *Kademlia) callable(val pot.PotVal) OverlayAddr { // 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)) + log.Trace(fmt.Sprintf("long time since last try (at %v) needed before retry %v, wait only warrants %v", timeAgo, e.retries, retries)) return nil } e.retries++ log.Trace(fmt.Sprintf("peer %v is callable", e)) - return val.(OverlayAddr) + return e.addr() } +// BaseAddr return the kademlia base addres func (self *Kademlia) BaseAddr() []byte { return self.base } -// kademlia table + kaddb table displayed with ascii +// String returns kademlia table + kaddb table displayed with ascii 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), self.BaseAddr())) + rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %x", time.Now().UTC().Format(time.UnixDate), self.BaseAddr()[:3])) 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) @@ -468,12 +480,12 @@ 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 { + self.conns.EachBin(pot.NewBytesVal(self.base, 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")) + v.(*entry).conn().Drop(fmt.Errorf("bucket full")) n++ return n < extra }) diff --git a/swarm/network/kademlia_test.go b/swarm/network/kademlia_test.go index c6bc875976..1d1ebfba17 100644 --- a/swarm/network/kademlia_test.go +++ b/swarm/network/kademlia_test.go @@ -17,13 +17,21 @@ package network import ( "fmt" + "os" "sync" "testing" "time" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/pot" ) +func init() { + h := log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true))) + // h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true))) + log.Root().SetHandler(h) +} + func testKadPeerAddr(s string) *bzzAddr { a := pot.NewHashAddress(s).Bytes() return &bzzAddr{OAddr: a, UAddr: a} @@ -57,23 +65,24 @@ type dropError struct { } func (self *testDropPeer) Drop(err error) { - err2 := &dropError{err, overlayStr(self)} + err2 := &dropError{err, binStr(self)} self.dropc <- err2 } -func (self *testDiscPeer) NotifyProx(po uint8) error { - key := overlayStr(self) +func (self *testDiscPeer) NotifyDepth(po uint8) error { + key := binStr(self) self.lock.Lock() defer self.lock.Unlock() self.notifications[key] = po return nil } -func (self *testDiscPeer) NotifyPeer(p OverlayPeer, po uint8) error { - key := overlayStr(self) - key += overlayStr(p) +func (self *testDiscPeer) NotifyPeer(p OverlayAddr, po uint8) error { + key := binStr(self) + key += binStr(p) self.lock.Lock() defer self.lock.Unlock() + log.Trace(fmt.Sprintf("key %v=>%v", key, po)) self.notifications[key] = po return nil } @@ -108,34 +117,6 @@ func (k *testKademlia) newTestKadPeer(s string) Peer { return Peer(dp) } -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 "" - // } - // var p Peer - // s, ok := a.(*KadPeer) - // if ok { - // p = s.Peer - // } else { - // p = a.(*testDiscPeer).Peer - // } - // log.Error(fmt.Sprintf("PeerAddr: %v (%T)", p, p)) - // if p == (Peer)(nil) || p == (*testDiscPeer)(nil) || p == (*bzzPeer)(nil) { - // 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.Over())[:6] - // } - // return pot.ToBin(a.Over())[:6] - return pot.ToBin(a.Address()) -} - func (k *testKademlia) On(ons ...string) *testKademlia { for _, s := range ons { p := k.newTestKadPeer(s) @@ -160,14 +141,16 @@ func (k *testKademlia) Register(regs ...string) *testKademlia { ch <- testKadPeerAddr(s) } }() - k.Kademlia.Register(ch) + err := k.Kademlia.Register(ch) + log.Trace(fmt.Sprintf("register %v addresses: %v", len(regs), err)) + return k } func testSuggestPeer(t *testing.T, k *testKademlia, expAddr string, expPo int, expWant bool) error { addr, o, want := k.SuggestPeer() - if overlayStr(addr) != expAddr { - return fmt.Errorf("incorrect peer address suggested. expected %v, got %v", expAddr, overlayStr(addr)) + if binStr(addr) != expAddr { + return fmt.Errorf("incorrect peer address suggested. expected %v, got %v", expAddr, binStr(addr)) } if o != expPo { return fmt.Errorf("incorrect prox order suggested. expected %v, got %v", expPo, o) @@ -178,6 +161,13 @@ func testSuggestPeer(t *testing.T, k *testKademlia, expAddr string, expPo int, e return nil } +func binStr(a OverlayPeer) string { + if a == nil { + return "" + } + return pot.ToBin(a.Address())[:6] +} + func TestSuggestPeerFindPeers(t *testing.T) { // 2 row gap, unsaturated proxbin, no callables -> want PO 0 k := newTestKademlia("000000").On("001000") @@ -225,7 +215,6 @@ func TestSuggestPeerFindPeers(t *testing.T) { // second time disconnected peer not callable // with reasonably set Interval - // err = testSuggestPeer(t, k, "010000", 2, true) err = testSuggestPeer(t, k, "", 1, true) if err != nil { t.Fatal(err.Error()) @@ -240,16 +229,16 @@ func TestSuggestPeerFindPeers(t *testing.T) { } k.On("010000") - k.Off("010000") - // PO1 disconnects // new closer peer appears, it is immediately wanted - // k.Off("010000") k.Register("000101") err = testSuggestPeer(t, k, "000101", 0, false) if err != nil { t.Fatal(err.Error()) } + // PO1 disconnects + k.On("000101") + k.Off("010000") // second time, gap filling err = testSuggestPeer(t, k, "010000", 0, false) if err != nil { @@ -268,6 +257,19 @@ func TestSuggestPeerFindPeers(t *testing.T) { t.Fatal(err.Error()) } + k.Register("010001") + err = testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + + k.On("100001") + log.Trace("Kad:\n%v", k.String()) + err = testSuggestPeer(t, k, "010001", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + k.On("100001") k.On("010001") err = testSuggestPeer(t, k, "", 0, false) @@ -276,7 +278,18 @@ func TestSuggestPeerFindPeers(t *testing.T) { } k.MinBinSize = 3 + k.Register("100010") + err = testSuggestPeer(t, k, "100010", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + k.On("100010") + err = testSuggestPeer(t, k, "", 1, true) + if err != nil { + t.Fatal(err.Error()) + } + k.On("010010") err = testSuggestPeer(t, k, "", 2, true) if err != nil { @@ -284,8 +297,16 @@ func TestSuggestPeerFindPeers(t *testing.T) { } k.On("001010") + err = testSuggestPeer(t, k, "", 3, true) + if err != nil { + log.Trace("Kad:\n%v", k.String()) + t.Fatal(err.Error()) + } + + k.On("000110") err = testSuggestPeer(t, k, "", 0, false) if err != nil { + log.Trace("Kad:\n%v", k.String()) t.Fatal(err.Error()) } @@ -443,7 +464,7 @@ func TestNotifications(t *testing.T) { k.Discovery = true k.MinProxBinSize = 3 k.On("010000", "001000") - time.Sleep(100 * time.Millisecond) + time.Sleep(1000 * time.Millisecond) err := k.checkNotifications( []*testPeerNotification{ &testPeerNotification{"010000", "001000", 1}, diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index ac24790721..fd42276589 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -88,6 +88,7 @@ type Conn interface { Send(interface{}) error // can send messages Drop(error) // disconnect this peer Run(func(interface{}) error) error // the run function to run a protocol + Off() OverlayAddr } // TODO: implement store for exec nodes @@ -107,6 +108,18 @@ type BzzConfig struct { Store Store } +// Bzz is the swarm protocol bundle +type Bzz struct { + Kademlia *Kademlia + Hive *Hive + Pss *Pss + + localAddr *bzzAddr + mtx sync.Mutex + handshakes map[discover.NodeID]*bzzHandshake +} + +// NewBzz is the swarm protocol constructor func NewBzz(config *BzzConfig) *Bzz { kademlia := NewKademlia(config.OverlayAddr, config.KadParams) bzz := &Bzz{ @@ -121,16 +134,10 @@ func NewBzz(config *BzzConfig) *Bzz { return bzz } -type Bzz struct { - Kademlia *Kademlia - Hive *Hive - Pss *Pss - - localAddr *bzzAddr - mtx sync.Mutex - handshakes map[discover.NodeID]*bzzHandshake -} - +// Bzz implements the node.Service interface, offers Protocols +// * handshake/hive +// * discovery +// * pss func (b *Bzz) Protocols() []p2p.Protocol { return []p2p.Protocol{ { @@ -156,6 +163,9 @@ func (b *Bzz) Protocols() []p2p.Protocol { } } +// Bzz implements the node.Service interface, offers APIs: +// * hive +// * pss func (b *Bzz) APIs() []rpc.API { return []rpc.API{{ Namespace: "hive", @@ -200,7 +210,7 @@ func (b *Bzz) runProtocol(spec *protocols.Spec, run func(*bzzPeer) error) func(* // the handshake has succeeded so run the service peer := &bzzPeer{ - Conn: protocols.NewPeer(p, rw, spec), + Peer: protocols.NewPeer(p, rw, spec), localAddr: b.localAddr, bzzAddr: handshake.peerAddr, } @@ -227,15 +237,15 @@ func (b *Bzz) getHandshake(peerID discover.NodeID) *bzzHandshake { // 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 + *protocols.Peer // 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 } -func newBzzPeer(conn Conn, over, under []byte) *bzzPeer { +func newBzzPeer(p *protocols.Peer, over, under []byte) *bzzPeer { return &bzzPeer{ - Conn: conn, + Peer: p, localAddr: &bzzAddr{over, under}, } } @@ -318,23 +328,16 @@ func (self *bzzAddr) Address() []byte { return self.OAddr } -func (self *bzzAddr) Bytes() []byte { - return self.OAddr -} +// Over returns the overlay address func (self *bzzAddr) Over() []byte { return self.OAddr } +// Under retrun the underlay address func (self *bzzAddr) Under() []byte { return self.UAddr } -func (self *bzzAddr) On(p OverlayConn) OverlayConn { - bp := p.(*bzzPeer) - bp.bzzAddr = self - return bp -} - func (self *bzzAddr) Update(a OverlayAddr) OverlayAddr { return &bzzAddr{self.OAddr, a.(Addr).Under()} } diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index 15875eb9ce..6848f49af0 100644 --- a/swarm/network/protocol_test.go +++ b/swarm/network/protocol_test.go @@ -5,7 +5,6 @@ import ( "sync" "testing" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" @@ -73,7 +72,7 @@ func newBzzBaseTester(t *testing.T, n int, addr *bzzAddr, spec *protocols.Spec, protocall := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { return srv(&bzzPeer{ - Conn: protocols.NewPeer(p, rw, spec), + Peer: protocols.NewPeer(p, rw, spec), localAddr: addr, bzzAddr: NewAddrFromNodeId(&adapters.NodeId{NodeID: p.ID()}), }) @@ -103,6 +102,9 @@ func newBzzTester(t *testing.T, n int, addr *bzzAddr, pp *p2ptest.TestPeerPool, extraservices := func(p *bzzPeer) error { pp.Add(p) defer pp.Remove(p) + if services == nil { + return nil + } return services(p) } return newBzzBaseTester(t, n, addr, spec, extraservices) @@ -183,67 +185,3 @@ func TestBzzHandshakeSuccess(t *testing.T) { &bzzHandshake{Version: 0, NetworkId: 322, Addr: NewAddrFromNodeId(id)}, ) } - -func TestBzzPeerPoolAdd(t *testing.T) { - pp := p2ptest.NewTestPeerPool() - addr := RandomAddr() - s := newBzzTester(t, 1, addr, pp, nil, nil) - defer s.Stop() - - id := s.Ids[0] - log.Trace(fmt.Sprintf("handshake with %v", id)) - s.runHandshakes() - - if !pp.Has(id) { - t.Fatalf("peer '%v' not added: %v", id, pp) - } -} - -func TestBzzPeerPoolRemove(t *testing.T) { - addr := RandomAddr() - pp := p2ptest.NewTestPeerPool() - s := newBzzTester(t, 1, addr, pp, nil, nil) - defer s.Stop() - - s.runHandshakes() - - id := s.Ids[0] - pp.Get(id).Drop(fmt.Errorf("p2p: read or write on closed message pipe")) - s.TestDisconnected(&p2ptest.Disconnect{id, fmt.Errorf("p2p: read or write on closed message pipe")}) - if pp.Has(id) { - t.Fatalf("peer '%v' not removed: %v", id, pp) - } -} - -func TestBzzPeerPoolBothAddRemove(t *testing.T) { - addr := RandomAddr() - pp := p2ptest.NewTestPeerPool() - s := newBzzTester(t, 1, addr, pp, nil, nil) - defer s.Stop() - - s.runHandshakes() - - id := s.Ids[0] - if !pp.Has(id) { - t.Fatalf("peer '%v' not added: %v", id, pp) - } - - pp.Get(id).Drop(fmt.Errorf("p2p: read or write on closed message pipe")) - s.TestDisconnected(&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("p2p: read or write on closed message pipe")}) - if pp.Has(id) { - t.Fatalf("peer '%v' not removed: %v", id, pp) - } -} - -func TestBzzPeerPoolNotAdd(t *testing.T) { - addr := RandomAddr() - pp := p2ptest.NewTestPeerPool() - s := newBzzTester(t, 1, addr, pp, nil, nil) - defer s.Stop() - - id := s.Ids[0] - s.testHandshake(correctBzzHandshake(addr), &bzzHandshake{Version: 0, NetworkId: 321, Addr: 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_test.go b/swarm/network/pss_test.go index 2fc7ddc11b..3c059ac972 100644 --- a/swarm/network/pss_test.go +++ b/swarm/network/pss_test.go @@ -12,7 +12,9 @@ const ( ) func init() { - h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true))) + h := log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true))) + // + // h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true))) log.Root().SetHandler(h) }