swarm/network: kademlia implementation initial commit

p2p/adapters/msgpipes.go

@@ -31,7 +31,7 @@ import (
 // peer session test
 // ExpectMsg(p2p.MsgReader, uint64, interface{}) error
 // SendMsg(p2p.MsgWriter, uint64, interface{}) error
-type SimPipe struct{
+type SimPipe struct {
 	rw p2p.MsgReadWriter
 }
 

p2p/protocols/protocol.go

@@ -164,7 +164,12 @@ func NewProtocol(protocolname string, protocolversion uint, run func(*Peer) erro
 
 		m := na.Messenger(rw)
 
-		peer := NewPeer(p, ct, m, func() {})
+		disc := func() {
+			id := p.ID()
+			na.Disconnect(id[:])
+		}
+
+		peer := NewPeer(p, ct, m, disc)
 
 		return run(peer)
 
@@ -217,7 +222,7 @@ func (self *Peer) Register(msg interface{}, handler func(interface{}) error) uin
 	if !found {
 		panic(fmt.Sprintf("message type '%v' unknown ", typ))
 	}
-	glog.V(logger.Debug).Infof("registered handle for %v %v", msg, typ)
+	glog.V(logger.Detail).Infof("registered handle for %v %v", msg, typ)
 	self.handlers[typ] = append(self.handlers[typ], handler)
 	return code
 }
@@ -253,7 +258,7 @@ func (self *Peer) Send(msg interface{}) error {
 	if !found {
 		return errorf(ErrInvalidMsgType, "%v", code)
 	}
-	glog.V(logger.Debug).Infof("=> %v (%d)", msg, code)
+	glog.V(logger.Detail).Infof("=> %v (%d)", msg, code)
 	err := self.m.SendMsg(uint64(code), msg)
 	if err != nil {
 		self.Drop()
@@ -272,7 +277,7 @@ func (self *Peer) handleIncoming() (interface{}, error) {
 	if err != nil {
 		return nil, err
 	}
-	glog.V(logger.Debug).Infof("<= %v", msg)
+	glog.V(logger.Detail).Infof("<= %v", msg)
 	// make sure that the payload has been fully consumed
 	defer msg.Discard()
 
@@ -294,7 +299,7 @@ func (self *Peer) handleIncoming() (interface{}, error) {
 	if err := msg.Decode(val.Interface()); err != nil {
 		return nil, errorf(ErrDecode, "<= %v: %v", msg, err)
 	}
-	glog.V(logger.Debug).Infof("<= %v %v (%d)", req, typ, msg.Code)
+	glog.V(logger.Detail).Infof("<= %v %v (%d)", req, typ, msg.Code)
 
 	// call the registered handler callbacks
 	// a registered callback take the decoded message as argument as an interface

p2p/simulations/cytoscape.go

@@ -38,7 +38,7 @@ func UpdateCy(conf *CyConfig, j *Journal) (*CyUpdate, error) {
 	removed := []string{}
 	messaged := []string{}
 	var el *CyElement
-	update := func(e *event.Event) bool {
+	update := func(e *event.TypeMuxEvent) bool {
 		entry := e.Data
 		var action string
 		if ev, ok := entry.(*NodeEvent); ok {

p2p/simulations/journal.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/logger"
 	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p/adapters"
 )
@@ -23,7 +24,7 @@ type Journal struct {
 	counter int
 	cursor  int
 	quitc   chan bool
-	Events  []*event.Event
+	Events  []*event.TypeMuxEvent
 }
 
 // NewJournal constructor
@@ -41,7 +42,7 @@ func NewJournal() *Journal {
 // used for journalling history of a network
 // the goroutine terminates when the journal is closed
 func (self *Journal) Subscribe(eventer *event.TypeMux, types ...interface{}) {
-	glog.V(6).Infof("subscribe")
+	glog.V(logger.Info).Infof("subscribe")
 	sub := eventer.Subscribe(types...)
 	go func() {
 		defer sub.Unsubscribe()
@@ -99,7 +100,7 @@ func (self *Journal) Close() {
 	close(self.quitc)
 }
 
-func (self *Journal) append(evs ...*event.Event) {
+func (self *Journal) append(evs ...*event.TypeMuxEvent) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	self.Events = append(self.Events, evs...)
@@ -118,7 +119,7 @@ func (self *Journal) WaitEntries(n int) {
 	}
 }
 
-func (self *Journal) Read(f func(*event.Event) bool) (read int) {
+func (self *Journal) Read(f func(*event.TypeMuxEvent) bool) (read int) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	ok := true
@@ -145,7 +146,7 @@ func (self *Journal) TimedRead(acc float64, f func(interface{}) bool) (read int)
 	var lastEvent time.Time
 	timer := time.NewTimer(0)
 	var data interface{}
-	h := func(ev *event.Event) bool {
+	h := func(ev *event.TypeMuxEvent) bool {
 		// wait for the interval time passes event time
 		if ev.Time.Before(lastEvent) {
 			panic("events not ordered")

p2p/simulations/journal_test.go

@@ -10,11 +10,11 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/adapters"
 )
 
-func testEvents(intervals ...int) (events []*event.Event) {
+func testEvents(intervals ...int) (events []*event.TypeMuxEvent) {
 	t := time.Now()
 	for _, interval := range intervals {
 		t = t.Add(time.Duration(interval) * time.Millisecond)
-		events = append(events, &event.Event{
+		events = append(events, &event.TypeMuxEvent{
 			Time: t,
 			Data: interface{}(&NodeEvent{
 				Type:   "node",

p2p/simulations/network.go

@@ -32,6 +32,7 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/logger"
 	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/adapters"
@@ -66,7 +67,7 @@ func NewNetworkController(conf *NetworkConfig, eventer *event.TypeMux, journal *
 			// GET /<networkId>/
 			Retrieve: &ResourceHandler{
 				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
-					glog.V(6).Infof("msg: %v", msg)
+					glog.V(logger.Detail).Infof("msg: %v", msg)
 					cyConfig, ok := msg.(*CyConfig)
 					if ok {
 						return UpdateCy(cyConfig, journal)
@@ -75,7 +76,7 @@ func NewNetworkController(conf *NetworkConfig, eventer *event.TypeMux, journal *
 					if ok {
 						return Snapshot(snapshotConfig, journal)
 					}
-					return nil, fmt.Errorf("invalId json body: must be CyConfig or SnapshotConfig")
+					return nil, fmt.Errorf("invalid json body: must be CyConfig or SnapshotConfig")
 				},
 				Type: reflect.TypeOf(&CyConfig{}),
 			},
@@ -89,6 +90,8 @@ func NewNetworkController(conf *NetworkConfig, eventer *event.TypeMux, journal *
 		},
 	)
 	// subscribe to all event entries (generated)
+	glog.V(logger.Info).Infof("subscribe to journal")
+
 	journal.Subscribe(eventer, ConnectivityEvents...)
 	// self.SetResource("nodes", NewNodesController(eventer))
 	// self.SetResource("connections", NewConnectionsController(eventer))
@@ -349,7 +352,7 @@ func (self *Network) Start(id *adapters.NodeId) error {
 		}
 	}
 	node.Up = true
-	glog.V(6).Infof("started node %v: %v", id, node.Up)
+	glog.V(logger.Info).Infof("started node %v: %v", id, node.Up)
 
 	self.events.Post(&NodeEvent{
 		Action: "up",
@@ -376,6 +379,8 @@ func (self *Network) Stop(id *adapters.NodeId) error {
 		}
 	}
 	node.Up = false
+	glog.V(logger.Info).Infof("stop node %v: %v", id, node.Up)
+
 	self.events.Post(&NodeEvent{
 		Action: "down",
 		Type:   "node",

p2p/simulations/session_controller.go

@@ -11,9 +11,9 @@ import (
 
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/logger"
 	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p/adapters"
-	
 )
 
 type returnHandler func(body io.Reader) (resp io.ReadSeeker, err error)
@@ -96,11 +96,11 @@ func NewSessionController() (*ResourceController, chan bool) {
 					journal := NewJournal()
 					net := NewNetwork(nil, &event.TypeMux{})
 					net.SetNaf(net.NewGenericSimNode)
+					glog.V(logger.Info).Infof("new network controller on %v", conf.Id)
 					m := NewNetworkController(conf, net.Events(), journal)
 					if len(conf.Id) == 0 {
 						conf.Id = fmt.Sprintf("%d", parent.id)
 					}
-					glog.V(6).Infof("new network controller on %v", conf.Id)
 					if parent != nil {
 						parent.SetResource(conf.Id, m)
 					}
@@ -111,12 +111,12 @@ func NewSessionController() (*ResourceController, chan bool) {
 							Create: &ResourceHandler{
 								Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
 									journaldump := []string{}
-									eventfmt := func(e *event.Event) bool {
+									eventfmt := func(e *event.TypeMuxEvent) bool {
 										journaldump = append(journaldump, fmt.Sprintf("%v", e))
 										return true
 									}
 									journal.Read(eventfmt)
-									return struct{Results []string}{Results: journaldump,}, nil
+									return struct{ Results []string }{Results: journaldump}, nil
 								},
 							},
 						},
@@ -126,15 +126,9 @@ func NewSessionController() (*ResourceController, chan bool) {
 						&ResourceHandlers{
 							Create: &ResourceHandler{
 								Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
-									var nodeid *adapters.NodeId
+									nodeid := adapters.RandomNodeId()
-									
-									nodeid = adapters.RandomNodeId()
-									
 									net.NewNode(&NodeConfig{Id: nodeid})
-									glog.V(6).Infof("added node %v to network %v", nodeid, net)
-									
 									return &NodeConfig{Id: nodeid}, nil
-									
 								},
 							},
 							Retrieve: &ResourceHandler{
@@ -147,7 +141,7 @@ func NewSessionController() (*ResourceController, chan bool) {
 									var othernode *Node
 
 									args := msg.(*NodeIF)
-									onenode := net.Nodes[args.One - 1]
+									onenode := net.Nodes[args.One-1]
 
 									if args.Other == 0 {
 										if net.Start(onenode.Id) != nil {
@@ -155,7 +149,7 @@ func NewSessionController() (*ResourceController, chan bool) {
 										}
 										return &NodeResult{Nodes: []*Node{onenode}}, nil
 									} else {
-										othernode = net.Nodes[args.Other - 1]
+										othernode = net.Nodes[args.Other-1]
 										net.Connect(onenode.Id, othernode.Id)
 										return &NodeResult{Nodes: []*Node{onenode, othernode}}, nil
 									}
@@ -172,7 +166,7 @@ func NewSessionController() (*ResourceController, chan bool) {
 
 			Destroy: &ResourceHandler{
 				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
-					glog.V(6).Infof("destroy handler called")
+					glog.V(logger.Debug).Infof("destroy handler called")
 					// this can quit the entire app (shut down the backend server)
 					quitc <- true
 					return empty, nil

p2p/simulations/session_controller_test.go

@@ -2,13 +2,13 @@ package simulations
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
 	"testing"
 	"time"
-	"encoding/json"
 
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/logger/glog"
@@ -27,7 +27,7 @@ var quitc chan bool
 var controller *ResourceController
 
 func init() {
-	glog.SetV(6)
+	glog.SetV(0)
 	glog.SetToStderr(true)
 	controller, quitc = NewSessionController()
 	StartRestApiServer(port, controller)
@@ -59,8 +59,8 @@ func TestDelete(t *testing.T) {
 }
 
 func TestCreate(t *testing.T) {
-	s, err := json.Marshal(&struct{Id string}{Id: "testnetwork"})
+	s, err := json.Marshal(&struct{ Id string }{Id: "testnetwork"})
-	req, err := http.NewRequest("POST", domain + ":" + port, bytes.NewReader(s))
+	req, err := http.NewRequest("POST", domain+":"+port, bytes.NewReader(s))
 	if err != nil {
 		t.Fatalf("unexpected error creating request: %v", err)
 	}
@@ -68,7 +68,7 @@ func TestCreate(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unexpected error on http.Client request: %v", err)
 	}
-	req, err = http.NewRequest("POST", domain + ":" + port + "/testnetwork/debug/", nil)
+	req, err = http.NewRequest("POST", domain+":"+port+"/testnetwork/debug/", nil)
 	if err != nil {
 		t.Fatalf("unexpected error creating request: %v", err)
 	}
@@ -76,44 +76,34 @@ func TestCreate(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unexpected error on http.Client request: %v", err)
 	}
-	body, err := ioutil.ReadAll(resp.Body)
+	_, err = ioutil.ReadAll(resp.Body)
 	if err != nil {
 		t.Fatalf("error reading response body: %v", err)
 	}
-	t.Logf("%s", body)
 }
 
 func TestNodes(t *testing.T) {
 	networkname := "testnetworkfornodes"
 
-	s, err := json.Marshal(&struct{Id string}{Id: networkname})
+	s, err := json.Marshal(&struct{ Id string }{Id: networkname})
-	req, err := http.NewRequest("POST", domain + ":" + port, bytes.NewReader(s))
+	req, err := http.NewRequest("POST", domain+":"+port, bytes.NewReader(s))
 	if err != nil {
 		t.Fatalf("unexpected error creating request: %v", err)
 	}
-	resp, err := (&http.Client{}).Do(req)
+	_, err = (&http.Client{}).Do(req)
 	if err != nil {
 		t.Fatalf("unexpected error on http.Client request: %v", err)
 	}
 	for i := 0; i < 3; i++ {
-		req, err = http.NewRequest("POST", domain + ":" + port + "/" + networkname + "/node/", nil)
+		req, err = http.NewRequest("POST", domain+":"+port+"/"+networkname+"/node/", nil)
 		if err != nil {
 			t.Fatalf("unexpected error creating request: %v", err)
 		}
-		resp, err = (&http.Client{}).Do(req)
+		_, err = (&http.Client{}).Do(req)
 		if err != nil {
 			t.Fatalf("unexpected error on http.Client request: %v", err)
 		}
-		t.Logf("%s", resp)
 	}
-	/*
-	body, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		t.Fatalf("error reading response body: %v", err)
-	}
-	
-	t.Logf("%s", body)
-	*/
 }
 
 func testResponse(t *testing.T, method, addr string, r io.ReadSeeker) []byte {

pot/address.go

@@ -1,4 +1,4 @@
-// Copyright 2016 The go-ethereum Authors
+// Copyright 2017 The go-ethereum Authors
 // This file is part of the go-ethereum library.
 //
 // The go-ethereum library is free software: you can redistribute it and/or modify
@@ -13,21 +13,26 @@
 //
 // You should have received a copy of the GNU Lesser General Public License
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
+package pot
-package network
 
 import (
+	"encoding/binary"
 	"fmt"
 	"math/rand"
+	"strconv"
 	"strings"
 
 	"github.com/ethereum/go-ethereum/common"
-	// "github.com/ethereum/go-ethereum/logger/glog"
 )
 
 var (
 	zeroAddr = &common.Hash{}
-	zeros    = zeroAddr.Hex()[2:]
+	zerosHex = zeroAddr.Hex()[2:]
+	zerosBin = Address{}.Bin()
+)
+
+var (
+	addrlen = keylen
 )
 
 type Address common.Hash
@@ -74,11 +79,18 @@ func proximity(one, other Address) (ret int, eq bool) {
 }
 
 // posProximity(a, b, pos) returns proximity order of b wrt a (symmetric) pretending
-// the first pos bits match, checking only bits at index >= pos
+// the first pos bits match, checking only bits kzindex >= pos
 func posProximity(one, other Address, pos int) (ret int, eq bool) {
 	for i := pos / 8; i < len(one); i++ {
+		if one[i] == other[i] {
+			continue
+		}
 		oxo := one[i] ^ other[i]
-		for j := pos % 8; j < 8; j++ {
+		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
 			}
@@ -184,27 +196,53 @@ func RandomAddress() Address {
 	return RandomAddressAt(Address{}, -1)
 }
 
-// wraps an Address to implement the PbVal interface
+// wraps an Address to implement the PotVal interface
-type PbAddress struct {
+type HashAddress struct {
 	Address
 }
 
-// Prefix(addr, pos) return the proximity order of addr wrt to
+func (a *HashAddress) String() string {
-// the pinned address of the tree
+	return a.Address.Bin()
-// assuming it is greater than or  equal to pos
-func (self *PbAddress) Prefix(val PbVal, pos int) (po int, eq bool) {
-	return posProximity(self.Address, val.(*PbAddress).Address, pos)
 }
 
-type BinAddr struct {
+func NewHashAddress(s string) *HashAddress {
+	ha := [32]byte{}
+
+	t := s + string(zerosBin)[:len(zerosBin)-len(s)]
+	for i := 0; i < len(t)/64; i++ {
+		n, err := strconv.ParseUint(t[i*64:(i+1)*64], 2, 64)
+		if err != nil {
+			panic("wrong format: " + err.Error())
+		}
+		binary.BigEndian.PutUint64(ha[i*8:(i+1)*8], uint64(n))
+	}
+	h := common.Hash{}
+	copy(h[:], ha[:])
+	return &HashAddress{Address(h)}
+}
+
+func NewHashAddressFromBytes(b []byte) *HashAddress {
+	h := common.Hash{}
+	copy(h[:], b)
+	return &HashAddress{Address(h)}
+}
+
+// PO(addr, pos) return the proximity order of addr wrt to
+// the pinned address of the tree
+// assuming it is greater than or  equal to pos
+func (self *HashAddress) PO(val PotVal, pos int) (po int, eq bool) {
+	return posProximity(self.Address, val.(*HashAddress).Address, pos)
+}
+
+type BoolAddress struct {
 	addr []bool
 }
 
-func NewBinAddr(s string) *BinAddr {
+func NewBoolAddress(s string) *BoolAddress {
-	return NewBinAddrXOR(s, zeros[:len(s)])
+	return NewBoolAddressXOR(s, zerosBin[:len(s)])
 }
 
-func NewBinAddrXOR(s, t string) *BinAddr {
+func NewBoolAddressXOR(s, t string) *BoolAddress {
 	if len(s) != len(t) {
 		panic("lengths do not match")
 	}
@@ -212,24 +250,23 @@ func NewBinAddrXOR(s, t string) *BinAddr {
 	for i, _ := range addr {
 		addr[i] = s[i] != t[i]
 	}
-	return &BinAddr{addr}
+	return &BoolAddress{addr}
 }
 
-func (self *BinAddr) String() string {
+func (self *BoolAddress) String() string {
 	a := self.addr
-	s := []byte(zeros)[:len(a)]
+	s := []byte(zerosBin)[:len(a)]
 	for i, one := range a {
 		if one {
 			s[i] = byte('1')
-			// glog.V(6).Infof("%v", s)
 		}
 	}
 	return string(s)
 }
 
-func (self *BinAddr) Prefix(val PbVal, pos int) (po int, eq bool) {
+func (self *BoolAddress) PO(val PotVal, pos int) (po int, eq bool) {
 	a := self.addr
-	b := val.(*BinAddr).addr
+	b := val.(*BoolAddress).addr
 	for po = pos; po < len(b); po++ {
 		if a[po] != b[po] {
 			return po, false

pot/address_test.go

@@ -1,4 +1,4 @@
-// Copyright 2016 The go-ethereum Authors
+// Copyright 2017 The go-ethereum Authors
 // This file is part of the go-ethereum library.
 //
 // The go-ethereum library is free software: you can redistribute it and/or modify
@@ -13,8 +13,7 @@
 //
 // You should have received a copy of the GNU Lesser General Public License
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
+package pot
-package network
 
 import (
 	"math/rand"
@@ -95,3 +94,48 @@ func TestRandomAddressAt(t *testing.T) {
 		}
 	}
 }
+
+const (
+	maxTestPOs   = 1000
+	testPOkeylen = 9
+)
+
+func TestPOs(t *testing.T) {
+	for i := 0; i < maxTestPOs; i++ {
+		length := rand.Intn(256) + 1
+		v0 := RandomAddress().Bin()[:length]
+		v1 := RandomAddress().Bin()[:length]
+		a0 := NewBoolAddress(v0)
+		a1 := NewBoolAddress(v1)
+		b0 := NewHashAddress(v0)
+		b1 := NewHashAddress(v1)
+		pos := rand.Intn(length) + 1
+		apo, aeq := a0.PO(a1, pos)
+		bpo, beq := b0.PO(b1, pos)
+		if bpo == 256 {
+			bpo = length
+		}
+		a0s := a0.String()
+		if a0s != v0 {
+			t.Fatalf("incorrect bool address. expected %v, got %v", v0, a0s)
+		}
+		a1s := a1.String()
+		if a1s != v1 {
+			t.Fatalf("incorrect bool address. expected %v, got %v", v1, a1s)
+		}
+		b0s := b0.String()[:length]
+		if b0s != v0 {
+			t.Fatalf("incorrect hash address. expected %v, got %v", v0, b0s)
+		}
+		b1s := b1.String()[:length]
+		if b1s != v1 {
+			t.Fatalf("incorrect hash address. expected %v, got %v", v1, b1s)
+		}
+		if apo != bpo {
+			t.Fatalf("PO does not match for %v X %v (pos: %v): expected %v, got %v", v0, v1, pos, apo, bpo)
+		}
+		if aeq != beq {
+			t.Fatalf("PO equality does not match for %v X %v (pos: %v): expected %v, got %v", v0, v1, pos, aeq, beq)
+		}
+	}
+}

pot/doc.go

@@ -0,0 +1,65 @@
+/*
+POT: proximity order tree implements a container similar to a binary tree.
+Value types implement the PoVal interface which provides the PO (proximity order)
+comparison operator.
+Each fork in the trie is itself a value. Values of the subtree contained under
+a node all share the same order when compared to other elements in the tree.
+
+Example of proximity order is the length of the common prefix over bitvectors.
+(which is equivalent to the order of magnitude of the XOR distance over integers).
+
+The package provides two implementations of PoVal,
+
+* BoolAddress: an arbitrary length boolean vector
+* HashAddress: a bitvector based address derived from 256 bit hash (common.Hash)
+
+If the address space if limited, equality is defined as the maximum proximity order.
+
+Arbitrary value types can extend these base types or define their own PO method.
+
+The container offers applicative (funcional) style methods on PO trees:
+* adding/removing en element
+* swap (value based add/remove)
+* merging two PO trees (union)
+
+as well as iterator accessors that respect proximity order
+
+When synchronicity of membership if not 100% requirement (e.g. used as a database
+of network connections), applicative structures have the advantage that nodes
+are immutable therefore manipulation does not need locking allowing for parallel retrievals.
+For the use case where the entire container is supposed to allow changes by parallel routines, instance methods with write locks and access methods with readlock are provided. The latter only locks while reading the root node.
+
+Pot
+* retrieval, insertion and deletion by key involves log(n) pointer lookups
+* for any item retrieval  (defined as common prefix on the binary key)
+* provide syncronous iterators respecting proximity ordering  wrt any item
+* provide asyncronous iterator (for parallel execution of operations) over n items
+* allows cheap iteration over ranges
+* asymmetric parallellised merge (union)
+
+Note:
+* as is, union only makes sense for set representations since which of two values with equal keys survives is random
+* intersection is not implemented
+* simple get accessor is not implemented (but derivable from EachNeighbour)
+
+Pinned value on the node implies no need to copy keys of the item type.
+
+Note that
+* the same set of values allows for a large number of alternative
+POT representations.
+* values on the top are accessed faster than lower ones and the steps needed to retrieve items has a logarithmic distribution.
+
+As a consequence on can organise the tree so that items that need faster access are torwards the top.
+In particular for any subset where popularity has a power distriution that is independent of
+proximity order (content addressed storage of chunks), it is in principle possible to create a pot where the steps needed to access an item is inversely proportional to its popularity.
+Such organisation is not implemented as yet.
+
+TODO:
+* swap
+* get
+* overwrite-style merge
+* intersection
+* access frequency based optimisations
+
+*/
+package pot

pot/pot.go

@@ -0,0 +1,793 @@
+// Copyright 2017 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+package pot
+
+import (
+	"fmt"
+	"sync"
+)
+
+const (
+	// keylen = 4
+	keylen    = 256
+	maxkeylen = 256
+)
+
+// Pot is the root node type, allows locked non-applicative manipulation
+type Pot struct {
+	lock sync.RWMutex
+	*pot
+}
+
+// pot is the node type (same for root, branching node and leaf)
+type pot struct {
+	pin  PotVal
+	bins []*pot
+	size int
+	po   int
+}
+
+// PotVal is the interface the generic container item should implement
+type PotVal interface {
+	PO(PotVal, int) (po int, eq bool)
+	String() string
+}
+
+// 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
+func NewPot(v PotVal, po int) *Pot {
+	var size int
+	if v != nil {
+		size++
+	}
+	return &Pot{
+		pot: &pot{
+			pin:  v,
+			po:   po,
+			size: size,
+		},
+	}
+}
+
+// Pin() returns the pinned element (key) of the Pot
+func (t *Pot) Pin() PotVal {
+	return t.pin
+}
+
+// Size() returns the number of values in the Pot
+func (t *Pot) Size() int {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+	return t.size
+}
+
+// Add(v) inserts v into the Pot and
+// returns the proximity order of v and a boolean
+// indicating if the item was found
+// Add locks the Pot while using applicative add on its pot
+func (t *Pot) Add(val PotVal) (po int, found bool) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.pot, po, found = add(t.pot, val)
+	return po, found
+}
+
+// Add(t, v) returns a new Pot that contains all the elements of t
+// plus the value v, using the applicative add
+// the second return value is the proximity order of the inserted element
+// the third is boolean indicating if the item was found
+// it only readlocks the Pot while reading its pot
+func Add(t *Pot, val PotVal) (*Pot, int, bool) {
+	t.lock.RLock()
+	n := t.pot
+	t.lock.RUnlock()
+	r, po, found := add(n, val)
+	return &Pot{pot: r}, po, found
+}
+
+func add(t *pot, val PotVal) (*pot, int, bool) {
+	var r *pot
+	if t == nil || t.pin == nil {
+		r = &pot{
+			pin:  val,
+			size: t.size + 1,
+			po:   t.po,
+			bins: t.bins,
+		}
+		return r, 0, false
+	}
+	po, found := val.PO(t.pin, t.po)
+	if found {
+		r = &pot{
+			pin:  val,
+			size: t.size,
+			po:   t.po,
+			bins: t.bins,
+		}
+		return r, po, true
+	}
+
+	var p *pot
+	var i, j int
+	size := t.size
+	for i < len(t.bins) {
+		n := t.bins[i]
+		if n.po == po {
+			p, _, found = add(n, val)
+			if !found {
+				size++
+			}
+			j++
+			break
+		}
+		if n.po > po {
+			break
+		}
+		i++
+		j++
+	}
+	if p == nil {
+		size++
+		p = &pot{
+			pin:  val,
+			size: 1,
+			po:   po,
+		}
+	}
+
+	bins := append([]*pot{}, t.bins[:i]...)
+	bins = append(bins, p)
+	bins = append(bins, t.bins[j:]...)
+	r = &pot{
+		pin:  t.pin,
+		size: size,
+		po:   t.po,
+		bins: bins,
+	}
+
+	return r, po, found
+}
+
+// T.Re move(v) deletes v from the Pot and returns
+// the proximity order of v and a boolean value indicating
+// if the value was found
+// Remove locks Pot while using applicative remove on its pot
+func (t *Pot) Remove(val PotVal) (po int, found bool) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.pot, po, found = remove(t.pot, val)
+	return po, found
+}
+
+// Remove(t, v) returns a new Pot that contains all the elements of t
+// minus the value v, using the applicative remove
+// the second return value is the proximity order of the inserted element
+// the third is boolean indicating if the item was found
+// it only readlocks the Pot while reading its pot
+func Remove(t *Pot, v PotVal) (*Pot, int, bool) {
+	t.lock.RLock()
+	n := t.pot
+	t.lock.RUnlock()
+	r, po, found := remove(n, v)
+	return &Pot{pot: r}, po, found
+}
+
+func remove(t *pot, val PotVal) (r *pot, po int, found bool) {
+	size := t.size
+	po, found = val.PO(t.pin, t.po)
+	if found {
+		size--
+		if size == 0 {
+			r = &pot{
+				po: t.po,
+			}
+			return r, po, true
+		}
+		i := len(t.bins) - 1
+		last := t.bins[i]
+		r = &pot{
+			pin:  last.pin,
+			bins: append(t.bins[:i], last.bins...),
+			size: size,
+			po:   t.po,
+		}
+		return r, t.po, true
+	}
+
+	var p *pot
+	var i, j int
+	for i < len(t.bins) {
+		n := t.bins[i]
+		if n.po == po {
+			p, po, found = remove(n, val)
+			if found {
+				size--
+			}
+			j++
+			break
+		}
+		if n.po > po {
+			return t, po, false
+		}
+		i++
+		j++
+	}
+	bins := t.bins[:i]
+	if p != nil && p.pin != nil {
+		bins = append(bins, p)
+	}
+	bins = append(bins, t.bins[j:]...)
+	r = &pot{
+		pin:  val,
+		size: size,
+		po:   t.po,
+		bins: bins,
+	}
+	return r, po, found
+}
+
+// Swap(k, f) looks up the item at k
+// and applies the function f to the value v at k or nil if the item is not found
+// if f returns nil, the element is removed
+// 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) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	var t0 *pot
+	t0, po, found, change = swap(t.pot, val, f)
+	if change {
+		t.pot = t0
+	}
+	return po, found, change
+}
+
+func swap(t *pot, k PotVal, f func(v PotVal) PotVal) (r *pot, po int, found bool, change bool) {
+	var val PotVal
+	if t == nil || t.pin == nil {
+		val = f(nil)
+		if val == nil {
+			return t, t.po, false, false
+		}
+		if _, eq := val.PO(k, t.po); !eq {
+			panic("value key mismatch")
+		}
+		r = &pot{
+			pin:  val,
+			size: t.size + 1,
+			po:   t.po,
+			bins: t.bins,
+		}
+		return r, t.po, false, true
+	}
+	size := t.size
+	if k == nil {
+		panic("k is nil")
+	}
+	po, found = k.PO(t.pin, t.po)
+	if found {
+		val = f(t.pin)
+		if val == nil {
+			size--
+			if size == 0 {
+				r = &pot{
+					po: t.po,
+				}
+				return r, po, true, true
+			}
+			i := len(t.bins) - 1
+			last := t.bins[i]
+			r = &pot{
+				pin:  last.pin,
+				bins: append(t.bins[:i], last.bins...),
+				size: size,
+				po:   t.po,
+			}
+			return r, t.po, true, true
+			// remove element
+		} else if val == t.pin {
+			return nil, po, true, false
+		} else { // add element
+			r = &pot{
+				pin:  val,
+				size: t.size,
+				po:   t.po,
+				bins: t.bins,
+			}
+			return r, po, true, true
+		}
+	}
+
+	var p *pot
+	var i, j int
+	for i < len(t.bins) {
+		n := t.bins[i]
+		if n.po == po {
+			p, po, found, change = swap(n, k, f)
+			if !change {
+				return nil, po, found, false
+			}
+			size += p.size - n.size
+			j++
+			break
+		}
+		if n.po > po {
+			break
+		}
+		i++
+		j++
+	}
+	if p == nil {
+		val := f(nil)
+		if val == nil {
+			return nil, po, false, false
+		}
+		size++
+		p = &pot{
+			pin:  val,
+			size: 1,
+			po:   po,
+		}
+	}
+
+	bins := append([]*pot{}, t.bins[:i]...)
+	if p.pin != nil {
+		bins = append(bins, p)
+	}
+	bins = append(bins, t.bins[j:]...)
+	r = &pot{
+		pin:  t.pin,
+		size: size,
+		po:   t.po,
+		bins: bins,
+	}
+
+	return r, po, found, true
+}
+
+// t0.Merge(t1) changes t0 to contain all the elements of t1
+// it locks t0, but only readlocks t1 while taking its pot
+// uses applicative union
+func (t *Pot) Merge(t1 *Pot) (c int) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t1.lock.RLock()
+	n1 := t1.pot
+	t1.lock.RUnlock()
+	t.pot, c = union(t.pot, n1)
+	return c
+}
+
+// Union(t0, t1) return the union of t0 and t1
+// it only readlocks the Pot-s to read their pots and
+// calculates the union using the applicative union
+// the second return value is the number of common elements
+func Union(t0, t1 *Pot) (*Pot, int) {
+	t0.lock.RLock()
+	n0 := t0.pot
+	t0.lock.RUnlock()
+	t1.lock.RLock()
+	n1 := t1.pot
+	t1.lock.RUnlock()
+
+	p, c := union(n0, n1)
+	return &Pot{
+		pot: p,
+	}, c
+}
+
+func union(t0, t1 *pot) (*pot, int) {
+	po, eq := t0.pin.PO(t1.pin, 0)
+	var pin PotVal
+	var bins []*pot
+	var mis []int
+	wg := &sync.WaitGroup{}
+	pin0 := t0.pin
+	pin1 := t1.pin
+	bins0 := t0.bins
+	bins1 := t1.bins
+	var i0, i1 int
+	var common int
+
+	for {
+		l0 := len(bins0)
+		l1 := len(bins1)
+		var n0, n1 *pot
+		var p0, p1 int
+		var a0, a1 bool
+
+		for {
+
+			if !a0 && i0 < l0 && bins0[i0].po <= po {
+				n0 = bins0[i0]
+				p0 = n0.po
+				a0 = p0 == po
+			} else {
+				a0 = true
+			}
+
+			if !a1 && i1 < l1 && bins1[i1].po <= po {
+				n1 = bins1[i1]
+				p1 = n1.po
+				a1 = p1 == po
+			} else {
+				a1 = true
+			}
+			if a0 && a1 {
+				break
+			}
+
+			switch {
+			case (p0 < p1 || a1) && !a0:
+				bins = append(bins, n0)
+				i0++
+				n0 = nil
+			case (p1 < p0 || a0) && !a1:
+				bins = append(bins, n1)
+				i1++
+				n1 = nil
+			case p1 < po:
+				bl := len(bins)
+				bins = append(bins, nil)
+				ml := len(mis)
+				mis = append(mis, 0)
+				wg.Add(1)
+				go func(b, m int, m0, m1 *pot) {
+					defer wg.Done()
+					bins[b], mis[m] = union(m0, m1)
+				}(bl, ml, n0, n1)
+				i0++
+				i1++
+				n0 = nil
+				n1 = nil
+			}
+		}
+
+		if eq {
+			common++
+			pin = pin1
+			break
+		}
+
+		var size0 int
+		for _, n := range bins0[i0:] {
+			size0 += n.size
+		}
+
+		np := &pot{
+			pin:  pin0,
+			bins: bins0[i0:],
+			size: size0 + 1,
+			po:   po,
+		}
+
+		bins2 := []*pot{np}
+		if n0 == nil {
+			pin0 = pin1
+			po = maxkeylen + 1
+			eq = true
+			common--
+		} else {
+			bins2 = append(bins2, n0.bins...)
+			pin0 = pin1
+			pin1 = n0.pin
+			po, eq = pin0.PO(pin1, n0.po)
+		}
+		bins0 = bins1
+		bins1 = bins2
+		i0 = i1
+		i1 = 0
+
+	}
+
+	wg.Wait()
+	for _, c := range mis {
+		common += c
+	}
+	n := &pot{
+		pin:  pin,
+		bins: bins,
+		size: t0.size + t1.size - common,
+		po:   t0.po,
+	}
+	return n, common
+}
+
+// Each(f) is a synchronous iterator over the bins of a node
+// it does NOT include the pinned item of the root
+// respecting an ordering
+// proximity > pinnedness
+func (t *Pot) Each(f func(PotVal, int) bool) bool {
+	t.lock.RLock()
+	n := t.pot
+	t.lock.RUnlock()
+	return n.each(f)
+}
+
+func (t *pot) each(f func(PotVal, int) bool) bool {
+	var next bool
+	for _, n := range t.bins {
+		next = n.each(f)
+		if !next {
+			return false
+		}
+	}
+	next = f(t.pin, t.po)
+	if !next {
+		return false
+	}
+
+	return true
+}
+
+// EachBin iterates over bins of the top node and offers iterators to the caller on each
+// subtree passing the proximity order and the size
+// the iteration continues until the function's return value is false
+// or there are no more subtries
+func (t *Pot) EachBin(po int, f func(int, int, func(func(val PotVal, i int) bool) bool) bool) {
+	for _, n := range t.bins {
+		if n.po < po {
+			continue
+		}
+		if !f(n.po, n.size, n.each) {
+			break
+		}
+	}
+}
+
+// syncronous iterator over neighbours of any target val
+// even if an item at val's exact address is in the pbtree,
+// it is not included in the iteration: $val \not\in Neighbours(val)$
+func (t *Pot) EachNeighbour(val PotVal, f func(PotVal, int) bool) bool {
+	t.lock.RLock()
+	n := t.pot
+	t.lock.RUnlock()
+	return n.eachNeighbour(val, f)
+}
+
+func (t *pot) eachNeighbour(val PotVal, f func(PotVal, int) bool) bool {
+	var next bool
+	l := len(t.bins)
+	var n *pot
+	ir := l
+	il := l
+	po, eq := val.PO(t.pin, t.po)
+	if !eq {
+		n, il = t.getPos(po)
+		if n != nil {
+			next = n.eachNeighbour(val, f)
+			if !next {
+				return false
+			}
+			ir = il
+		} else {
+			ir = il - 1
+		}
+	}
+
+	next = f(t.pin, po)
+	if !next {
+		return false
+	}
+
+	for i := l - 1; i > ir; i-- {
+		next = t.bins[i].each(func(v PotVal, _ int) bool {
+			return f(v, po)
+		})
+		if !next {
+			return false
+		}
+	}
+
+	for i := il - 1; i >= 0; i-- {
+		n := t.bins[i]
+		next = n.each(func(v PotVal, _ int) bool {
+			return f(v, n.po)
+		})
+		if !next {
+			return false
+		}
+	}
+	return true
+}
+
+func (t *Pot) EachNeighbourAsync(val PotVal, max int, maxPos int, f func(PotVal, int), wait bool) {
+	t.lock.RLock()
+	n := t.pot
+	t.lock.RUnlock()
+	if max > t.size {
+		max = t.size
+	}
+	var wg *sync.WaitGroup
+	if wait {
+		wg = &sync.WaitGroup{}
+	}
+	_ = n.eachNeighbourAsync(val, max, maxPos, f, wg)
+	if wait {
+		wg.Wait()
+	}
+}
+
+func (t *pot) eachNeighbourAsync(val PotVal, max int, maxPos int, f func(PotVal, int), wg *sync.WaitGroup) (extra int) {
+
+	l := len(t.bins)
+	var n *pot
+	il := l
+	ir := l
+	// ic := l
+
+	po, eq := val.PO(t.pin, t.po)
+
+	// if po is too close, set the pivot branch (pom) to maxPos
+	pom := po
+	if pom > maxPos {
+		pom = maxPos
+	}
+	n, il = t.getPos(pom)
+	ir = il
+	// if pivot branch exists and po is not too close, iterate on the pivot branch
+	if pom == po {
+		if n != nil {
+
+			m := n.size
+			if max < m {
+				m = max
+			}
+			max -= m
+
+			extra = n.eachNeighbourAsync(val, m, maxPos, f, wg)
+
+		} else {
+			if !eq {
+				ir--
+			}
+		}
+	} else {
+		extra++
+		max--
+		if n != nil {
+			il++
+		}
+		// before checking max, add up the extra elements
+		// on the close branches that are skipped (if po is too close)
+		for i := l - 1; i >= il; i-- {
+			s := t.bins[i]
+			m := s.size
+			if max < m {
+				m = max
+			}
+			max -= m
+			extra += m
+		}
+	}
+
+	var m int
+	if pom == po {
+
+		m, max, extra = need(1, max, extra)
+		if m <= 0 {
+			return
+		}
+
+		if wg != nil {
+			wg.Add(1)
+		}
+		go func() {
+			if wg != nil {
+				defer wg.Done()
+			}
+			f(t.pin, po)
+		}()
+
+		// otherwise iterats
+		for i := l - 1; i > ir; i-- {
+			n := t.bins[i]
+
+			m, max, extra = need(n.size, max, extra)
+			if m <= 0 {
+				return
+			}
+
+			if wg != nil {
+				wg.Add(m)
+			}
+			go func(pn *pot, pm int) {
+				pn.each(func(v PotVal, _ int) bool {
+					if wg != nil {
+						defer wg.Done()
+					}
+					f(v, po)
+					pm--
+					return pm > 0
+				})
+			}(n, m)
+
+		}
+	}
+
+	// iterate branches that are farther tham pom with their own po
+	for i := il - 1; i >= 0; i-- {
+		n := t.bins[i]
+		// the first time max is less than the size of the entire branch
+		// wait for the pivot thread to release extra elements
+		m, max, extra = need(n.size, max, extra)
+		if m <= 0 {
+			return
+		}
+
+		if wg != nil {
+			wg.Add(m)
+		}
+		go func(pn *pot, pm int) {
+			pn.each(func(v PotVal, _ int) bool {
+				if wg != nil {
+					defer wg.Done()
+				}
+				f(v, pn.po)
+				pm--
+				return pm > 0
+			})
+		}(n, m)
+
+	}
+	return max + extra
+
+}
+
+// getPos(n) returns the forking node at PO n and its index if it exists
+// otherwise nil
+// caller is suppoed to hold the lock
+func (t *pot) getPos(po int) (n *pot, i int) {
+	for i, n = range t.bins {
+		if po > n.po {
+			continue
+		}
+		if po < n.po {
+			return nil, i
+		}
+		return n, i
+	}
+	return nil, len(t.bins)
+}
+
+// need(m, max, extra) uses max m out of extra, and then max
+// if needed, returns the adjusted counts
+func need(m, max, extra int) (int, int, int) {
+	if m <= extra {
+		return m, max, extra - m
+	}
+	max += extra - m
+	if max <= 0 {
+		return m + max, 0, 0
+	}
+	return m, max, 0
+}
+
+func (t *pot) String() string {
+	return t.sstring("")
+}
+
+func (t *pot) sstring(indent string) string {
+	var s string
+	indent += "  "
+	s += fmt.Sprintf("%v%v (%v) %v \n", indent, t.pin, t.po, t.size)
+	for _, n := range t.bins {
+		s += fmt.Sprintf("%v%v\n", indent, n.sstring(indent))
+	}
+	return s
+}

pot/pot_test.go

@@ -1,4 +1,19 @@
-package network
+// Copyright 2017 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+package pot
 
 import (
 	"errors"
@@ -12,53 +27,85 @@ import (
 	"github.com/ethereum/go-ethereum/logger/glog"
 )
 
+const (
+	maxEachNeighbourTests = 420
+	maxEachNeighbour      = 420
+)
+
 func init() {
-	glog.SetV(4)
+	glog.SetV(0)
 	glog.SetToStderr(true)
 }
 
+type testBVAddr struct {
+	*HashAddress
+	i int
+}
+
+func NewTestBVAddr(s string, i int) *testBVAddr {
+	return &testBVAddr{NewHashAddress(s), i}
+}
+
+func (a *testBVAddr) String() string {
+	return a.HashAddress.String()[:keylen]
+}
+
+func (self *testBVAddr) PO(val PotVal, po int) (int, bool) {
+	return self.HashAddress.PO(val.(*testBVAddr).HashAddress, po)
+}
+
 type testAddr struct {
-	*BinAddr
+	*BoolAddress
 	i int
 }
 
 func NewTestAddr(s string, i int) *testAddr {
-	return &testAddr{NewBinAddr(s), i}
+	return &testAddr{NewBoolAddress(s), i}
 }
 
-func str(v PbVal) string {
+func (self *testAddr) PO(val PotVal, po int) (int, bool) {
+	return self.BoolAddress.PO(val.(*testAddr).BoolAddress, po)
+}
+
+func str(v PotVal) string {
 	if v == nil {
 		return ""
 	}
 	return v.(*testAddr).String()
 }
 
-func indexes(t *PbTree) (i []int, pos []int) {
+func randomTestAddr(n int, i int) *testAddr {
-	t.Each(func(v PbVal, po int) bool {
+	v := RandomAddress().Bin()[:n]
-		a := v.(*testAddr)
+	return NewTestAddr(v, i)
-		i = append(i, a.i)
-		pos = append(pos, po)
-		return true
-	})
-	return i, pos
 }
 
-func add(t *PbTree, n int, values ...string) {
+func randomTestBVAddr(n int, i int) *testBVAddr {
+	v := RandomAddress().Bin()[:n]
+	return NewTestBVAddr(v, i)
+}
+
+func indexes(t *Pot) (i []int, po []int) {
+	t.Each(func(v PotVal, p int) bool {
+		a := v.(*testAddr)
+		i = append(i, a.i)
+		po = append(po, p)
+		return true
+	})
+	return i, po
+}
+
+func testAdd(t *Pot, n int, values ...string) {
 	for i, val := range values {
 		t.Add(NewTestAddr(val, i+n))
 	}
 }
 
-func (self *testAddr) Prefix(val PbVal, pos int) (po int, eq bool) {
+// func RandomBoolAddress()
-	return self.BinAddr.Prefix(val.(*testAddr).BinAddr, pos)
+func TestPotAdd(t *testing.T) {
-}
+	n := NewPot(NewTestAddr("001111", 0), 0)
-
-// func RandomBinAddr()
-func TestPbTreeAdd(t *testing.T) {
-	n := NewPbTree(NewTestAddr("001111", 0), 0)
 	// Pin set correctly
 	exp := "001111"
-	got := str(n.Pin())
+	got := str(n.Pin())[:6]
 	if got != exp {
 		t.Fatalf("incorrect pinned value. Expected %v, got %v", exp, got)
 	}
@@ -66,71 +113,125 @@ func TestPbTreeAdd(t *testing.T) {
 	goti := n.Size()
 	expi := 1
 	if goti != expi {
-		t.Fatalf("incorrect number of elements in PbTree. Expected %v, got %v", expi, goti)
+		t.Fatalf("incorrect number of elements in Pot. Expected %v, got %v", expi, goti)
 	}
 
-	add(n, 1, "011111", "001111", "011111", "000111")
+	testAdd(n, 1, "011111", "001111", "011111", "000111")
 	// check size
 	goti = n.Size()
 	expi = 3
 	if goti != expi {
-		t.Fatalf("incorrect number of elements in PbTree. Expected %v, got %v", expi, goti)
+		t.Fatalf("incorrect number of elements in Pot. Expected %v, got %v", expi, goti)
 	}
-	inds, pos := indexes(n)
+	inds, po := indexes(n)
 	got = fmt.Sprintf("%v", inds)
 	exp = "[3 4 2]"
 	if got != exp {
-		t.Fatalf("incorrect indexes in iteration over PbTree. Expected %v, got %v", exp, got)
+		t.Fatalf("incorrect indexes in iteration over Pot. Expected %v, got %v", exp, got)
 	}
-	got = fmt.Sprintf("%v", pos)
+	got = fmt.Sprintf("%v", po)
 	exp = "[1 2 0]"
 	if got != exp {
-		t.Fatalf("incorrect po-s in iteration over PbTree. Expected %v, got %v", exp, got)
+		t.Fatalf("incorrect po-s in iteration over Pot. Expected %v, got %v", exp, got)
 	}
 }
 
-// func RandomBinAddr()
+// func RandomBoolAddress()
-func TestPbTreeRemove(t *testing.T) {
+func TestPotRemove(t *testing.T) {
-	n := NewPbTree(NewTestAddr("001111", 0), 0)
+	n := NewPot(NewTestAddr("001111", 0), 0)
 	n.Remove(NewTestAddr("001111", 0))
 	exp := ""
 	got := str(n.Pin())
 	if got != exp {
 		t.Fatalf("incorrect pinned value. Expected %v, got %v", exp, got)
 	}
-	add(n, 1, "000000", "011111", "001111", "000111")
+	testAdd(n, 1, "000000", "011111", "001111", "000111")
 	n.Remove(NewTestAddr("001111", 0))
 	goti := n.Size()
 	expi := 3
 	if goti != expi {
-		t.Fatalf("incorrect number of elements in PbTree. Expected %v, got %v", expi, goti)
+		t.Fatalf("incorrect number of elements in Pot. Expected %v, got %v", expi, goti)
 	}
-	inds, pos := indexes(n)
+	inds, po := indexes(n)
 	got = fmt.Sprintf("%v", inds)
-	exp = "[2 4 1]"
+	exp = "[2 4 0]"
 	if got != exp {
-		t.Fatalf("incorrect indexes in iteration over PbTree. Expected %v, got %v", exp, got)
+		t.Fatalf("incorrect indexes in iteration over Pot. Expected %v, got %v", exp, got)
 	}
-	got = fmt.Sprintf("%v", pos)
+	got = fmt.Sprintf("%v", po)
 	exp = "[1 3 0]"
 	if got != exp {
-		t.Fatalf("incorrect po-s in iteration over PbTree. Expected %v, got %v", exp, got)
+		t.Fatalf("incorrect po-s in iteration over Pot. Expected %v, got %v", exp, got)
 	}
 	// remove again
 	n.Remove(NewTestAddr("001111", 0))
-	inds, pos = indexes(n)
+	inds, po = indexes(n)
 	got = fmt.Sprintf("%v", inds)
-	exp = "[2 4 1]"
+	exp = "[2 4]"
 	if got != exp {
-		t.Fatalf("incorrect indexes in iteration over PbTree. Expected %v, got %v", exp, got)
+		t.Fatalf("incorrect indexes in iteration over Pot. Expected %v, got %v", exp, got)
 	}
 
 }
 
-func checkPo(val PbVal) func(PbVal, int) error {
+func TestPotSwap(t *testing.T) {
-	return func(v PbVal, po int) error {
+	max := maxEachNeighbour
+	n := NewPot(nil, 0)
+	var m []*testBVAddr
+	for j := 0; j < 2*max; {
+		v := randomTestBVAddr(keylen, j)
+		_, found := n.Add(v)
+		if !found {
+			m = append(m, v)
+			j++
+		}
+	}
+	k := make(map[string]*testBVAddr)
+	for j := 0; j < max; {
+		v := randomTestBVAddr(keylen, 1)
+		_, found := k[v.String()]
+		if !found {
+			k[v.String()] = v
+			j++
+		}
+	}
+	for _, v := range k {
+		m = append(m, v)
+	}
+	f := func(v PotVal) PotVal {
+		tv := v.(*testBVAddr)
+		if tv.i < max {
+			return nil
+		}
+		tv.i = 0
+		return v
+	}
+	for _, val := range m {
+		n.Swap(val, func(v PotVal) PotVal {
+			if v == nil {
+				return val
+			}
+			return f(v)
+		})
+	}
+	sum := 0
+	n.Each(func(v PotVal, i int) bool {
+		sum++
+		tv := v.(*testBVAddr)
+		if tv.i > 1 {
+			t.Fatalf("item value incorrect, expected 0, got %v", tv.i)
+		}
+		return true
+	})
+	if sum != 2*max {
+		t.Fatalf("incorrect number of elements. expected %v, got %v", max, sum)
+	}
+}
+
+func checkPo(val PotVal) func(PotVal, int) error {
+	return func(v PotVal, po int) error {
 		// check the po
-		exp, _ := val.Prefix(v, 0)
+		exp, _ := val.PO(v, 0)
 		if po != exp {
 			return fmt.Errorf("incorrect prox order for item %v in neighbour iteration for %v. Expected %v, got %v", v, val, exp, po)
 		}
@@ -138,19 +239,19 @@ func checkPo(val PbVal) func(PbVal, int) error {
 	}
 }
 
-func checkOrder(val PbVal) func(PbVal, int) error {
+func checkOrder(val PotVal) func(PotVal, int) error {
-	var pos int = keylen
+	var po int = keylen
-	return func(v PbVal, po int) error {
+	return func(v PotVal, p int) error {
-		if pos < po {
+		if po < p {
-			return fmt.Errorf("incorrect order for item %v in neighbour iteration for %v. PO %v > %v (previous max)", v, val, po, pos)
+			return fmt.Errorf("incorrect order for item %v in neighbour iteration for %v. PO %v > %v (previous max)", v, val, p, po)
 		}
-		pos = po
+		po = p
 		return nil
 	}
 }
 
-func checkValues(m map[string]bool, val PbVal) func(PbVal, int) error {
+func checkValues(m map[string]bool, val PotVal) func(PotVal, int) error {
-	return func(v PbVal, po int) error {
+	return func(v PotVal, po int) error {
 		duplicate, ok := m[v.String()]
 		if !ok {
 			return fmt.Errorf("alien value %v", v)
@@ -165,10 +266,10 @@ func checkValues(m map[string]bool, val PbVal) func(PbVal, int) error {
 
 var errNoCount = errors.New("not count")
 
-func testPbTreeEachNeighbour(n *PbTree, val PbVal, expCount int, fs ...func(PbVal, int) error) error {
+func testPotEachNeighbour(n *Pot, val PotVal, expCount int, fs ...func(PotVal, int) error) error {
 	var err error
 	var count int
-	n.EachNeighbour(val, func(v PbVal, po int) bool {
+	n.EachNeighbour(val, func(v PotVal, po int) bool {
 		for _, f := range fs {
 			err = f(v, po)
 			if err != nil {
@@ -187,29 +288,19 @@ func testPbTreeEachNeighbour(n *PbTree, val PbVal, expCount int, fs ...func(PbVa
 	return err
 }
 
-const (
+func TestPotMerge(t *testing.T) {
-	maxEachNeighbourTests = 500
-	maxEachNeighbour      = 4
-	keylen                = 4
-)
-
-func randomTestAddr(n int, i int) *testAddr {
-	v := RandomAddress().Bin()[:n]
-	return NewTestAddr(v, i)
-}
-
-func TestPbTreeMerge(t *testing.T) {
 	for i := 0; i < maxEachNeighbourTests; i++ {
 		max0 := rand.Intn(maxEachNeighbour) + 1
 		max1 := rand.Intn(maxEachNeighbour) + 1
-		n0 := NewPbTree(nil, 0)
+		n0 := NewPot(nil, 0)
-		n1 := NewPbTree(nil, 0)
+		n1 := NewPot(nil, 0)
+		glog.V(3).Infof("round %v: %v - %v", i, max0, max1)
 		m := make(map[string]bool)
 		for j := 0; j < max0; {
-			v := randomTestAddr(keylen, j)
+			v := randomTestBVAddr(keylen, j)
+			// v := randomTestBVAddr(keylen, j)
 			_, found := n0.Add(v)
 			if !found {
-				glog.V(4).Infof("%v: add %v", j, v)
 				m[v.String()] = false
 				j++
 			}
@@ -217,46 +308,66 @@ func TestPbTreeMerge(t *testing.T) {
 		expAdded := 0
 
 		for j := 0; j < max1; {
-			v := randomTestAddr(keylen, j)
+			v := randomTestBVAddr(keylen, j)
+			// v := randomTestBVAddr(keylen, j)
 			_, found := n1.Add(v)
-			glog.V(4).Infof("%v: add %v", j, v)
 			if !found {
 				j++
 			}
 			_, found = m[v.String()]
 			if !found {
 				expAdded++
-				glog.V(4).Infof("%v: newly added %v", j-1, v)
 				m[v.String()] = false
 			}
 		}
+		if i < 6 {
+			continue
+		}
 		expSize := len(m)
-
 		glog.V(4).Infof("%v-0: pin: %v, size: %v", i, n0.Pin(), max0)
 		glog.V(4).Infof("%v-1: pin: %v, size: %v", i, n1.Pin(), max1)
-		glog.V(4).Infof("%v: %v", i, expSize)
+		glog.V(4).Infof("%v: merged tree size: %v, newly added: %v", i, expSize, expAdded)
-		added := n0.Merge(n1)
+		n, common := Union(n0, n1)
-		size := n0.Size()
+		added := n1.Size() - common
+		size := n.Size()
+
 		if expSize != size {
-			t.Fatalf("incorrect number of elements in merged pbTree, expected %v, got %v\n%v", expSize, size, n0)
+			t.Fatalf("%v: incorrect number of elements in merged pot, expected %v, got %v\n%v", i, expSize, size, n)
 		}
 		if expAdded != added {
-			t.Fatalf("incorrect number of added elements in merged pbTree, expected %v, got %v", expAdded, added)
+			t.Fatalf("%v: incorrect number of added elements in merged pot, expected %v, got %v", i, expAdded, added)
+		}
+		if !checkDuplicates(n.pot) {
+			t.Fatalf("%v: merged pot contains duplicates: \n%v", i, n)
 		}
 		for k, _ := range m {
-			_, found := n0.Add(NewTestAddr(k, 0))
+			_, found := n.Add(NewTestBVAddr(k, 0))
 			if !found {
-				t.Fatalf("merged pbTree missing element %v", k)
+				t.Fatalf("%v: merged pot (size:%v, added: %v) missing element %v\n%v", i, size, added, k, n)
 			}
 		}
 	}
 }
 
-func TestPbTreeEachNeighbourSync(t *testing.T) {
+func checkDuplicates(t *pot) bool {
+	var po int = -1
+	for _, c := range t.bins {
+		if c == nil {
+			return false
+		}
+		if c.po <= po || !checkDuplicates(c) {
+			return false
+		}
+		po = c.po
+	}
+	return true
+}
+
+func TestPotEachNeighbourSync(t *testing.T) {
 	for i := 0; i < maxEachNeighbourTests; i++ {
 		max := rand.Intn(maxEachNeighbour/2) + maxEachNeighbour/2
 		pin := randomTestAddr(keylen, 0)
-		n := NewPbTree(pin, 0)
+		n := NewPot(pin, 0)
 		m := make(map[string]bool)
 		m[pin.String()] = false
 		for j := 1; j <= max; j++ {
@@ -271,15 +382,15 @@ func TestPbTreeEachNeighbourSync(t *testing.T) {
 		}
 		count := rand.Intn(size/2) + size/2
 		val := randomTestAddr(keylen, max+1)
-		glog.V(4).Infof("%v: pin: %v, size: %v, val: %v, count: %v", i, n.Pin(), size, val, count)
+		glog.V(3).Infof("%v: pin: %v, size: %v, val: %v, count: %v", i, n.Pin(), size, val, count)
-		err := testPbTreeEachNeighbour(n, val, count, checkPo(val), checkOrder(val), checkValues(m, val))
+		err := testPotEachNeighbour(n, val, count, checkPo(val), checkOrder(val), checkValues(m, val))
 		if err != nil {
 			t.Fatal(err)
 		}
 		minPoFound := keylen
 		maxPoNotFound := 0
 		for k, found := range m {
-			po, _ := val.Prefix(NewTestAddr(k, 0), 0)
+			po, _ := val.PO(NewTestAddr(k, 0), 0)
 			if found {
 				if po < minPoFound {
 					minPoFound = po
@@ -296,10 +407,10 @@ func TestPbTreeEachNeighbourSync(t *testing.T) {
 	}
 }
 
-func TestPbTreeEachNeighbourAsync(t *testing.T) {
+func TestPotEachNeighbourAsync(t *testing.T) {
 	for i := 0; i < maxEachNeighbourTests; i++ {
 		max := rand.Intn(maxEachNeighbour/2) + maxEachNeighbour/2
-		n := NewPbTree(randomTestAddr(keylen, 0), 0)
+		n := NewPot(randomTestAddr(keylen, 0), 0)
 		var size int = 1
 		for j := 1; j <= max; j++ {
 			v := randomTestAddr(keylen, j)
@@ -320,9 +431,9 @@ func TestPbTreeEachNeighbourAsync(t *testing.T) {
 		mu := sync.Mutex{}
 		m := make(map[string]bool)
 		maxPos := rand.Intn(keylen)
-		glog.V(5).Infof("%v: pin: %v, size: %v, val: %v, count: %v, maxPos: %v", i, n.Pin(), size, val, count, maxPos)
+		glog.V(3).Infof("%v: pin: %v, size: %v, val: %v, count: %v, maxPos: %v", i, n.Pin(), size, val, count, maxPos)
 		msize := 0
-		remember := func(v PbVal, po int) error {
+		remember := func(v PotVal, po int) error {
 			// mu.Lock()
 			// defer mu.Unlock()
 			if po > maxPos {
@@ -337,12 +448,12 @@ func TestPbTreeEachNeighbourAsync(t *testing.T) {
 		if i == 0 {
 			continue
 		}
-		err := testPbTreeEachNeighbour(n, val, count, remember)
+		err := testPotEachNeighbour(n, val, count, remember)
 		if err != nil {
 			glog.V(6).Info(err)
 		}
 		d := 0
-		forget := func(v PbVal, po int) {
+		forget := func(v PotVal, po int) {
 			mu.Lock()
 			defer mu.Unlock()
 			d++
@@ -355,7 +466,6 @@ func TestPbTreeEachNeighbourAsync(t *testing.T) {
 			t.Fatalf("incorrect number of neighbour calls in async iterator. expected %v, got %v", msize, d)
 		}
 		if len(m) != 0 {
-
 			t.Fatalf("incorrect neighbour calls in async iterator. %v items missed:\n%v", len(m), n)
 		}
 	}
@@ -364,7 +474,7 @@ func TestPbTreeEachNeighbourAsync(t *testing.T) {
 func benchmarkEachNeighbourSync(t *testing.B, max, count int, d time.Duration) {
 	t.ReportAllocs()
 	pin := randomTestAddr(keylen, 0)
-	n := NewPbTree(pin, 0)
+	n := NewPot(pin, 0)
 	for j := 1; j <= max; {
 		v := randomTestAddr(keylen, j)
 		_, found := n.Add(v)
@@ -376,7 +486,7 @@ func benchmarkEachNeighbourSync(t *testing.B, max, count int, d time.Duration) {
 	for i := 0; i < t.N; i++ {
 		val := randomTestAddr(keylen, max+1)
 		m := 0
-		n.EachNeighbour(val, func(v PbVal, po int) bool {
+		n.EachNeighbour(val, func(v PotVal, po int) bool {
 			time.Sleep(d)
 			m++
 			if m == count {
@@ -394,7 +504,7 @@ func benchmarkEachNeighbourSync(t *testing.B, max, count int, d time.Duration) {
 func benchmarkEachNeighbourAsync(t *testing.B, max, count int, d time.Duration) {
 	t.ReportAllocs()
 	pin := randomTestAddr(keylen, 0)
-	n := NewPbTree(pin, 0)
+	n := NewPot(pin, 0)
 	for j := 1; j <= max; {
 		v := randomTestAddr(keylen, j)
 		_, found := n.Add(v)
@@ -405,7 +515,7 @@ func benchmarkEachNeighbourAsync(t *testing.B, max, count int, d time.Duration)
 	t.ResetTimer()
 	for i := 0; i < t.N; i++ {
 		val := randomTestAddr(keylen, max+1)
-		n.EachNeighbourAsync(val, count, keylen, func(v PbVal, po int) {
+		n.EachNeighbourAsync(val, count, keylen, func(v PotVal, po int) {
 			time.Sleep(d)
 		}, true)
 	}

swarm/network/hive.go

@@ -17,10 +17,8 @@
 package network
 
 import (
-	"fmt"
-	// "math/rand"
-	// "sort"
 	"bytes"
+	"fmt"
 	"sync"
 	"time"
 
@@ -42,24 +40,24 @@ peer connections and disconnections are reported and registered
 to keep the nodetable uptodate
 */
 type Overlay interface {
-	Register(NodeAddr) error
+	Register(...PeerAddr) error
-	On(Node) (Node, error)
-	Off(Node)
 
-	EachNode([]byte, int, func(Node) bool)
+	On(Peer)
-	EachNodeAddr([]byte, int, func(NodeAddr) bool)
+	Off(Peer)
 
-	SuggestNodeAddr() NodeAddr
+	EachLivePeer([]byte, int, func(Peer) bool)
-	SuggestOrder() int
+	EachPeer([]byte, int, func(PeerAddr) bool)
 
-	Info() string
+	SuggestPeer() (PeerAddr, int, bool)
+
+	String() string
 }
 
 // Hive implements the PeerPool interface
 type Hive struct {
 	*HiveParams // settings
 	Overlay     // the overlay topology driver
-	peers       map[discover.NodeID]Node
+	peers       map[discover.NodeID]Peer
 
 	lock   sync.Mutex
 	quit   chan bool
@@ -94,7 +92,7 @@ func NewHive(params *HiveParams, overlay Overlay) *Hive {
 	return &Hive{
 		HiveParams: params,
 		Overlay:    overlay,
-		peers:      make(map[discover.NodeID]Node),
+		peers:      make(map[discover.NodeID]Peer),
 	}
 }
 
@@ -118,6 +116,10 @@ and correctness of responses should be checked
 If the proxBin of peers in the response is incorrect the sender should be
 disconnected
 */
+
+// peersMsg encapsulates an array of peer addresses
+// used for communicating about known peers
+// relevvant for bootstrapping connectivity and updating peersets
 type peersMsg struct {
 	Peers []*peerAddr
 }
@@ -146,8 +148,7 @@ func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Ti
 	self.toggle = make(chan bool)
 	self.more = make(chan bool)
 	self.quit = make(chan bool)
-	order := -1
+	glog.V(logger.Debug).Infof("hive started")
-	glog.V(logger.Detail).Infof("hive started")
 	// this loop is doing bootstrapping and maintains a healthy table
 	go self.keepAlive(af)
 	go func() {
@@ -159,27 +160,25 @@ func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Ti
 				return
 			}
 			glog.V(logger.Detail).Infof("hive delegate to overlay driver: suggest addr to connect to")
-			addr := self.SuggestNodeAddr()
+			addr, order, want := self.SuggestPeer()
 
 			if addr != nil {
 				glog.V(logger.Detail).Infof("========> connect to bee %v", addr)
 				err := connectPeer(NodeId(addr).NodeID.String())
 				if err != nil {
 					glog.V(logger.Detail).Infof("===X====> connect to bee %v failed: %v", addr, err)
-
 				}
 			}
-			glog.V(logger.Detail).Infof("hive delegate to overlay driver: suggest order for getPeersMsg")
+			if want {
-			order = self.SuggestOrder()
 				req := &getPeersMsg{
 					Order: uint(order),
 					Max:   self.MaxPeersPerRequest,
 				}
 				var i uint
 				var err error
-			glog.V(logger.Debug).Infof("requesting bees of PO%03d from %v (each max %v)", order, self.PeersBroadcastSetSize, self.MaxPeersPerRequest)
+				glog.V(logger.Detail).Infof("requesting bees of PO%03d from %v (each max %v)", order, self.PeersBroadcastSetSize, self.MaxPeersPerRequest)
-			self.EachNode(nil, order, func(n Node) bool {
+				self.EachLivePeer(nil, order, func(n Peer) bool {
-				glog.V(logger.Debug).Infof("%T sent to %v", req, n.ID())
+					glog.V(logger.Detail).Infof("%T sent to %v", req, n.ID())
 					err = n.Send(req)
 					if err == nil {
 						i++
@@ -189,20 +188,18 @@ func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Ti
 					}
 					return true
 				})
-			glog.V(logger.Debug).Infof("sent %T to %d/%d peers", req, i, self.PeersBroadcastSetSize)
+				glog.V(logger.Detail).Infof("sent %T to %d/%d peers", req, i, self.PeersBroadcastSetSize)
-			// only switch off if full
-			var need bool
-			if order < 256 || addr != nil {
-				need = true
 			}
+			glog.V(logger.Info).Infof("%v", self)
+
 			select {
-			case self.toggle <- need:
+			case self.toggle <- want:
-				glog.V(logger.Debug).Infof("keep hive alive: %v", need)
+				glog.V(logger.Detail).Infof("keep hive alive: %v", want)
 			case <-self.quit:
 				return
 			}
 		}
-		glog.V(logger.Debug).Infof("%v", self.Info())
+		glog.V(logger.Warn).Infof("%v", self)
 	}()
 	return
 }
@@ -217,12 +214,12 @@ func (self *Hive) ticker() <-chan time.Time {
 // wake state is toggled by writing to self.toggle
 // it restarts if the table becomes non-full again due to disconnections
 func (self *Hive) keepAlive(af func() <-chan time.Time) {
-	glog.V(logger.Debug).Infof("keep alive loop started")
+	glog.V(logger.Detail).Infof("keep alive loop started")
 	alarm := af()
 	for {
 		select {
 		case <-alarm:
-			glog.V(logger.Debug).Infof("wake up: make hive alive")
+			glog.V(logger.Detail).Infof("wake up: make hive alive")
 			self.wake()
 		case need := <-self.toggle:
 			if alarm == nil && need {
@@ -246,45 +243,19 @@ func (self *Hive) Stop() {
 func (self *Hive) wake() {
 	select {
 	case self.more <- true:
-		glog.V(logger.Debug).Infof("hive woken up")
+		glog.V(logger.Detail).Infof("hive woken up")
 	case <-self.quit:
 	default:
-		glog.V(logger.Debug).Infof("hive already awake")
+		glog.V(logger.Detail).Infof("hive already awake")
 	}
 }
 
-// func (self *Hive) anyN(n int, peers []Node) []Node {
+// Add is called at the end of a successful protocol handshake
-// 	self.lock.Lock()
+// to register a connected (live) peer
-// 	defer self.lock.Unlock()
+func (self *Hive) Add(p Peer) error {
-// 	pick := rand.Perm(len(peers))
-// 	sort.Ints(pick)
-// 	var nodes []Node
-// 	j := 0
-// 	i := 0
-// 	for i, node := range peers {
-// 		if i == pick[j] {
-// 			j++
-// 			nodes = append(nodes, node)
-// 			if j == n {
-// 				break
-// 			}
-// 		}
-// 	}
-// 	return nodes
-// }
-
-// Add is called at the end of a successful protocol handshake to register a peer onlune
-func (self *Hive) Add(p Node) error {
 	defer self.wake()
-	glog.V(logger.Detail).Infof("add new bee %v", p)
+	glog.V(logger.Debug).Infof("to add new bee %v", p)
-	drop, err := self.On(p)
+	self.On(p)
-	if err != nil {
-		return err
-	}
-	if drop != nil {
-		drop.Drop()
-		return nil
-	}
 
 	self.lock.Lock()
 	self.peers[p.ID()] = p
@@ -297,7 +268,7 @@ func (self *Hive) Add(p Node) error {
 }
 
 // Remove called after peer is disconnected
-func (self *Hive) Remove(p Node) {
+func (self *Hive) Remove(p Peer) {
 	defer self.wake()
 	glog.V(logger.Debug).Infof("remove bee %v", p)
 	self.Off(p)
@@ -306,37 +277,37 @@ func (self *Hive) Remove(p Node) {
 	self.lock.Unlock()
 }
 
-// func (self *Hive) Get(n string) Node {
-// 	return Node(self.peers[n])
-// }
-
 // handlePeersMsg called by the protocol when receiving peerset (for target address)
-// list of nodes ([]NodeAddr in peersMsg is added to the overlay db
+// list of nodes ([]PeerAddr in peersMsg) is added to the overlay db using the
-func (self *Hive) handlePeersMsg(p Node) func(interface{}) error {
+// Register interface method
+func (self *Hive) handlePeersMsg(p Peer) func(interface{}) error {
 	return func(msg interface{}) error {
 		// wake up the hive on news of new arrival
 		defer self.wake()
 		// register all addresses
-		var err error
+		var nas []PeerAddr
-		req := msg.(*peersMsg)
+		for _, na := range msg.(*peersMsg).Peers {
-		for _, p := range req.Peers {
+			nas = append(nas, PeerAddr(na))
-			err = self.Register(p)
-			// TODO: these are known to our peer, so do not resend during the session
 		}
-		// FIXME: only the last error is returned
+		return self.Register(nas...)
-		return err
 	}
 }
 
-// HandleGetPeersMsg called by the protocol when receiving peerset (for target address)
+// handleGetPeersMsg is called by the protocol when receiving a
-// peersMsgData is converted to a slice of NodeRecords for Kademlia
+// peerset (for target address) request
-// this is to store all thats needed
+// peers suggestions are retrieved from the overlay topology driver
-func (self *Hive) handleGetPeersMsg(p Node) func(interface{}) error {
+// using the EachLivePeer interface iterator method
+// peers sent are remembered throughout a session and not sent twice
+func (self *Hive) handleGetPeersMsg(p Peer) func(interface{}) error {
 	return func(msg interface{}) error {
 		req := msg.(*getPeersMsg)
 		var peers []*peerAddr
-		self.EachNode(p.OverlayAddr(), int(req.Order), func(n Node) bool {
+		alreadySent := p.Peers()
-			if bytes.Compare(n.OverlayAddr(), p.OverlayAddr()) != 0 {
+		self.EachLivePeer(p.OverlayAddr(), int(req.Order), func(n Peer) bool {
+			if bytes.Compare(n.OverlayAddr(), p.OverlayAddr()) != 0 &&
+				// only send peers we have not sent before in this session
+				!alreadySent[n.ID()] {
+				alreadySent[n.ID()] = true
 				peers = append(peers, &peerAddr{n.OverlayAddr(), n.UnderlayAddr()})
 			}
 			return len(peers) < int(req.Max)
@@ -353,6 +324,10 @@ func (self *Hive) handleGetPeersMsg(p Node) func(interface{}) error {
 	}
 }
 
+func (self *Hive) NodeInfo() interface{} {
+	return interface{}(self.String())
+}
+
 func (self *Hive) PeerInfo(id discover.NodeID) interface{} {
 	self.lock.Lock()
 	defer self.lock.Unlock()

swarm/network/hive_test.go

@@ -36,7 +36,7 @@ func (self *testConnect) connect(na string) error {
 	return nil
 }
 
-func newBzzHiveTester(t *testing.T, n int, addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Node) error) *bzzTester {
+func newBzzHiveTester(t *testing.T, n int, addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester {
 	s := p2ptest.NewProtocolTester(t, NodeId(addr), n, newTestBzzProtocol(addr, pp, ct, services))
 	return &bzzTester{
 		addr:            addr,
@@ -55,7 +55,7 @@ func TestOverlayRegistration(t *testing.T) {
 
 	// connect to the other peer
 	id := s.Ids[0]
-	raddr := NodeIdToAddr(id)
+	raddr := NewPeerAddrFromNodeId(id)
 	s.runHandshakes()
 
 	// hive should have called the overlay
@@ -74,9 +74,9 @@ func TestRegisterAndConnect(t *testing.T) {
 	// register the node with the peerPool
 	id := p2ptest.RandomNodeId()
 	s.Start(id)
-	raddr := NodeIdToAddr(id)
+	raddr := NewPeerAddrFromNodeId(id)
 	pp.Register(raddr)
-	glog.V(5).Infof("%v", pp.Info())
+	glog.V(5).Infof("%v", pp)
 	// start the hive and wait for the connection
 	tc := &testConnect{
 		connectf: func(c string) error {

swarm/network/kademlia.go

@@ -0,0 +1,352 @@
+// Copyright 2017 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+package network
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/pot"
+)
+
+/*
+
+Taking the proximity order relative to a fix point x classifies the points in
+the space (n byte long byte sequences) into bins. Items in each are at
+most half as distant from x as items in the previous bin. Given a sample of
+uniformly distributed items (a hash function over arbitrary sequence) the
+proximity scale maps onto series of subsets with cardinalities on a negative
+exponential scale.
+
+It also has the property that any two item belonging to the same bin are at
+most half as distant from each other as they are from x.
+
+If we think of random sample of items in the bins as connections in a network of interconnected nodes than relative proximity can serve as the basis for local
+decisions for graph traversal where the task is to find a route between two
+points. Since in every hop, the finite distance halves, there is
+a guaranteed constant maximum limit on the number of hops needed to reach one
+node from the other.
+
+proxLimit is dynamically adjusted so that
+1) there is no empty buckets in bin < proxLimit and
+2) the sum of all items are the minimum possible but higher than ProxBinSize
+adjust Prox (proxLimit and proxSize after an insertion/removal of nodes)
+caller holds the lock
+
+*/
+
+// default values for Kademlia Parameters
+const (
+	bucketSize   = 4
+	proxBinSize  = 2
+	maxProx      = 8
+	connRetryExp = 2
+)
+
+var (
+	purgeInterval        = 42 * time.Hour
+	initialRetryInterval = 42 * time.Millisecond
+	maxIdleInterval      = 42 * 1000 * time.Millisecond
+)
+
+type KadParams struct {
+	// adjustable parameters
+	MaxProx              int
+	ProxBinSize          int
+	BucketSize           int
+	PurgeInterval        time.Duration
+	InitialRetryInterval time.Duration
+	MaxIdleInterval      time.Duration
+	ConnRetryExp         int
+}
+
+// NewKadParams() returns a params struct with default values
+func NewKadParams() *KadParams {
+	return &KadParams{
+		MaxProx:              maxProx,
+		ProxBinSize:          proxBinSize,
+		BucketSize:           bucketSize,
+		PurgeInterval:        purgeInterval,
+		InitialRetryInterval: initialRetryInterval,
+		MaxIdleInterval:      maxIdleInterval,
+		ConnRetryExp:         connRetryExp,
+	}
+}
+
+// Kademlia is a table of live peers and a db of known peers
+type Kademlia struct {
+	addr         *pot.HashAddress // immutable baseaddress of the table
+	*KadParams                    // Kademlia configuration parameters
+	conns, peers *pot.Pot         // pots container for peers
+}
+
+// NewKademlia(addr, params) 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()
+	}
+	base := pot.NewHashAddressFromBytes(addr)
+	return &Kademlia{
+		addr:      base,
+		KadParams: params,
+		conns:     pot.NewPot(nil, 0),
+		peers:     pot.NewPot(nil, 0),
+	}
+}
+
+// KadPeer represents a Kademlia Peer and extends
+// * PeerAddr interface (overlay and underlay addresses)
+// * Peer interface (id, last seen, drop)
+// * HashAddress as derived from PeerAddr overlay implement pot.PoVal interface
+type KadPeer struct {
+	*pot.HashAddress
+	PeerAddr
+	Peer            Peer
+	seenAt, readyAt time.Time
+}
+
+func (self *KadPeer) String() string {
+	return string(self.OverlayAddr())
+}
+
+func (self *Kademlia) callable(kp *KadPeer) bool {
+	if kp.Peer != nil || kp.readyAt.After(time.Now()) {
+		return false
+	}
+	delta := time.Since(time.Time(kp.seenAt))
+	if delta < self.InitialRetryInterval {
+		delta = self.InitialRetryInterval
+	}
+	interval := time.Duration(delta * time.Duration(self.ConnRetryExp))
+
+	glog.V(logger.Detail).Infof("peer %v ready to be tried. seen at %v (%v ago), scheduled at %v", kp, kp.seenAt, delta, kp.readyAt)
+
+	// scheduling next check
+	kp.readyAt = time.Now().Add(interval)
+	return true
+}
+
+// NewKadPeer(na) creates a kademlia peer from a PeerAddr interface
+func NewKadPeer(na PeerAddr) *KadPeer {
+	return &KadPeer{
+		HashAddress: pot.NewHashAddressFromBytes(na.OverlayAddr()),
+		PeerAddr:    na,
+	}
+}
+
+// Register(nas) enters each PeerAddr as kademlia peers into the
+// database of known peers
+func (self *Kademlia) Register(nas ...PeerAddr) error {
+	np := pot.NewPot(nil, 0)
+	for _, na := range nas {
+		p := NewKadPeer(na)
+		np, _, _ = pot.Add(np, pot.PotVal(p))
+	}
+	common := self.peers.Merge(np)
+	glog.V(6).Infof("add peers: %v out of %v new", np.Size()-common, np.Size())
+	return nil
+}
+
+// On(p) inserts the peer as a kademlia peer into the live peers
+func (self *Kademlia) On(p Peer) {
+	kp := NewKadPeer(p)
+	self.conns.Swap(kp, func(v pot.PotVal) pot.PotVal {
+		if v == nil {
+			self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal {
+				if v != nil {
+					kp = v.(*KadPeer)
+				}
+				kp.Peer = p
+				return pot.PotVal(kp)
+			})
+			return pot.PotVal(kp)
+		}
+		return v
+	})
+}
+
+// Off removes a peer from among live peers
+func (self *Kademlia) Off(p Peer) {
+	kp := NewKadPeer(p)
+	self.conns.Remove(kp)
+	kp.Peer = nil
+}
+
+// EachLivePeer(base, po, f) is an iterator applying f to each live peer
+// that has proximity order po as measure from the base
+func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer) bool) {
+	var p pot.PotVal
+	if base == nil {
+		p = pot.PotVal(self.addr)
+	} else {
+		p = pot.NewHashAddressFromBytes(base)
+	}
+	self.conns.EachNeighbour(p, func(val pot.PotVal, po int) bool {
+		if po == o {
+			return f(val.(*KadPeer).Peer)
+		}
+		return po < o
+	})
+}
+
+// EachPeer(base, po, f) is an iterator applying f to each known peer
+// that has proximity order po as measure from the base
+func (self *Kademlia) EachPeer(base []byte, o int, f func(PeerAddr) bool) {
+	var p pot.PotVal
+	if base == nil {
+		p = pot.PotVal(self.addr)
+	} else {
+		p = pot.NewHashAddressFromBytes(base)
+	}
+	self.peers.EachNeighbour(p, func(val pot.PotVal, po int) bool {
+		if po == o {
+			return f(val.(*KadPeer).PeerAddr)
+		}
+		return po < o
+	})
+}
+
+// proxLimit() returns the proximity order that defines the distance of
+// the nearest neighbour set with cardinality >= ProxBinSize
+func (self *Kademlia) proxLimit() int {
+	var proxLimit int
+	var size int
+	f := func(v pot.PotVal, i int) bool {
+		size++
+		proxLimit = i
+		return size <= self.BucketSize
+	}
+	self.conns.EachNeighbour(pot.PotVal(self.addr), f)
+	return proxLimit
+}
+
+// SuggestPeer returns a known peer for the lowest proximity bin for the
+// lowest bincount below proxLimit
+// naturally if there is an empty row it returns a peer for that
+//
+func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) {
+	minsize := self.BucketSize
+	proxLimit := self.proxLimit()
+	var bpo []int
+	self.conns.EachBin(0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool {
+		if size < minsize {
+			minsize = size
+			bpo = append(bpo, po)
+		}
+		return size > 0 && po < proxLimit
+	})
+	// all buckets are full
+	// minsize == self.BucketSize
+	if len(bpo) == 0 {
+		return nil, 0, false
+	}
+	// as long as we got candidate peers to connect to
+	// just ask for closest peers
+	o = 256
+	// try to select a peer
+	for i := len(bpo) - 1; i >= 0; i-- {
+		// find the first callable peer
+		self.peers.EachBin(bpo[i], func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool {
+			f(func(val pot.PotVal, i int) bool {
+				cp := val.(*KadPeer)
+				if self.callable(cp) {
+					p = cp
+					return false
+				}
+				return true
+			})
+			return false
+		})
+		// found a candidate
+		if p != nil {
+			break
+		}
+		// cannot find a candidate, ask for more for this proximity bin specifically
+		o = bpo[i]
+	}
+	return p, o, true
+}
+
+// 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.addr.String()[:6]))
+	rows = append(rows, fmt.Sprintf("population: %d (%d), ProxBinSize: %d, BucketSize: %d", self.conns.Size(), self.peers.Size(), self.ProxBinSize, self.BucketSize))
+
+	liverows := make([]string, self.MaxProx)
+	peersrows := make([]string, self.MaxProx)
+	var proxLimit int
+	rest := self.conns.Size()
+	self.conns.EachBin(0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool {
+		var rowlen int
+		row := []string{fmt.Sprintf("%2d", size)}
+		rest -= size
+		f(func(val pot.PotVal, vpo int) bool {
+			row = append(row, val.(*KadPeer).String()[:6])
+			rowlen++
+			return rowlen < 4
+		})
+		if rowlen == 0 || rest < self.ProxBinSize {
+			proxLimit = po - 1
+		}
+		for ; rowlen < 4; rowlen++ {
+			row = append(row, "      ")
+		}
+		liverows[po] = strings.Join(row, " ")
+		return true
+	})
+
+	self.peers.EachBin(0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool {
+		var rowlen int
+		row := []string{fmt.Sprintf("| %2d", size)}
+		f(func(val pot.PotVal, vpo int) bool {
+			kp := val.(*KadPeer)
+			if kp.Peer != nil {
+				return true
+			}
+			row = append(row, kp.String()[:6])
+			rowlen++
+			return rowlen < 4
+		})
+		peersrows[po] = strings.Join(row, " ")
+		return true
+	})
+
+	for i := 0; i < self.MaxProx; i++ {
+		if i == proxLimit {
+			rows = append(rows, fmt.Sprintf("============ PROX LIMIT: %d ==========================================", i))
+		}
+		left := liverows[i]
+		right := peersrows[i]
+		if len(left) == 0 {
+			left = "0                                           "
+		}
+		if len(right) == 0 {
+			right = "0                                           "
+		}
+		rows = append(rows, fmt.Sprintf("%03d %v | %v", i, left, right))
+	}
+	rows = append(rows, "=========================================================================")
+	return strings.Join(rows, "\n")
+}

swarm/network/pbtree.go

@@ -1,603 +0,0 @@
-package network
-
-import (
-	"fmt"
-	"sync"
-
-	"github.com/ethereum/go-ethereum/logger/glog"
-)
-
-/*
-PbTree implements a pinned binary tree.
-It is a generic container type for objects implementing the PbVal interface
-Each item is pinned to the node at pos x such that all items pinned to all
-ancestor nodes share an at least x bits long key prefix
-
-PbTree
-* does not need to copy keys of the item type.
-* retrieval, insertion and deletion by key involves log(n) pointer lookups
-* for any item retrieval respects proximity order on logarithmic distance
-* provide syncronous iterators  respecting proximity ordering  wrt any item
-* provide asyncronous iterator (for parallel execution of operations) over n items
-* allows cheap iteration over ranges
-* TODO: asymmetric parallelisable merge
-
-*/
-// PbTree is the root node type k(same for root, branching node and leat)
-type PbTree struct {
-	lock sync.RWMutex
-	*pbTree
-}
-
-// pbTree is the node type (same for root, branching node and leat)
-type pbTree struct {
-	pin  PbVal
-	bins []*pbTree
-	size int
-	pos  int
-}
-
-// PbVal is the interface the generic container item should implement
-type PbVal interface {
-	Prefix(PbVal, int) (pos int, eq bool)
-	String() string
-}
-
-// PbTree constructor. Requires  value of type PbVal to pin
-// and pos to point to a span in the PbVal key
-// The pinned item counts towards the size
-func NewPbTree(v PbVal, pos int) *PbTree {
-	return &PbTree{
-		pbTree: &pbTree{
-			pin:  v,
-			pos:  pos,
-			size: 1,
-		},
-	}
-}
-
-// Pin() returns the pinned element (key) of the PbTree
-func (t *PbTree) Pin() PbVal {
-	return t.pin
-}
-
-// Size() returns the number of values in the PbTree
-func (t *PbTree) Size() int {
-	t.lock.RLock()
-	defer t.lock.RUnlock()
-	return t.size
-}
-
-// Add(v) inserts v into the PbTree
-func (t *PbTree) Add(val PbVal) (pos int, found bool) {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-	return t.add(val)
-}
-
-func (t *pbTree) add(val PbVal) (pos int, found bool) {
-	if t.pin == nil {
-		t.pin = val
-		t.size = 1
-		return 0, false
-	}
-	pos, found = val.Prefix(t.pin, t.pos)
-	if found {
-		t.pin = val
-		return pos, true
-	}
-
-	n, j := t.getPos(pos)
-	if n != nil {
-		p, f := n.add(val)
-		if !f {
-			t.size++
-		}
-		return p, f
-	}
-	// insert empty sub-pbTree and pin it to val
-	ins := &pbTree{
-		pin:  val,
-		pos:  pos,
-		size: 1,
-	}
-	t.size++
-	t.bins = append(t.bins, nil)
-	copy(t.bins[j+1:], t.bins[j:])
-	t.bins[j] = ins
-	return pos, false
-}
-
-// Remove(v) deletes v from the PbTree and returns
-// the proximity order of v
-func (t *PbTree) Remove(val PbVal) (pos int, found bool) {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-	return t.remove(val)
-}
-
-func (t *pbTree) remove(val PbVal) (pos int, found bool) {
-	pos, found = val.Prefix(t.pin, t.pos)
-	if found {
-		t.size -= 1
-		if t.size == 0 {
-			t.pin = nil
-			return t.pos, true
-		}
-		i := len(t.bins) - 1
-		last := t.bins[i]
-		t.bins = append(t.bins[:i], last.bins...)
-		t.pin = last.pin
-		return t.pos, true
-	}
-	for j, n := range t.bins {
-		if n.pos == pos {
-			p, f := n.remove(val)
-			if f {
-				t.size--
-			}
-			last := len(t.bins) - 1
-			copy(t.bins[j:], t.bins[j+1:])
-			t.bins[last] = nil // or the zero value of T
-			t.bins = t.bins[:last]
-			return p, f
-		}
-		if n.pos > pos {
-			return 0, false
-		}
-	}
-	return 0, false
-}
-
-func (t *PbTree) Merge(t1 *PbTree) int {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-	t1.lock.RLock()
-	defer t1.lock.RUnlock()
-	return t.merge(t1.pbTree)
-}
-
-func (t *pbTree) merge(t1 *pbTree) int {
-	if t.pin == nil {
-		if t1.pin == nil {
-			return 0
-		}
-		t.pin = t1.pin
-		t.size = 1
-		return t.merge(t1)
-	}
-	i := 0
-	j := 0
-	added := 0
-
-	var bins []*pbTree
-	var m, t2 *pbTree
-	var is []int
-	pos, _ := t.pin.Prefix(t1.pin, 0)
-	n, l := t.getPos(pos)
-	for {
-		glog.V(4).Infof("%v-%v, i: %v, j: %v, pos: %v, n: %v, l: %v", t, t1, i, j, pos, n, l)
-
-		if i == len(t.bins) && j == len(t1.bins) {
-			if l < len(t.bins) {
-				glog.V(4).Infof("l < len(t.bins): break")
-				break
-			}
-		}
-
-		if i == l && j <= len(t1.bins) {
-			if m == nil {
-				if n == nil {
-					n = &pbTree{
-						pin:  t1.pin,
-						size: 1,
-						pos:  pos,
-					}
-					added++
-				} else {
-					_, found := n.add(t1.pin)
-					if !found {
-						added++
-					}
-					glog.V(4).Infof("%v-%v: i: %v, j: %v. adding t1.pin (found: %v) to n: %v", t.pin, t1.pin, i, j, found, n)
-				}
-				m = n
-				bins = append(bins, n)
-			}
-			if j < len(t1.bins) {
-				if t2 == nil && t1.bins[j].pos == 0 {
-					t2 = t1.bins[j]
-					_, found := n.add(t2.pin)
-					if !found {
-						added++
-					}
-					glog.V(4).Infof("%v-%v: i: %v, j: %v. will merge into 0 the 0 pos branch from 1: %v", t.pin, t1.pin, i, j, t1.bins[j])
-					glog.V(4).Infof("%v-%v: i: %v, j: %v. adding t2.pin: %v (found: %v) to n: %v", t.pin, t1.pin, i, j, t2.pin, found, n)
-				} else {
-					glog.V(4).Infof("%v-%v: i: %v, j: %v. merge into 0 from 1: %v", t.pin, t1.pin, i, j, t1.bins[j])
-					added += n.merge(t1.bins[j])
-				}
-				j++
-				continue
-			}
-			if t2 == nil && l == len(t.bins) {
-				break
-			}
-			if l < len(t.bins) {
-				i++
-			}
-			glog.V(4).Infof("%v-%v: i: %v, j: %v. t1 reset to %v", t.pin, t1.pin, i, j, t2)
-			if t2 != nil {
-				t1 = t2
-				j = 0
-				m = nil
-				t2 = nil
-				pos, _ = t.pin.Prefix(t1.pin, 0)
-				n, l = t.getPos(pos)
-			}
-			continue
-		}
-
-		if j == len(t1.bins) || t1.bins[j].pos > t.bins[i].pos {
-			glog.V(4).Infof("%v-%v: i: %v, j: %v. insert from 0: %v", t.pin, t1.pin, i, j, t.bins[i])
-			bins = append(bins, t.bins[i])
-			i++
-			continue
-		}
-
-		if i < l && t1.bins[j].pos < t.bins[i].pos {
-			glog.V(4).Infof("%v-%v: i: %v, j: %v. insert from 1: %v", t.pin, t1.pin, i, j, t1.bins[j])
-			m := &pbTree{}
-			added += m.merge(t1.bins[j])
-			bins = append(bins, n)
-			j++
-			continue
-		}
-
-		glog.V(4).Infof("%v-%v: i: %v, j: %v. merge: %v", t.pin, t1.pin, i, j, t.bins[i], t1.bins[j])
-		bins = append(bins, t.bins[i])
-		is = append(is, i)
-		i++
-		j++
-	}
-	t.bins = bins
-	wg := sync.WaitGroup{}
-	if len(is) > 0 {
-		wg.Add(len(is))
-		for _, i := range is {
-			go func(k int) {
-				defer wg.Done()
-				is[k] = bins[k].merge(t1.bins[k])
-			}(i)
-		}
-		wg.Wait()
-		for _, a := range is {
-			added += a
-		}
-	}
-	glog.V(4).Infof("%v-%v: added: %v", t.pin, t1.pin, added)
-	t.size += added
-	return added
-}
-
-// func (t *PbTree) Traverse(f func(val PbVal, pos int) (next bool, fork bool)) *PbTree {
-//   t.lock.Lock()
-//   defer t.lock.Unlock()
-//   return t.traverse(f)
-// }
-
-// func (t *pbTree) traverse(n *pbTree, f func(val PbVal, pos int) (next bool, fork bool)) *PbTree {
-
-// next, stop = pinf(t.pin,t.pos)
-//  if !next
-// }
-
-// Each(f) is a synchronous iterator over the bins of a node
-// it does NOT include the pinned item of the root
-// respecting an ordering
-// proximity > pinnedness
-func (t *PbTree) Each(f func(PbVal, int) bool) bool {
-	t.lock.RLock()
-	defer t.lock.RUnlock()
-	return t.each(f)
-}
-
-func (t *pbTree) each(f func(PbVal, int) bool) bool {
-	var next bool
-	for _, n := range t.bins {
-		next = n.each(f)
-		if !next {
-			return false
-		}
-	}
-	next = f(t.pin, t.pos)
-	if !next {
-		return false
-	}
-
-	return true
-}
-
-// syncronous iterator over neighbours of any target val
-// even if an item at val's exact address is in the pbtree,
-// it is not included in the iteration: $val \not\in Neighbours(val)$
-func (t *PbTree) EachNeighbour(val PbVal, f func(PbVal, int) bool) bool {
-	t.lock.RLock()
-	defer t.lock.RUnlock()
-	return t.eachNeighbour(val, f)
-}
-
-func (t *pbTree) eachNeighbour(val PbVal, f func(PbVal, int) bool) bool {
-	var next bool
-	l := len(t.bins)
-	var n *pbTree
-	ir := l
-	il := l
-	pos, eq := val.Prefix(t.pin, t.pos)
-	if !eq {
-		n, il = t.getPos(pos)
-		if n != nil {
-			next = n.eachNeighbour(val, f)
-			if !next {
-				return false
-			}
-			ir = il
-		} else {
-			ir = il - 1
-		}
-	}
-
-	next = f(t.pin, pos)
-	if !next {
-		return false
-	}
-
-	for i := l - 1; i > ir; i-- {
-		next = t.bins[i].each(func(v PbVal, _ int) bool {
-			return f(v, pos)
-		})
-		if !next {
-			return false
-		}
-	}
-
-	for i := il - 1; i >= 0; i-- {
-		n := t.bins[i]
-		next = n.each(func(v PbVal, _ int) bool {
-			return f(v, n.pos)
-		})
-		if !next {
-			return false
-		}
-	}
-	return true
-}
-
-func (t *PbTree) EachNeighbourAsync(val PbVal, max int, maxPos int, f func(PbVal, int), wait bool) {
-	t.lock.RLock()
-	defer t.lock.RUnlock()
-	if max > t.size {
-		max = t.size
-	}
-	var wg *sync.WaitGroup
-	if wait {
-		wg = &sync.WaitGroup{}
-	}
-	_ = t.eachNeighbourAsync(val, max, maxPos, f, wg)
-	if wait {
-		wg.Wait()
-	}
-}
-
-func (t *pbTree) eachNeighbourAsync(val PbVal, max int, maxPos int, f func(PbVal, int), wg *sync.WaitGroup) (extra int) {
-
-	l := len(t.bins)
-	var n *pbTree
-	il := l
-	ir := l
-	// ic := l
-
-	pos, eq := val.Prefix(t.pin, t.pos)
-	glog.V(4).Infof("pin %v: each neighbour iteration async. count: %v/%v, t.pos: %v, pos: %v, maxPos: %v", t.pin, max, t.size, t.pos, pos, maxPos)
-
-	// if pos is too close, set the pivot branch (pom) to maxPos
-	pom := pos
-	if pom > maxPos {
-		pom = maxPos
-	}
-	n, il = t.getPos(pom)
-	ir = il
-	// if pivot branch exists and pos is not too close, iterate on the pivot branch
-	if pom == pos {
-		if n != nil {
-
-			m := n.size
-			if max < m {
-				m = max
-			}
-			max -= m
-
-			glog.V(4).Infof("pin %v recursive branch %v pos: %v/%v (%v), count: %v/%v", t.pin, n.pin, n.pos, maxPos, il, m, max)
-
-			extra = n.eachNeighbourAsync(val, m, maxPos, f, wg)
-
-		} else {
-			if !eq {
-				ir--
-			}
-		}
-	} else {
-		extra++
-		max--
-		if n != nil {
-			il++
-		}
-		// before checking max, add up the extra elements
-		// on the close branches that are skipped (if pos is too close)
-		for i := l - 1; i >= il; i-- {
-			s := t.bins[i]
-			m := s.size
-			if max < m {
-				m = max
-			}
-			max -= m
-			extra += m
-		}
-		glog.V(4).Infof("count extra pos: %v/%v -> %v", pos, maxPos, extra)
-	}
-	glog.V(4).Infof("branch %v: %v/%v, il: %v, ir: %v, l: %v, extra: %v", t.pin, pos, maxPos, il, ir, l, extra)
-
-	var m int
-	// if max <= 0 {
-	// 	return
-	// }
-	// unless pos was too close, call f on the pinned element
-	if pom == pos {
-
-		glog.V(4).Infof("pinned val %v, t.pos: %v, pos: %v (%v), count: %v, max: %v", t.pin, pos, maxPos, "pin", 1, max)
-		glog.V(4).Infof("BEFORE %v %v %v", 1, max, extra)
-		m, max, extra = need(1, max, extra)
-		if m <= 0 {
-			return
-		}
-		glog.V(4).Infof("AFTER %v %v %v", 1, max, extra)
-		glog.V(4).Infof("pinned val %v, t.pos: %v, pos: %v (%v), count: %v, max: %v", t.pin, pos, maxPos, "pin", 1, max)
-
-		if wg != nil {
-			wg.Add(1)
-		}
-		go func() {
-			if wg != nil {
-				defer wg.Done()
-			}
-			f(t.pin, pos)
-		}()
-
-		// otherwise iterats
-		glog.V(4).Infof("closer branches %v: %v/%v, il: %v, ir: %v, l: %v", t.pin, pos, maxPos, il, ir, l)
-		for i := l - 1; i > ir; i-- {
-			n := t.bins[i]
-
-			glog.V(4).Infof("branch %v closer branch %v pos: %v/%v (%v), count: %v, size: %v, max: %v", t.pin, n.pin, pos, maxPos, i, m, n.size, max)
-			glog.V(4).Infof("BEFORE %v %v %v", n.size, max, extra)
-			m, max, extra = need(n.size, max, extra)
-			if m <= 0 {
-				glog.V(4).Infof("branch %v closer branch %v NOT ADDED pos: %v/%v (%v), count: %v, size: %v, max: %v", t.pin, n.pin, pos, maxPos, i, m, n.size, max)
-				return
-			}
-			glog.V(4).Infof("AFTER %v %v %v", m, max, extra)
-
-			glog.V(4).Infof("branch %v closer branch %v pos: %v/%v (%v), count: %v, size: %v, max: %v", t.pin, n.pin, pos, maxPos, i, m, n.size, max)
-
-			if wg != nil {
-				wg.Add(m)
-			}
-			go func(pn *pbTree, pm int) {
-				pn.each(func(v PbVal, _ int) bool {
-					if wg != nil {
-						defer wg.Done()
-					}
-					glog.V(4).Infof("branch %v call f on %v pos: %v/%v (%v), count: %v/%v", pn.pin, v, pos, maxPos, i, pm, max)
-					f(v, pos)
-					pm--
-					return pm > 0
-				})
-			}(n, m)
-
-		}
-	}
-
-	// if max <= 0 {
-	// 	return
-	// }
-	// iterate branches that are farther tham pom with their own po
-	glog.V(4).Infof("further branches %v: %v/%v, il: %v, ir: %v, l: %v, extra: %v", t.pin, pos, maxPos, il, ir, l, extra)
-	for i := il - 1; i >= 0; i-- {
-		n := t.bins[i]
-		// the first time max is less than the size of the entire branch
-		// wait for the pivot thread to release extra elements
-		glog.V(4).Infof("branch %v further branch %v pos: %v/%v (%v), count: %v, size: %v, max: %v", t.pin, n.pin, n.pos, maxPos, i, m, n.size, max)
-		glog.V(4).Infof("BEFORE %v %v %v", n.size, max, extra)
-		m, max, extra = need(n.size, max, extra)
-		if m <= 0 {
-			return
-		}
-		glog.V(4).Infof("AFTER %v %v %v", m, max, extra)
-		glog.V(4).Infof("branch %v further branch %v pos: %v/%v (%v), count: %v, size: %v, max: %v", t.pin, n.pin, n.pos, maxPos, i, m, n.size, max)
-
-		if wg != nil {
-			wg.Add(m)
-		}
-		go func(pn *pbTree, pm int) {
-			pn.each(func(v PbVal, _ int) bool {
-				if wg != nil {
-					defer wg.Done()
-				}
-				f(v, pn.pos)
-				glog.V(4).Infof("branch %v call f on %v pos: %v/%v (%v), count: %v/%v", pn.pin, v, pn.pos, maxPos, i, pm, max)
-				pm--
-				return pm > 0
-			})
-		}(n, m)
-
-	}
-	return max + extra
-
-}
-
-// getPos(n) returns the forking node at PO n and its index if it exists
-// otherwise nil
-// caller is supposed to hold the lock
-func (t *pbTree) getPos(pos int) (n *pbTree, i int) {
-	for i, n = range t.bins {
-		if pos > n.pos {
-			continue
-		}
-		if pos < n.pos {
-			return nil, i
-		}
-		return n, i
-	}
-	return nil, len(t.bins)
-}
-
-// need(m, max, extra) uses max m out of extra, and then max
-// if needed, returns the adjusted counts
-func need(m, max, extra int) (int, int, int) {
-	if m <= extra {
-		return m, max, extra - m
-	}
-	max += extra - m
-	if max <= 0 {
-		return m + max, 0, 0
-	}
-	return m, max, 0
-}
-
-// func need(max int, more chan int) int {
-// 	// if max <= 0 {
-// 	c, ok := <-more
-// 	if ok {
-// 		defer close(more)
-// 		if c > 0 {
-// 			glog.V(4).Infof("need: %v + %v", max, c)
-// 			return max + c
-// 		}
-// 	}
-// 	// }
-// 	return max
-// }
-
-func (t *pbTree) String() string {
-	return t.sstring("")
-}
-
-func (t *pbTree) sstring(indent string) string {
-	var s string
-	indent += "  "
-	s += fmt.Sprintf("%v%v (%v) %v \n", indent, t.pin, t.pos, t.size)
-	for _, n := range t.bins {
-		s += fmt.Sprintf("%v%v\n", indent, n.sstring(indent))
-	}
-	return s
-}

swarm/network/protocol.go

@@ -37,31 +37,36 @@ const (
 )
 
 // bzz is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer)
-type bzz struct {
+type bzzPeer struct {
 	*protocols.Peer
 	hive       PeerPool
 	network    adapters.NodeAdapter
 	localAddr  *peerAddr
 	*peerAddr  // remote address
 	lastActive time.Time
+	peers      map[discover.NodeID]bool
 }
 
-func (self *bzz) LastActive() time.Time {
+func (self *bzzPeer) LastActive() time.Time {
 	return self.lastActive
 }
 
-// implemented by peerAddr and peerAddr
+func (self *bzzPeer) Peers() map[discover.NodeID]bool {
-type NodeAddr interface {
+	return self.peers
+}
+
+// implemented by peerAddr
+type PeerAddr interface {
 	OverlayAddr() []byte
 	UnderlayAddr() []byte
 }
 
-// the Node interface that peerPool needs
+// the Peer interface that peerPool needs
-type Node interface {
+type Peer interface {
-	NodeAddr
+	PeerAddr
 	String() string      // pretty printable the Node
 	ID() discover.NodeID // the key that uniquely identifies the Node for the peerPool
-
+	Peers() map[discover.NodeID]bool
 	Send(interface{}) error                               // can send messages
 	Drop()                                                // disconnect this peer
 	Register(interface{}, func(interface{}) error) uint64 // register message-handler callbacks
@@ -70,12 +75,12 @@ type Node interface {
 // PeerPool is the interface for the connectivity manager
 // directly interacts with the p2p server to suggest connections
 type PeerPool interface {
-	Add(Node) error
+	Add(Peer) error
-	Remove(Node)
+	Remove(Peer)
 }
 
 type PeerInfo interface {
-	Info() interface{}
+	NodeInfo() interface{}
 	PeerInfo(discover.NodeID) interface{}
 }
 
@@ -86,11 +91,19 @@ func BzzCodeMap(msgs ...interface{}) *protocols.CodeMap {
 	return ct
 }
 
-func Bzz(localAddr []byte, hive PeerPool, na adapters.NodeAdapter, ct *protocols.CodeMap, services func(Node) error) *p2p.Protocol {
+// Bzz is the protocol constructor
+// returns p2p.Protocol that is to be offered by the node.Service
+func Bzz(localAddr []byte, hive PeerPool, na adapters.NodeAdapter, ct *protocols.CodeMap, services func(Peer) error) *p2p.Protocol {
 	run := func(p *protocols.Peer) error {
 		addr := &peerAddr{localAddr, na.LocalAddr()}
 
-		bee := &bzz{Peer: p, hive: hive, network: na, localAddr: addr}
+		bee := &bzzPeer{
+			Peer:      p,
+			hive:      hive,
+			network:   na,
+			localAddr: addr,
+			peers:     make(map[discover.NodeID]bool),
+		}
 		// protocol handshake and its validation
 		// sets remote peer address
 		err := bee.bzzHandshake()
@@ -118,70 +131,16 @@ func Bzz(localAddr []byte, hive PeerPool, na adapters.NodeAdapter, ct *protocols
 		return bee.Run()
 	}
 
-	return protocols.NewProtocol(ProtocolName, Version, run, na, ct)
+	proto := protocols.NewProtocol(ProtocolName, Version, run, na, ct)
-}
-
-// Bzz is the protocol constructor
-// returns p2p.Protocol that is to be offered by the node.Service
-/*func Bzz(localAddr []byte, hive PeerPool, na adapters.NodeAdapter, m adapters.Messenger, ct *protocols.CodeMap, services func(Node) error) *p2p.Protocol {
-	// handle handshake
-
-	run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-		glog.V(6).Infof("protocol starting on %v connected to %v", localAddr, p.ID())
-
-		id := p.ID()
-		peer := protocols.NewPeer(p, rw, ct, m, func() { na.Disconnect(id[:]) })
-		addr := &peerAddr{localAddr, na.LocalAddr()}
-
-		bee := &bzz{Peer: peer, hive: hive, network: na, localAddr: addr}
-		// protocol handshake and its validation
-		// sets remote peer address
-		err := bee.bzzHandshake()
-		if err != nil {
-			glog.V(6).Infof("handshake error in peer %v: %v", bee.ID(), err)
-			return err
-		}
-
-		// mount external service models on the peer connection (swap, sync)
-		if services != nil {
-			err = services(bee)
-			if err != nil {
-				glog.V(6).Infof("protocol service error for peer %v: %v", bee.ID(), err)
-				return err
-			}
-		}
-
-		err = hive.Add(bee)
-		if err != nil {
-			glog.V(6).Infof("failed to add peer '%v' to hive: %v", bee.ID(), err)
-			return err
-		}
-
-		defer hive.Remove(bee)
-		return bee.Run()
-	}
-
-	var info func() interface{}
-
-	var peerInfo func(discover.NodeID) interface{}
 
 	if o, ok := hive.(PeerInfo); ok {
-		info = o.Info
+		proto.NodeInfo = o.NodeInfo
-		peerInfo = o.PeerInfo
+		proto.PeerInfo = o.PeerInfo
 	}
 
-	return &p2p.Protocol{
+	return proto
-		Name:     ProtocolName,
-		Version:  Version,
-		Length:   ct.Length(),
-		Run:      run,
-		NodeInfo: info,
-		PeerInfo: peerInfo,
-	}
 }
 
-*/
-
 /*
  Handshake
 
@@ -199,6 +158,7 @@ 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 {
 	OAddr []byte
 	UAddr []byte
@@ -219,7 +179,7 @@ func (self *peerAddr) String() string {
 // bzzHandshake negotiates the bzz master handshake
 // and validates the response, returns error when
 // mismatch/incompatibility is evident
-func (self *bzz) bzzHandshake() error {
+func (self *bzzPeer) bzzHandshake() error {
 
 	lhs := &bzzHandshake{
 		Version:   uint64(Version),
@@ -236,7 +196,7 @@ func (self *bzz) bzzHandshake() error {
 	rhs := hs.(*bzzHandshake)
 	err = checkBzzHandshake(rhs)
 	if err != nil {
-		glog.V(6).Infof("handshake between %v and %v  failed: %v", self.localAddr, self.peerAddr)
+		glog.V(6).Infof("handshake between %v and %v  failed: %v", self.localAddr, self.peerAddr, err)
 		return err
 	}
 
@@ -254,6 +214,7 @@ func (self *bzz) bzzHandshake() error {
 
 }
 
+// checkBzzHandshake checks for the validity and compatibility of the remote handshake
 func checkBzzHandshake(rhs *bzzHandshake) error {
 
 	if NetworkId != rhs.NetworkId {
@@ -267,6 +228,7 @@ func checkBzzHandshake(rhs *bzzHandshake) error {
 	return nil
 }
 
+// RandomAddr is a utility method generating an address from a public key
 func RandomAddr() *peerAddr {
 	key, err := crypto.GenerateKey()
 	if err != nil {
@@ -281,11 +243,14 @@ func RandomAddr() *peerAddr {
 	}
 }
 
-func NodeId(addr NodeAddr) *adapters.NodeId {
+// NodeId transforms the underlay address to an adapters.NodeId
+func NodeId(addr PeerAddr) *adapters.NodeId {
 	return adapters.NewNodeId(addr.UnderlayAddr())
 }
 
-func NodeIdToAddr(n *adapters.NodeId) *peerAddr {
+// NewPeerAddrFromNodeId constucts a peerAddr from an adapters.NodeId
+// the overlay address is derived as the hash of the nodeId
+func NewPeerAddrFromNodeId(n *adapters.NodeId) *peerAddr {
 	id := n.NodeID
 	return &peerAddr{
 		OAddr: crypto.Keccak256(id[:]),

swarm/network/protocol_test.go

@@ -37,13 +37,13 @@ func bzzHandshakeExchange(lhs, rhs *bzzHandshake, id *adapters.NodeId) []p2ptest
 	}
 }
 
-func newTestBzzProtocol(addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Node) error) func(adapters.NodeAdapter) adapters.ProtoCall {
+func newTestBzzProtocol(addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Peer) error) func(adapters.NodeAdapter) adapters.ProtoCall {
 	if ct == nil {
 		ct = BzzCodeMap()
 	}
 	ct.Register(p2ptest.FlushMsg)
 	return func(na adapters.NodeAdapter) adapters.ProtoCall {
-		srv := func(p Node) error {
+		srv := func(p Peer) error {
 			if services != nil {
 				err := services(p)
 				if err != nil {
@@ -59,7 +59,7 @@ func newTestBzzProtocol(addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, serv
 			return nil
 		}
 
-		protocol := Bzz(addr.OverlayAddr(), pp, na, na.Messenger(), ct, srv)
+		protocol := Bzz(addr.OverlayAddr(), pp, na, ct, srv)
 		return protocol.Run
 	}
 }
@@ -97,7 +97,7 @@ func (s *bzzTester) runHandshakes(ids ...*adapters.NodeId) {
 	for _, id := range ids {
 		glog.V(6).Infof("\n\n\nrun handshake with %v", id)
 		time.Sleep(1)
-		s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NodeIdToAddr(id)))
+		s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewPeerAddrFromNodeId(id)))
 		time.Sleep(1)
 	}
 	glog.V(6).Infof("flush %v", ids)
@@ -108,7 +108,7 @@ func correctBzzHandshake(addr *peerAddr) *bzzHandshake {
 	return &bzzHandshake{0, 322, addr}
 }
 
-func newBzzTester(t *testing.T, addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Node) error) *bzzTester {
+func newBzzTester(t *testing.T, addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester {
 	s := p2ptest.NewProtocolTester(t, NodeId(addr), 1, newTestBzzProtocol(addr, pp, ct, services))
 	return &bzzTester{
 		addr:            addr,
@@ -124,7 +124,7 @@ func TestBzzHandshakeNetworkIdMismatch(t *testing.T) {
 	id := s.Ids[0]
 	s.testHandshake(
 		correctBzzHandshake(addr),
-		&bzzHandshake{0, 321, NodeIdToAddr(id)},
+		&bzzHandshake{0, 321, NewPeerAddrFromNodeId(id)},
 		&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")},
 	)
 }
@@ -136,7 +136,7 @@ func TestBzzHandshakeVersionMismatch(t *testing.T) {
 	id := s.Ids[0]
 	s.testHandshake(
 		correctBzzHandshake(addr),
-		&bzzHandshake{1, 322, NodeIdToAddr(id)},
+		&bzzHandshake{1, 322, NewPeerAddrFromNodeId(id)},
 		&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("version mismatch 1 (!= 0)")},
 	)
 }
@@ -148,7 +148,7 @@ func TestBzzHandshakeSuccess(t *testing.T) {
 	id := s.Ids[0]
 	s.testHandshake(
 		correctBzzHandshake(addr),
-		&bzzHandshake{0, 322, NodeIdToAddr(id)},
+		&bzzHandshake{0, 322, NewPeerAddrFromNodeId(id)},
 	)
 }
 
@@ -203,7 +203,7 @@ func TestBzzPeerPoolNotAdd(t *testing.T) {
 	s := newBzzTester(t, addr, pp, nil, nil)
 
 	id := s.Ids[0]
-	s.testHandshake(correctBzzHandshake(addr), &bzzHandshake{0, 321, NodeIdToAddr(id)}, &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")})
+	s.testHandshake(correctBzzHandshake(addr), &bzzHandshake{0, 321, NewPeerAddrFromNodeId(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)
 	}
@@ -212,21 +212,21 @@ func TestBzzPeerPoolNotAdd(t *testing.T) {
 // TestPeerPool is an example peerPool to demonstrate registration of peer connections
 type TestPeerPool struct {
 	lock  sync.Mutex
-	peers map[discover.NodeID]Node
+	peers map[discover.NodeID]Peer
 }
 
 func NewTestPeerPool() *TestPeerPool {
-	return &TestPeerPool{peers: make(map[discover.NodeID]Node)}
+	return &TestPeerPool{peers: make(map[discover.NodeID]Peer)}
 }
 
-func (self *TestPeerPool) Add(p Node) error {
+func (self *TestPeerPool) Add(p Peer) error {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	self.peers[p.ID()] = p
 	return nil
 }
 
-func (self *TestPeerPool) Remove(p Node) {
+func (self *TestPeerPool) Remove(p Peer) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	// glog.V(6).Infof("removing peer %v", p.ID())
@@ -240,7 +240,7 @@ func (self *TestPeerPool) Has(n *adapters.NodeId) bool {
 	return ok
 }
 
-func (self *TestPeerPool) Get(n *adapters.NodeId) Node {
+func (self *TestPeerPool) Get(n *adapters.NodeId) Peer {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	return self.peers[n.NodeID]

swarm/network/simulations/overlay.go

@@ -12,7 +12,9 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/logger"
 	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/adapters"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
@@ -23,7 +25,6 @@ import (
 type Network struct {
 	*simulations.Network
 	hives []*network.Hive
-	messenger *adapters.SimPipe
 }
 
 // SimNode is the adapter used by Swarm simulations.
@@ -34,7 +35,7 @@ type SimNode struct {
 
 // the hive update ticker for hive
 func af() <-chan time.Time {
-	return time.NewTicker(5 * time.Second).C
+	return time.NewTicker(1 * time.Second).C
 }
 
 // Start() starts up the hive
@@ -54,28 +55,32 @@ func (self *SimNode) Stop() error {
 	return nil
 }
 
+func (self *SimNode) RunProtocol(id *adapters.NodeId, rw, rrw p2p.MsgReadWriter, runc chan bool) error {
+	return self.NodeAdapter.(adapters.ProtocolRunner).RunProtocol(id, rw, rrw, runc)
+}
+
 // NewSimNode creates adapters for nodes in the simulation.
 func (self *Network) NewSimNode(conf *simulations.NodeConfig) adapters.NodeAdapter {
 	id := conf.Id
-	na := adapters.NewSimNode(id, self.Network, self.messenger)
+	na := adapters.NewSimNode(id, self.Network, adapters.NewSimPipe)
-	addr := network.NodeIdToAddr(id)
+	addr := network.NewPeerAddrFromNodeId(id)
+	// to := network.NewKademlia(addr.OverlayAddr(), nil)   // overlay topology driver
 	to := network.NewTestOverlay(addr.OverlayAddr())   // overlay topology driver
 	pp := network.NewHive(network.NewHiveParams(), to) // hive
 	self.hives = append(self.hives, pp)                // remember hive
 	// bzz protocol Run function. messaging through SimPipe
 	ct := network.BzzCodeMap(network.HiveMsgs...) // bzz protocol code map
-	na.Run = network.Bzz(addr.OverlayAddr(), pp, na, &adapters.SimPipe{}, ct, nil).Run
+	na.Run = network.Bzz(addr.OverlayAddr(), pp, na, ct, nil).Run
 	return &SimNode{
 		hive:        pp,
 		NodeAdapter: na,
 	}
 }
 
-func NewNetwork(network *simulations.Network, messenger *adapters.SimPipe) *Network {
+func NewNetwork(network *simulations.Network) *Network {
 	n := &Network{
 		// hives:
 		Network: network,
-		messenger: messenger,
 	}
 	n.SetNaf(n.NewSimNode)
 	return n
@@ -91,22 +96,21 @@ func NewSessionController() (*simulations.ResourceController, chan bool) {
 			Create: &simulations.ResourceHandler{
 				Handle: func(msg interface{}, parent *simulations.ResourceController) (interface{}, error) {
 					conf := msg.(*simulations.NetworkConfig)
-					messenger := &adapters.SimPipe{}
 					net := simulations.NewNetwork(nil, &event.TypeMux{})
-					ppnet := NewNetwork(net, messenger)
+					ppnet := NewNetwork(net)
 					c := simulations.NewNetworkController(conf, net.Events(), simulations.NewJournal())
 					if len(conf.Id) == 0 {
 						conf.Id = fmt.Sprintf("%d", 0)
 					}
-					glog.V(6).Infof("new network controller on %v", conf.Id)
+					glog.V(logger.Debug).Infof("new network controller on %v", conf.Id)
 					if parent != nil {
 						parent.SetResource(conf.Id, c)
 					}
-					ids := p2ptest.RandomNodeIds(10)
+					ids := p2ptest.RandomNodeIds(5)
 					for _, id := range ids {
 						ppnet.NewNode(&simulations.NodeConfig{Id: id})
 						ppnet.Start(id)
-						glog.V(6).Infof("node %v starting up", id)
+						glog.V(logger.Debug).Infof("node %v starting up", id)
 					}
 					// the nodes only know about their 2 neighbours (cyclically)
 					for i, _ := range ids {
@@ -116,7 +120,7 @@ func NewSessionController() (*simulations.ResourceController, chan bool) {
 						} else {
 							peerId = ids[i-1]
 						}
-						err := ppnet.hives[i].Register(network.NodeIdToAddr(peerId))
+						err := ppnet.hives[i].Register(network.NewPeerAddrFromNodeId(peerId))
 						if err != nil {
 							panic(err.Error())
 						}
@@ -124,12 +128,11 @@ func NewSessionController() (*simulations.ResourceController, chan bool) {
 					return struct{}{}, nil
 				},
 				Type: reflect.TypeOf(&simulations.NetworkConfig{}),
-				// Type: reflect.TypeOf(&simulations.NetworkConfig{}),
 			},
 			// DELETE /
 			Destroy: &simulations.ResourceHandler{
 				Handle: func(msg interface{}, parent *simulations.ResourceController) (interface{}, error) {
-					glog.V(6).Infof("destroy handler called")
+					glog.V(logger.Debug).Infof("destroy handler called")
 					// this can quit the entire app (shut down the backend server)
 					quitc <- true
 					return struct{}{}, nil
@@ -142,7 +145,7 @@ func NewSessionController() (*simulations.ResourceController, chan bool) {
 // var server
 func main() {
 	runtime.GOMAXPROCS(runtime.NumCPU())
-	glog.SetV(6)
+	glog.SetV(logger.Info)
 	glog.SetToStderr(true)
 
 	c, quitc := NewSessionController()

swarm/network/test_overlay.go

@@ -15,28 +15,33 @@ const orders = 8
 type testOverlay struct {
 	mu     sync.Mutex
 	addr   []byte
-	pos    [][]*testNodeAddr
+	pos    [][]*testPeerAddr
-	posMap map[string]*testNodeAddr
+	posMap map[string]*testPeerAddr
 }
 
-type testNodeAddr struct {
+type testPeerAddr struct {
-	NodeAddr
+	PeerAddr
-	Node Node
+	Peer Peer
 }
 
-func (self *testOverlay) Register(na NodeAddr) error {
+func (self *testOverlay) Register(nas ...PeerAddr) error {
 	self.mu.Lock()
 	defer self.mu.Unlock()
-	return self.register(na)
+	return self.register(nas...)
 }
 
-func (self *testOverlay) register(na NodeAddr) error {
+func (self *testOverlay) register(nas ...PeerAddr) error {
-	tna := &testNodeAddr{NodeAddr: na}
+	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)
 		glog.V(6).Infof("PO: %v, orders: %v", o, orders)
 		self.pos[o] = append(self.pos[o], tna)
+	}
 	return nil
 }
 
@@ -44,7 +49,7 @@ func order(addr []byte) int {
 	return int(addr[0]) / 32
 }
 
-func (self *testOverlay) On(n Node) (Node, error) {
+func (self *testOverlay) On(n Peer) {
 	self.mu.Lock()
 	defer self.mu.Unlock()
 	addr := n.OverlayAddr()
@@ -52,20 +57,15 @@ func (self *testOverlay) On(n Node) (Node, error) {
 	if na == nil {
 		self.register(n)
 		na = self.posMap[string(addr)]
-	} else if na.Node != nil {
+	} else if na.Peer != nil {
-		return nil, nil
+		return
 	}
 	glog.V(6).Infof("Online: %v", fmt.Sprintf("%x", addr[:4]))
-	na.Node = n
+	na.Peer = n
-	o := order(addr)
+	return
-	ons := self.on(self.pos[o])
-	if len(ons) > 2 {
-		return ons[0], nil
-	}
-	return nil, nil
 }
 
-func (self *testOverlay) Off(n Node) {
+func (self *testOverlay) Off(n Peer) {
 	self.mu.Lock()
 	defer self.mu.Unlock()
 	addr := n.OverlayAddr()
@@ -73,37 +73,38 @@ func (self *testOverlay) Off(n Node) {
 	if na == nil {
 		return
 	}
-	na.Node = nil
+	delete(self.posMap, string(addr))
+	na.Peer = nil
 }
 
 // caller must hold the lock
-func (self *testOverlay) on(po []*testNodeAddr) (nodes []Node) {
+func (self *testOverlay) on(po []*testPeerAddr) (nodes []Peer) {
 	for _, na := range po {
-		if na.Node != nil {
+		if na.Peer != nil {
-			nodes = append(nodes, na.Node)
+			nodes = append(nodes, na.Peer)
 		}
 	}
 	return nodes
 }
 
 // caller must hold the lock
-func (self *testOverlay) off(po []*testNodeAddr) (nas []NodeAddr) {
+func (self *testOverlay) off(po []*testPeerAddr) (nas []PeerAddr) {
 	for _, na := range po {
-		if na.Node == nil {
+		if na.Peer == nil {
-			nas = append(nas, NodeAddr(na))
+			nas = append(nas, PeerAddr(na))
 		}
 	}
 	return nas
 }
 
-func (self *testOverlay) EachNode(base []byte, o int, f func(Node) bool) {
+func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer) bool) {
 	if base == nil {
 		base = self.addr
 	}
 	for i := o; i < len(self.pos); i++ {
 		for _, na := range self.pos[i] {
-			if na.Node != nil {
+			if na.Peer != nil {
-				if !f(na.Node) {
+				if !f(na.Peer) {
 					return
 				}
 			}
@@ -111,7 +112,7 @@ func (self *testOverlay) EachNode(base []byte, o int, f func(Node) bool) {
 	}
 }
 
-func (self *testOverlay) EachNodeAddr(base []byte, o int, f func(NodeAddr) bool) {
+func (self *testOverlay) EachPeer(base []byte, o int, f func(PeerAddr) bool) {
 	if base == nil {
 		base = self.addr
 	}
@@ -124,53 +125,39 @@ func (self *testOverlay) EachNodeAddr(base []byte, o int, f func(NodeAddr) bool)
 	}
 }
 
-func (self *testOverlay) SuggestNodeAddr() NodeAddr {
+func (self *testOverlay) SuggestPeer() (PeerAddr, int, bool) {
 	self.mu.Lock()
 	defer self.mu.Unlock()
-	for _, po := range self.pos {
+	for i, po := range self.pos {
 		ons := self.on(po)
 		if len(ons) < 2 {
 			offs := self.off(po)
 			if len(offs) > 0 {
 				glog.V(6).Infof("node %v is off", offs[0])
-				return offs[0]
+				return offs[0], i, true
 			}
 		}
 	}
-	return nil
+	return nil, 0, true
 }
 
-func (self *testOverlay) SuggestOrder() int {
+func (self *testOverlay) String() string {
-	self.mu.Lock()
-	defer self.mu.Unlock()
-	for o, po := range self.pos {
-		off := self.off(po)
-		if len(off) < 5 {
-			glog.V(6).Infof("suggest PO%02d / %v", o, len(self.pos)-1)
-			return o
-		}
-	}
-	return 256
-
-}
-
-func (self *testOverlay) Info() string {
 	self.mu.Lock()
 	defer self.mu.Unlock()
 	var t []string
 	var ons, offs int
-	var ns []Node
+	var ns []Peer
-	var nas []NodeAddr
+	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, "|")
-		nas = self.off(po)
 		offs = len(nas)
 		for _, na := range nas {
 			addr := na.OverlayAddr()
@@ -184,7 +171,7 @@ func (self *testOverlay) Info() string {
 func NewTestOverlay(addr []byte) *testOverlay {
 	return &testOverlay{
 		addr:   addr,
-		posMap: make(map[string]*testNodeAddr),
+		posMap: make(map[string]*testPeerAddr),
-		pos:    make([][]*testNodeAddr, orders),
+		pos:    make([][]*testPeerAddr, orders),
 	}
 }