From 8cd9adcf38684beed6c5d141a92c395e16493e20 Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 2 Mar 2017 18:01:43 +0700 Subject: [PATCH] extending kademlia with bootstrapping, refactor discovery from hive --- p2p/simulations/examples/connectivity.go | 4 +- p2p/simulations/journal.go | 215 ---------------- p2p/simulations/mocker.go | 262 +++++++++++++++++++ p2p/simulations/network.go | 197 +++++++++++--- p2p/simulations/session_controller.go | 76 +----- p2p/simulations/session_controller_test.go | 15 +- p2p/testing/sessions.go | 2 +- pot/address.go | 6 +- pot/pot.go | 116 +++++++-- swarm/network/discovery.go | 157 ++++++++++++ swarm/network/discovery_test.go | 31 +++ swarm/network/hive.go | 127 ++-------- swarm/network/hive_test.go | 3 +- swarm/network/kademlia.go | 249 +++++++++++------- swarm/network/kademlia_test.go | 282 +++++++++++++++++++++ swarm/network/protocol.go | 2 +- swarm/network/simulations/overlay.go | 108 ++++---- swarm/network/test_overlay.go | 8 +- 18 files changed, 1251 insertions(+), 609 deletions(-) create mode 100644 p2p/simulations/mocker.go create mode 100644 swarm/network/discovery.go create mode 100644 swarm/network/discovery_test.go create mode 100644 swarm/network/kademlia_test.go diff --git a/p2p/simulations/examples/connectivity.go b/p2p/simulations/examples/connectivity.go index 9307637a02..5d05b5d478 100644 --- a/p2p/simulations/examples/connectivity.go +++ b/p2p/simulations/examples/connectivity.go @@ -7,14 +7,12 @@ import ( "github.com/ethereum/go-ethereum/p2p/simulations" ) -// var server func main() { runtime.GOMAXPROCS(runtime.NumCPU()) glog.SetV(6) glog.SetToStderr(true) - c, quitc := simulations.NewSessionController() - + c, quitc := simulations.NewSessionController(simulations.DefaultNet) simulations.StartRestApiServer("8888", c) // wait until server shuts down <-quitc diff --git a/p2p/simulations/journal.go b/p2p/simulations/journal.go index da48e8fc6f..9b0b2339a9 100644 --- a/p2p/simulations/journal.go +++ b/p2p/simulations/journal.go @@ -4,7 +4,6 @@ import ( "bytes" "encoding/json" "fmt" - "math/rand" "reflect" "sync" "time" @@ -262,79 +261,6 @@ func NewJournalPlayerController(conf *JournalPlayConfig) Controller { return self } -type MockerConfig struct { - // TODO: frequency/volume etc. - Id string - NodeCount int - UpdateInterval time.Duration -} - -func NewMockersController(eventer *event.TypeMux) Controller { - self := NewResourceContoller( - &ResourceHandlers{ - // Create: n.StartNode, NodeConfig - Create: &ResourceHandler{ - Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { - conf := msg.(*MockerConfig) - if conf.NodeCount == 0 { - conf.NodeCount = 100 - } - ids := RandomNodeIds(conf.NodeCount) - if conf.UpdateInterval == 0 { - conf.UpdateInterval = 1 * time.Second - } - ticker := time.NewTicker(conf.UpdateInterval) - go MockEvents(eventer, ids, ticker.C) - c := NewMockerController(conf, ticker) - if len(conf.Id) == 0 { - conf.Id = fmt.Sprintf("%d", parent.id) - } - glog.V(6).Infof("new mocker controller on %v", conf.Id) - if parent != nil { - parent.SetResource(conf.Id, c) - } - parent.id++ - return empty, nil - }, - Type: reflect.TypeOf(&MockerConfig{}), - }, - }) - return self -} - -func NewMockerController(conf *MockerConfig, ticker *time.Ticker) Controller { - self := NewResourceContoller( - &ResourceHandlers{ - // GET /0/mockevents/ - Retrieve: &ResourceHandler{ - Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { - return nil, fmt.Errorf("info about mocker not implemented") - }, - }, - // DELETE /0/mockevents/ - Destroy: &ResourceHandler{ - Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { - ticker.Stop() //terminate MockEvents routine - parent.DeleteResource(conf.Id) - return empty, nil - }, - }, - }) - return self -} - -// deltas: changes in the number of cumulative actions: non-negative integers. -// base unit is the fixed minimal interval between two measurements (time quantum) -// acceleration : to slow down you just set the base unit higher. -// to speed up: skip x number of base units -// frequency: given as the (constant or average) number of base units between measurements -// if resolution is expressed as the inverse of frequency = preserved information -// setting the acceleration -// beginning of the record (lifespan) of the network is index 0 -// acceleration means that snapshots are rarer so the same span can be generated by the journal -// then update logs can be compressed (toonly one state transition per affected node) -// epoch, epochcount - func ConnLabel(source, target *adapters.NodeId) string { var first, second *adapters.NodeId if bytes.Compare(source.Bytes(), target.Bytes()) > 0 { @@ -346,144 +272,3 @@ func ConnLabel(source, target *adapters.NodeId) string { } return fmt.Sprintf("%v-%v", first, second) } - -// MockEvents generates random connectivity events and posts them -// to the eventer -// The journal using the eventer can then be read to visualise or -// drive connections -func MockEvents(eventer *event.TypeMux, ids []*adapters.NodeId, ticker <-chan time.Time) { - - var onNodes []*Node - offNodes := ids - onConnsMap := make(map[string]int) - var onConns []*Conn - connsMap := make(map[string]int) - var conns []*Conn - // ids := RandomNodeIds(100) - switchonRate := 5 - dropoutRate := 100 - newConnCount := 1 // new connection per node per tick - connFailRate := 100 - disconnRate := 100 // fraction of all connections - nodesTarget := len(ids) / 2 - degreeTarget := 8 - convergenceRate := 5 - rounds := 0 - for _ = range ticker { - glog.V(6).Infof("rates: %v/%v, %v (%v/%v)", switchonRate, dropoutRate, newConnCount, connFailRate, disconnRate) - // here switchon rate will depend - nodesUp := len(offNodes) / switchonRate - missing := nodesTarget - len(onNodes) - if missing > 0 { - if nodesUp < missing { - nodesUp += (missing-nodesUp)/convergenceRate + 1 - } - } - - nodesDown := len(onNodes) / dropoutRate - - connsUp := len(onNodes) * newConnCount - connsUp = connsUp - connsUp/connFailRate - missing = nodesTarget*degreeTarget/2 - len(onConns) - if missing < connsUp { - connsUp = missing - if connsUp < 0 { - connsUp = 0 - } - } - connsDown := len(onConns) / disconnRate - glog.V(6).Infof("Nodes Up: %v, Down: %v [ON: %v/%v]\nConns Up: %v, Down: %v [ON: %v/%v(%v)]", nodesUp, nodesDown, len(onNodes), len(onNodes)+len(offNodes), connsUp, connsDown, len(onConns), len(conns)-len(onConns), len(conns)) - - for i := 0; len(onNodes) > 0 && i < nodesDown; i++ { - c := rand.Intn(len(onNodes)) - sn := onNodes[c] - err := eventer.Post(&NodeEvent{ - Type: "node", - Action: "down", - node: sn, - }) - if err != nil { - panic(err.Error()) - } - onNodes = append(onNodes[0:c], onNodes[c+1:]...) - offNodes = append(offNodes, sn.Id) - } - for i := 0; len(offNodes) > 0 && i < nodesUp; i++ { - c := rand.Intn(len(offNodes)) - sn := &Node{Id: offNodes[c]} - err := eventer.Post(&NodeEvent{ - Type: "node", - Action: "up", - node: sn, - }) - if err != nil { - panic(err.Error()) - } - onNodes = append(onNodes, sn) - offNodes = append(offNodes[0:c], offNodes[c+1:]...) - } - var found bool - var sc *Conn - for i := 0; len(onNodes) > 1 && i < connsUp; i++ { - sc = nil - n := rand.Intn(len(onNodes) - 1) - m := n + 1 + rand.Intn(len(onNodes)-n-1) - for i := m; i < len(onNodes); i++ { - lab := ConnLabel(onNodes[n].Id, onNodes[i].Id) - var j int - j, found = onConnsMap[lab] - if found { - continue - } - j, found = connsMap[lab] - if found { - sc = conns[j] - break - } - caller := onNodes[n].Id - callee := onNodes[i].Id - - sc := &Conn{ - One: caller, - Other: callee, - } - connsMap[lab] = len(conns) - conns = append(conns, sc) - break - } - - if sc == nil { - i-- - continue - } - lab := ConnLabel(sc.One, sc.Other) - onConnsMap[lab] = len(onConns) - onConns = append(onConns, sc) - err := eventer.Post(&ConnEvent{ - Type: "conn", - Action: "up", - conn: sc, - }) - if err != nil { - panic(err.Error()) - } - } - - for i := 0; len(onConns) > 0 && i < connsDown; i++ { - c := rand.Intn(len(onConns)) - conn := onConns[c] - onConns = append(onConns[0:c], onConns[c+1:]...) - lab := ConnLabel(conn.One, conn.Other) - delete(onConnsMap, lab) - err := eventer.Post(&ConnEvent{ - Type: "conn", - Action: "down", - conn: conn, - }) - if err != nil { - panic(err.Error()) - } - } - rounds++ - } -} diff --git a/p2p/simulations/mocker.go b/p2p/simulations/mocker.go new file mode 100644 index 0000000000..d288b569d2 --- /dev/null +++ b/p2p/simulations/mocker.go @@ -0,0 +1,262 @@ +package simulations + +import ( + "fmt" + "math/rand" + "reflect" + "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" +) + +func NewMockersController(eventer *event.TypeMux, defaultMockerConfig *MockerConfig) Controller { + self := NewResourceContoller( + &ResourceHandlers{ + // Create: n.StartNode, NodeConfig + Create: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + // conf := msg.(*MockerConfig) + conf := defaultMockerConfig + ids := RandomNodeIds(conf.NodeCount) + go MockEvents(eventer, ids, conf) + c := NewMockerController(conf) + if len(conf.Id) == 0 { + conf.Id = fmt.Sprintf("%d", parent.id) + } + glog.V(logger.Detail).Infof("new mocker controller on %v", conf.Id) + if parent != nil { + parent.SetResource(conf.Id, c) + parent.id++ + } + return empty, nil + }, + Type: reflect.TypeOf(&MockerConfig{}), + }, + }) + return self +} + +func NewMockerController(conf *MockerConfig) Controller { + self := NewResourceContoller( + &ResourceHandlers{ + // GET /0/mockevents/ + Retrieve: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + return nil, fmt.Errorf("info about mocker not implemented") + }, + }, + // DELETE /0/mockevents/ + Destroy: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + conf.ticker.Stop() + parent.DeleteResource(conf.Id) + return empty, nil + }, + }, + }) + return self +} + +type MockerConfig struct { + Id string + NodeCount int + UpdateInterval int + SwitchonRate int // fraction of off nodes switching on + DropoutRate int // fraction of on nodes dropping out + NewConnCount int // new connection per node per tick + ConnFailRate int // fraction of connections failing + DisconnRate int // fraction of all connections + NodesTarget int // total number of nodes to converge on + DegreeTarget int // number of connections per peer to converge on + ConvergenceRate int // speed of convergence + ticker *time.Ticker +} + +func DefaultMockerConfig() *MockerConfig { + return &MockerConfig{ + Id: "0", + NodeCount: 100, + UpdateInterval: 1000, + SwitchonRate: 5, + DropoutRate: 100, + NewConnCount: 1, // new connection per node per tick + ConnFailRate: 100, + DisconnRate: 100, // fraction of all connections + NodesTarget: 50, + DegreeTarget: 8, + ConvergenceRate: 5, + } +} + +// base unit is the fixed minimal interval between two measurements (time quantum) +// acceleration : to slow down you just set the base unit higher. +// to speed up: skip x number of base units +// frequency: given as the (constant or average) number of base units between measurements +// if resolution is expressed as the inverse of frequency = preserved information +// setting the acceleration +// beginning of the record (lifespan) of the network is index 0 +// acceleration means that snapshots are rarer so the same span can be generated by the journal +// then update logs can be compressed (to only one state transition per affected node) +// epoch, epochcount + +// MockEvents generates random connectivity events and posts them +// to the eventer +// The journal using the eventer can then be read to visualise or +// drive connections +func MockEvents(eventer *event.TypeMux, ids []*adapters.NodeId, conf *MockerConfig) { + + var onNodes []*Node + offNodes := ids + onConnsMap := make(map[string]int) + var onConns []*Conn + connsMap := make(map[string]int) + var conns []*Conn + + conf.ticker = time.NewTicker(time.Duration(conf.UpdateInterval) * time.Millisecond) + switchonRate := conf.SwitchonRate + dropoutRate := conf.DropoutRate + newConnCount := conf.NewConnCount + connFailRate := conf.ConnFailRate + disconnRate := conf.DisconnRate + nodesTarget := conf.NodesTarget + degreeTarget := conf.DegreeTarget + convergenceRate := conf.ConvergenceRate + + rounds := 0 + for _ = range conf.ticker.C { + glog.V(logger.Detail).Infof("rates: %v/%v, %v (%v/%v)", switchonRate, dropoutRate, newConnCount, connFailRate, disconnRate) + // here switchon rate will depend + nodesUp := len(offNodes) / switchonRate + missing := nodesTarget - len(onNodes) + if missing > 0 { + if nodesUp < missing { + nodesUp += (missing-nodesUp)/convergenceRate + 1 + } + } + + nodesDown := len(onNodes) / dropoutRate + + connsUp := len(onNodes) * newConnCount + connsUp = connsUp - connsUp/connFailRate + missing = nodesTarget*degreeTarget/2 - len(onConns) + if missing < connsUp { + connsUp = missing + if connsUp < 0 { + connsUp = 0 + } + } + connsDown := len(onConns) / disconnRate + glog.V(logger.Detail).Infof("Nodes Up: %v, Down: %v [ON: %v/%v]\nConns Up: %v, Down: %v [ON: %v/%v(%v)]", nodesUp, nodesDown, len(onNodes), len(onNodes)+len(offNodes), connsUp, connsDown, len(onConns), len(conns)-len(onConns), len(conns)) + + for i := 0; len(onNodes) > 0 && i < nodesDown; i++ { + c := rand.Intn(len(onNodes)) + sn := onNodes[c] + err := eventer.Post(&NodeEvent{ + Type: "node", + Action: "down", + node: sn, + }) + if err != nil { + panic(err.Error()) + } + onNodes = append(onNodes[0:c], onNodes[c+1:]...) + offNodes = append(offNodes, sn.Id) + } + var mustconnect []int + for i := 0; len(offNodes) > 0 && i < nodesUp; i++ { + c := rand.Intn(len(offNodes)) + sn := &Node{Id: offNodes[c]} + err := eventer.Post(&NodeEvent{ + Type: "node", + Action: "up", + node: sn, + }) + if err != nil { + panic(err.Error()) + } + mustconnect = append(mustconnect, len(onNodes)) + onNodes = append(onNodes, sn) + offNodes = append(offNodes[0:c], offNodes[c+1:]...) + } + var found bool + var sc *Conn + if connsUp < len(mustconnect) { + connsUp = len(mustconnect) + } + connected := make(map[int]bool) + for i := 0; len(onNodes) > 1 && i < connsUp; i++ { + sc = nil + var n int + if i < len(mustconnect) { + n = mustconnect[i] + } else { + n = rand.Intn(len(onNodes) - 1) + if connected[n] { + continue + } + } + m := n + rand.Intn(len(onNodes)-n) + // m := n + 1 + rand.Intn(len(onNodes)-n-1) + for k := m; k < len(onNodes); k++ { + lab := ConnLabel(onNodes[n].Id, onNodes[k].Id) + var j int + j, found = onConnsMap[lab] + if found { + continue + } + j, found = connsMap[lab] + if found { + sc = conns[j] + break + } + connected[k] = true + caller := onNodes[n].Id + callee := onNodes[k].Id + + sc := &Conn{ + One: caller, + Other: callee, + } + connsMap[lab] = len(conns) + conns = append(conns, sc) + break + } + + if sc == nil { + i-- + continue + } + lab := ConnLabel(sc.One, sc.Other) + onConnsMap[lab] = len(onConns) + onConns = append(onConns, sc) + err := eventer.Post(&ConnEvent{ + Type: "conn", + Action: "up", + conn: sc, + }) + if err != nil { + panic(err.Error()) + } + } + + for i := 0; len(onConns) > 0 && i < connsDown; i++ { + c := rand.Intn(len(onConns)) + conn := onConns[c] + onConns = append(onConns[0:c], onConns[c+1:]...) + lab := ConnLabel(conn.One, conn.Other) + delete(onConnsMap, lab) + err := eventer.Post(&ConnEvent{ + Type: "conn", + Action: "down", + conn: conn, + }) + if err != nil { + panic(err.Error()) + } + } + rounds++ + } +} diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 60260f4315..26e25d68fc 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -39,16 +39,20 @@ import ( "github.com/ethereum/go-ethereum/p2p/discover" ) -type NetworkQuery struct { - Type string -} - type NetworkConfig struct { // Type NetworkType // Config json.RawMessage // type-specific configs // type // Events []string - Id string + Id string + DefaultMockerConfig *MockerConfig + Backend bool +} + +type NetworkControl interface { + Events() *event.TypeMux + Config() *NetworkConfig + Subscribe(*event.TypeMux, ...interface{}) } // event types related to connectivity, i.e., nodes coming on dropping off @@ -60,8 +64,19 @@ var ConnectivityEvents = []interface{}{&NodeEvent{}, &ConnEvent{}, &MsgEvent{}} // // Events from the eventer go into the provided journal. The content of the journal can be // accessed through the HTTP API. -func NewNetworkController(conf *NetworkConfig, eventer *event.TypeMux, journal *Journal) Controller { - +// the events are generated by mockers and replayed journals +func NewNetworkController(net NetworkControl, nodesController *ResourceController) Controller { + journal := NewJournal() + eventer := &event.TypeMux{} + conf := net.Config() + if conf.Backend { + journal.Subscribe(net.Events(), ConnectivityEvents...) + // the network can subscribe to the eventer fed by mockers and players + net.Subscribe(eventer, ConnectivityEvents...) + } else { + // alternatively mocked and replayed events bypass the simulation network backend + journal.Subscribe(eventer, ConnectivityEvents...) + } self := NewResourceContoller( &ResourceHandlers{ // GET // @@ -89,30 +104,92 @@ 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)) - self.SetResource("mockevents", NewMockersController(eventer)) + self.SetResource("nodes", nodesController) + self.SetResource("debug", NewDebugController(journal)) + self.SetResource("mockevents", NewMockersController(eventer, conf.DefaultMockerConfig)) self.SetResource("journals", NewJournalPlayersController(eventer)) + return Controller(self) } +func NewNodesController(net *Network) *ResourceController { + return NewResourceContoller( + &ResourceHandlers{ + Create: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + nodeid := adapters.RandomNodeId() + net.NewNode(&NodeConfig{Id: nodeid}) + return &NodeConfig{Id: nodeid}, nil + }, + }, + Retrieve: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + ids := msg.([]string) + var result []interface{} + for _, id := range ids { + node := net.GetNode(adapters.NewNodeIdFromHex(id)) + if node != nil { + result = append(result, interface{}(node.String())) + } + } + return result, nil + }, + Type: reflect.TypeOf([]string{}), // this is input not output param structure + }, + Update: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + args := msg.(*NodeIF) + oneId := adapters.NewNodeIdFromHex(args.One) + if len(args.Other) == 0 { + if net.Start(oneId) != nil { + net.Stop(oneId) + } + return empty, nil + } else { + otherId := adapters.NewNodeIdFromHex(args.Other) + err := net.Connect(oneId, otherId) + return empty, err + } + }, + Type: reflect.TypeOf(&NodeIF{}), // this is input not output param structure + }, + }, + ) +} + +func NewDebugController(journal *Journal) Controller { + return NewResourceContoller( + &ResourceHandlers{ + Create: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + journaldump := []string{} + 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 + }, + }, + }, + ) +} + // Network models a p2p network // the actual logic of bringing nodes and connections up and down and // messaging is implemented in the particular NodeAdapter interface type Network struct { // input trigger events and other events - triggers *event.TypeMux // event triggers - events *event.TypeMux // events + events *event.TypeMux // generated events a journal can subsribe to lock sync.RWMutex nodeMap map[discover.NodeID]int connMap map[string]int Nodes []*Node `json:"nodes"` Conns []*Conn `json:"conns"` messenger func(p2p.MsgReadWriter) adapters.Messenger + quitc chan bool + conf *NetworkConfig // // adapters.Messenger // node adapter function that creates the node model for @@ -120,13 +197,14 @@ type Network struct { naf func(*NodeConfig) adapters.NodeAdapter } -func NewNetwork(triggers, events *event.TypeMux) *Network { +func NewNetwork(conf *NetworkConfig) *Network { return &Network{ - triggers: triggers, - events: events, + conf: conf, + events: &event.TypeMux{}, nodeMap: make(map[discover.NodeID]int), connMap: make(map[string]int), messenger: adapters.NewSimPipe, + quitc: make(chan bool), } } @@ -134,6 +212,62 @@ func (self *Network) SetNaf(naf func(*NodeConfig) adapters.NodeAdapter) { self.naf = naf } +// Subscribe takes an event.TypeMux and subscibes to types +// and launches a goroutine that reads control events from an eventer Subsription channel +// and executes the events +func (self *Network) Subscribe(eventer *event.TypeMux, types ...interface{}) { + glog.V(logger.Info).Infof("subscribe") + sub := eventer.Subscribe(types...) + go func() { + defer sub.Unsubscribe() + for { + select { + case ev := <-sub.Chan(): + self.execute(ev) + case <-self.quitc: + return + } + } + }() +} + +func (self *Network) execute(in *event.TypeMuxEvent) { + glog.V(logger.Detail).Infof("execute event %v", in) + ev := in.Data + if ne, ok := ev.(*NodeEvent); ok { + if ne.Action == "up" { + err := self.NewNode(&NodeConfig{Id: ne.node.Id}) + if err != nil { + glog.V(logger.Detail).Infof("error execute event %v: %v", ne, err) + } + err = self.Start(ne.node.Id) + if err != nil { + glog.V(logger.Detail).Infof("error execute event %v: %v", ne, err) + } + } else { + err := self.Stop(ne.node.Id) + if err != nil { + glog.V(logger.Detail).Infof("error execute event %v: %v", ne, err) + } + } + } else if ce, ok := ev.(*ConnEvent); ok { + if ce.Action == "up" { + err := self.Connect(ce.conn.One, ce.conn.Other) + if err != nil { + glog.V(logger.Detail).Infof("error execute event %v: %v", ne, err) + } + } else { + err := self.Disconnect(ce.conn.One, ce.conn.Other) + if err != nil { + glog.V(logger.Detail).Infof("error execute event %v: %v", ne, err) + } + } + } else { + glog.V(logger.Detail).Infof("event: %#v", ev) + panic("unhandled event") + } +} + // Events returns the output eventer of the Network. func (self *Network) Events() *event.TypeMux { return self.events @@ -244,9 +378,7 @@ func (self *Msg) String() string { func (self *Msg) event() *MsgEvent { return &MsgEvent{ Action: "up", - //Type: fmt.Sprintf("%d", self.Code), - Type: "msg", - msg: self, + msg: self, } } @@ -293,7 +425,10 @@ func (self *Network) NewNode(conf *NodeConfig) error { return nil } -func (self *Network) NewGenericSimNode(conf *NodeConfig) adapters.NodeAdapter { +func (self *Network) Config() *NetworkConfig { + return self.conf +} +func (self *Network) NewSimNode(conf *NodeConfig) adapters.NodeAdapter { id := conf.Id na := adapters.NewSimNode(id, self, self.messenger) return na @@ -424,7 +559,7 @@ func (self *Network) Connect(oneId, otherId *adapters.NodeId) error { // sets the Conn model to Down // the disconnect will be initiated (the connection is dropped by) the first node // it errors if either of the nodes is down (or does not exist) -func (self *Network) Disconnect(oneId, otherId *adapters.NodeId, disconnect bool) error { +func (self *Network) Disconnect(oneId, otherId *adapters.NodeId) error { conn := self.GetConn(oneId, otherId) if conn == nil { return fmt.Errorf("connection between %v and %v does not exist", oneId, otherId) @@ -436,20 +571,10 @@ func (self *Network) Disconnect(oneId, otherId *adapters.NodeId, disconnect bool if conn.One.NodeID != oneId.NodeID { rev = true } - // if Disconnect is externally triggered one needs to call the actual - // adapter's disconnect method - if disconnect { - var err error - if rev { - err = conn.other.na.Disconnect(oneId.Bytes()) - } else { - err = conn.one.na.Disconnect(otherId.Bytes()) - } - if err != nil { - return err - } + if rev { + return conn.other.na.Disconnect(oneId.Bytes()) } - return nil + return conn.one.na.Disconnect(otherId.Bytes()) // return self.DidDisconnect(oneId, otherId) } diff --git a/p2p/simulations/session_controller.go b/p2p/simulations/session_controller.go index 6769c5c9c5..66d15bc869 100644 --- a/p2p/simulations/session_controller.go +++ b/p2p/simulations/session_controller.go @@ -10,7 +10,6 @@ import ( "sync" "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" @@ -40,8 +39,8 @@ type NodeResult struct { } type NodeIF struct { - One uint - Other uint + One string + Other string MessageType uint8 } @@ -85,7 +84,12 @@ func NewResourceContoller(c *ResourceHandlers) *ResourceController { var empty = struct{}{} -func NewSessionController() (*ResourceController, chan bool) { +func DefaultNet(conf *NetworkConfig) (NetworkControl, *ResourceController) { + net := NewNetwork(conf) + return NetworkControl(net), NewNodesController(net) +} + +func NewSessionController(nethook func(*NetworkConfig) (NetworkControl, *ResourceController)) (*ResourceController, chan bool) { quitc := make(chan bool) return NewResourceContoller( &ResourceHandlers{ @@ -93,72 +97,12 @@ func NewSessionController() (*ResourceController, chan bool) { Create: &ResourceHandler{ Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { conf := msg.(*NetworkConfig) - journal := NewJournal() - net := NewNetwork(nil, &event.TypeMux{}) - net.SetNaf(net.NewGenericSimNode) + netC, nodesC := nethook(conf) 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) - } + m := NewNetworkController(netC, nodesC) if parent != nil { parent.SetResource(conf.Id, m) } - parent.id++ - - m.SetResource("debug", NewResourceContoller( - &ResourceHandlers{ - Create: &ResourceHandler{ - Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { - journaldump := []string{} - 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 - }, - }, - }, - )) - - m.SetResource("node", NewResourceContoller( - &ResourceHandlers{ - Create: &ResourceHandler{ - Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { - nodeid := adapters.RandomNodeId() - net.NewNode(&NodeConfig{Id: nodeid}) - return &NodeConfig{Id: nodeid}, nil - }, - }, - Retrieve: &ResourceHandler{ - Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { - return &NodeResult{Nodes: net.Nodes}, nil - }, - }, - Update: &ResourceHandler{ - Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { - var othernode *Node - - args := msg.(*NodeIF) - onenode := net.Nodes[args.One-1] - - if args.Other == 0 { - if net.Start(onenode.Id) != nil { - net.Stop(onenode.Id) - } - return &NodeResult{Nodes: []*Node{onenode}}, nil - } else { - othernode = net.Nodes[args.Other-1] - net.Connect(onenode.Id, othernode.Id) - return &NodeResult{Nodes: []*Node{onenode, othernode}}, nil - } - }, - Type: reflect.TypeOf(&NodeIF{}), // this is input not output param structure - }, - }, - )) - return empty, nil }, Type: reflect.TypeOf(&NetworkConfig{}), diff --git a/p2p/simulations/session_controller_test.go b/p2p/simulations/session_controller_test.go index fcd101bac6..d557a7028b 100644 --- a/p2p/simulations/session_controller_test.go +++ b/p2p/simulations/session_controller_test.go @@ -29,7 +29,7 @@ var controller *ResourceController func init() { glog.SetV(0) glog.SetToStderr(true) - controller, quitc = NewSessionController() + controller, quitc = NewSessionController(DefaultNet) StartRestApiServer(port, controller) } @@ -125,14 +125,12 @@ func testResponse(t *testing.T, method, addr string, r io.ReadSeeker) []byte { } func TestUpdate(t *testing.T) { - - ids := testIDs() - journal := testJournal(ids) - + t.Skip("...") conf := &NetworkConfig{ - Id: "0", + Id: "0", + Backend: false, } - mc := NewNetworkController(conf, &event.TypeMux{}, journal) + mc := NewNetworkController(NewNetwork(conf), nil) controller.SetResource(conf.Id, mc) exp := `{ "add": [ @@ -154,7 +152,8 @@ func TestUpdate(t *testing.T) { "remove": [], "message": [] }` - resp := testResponse(t, "GET", url(port, "0"), bytes.NewReader([]byte("{}"))) + s, _ := json.Marshal(&CyConfig{}) + resp := testResponse(t, "GET", url(port, "0"), bytes.NewReader(s)) if string(resp) != exp { t.Fatalf("incorrect response body. got\n'%v', expected\n'%v'", string(resp), exp) } diff --git a/p2p/testing/sessions.go b/p2p/testing/sessions.go index 13d37511b1..4054c6e8b5 100644 --- a/p2p/testing/sessions.go +++ b/p2p/testing/sessions.go @@ -31,7 +31,7 @@ type ExchangeSession struct { // higher level or network behaviour should be tested with network simulators func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(id adapters.NodeAdapter) adapters.ProtoCall) *ExchangeSession { simPipe := adapters.NewSimPipe - network := simulations.NewNetwork(nil, nil) + network := simulations.NewNetwork(&simulations.NetworkConfig{}) naf := func(conf *simulations.NodeConfig) adapters.NodeAdapter { na := adapters.NewSimNode(conf.Id, network, simPipe) if conf.Id.NodeID == id.NodeID { diff --git a/pot/address.go b/pot/address.go index a9ce9a4779..6f3019c182 100644 --- a/pot/address.go +++ b/pot/address.go @@ -59,6 +59,10 @@ func (a Address) Bin() string { return strings.Join(bs, "") } +func (a Address) Bytes() []byte { + return a[:] +} + /* Proximity(x, y) returns the proximity order of the MSB distance between x and y @@ -209,7 +213,7 @@ func NewHashAddress(s string) *HashAddress { ha := [32]byte{} t := s + string(zerosBin)[:len(zerosBin)-len(s)] - for i := 0; i < len(t)/64; i++ { + for i := 0; i < 4; i++ { n, err := strconv.ParseUint(t[i*64:(i+1)*64], 2, 64) if err != nil { panic("wrong format: " + err.Error()) diff --git a/pot/pot.go b/pot/pot.go index 43c3eddb5d..c2d92b60d3 100644 --- a/pot/pot.go +++ b/pot/pot.go @@ -110,7 +110,7 @@ func add(t *pot, val PotVal) (*pot, int, bool) { } return r, 0, false } - po, found := val.PO(t.pin, t.po) + po, found := t.pin.PO(val, t.po) if found { r = &pot{ pin: val, @@ -188,7 +188,7 @@ func Remove(t *Pot, v PotVal) (*Pot, int, bool) { func remove(t *pot, val PotVal) (r *pot, po int, found bool) { size := t.size - po, found = val.PO(t.pin, t.po) + po, found = t.pin.PO(val, t.po) if found { size-- if size == 0 { @@ -392,6 +392,12 @@ func Union(t0, t1 *Pot) (*Pot, int) { } func union(t0, t1 *pot) (*pot, int) { + if t0 == nil || t0.size == 0 { + return t1, 0 + } + if t1 == nil || t1.size == 0 { + return t0, 0 + } po, eq := t0.pin.PO(t1.pin, 0) var pin PotVal var bins []*pot @@ -509,7 +515,6 @@ func union(t0, t1 *pot) (*pot, int) { } // 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 { @@ -527,32 +532,94 @@ func (t *pot) each(f func(PotVal, int) bool) bool { return false } } - next = f(t.pin, t.po) - if !next { - return false - } - - return true + return f(t.pin, t.po) } -// EachBin iterates over bins of the top node and offers iterators to the caller on each +// EachFrom(f, start) is a synchronous iterator over the elements of a pot +// within the inclusive range starting from proximity order start +// the function argument is passed the value and the proximity order wrt the root pin +// it does NOT include the pinned item of the root +// respecting an ordering +// proximity > pinnedness +// the iteration ends if the function return false or there are no more elements +// end of a po range can be implemented since po is passed to the function +func (t *Pot) EachFrom(f func(PotVal, int) bool, po int) bool { + t.lock.RLock() + n := t.pot + t.lock.RUnlock() + return n.eachFrom(f, po) +} + +func (t *pot) eachFrom(f func(PotVal, int) bool, po int) bool { + var next bool + _, lim := t.getPos(po) + for i := lim; i < len(t.bins); i++ { + n := t.bins[i] + next = n.each(f) + if !next { + return false + } + } + return f(t.pin, t.po) +} + +// EachBin iterates over bins of the pivot 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 { +func (t *Pot) EachBin(val PotVal, po int, f func(int, int, func(func(val PotVal, i int) bool) bool) bool) { + t.lock.RLock() + n := t.pot + t.lock.RUnlock() + n.eachBin(val, po, f) +} + +func (t *pot) eachBin(val PotVal, po int, f func(int, int, func(func(val PotVal, i int) bool) bool) bool) { + if t == nil || t.size == 0 { + return + } + spr, _ := t.pin.PO(val, t.po) + _, lim := t.getPos(spr) + var size int + var n *pot + for i := 0; i < lim; i++ { + n = t.bins[i] + size += n.size if n.po < po { continue } if !f(n.po, n.size, n.each) { - break + return } } + if lim == len(t.bins) { + f(spr, 1, func(g func(PotVal, int) bool) bool { + return g(t.pin, spr) + }) + return + } + n = t.bins[lim] + + spo := spr + if n.po == spr { + spo++ + size += n.size + } + if !f(spr, t.size-size, func(g func(PotVal, int) bool) bool { + return t.eachFrom(func(v PotVal, j int) bool { + return g(v, spr) + }, spo) + }) { + return + } + if spo > spr { + n.eachBin(val, spo, f) + } } // 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)$ +// the order of elements retrieved reflect proximity order to the target +// TODO: add maximum proxbin to start range of iteration func (t *Pot) EachNeighbour(val PotVal, f func(PotVal, int) bool) bool { t.lock.RLock() n := t.pot @@ -561,12 +628,15 @@ func (t *Pot) EachNeighbour(val PotVal, f func(PotVal, int) bool) bool { } func (t *pot) eachNeighbour(val PotVal, f func(PotVal, int) bool) bool { + if t == nil || t.size == 0 { + return false + } var next bool l := len(t.bins) var n *pot ir := l il := l - po, eq := val.PO(t.pin, t.po) + po, eq := t.pin.PO(val, t.po) if !eq { n, il = t.getPos(po) if n != nil { @@ -606,6 +676,17 @@ func (t *pot) eachNeighbour(val PotVal, f func(PotVal, int) bool) bool { return true } +// EachNeighnbourAsync(val, max, maxPos, f, wait) is an asyncronous iterator +// over elements not closer than maxPos wrt val. +// val does not need to be match an element of the pot, but if it does, and +// maxPos is keylength than it is included in the iteration +// Calls to f are parallelised, the order of calls is undefined. +// proximity order is respected in that there is no element in the pot that +// is not visited if a closer node is visited. +// The iteration is finished when max number of nearest nodes is visited +// or if the entire there are no nodes not closer than maxPos that is not visited +// if wait is true, the iterator returns only if all calls to f are finished +// TODO: implement minPos for proper prox range iteration func (t *Pot) EachNeighbourAsync(val PotVal, max int, maxPos int, f func(PotVal, int), wait bool) { t.lock.RLock() n := t.pot @@ -631,7 +712,7 @@ func (t *pot) eachNeighbourAsync(val PotVal, max int, maxPos int, f func(PotVal, ir := l // ic := l - po, eq := val.PO(t.pin, t.po) + po, eq := t.pin.PO(val, t.po) // if po is too close, set the pivot branch (pom) to maxPos pom := po @@ -746,7 +827,6 @@ func (t *pot) eachNeighbourAsync(val PotVal, max int, maxPos int, f func(PotVal, } return max + extra - } // getPos(n) returns the forking node at PO n and its index if it exists diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go new file mode 100644 index 0000000000..e4633db8e0 --- /dev/null +++ b/swarm/network/discovery.go @@ -0,0 +1,157 @@ +package network + +import ( + "fmt" +) + +// discovery bzz hive extension for efficient peer relaying +// this will be triggered by p2p/protocol already + +type discPeer struct { + Peer + hive Hive + sub overlaySubscription + peers map[discover.NodeID]bool +} + +type overlaySubscription interface { + Subscribe(proxLimit uint) chan interface{} + SubscribeProxChange(proxLimit uint) chan interface{} +} + +// new discovery contructor +func NewDiscovery(p Peer, h Hive) error { + overlay, ok := h.Overlay.(overlaySubscription) + if !ok { + return fmt.Errorf("overlay does not support subscription") + } + self := &discPeer{ + sub: overlay, + hive: h, + Peer: p, + peers: make(map[discover.NodeID]Peer), + } + + c := sub.SubscribeProxChange() + go func() { + for { + select { + case <-h.quit: + return + case p := <-c: + resp := &peersMsg{ + Peers: []*peerAddr{p.PeerAddr.(*peerAddr)}, + } + p.Send(resp) + } + } + }() + p.Register(&subPeersMsg{}, self.handleSubPeersMsg) + p.Register(&peersMsg{}, self.handlePeersMsg) + p.Register(&getPeersMsg{}, self.handleGetPeersMsg) + + return nil +} + +/* +peersMsg is the message to pass peer information +It is always a response to a peersRequestMsg + +The encoding of a peer is identical to that in the devp2p base protocol peers +messages: [IP, Port, NodeID] +note that a node's DPA address is not the NodeID but the hash of the NodeID. + +To mitigate against spurious peers messages, requests should be remembered +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 +} + +func (self peersMsg) String() string { + return fmt.Sprintf("%T: %v", self, self.Peers) +} + +// getPeersMsg is sent to (random) peers to request (Max) peers of a specific order +type getPeersMsg struct { + Order uint + Max uint +} + +func (self getPeersMsg) String() string { + return fmt.Sprintf("%T: accept max %v peers of PO%03d", self, self.Max, self.Order) +} + +// subPeers msg is communicating the depth/sharpness/focus of the overlay table of a peer +type subPeersMsg struct { + MinProxBinSize uint + ProxLimit uint + // Offset uint + // Batch uint +} + +func (self subPeersMsg) String() string { + return fmt.Sprintf("request peers > PO%02d. ProxLimit: %02d", self.Request, self.ProxLimit) +} + +func (self *discPeer) handleSubPeersMsg(msg interface{}) error { + spm := msg.(*subPeersMsg) + self.sub.Subscribe(spm.ProxLimit) +} + +// handlePeersMsg called by the protocol when receiving peerset (for target address) +// list of nodes ([]PeerAddr in peersMsg) is added to the overlay db using the +// Register interface method +func (p *discPeer) handlePeersMsg(msg interface{}) error { + // register all addresses + var nas []PeerAddr + for _, na := range msg.(*peersMsg).Peers { + addr := PeerAddr(na) + nas = append(nas, addr) + p.peers[NodeId(addr)] = true + } + return p.hive.Register(nas...) +} + +// handleGetPeersMsg is called by the protocol when receiving a +// peerset (for target address) request +// peers suggestions are retrieved from the overlay topology driver +// using the EachLivePeer interface iterator method +// peers sent are remembered throughout a session and not sent twice +func (p *discPeer) handleGetPeersMsg(msg interface{}) error { + req := msg.(*getPeersMsg) + var peers []*peerAddr + alreadySent := p.peers + i := 0 + p.Overlay.EachLivePeer(p.OverlayAddr(), int(req.Order), func(n Peer, po int) bool { + i++ + 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 int(req.Order) == po && len(peers) < int(req.Max) + return len(peers) < int(req.Max) + }) + if len(peers) == 0 { + glog.V(logger.Debug).Infof("no peers found for %v", p) + return nil + } + glog.V(logger.Debug).Infof("%v peers sent to %v", len(peers), p) + resp := &peersMsg{ + Peers: peers, + } + err := p.Send(resp) + if err != nil { + return err + } + return nil +} diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go new file mode 100644 index 0000000000..3229188824 --- /dev/null +++ b/swarm/network/discovery_test.go @@ -0,0 +1,31 @@ +package network + +import ( + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/p2p/adapters" + // "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p/protocols" + p2ptest "github.com/ethereum/go-ethereum/p2p/testing" +) + +func TestDiscovery(t *testing.T) { + addr := RandomAddr() + to := NewTestOverlay(addr.OverlayAddr()) + pp := NewHive(NewHiveParams(), to) + ct := BzzCodeMap(HiveMsgs...) + s := newDiscoveryTester(t, 0, addr, pp, ct, nil) + s.TestExchanges(p2ptest.Exchange{ + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 1, + Msg: &subPeersMsg{uint(o)}, + Peer: id, + }, + }, + }) +} diff --git a/swarm/network/hive.go b/swarm/network/hive.go index b91d7fdd2a..3434776814 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -45,8 +45,8 @@ type Overlay interface { On(Peer) Off(Peer) - EachLivePeer([]byte, int, func(Peer) bool) - EachPeer([]byte, int, func(PeerAddr) bool) + EachLivePeer([]byte, int, func(Peer, int) bool) + EachPeer([]byte, int, func(PeerAddr, int) bool) SuggestPeer() (PeerAddr, int, bool) @@ -57,7 +57,7 @@ type Overlay interface { type Hive struct { *HiveParams // settings Overlay // the overlay topology driver - peers map[discover.NodeID]Peer + disc Discovery lock sync.Mutex quit chan bool @@ -66,15 +66,15 @@ type Hive struct { } const ( - peersBroadcastSetSize = 1 + peersBroadcastSetSize = 2 maxPeersPerRequest = 5 - callInterval = 3000000000 + callInterval = 1000 ) type HiveParams struct { PeersBroadcastSetSize uint MaxPeersPerRequest uint - CallInterval uint64 + CallInterval uint } func NewHiveParams() *HiveParams { @@ -92,7 +92,6 @@ func NewHive(params *HiveParams, overlay Overlay) *Hive { return &Hive{ HiveParams: params, Overlay: overlay, - peers: make(map[discover.NodeID]Peer), } } @@ -100,42 +99,7 @@ func NewHive(params *HiveParams, overlay Overlay) *Hive { var HiveMsgs = []interface{}{ &getPeersMsg{}, &peersMsg{}, -} - -/* -peersMsg is the message to pass peer information -It is always a response to a peersRequestMsg - -The encoding of a peer is identical to that in the devp2p base protocol peers -messages: [IP, Port, NodeID] -note that a node's DPA address is not the NodeID but the hash of the NodeID. - -To mitigate against spurious peers messages, requests should be remembered -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 -} - -func (self peersMsg) String() string { - return fmt.Sprintf("%T: %v", self, self.Peers) -} - -// getPeersMsg is sent to (random) peers to request (Max) peers of a specific order -type getPeersMsg struct { - Order uint - Max uint -} - -func (self getPeersMsg) String() string { - return fmt.Sprintf("%T: accept max %v peers of PO%03d", self, self.Max, self.Order) + &subPeersMsg{}, } // Start receives network info only at startup @@ -143,7 +107,7 @@ func (self getPeersMsg) String() string { // connectPeer is a function to connect to a peer based on its NodeID or enode URL // af() returns an arbitrary ticker channel // there are called on the p2p.Server which runs on the node -func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Time) (err error) { +func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Time) error { self.toggle = make(chan bool) self.more = make(chan bool) @@ -163,11 +127,13 @@ func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Ti addr, order, want := self.SuggestPeer() if addr != nil { - glog.V(logger.Detail).Infof("========> connect to bee %v", addr) + glog.V(logger.Info).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) } + } else { + glog.V(logger.Detail).Infof("cannot suggest peers") } if want { req := &getPeersMsg{ @@ -176,8 +142,7 @@ func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Ti } var i uint var err error - glog.V(logger.Detail).Infof("requesting bees of PO%03d from %v (each max %v)", order, self.PeersBroadcastSetSize, self.MaxPeersPerRequest) - self.EachLivePeer(nil, order, func(n Peer) bool { + self.EachLivePeer(nil, order, func(n Peer, po int) bool { glog.V(logger.Detail).Infof("%T sent to %v", req, n.ID()) err = n.Send(req) if err == nil { @@ -188,9 +153,8 @@ func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Ti } return true }) - glog.V(logger.Detail).Infof("sent %T to %d/%d peers", req, i, self.PeersBroadcastSetSize) + glog.V(logger.Info).Infof("requesting bees of PO%03d from %v/%v (each max %v)", order, i, self.PeersBroadcastSetSize, self.MaxPeersPerRequest) } - glog.V(logger.Info).Infof("%v", self) select { case self.toggle <- want: @@ -198,14 +162,14 @@ func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Ti case <-self.quit: return } + glog.V(logger.Info).Infof("%v", self) } - glog.V(logger.Warn).Infof("%v", self) }() - return + return nil } func (self *Hive) ticker() <-chan time.Time { - return time.NewTicker(time.Duration(self.CallInterval)).C + return time.NewTicker(time.Duration(self.CallInterval) * time.Millisecond).C } // keepAlive is a forever loop @@ -256,15 +220,13 @@ func (self *Hive) Add(p Peer) error { defer self.wake() glog.V(logger.Debug).Infof("to add new bee %v", p) self.On(p) + glog.V(logger.Warn).Infof("%v", self) - self.lock.Lock() - self.peers[p.ID()] = p - self.lock.Unlock() + // self.lock.Lock() + // self.peers[p.ID()] = p + // self.lock.Unlock() - p.Register(&peersMsg{}, self.handlePeersMsg(p)) - p.Register(&getPeersMsg{}, self.handleGetPeersMsg(p)) - - return nil + return NewDiscovery(p, self) } // Remove called after peer is disconnected @@ -277,53 +239,6 @@ func (self *Hive) Remove(p Peer) { self.lock.Unlock() } -// handlePeersMsg called by the protocol when receiving peerset (for target address) -// list of nodes ([]PeerAddr in peersMsg) is added to the overlay db using the -// 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 nas []PeerAddr - for _, na := range msg.(*peersMsg).Peers { - nas = append(nas, PeerAddr(na)) - } - return self.Register(nas...) - } -} - -// handleGetPeersMsg is called by the protocol when receiving a -// peerset (for target address) request -// peers suggestions are retrieved from the overlay topology driver -// using the EachLivePeer interface iterator method -// 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 - alreadySent := p.Peers() - 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) - }) - - resp := &peersMsg{ - Peers: peers, - } - err := p.Send(resp) - if err != nil { - return err - } - return nil - } -} - func (self *Hive) NodeInfo() interface{} { return interface{}(self.String()) } diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index 6ce641cf8b..9fe0ac5e05 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -7,13 +7,14 @@ import ( "github.com/ethereum/go-ethereum/p2p/adapters" // "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p/protocols" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" ) func init() { - glog.SetV(6) + glog.SetV(logger.Detail) glog.SetToStderr(true) } diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index a99e7eb646..c44adc066c 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -51,41 +51,27 @@ 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 + MaxProxDisplay int + MinProxBinSize int + MinBinSize int + MaxBinSize int + RetryInterval int + RetryExponent int + MaxRetries 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, + MaxProxDisplay: 8, + MinProxBinSize: 2, + MinBinSize: 2, + MaxBinSize: 4, + RetryInterval: 42000000000, + MaxRetries: 42, + RetryExponent: 2, } } @@ -119,36 +105,64 @@ func NewKademlia(addr []byte, params *KadParams) *Kademlia { type KadPeer struct { *pot.HashAddress PeerAddr - Peer Peer - seenAt, readyAt time.Time + Peer Peer + seenAt time.Time + retries int +} + +func (self *KadPeer) PO(val pot.PotVal, pos int) (po int, eq bool) { + kp, ok := val.(*KadPeer) + var ha *pot.HashAddress + if ok { + ha = kp.HashAddress + } else { + ha = val.(*pot.HashAddress) + } + return self.HashAddress.PO(pot.PotVal(ha), pos) } func (self *KadPeer) String() string { - return string(self.OverlayAddr()) + if self == nil { + return "" + } + // return string(self.OverlayAddr()) + return self.HashAddress.Address.String() } -func (self *Kademlia) callable(kp *KadPeer) bool { - if kp.Peer != nil || kp.readyAt.After(time.Now()) { - return false +func (self *Kademlia) callable(val pot.PotVal) *KadPeer { + kp := val.(*KadPeer) + // not callable if peer is live or exceeded maxRetries + if kp.Peer != nil || kp.retries > self.MaxRetries { + glog.V(logger.Detail).Infof("peer %v not callable", kp.PeerAddr) + return nil } - delta := time.Since(time.Time(kp.seenAt)) - if delta < self.InitialRetryInterval { - delta = self.InitialRetryInterval + // calculate the allowed number of retries based on time lapsed since last seen + timeAgo := time.Since(kp.seenAt) + var retries int + for delta := int(timeAgo) / self.RetryInterval; delta > 0; delta /= self.RetryExponent { + glog.V(logger.Detail).Infof("delta: %v", delta) + retries++ } - interval := time.Duration(delta * time.Duration(self.ConnRetryExp)) + // this is never called concurrently, so safe to increment + // peer can be retried again + if retries < kp.retries { + glog.V(logger.Detail).Infof("log time needed before retry %v, wait only warrants %v", kp.retries, retries) + return nil + } + kp.retries++ + glog.V(logger.Detail).Infof("peer %v is callable", kp.PeerAddr) - 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 + return kp } // NewKadPeer(na) creates a kademlia peer from a PeerAddr interface func NewKadPeer(na PeerAddr) *KadPeer { + // o := na.OverlayAddr() + // glog.V(logger.Detail).Infof("newkadpeer from peerAddr overlay address: %x", o[:6]) return &KadPeer{ HashAddress: pot.NewHashAddressFromBytes(na.OverlayAddr()), PeerAddr: na, + seenAt: time.Now(), } } @@ -161,16 +175,18 @@ func (self *Kademlia) Register(nas ...PeerAddr) error { 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()) + glog.V(logger.Detail).Infof("add peers: %v out of %v new; root %v", np.Size()-common, np.Size(), np.Pin().String()[:6]) 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 { + pp := NewKadPeer(p) + // var pp *KadPeer + kp := pp + self.conns.Swap(pp, func(v pot.PotVal) pot.PotVal { if v == nil { - self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal { + self.peers.Swap(pp, func(v pot.PotVal) pot.PotVal { if v != nil { kp = v.(*KadPeer) } @@ -181,18 +197,36 @@ func (self *Kademlia) On(p Peer) { } return v }) + kp.seenAt = time.Now() + kp.retries = 0 + f := func(val pot.PotVal, po int) { + vp := val.(*KadPeer).Peer + dp.NotifyPeer(kp.PeerAddr, po) + } + + self.conns.EachNeighbourAsync(pp, 256, 256, f, nil) } // Off removes a peer from among live peers func (self *Kademlia) Off(p Peer) { kp := NewKadPeer(p) - self.conns.Remove(kp) + self.conns.Swap(kp, func(v pot.PotVal) pot.PotVal { + if v != nil { + kp = v.(*KadPeer) + kp.Peer = nil + return nil + } + return nil + }) kp.Peer = nil + kp.retries = 0 + kp.seenAt = time.Now() } // 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) { +// that has proximity order po or less as measured from the base +// if base is nil, kademlia base address is used +func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int) bool) { var p pot.PotVal if base == nil { p = pot.PotVal(self.addr) @@ -200,16 +234,17 @@ func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer) bool) { p = pot.NewHashAddressFromBytes(base) } self.conns.EachNeighbour(p, func(val pot.PotVal, po int) bool { - if po == o { - return f(val.(*KadPeer).Peer) + if po > o { + return true } - return po < o + return f(val.(*KadPeer).Peer, po) }) } // 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) { +// that has proximity order po or less as measured from the base +// if base is nil, kademlia base address is used +func (self *Kademlia) EachPeer(base []byte, o int, f func(PeerAddr, int) bool) { var p pot.PotVal if base == nil { p = pot.PotVal(self.addr) @@ -217,22 +252,26 @@ func (self *Kademlia) EachPeer(base []byte, o int, f func(PeerAddr) bool) { p = pot.NewHashAddressFromBytes(base) } self.peers.EachNeighbour(p, func(val pot.PotVal, po int) bool { - if po == o { - return f(val.(*KadPeer).PeerAddr) + if po > o { + return true } - return po < o + return f(val.(*KadPeer).Peer, po) }) } // proxLimit() returns the proximity order that defines the distance of -// the nearest neighbour set with cardinality >= ProxBinSize +// the nearest neighbour set with cardinality >= MinProxBinSize +// if there is altogether less than MinProxBinSize peers it returns 0 func (self *Kademlia) proxLimit() int { + if self.conns.Size() < self.MinProxBinSize { + return 0 + } var proxLimit int var size int f := func(v pot.PotVal, i int) bool { size++ proxLimit = i - return size <= self.BucketSize + return size < self.MinProxBinSize } self.conns.EachNeighbour(pot.PotVal(self.addr), f) return proxLimit @@ -243,10 +282,36 @@ func (self *Kademlia) proxLimit() int { // 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 + minsize := self.MinBinSize proxLimit := self.proxLimit() + // if there is a callable neighbour within the current proxBin, connect + // this makes sure nearest neighbour set is fully connected + glog.V(logger.Detail).Infof("candidate prox peer checking above PO %v", proxLimit) + var ppo int + self.peers.EachNeighbour(self.addr, func(val pot.PotVal, po int) bool { + r := self.callable(val) + if r == nil { + glog.V(logger.Detail).Infof("candidate peer not callable: %#v", r) + return po > proxLimit + } + p = r + return false + }) + if p != nil { + glog.V(logger.Detail).Infof("candidate prox peer found: %v (%v), %#v", p, ppo, p) + return p, 0, false + } + glog.V(logger.Detail).Infof("no candidate prox peers to connect to (ProxLimit: %v, minProxSize: %v)", proxLimit, self.MinProxBinSize) + var bpo []int - self.conns.EachBin(0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + prev := -1 + self.conns.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + glog.V(logger.Detail).Infof("check PO%02d: ", po) + prev++ + if po > prev { + size = 0 + po = prev + } if size < minsize { minsize = size bpo = append(bpo, po) @@ -259,19 +324,20 @@ func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) { 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 + // dont ask for new peers (want = false) + // try to select a candidate 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 { + self.peers.EachBin(self.addr, bpo[i], func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + // for each bin we find callable candidate peers f(func(val pot.PotVal, i int) bool { - cp := val.(*KadPeer) - if self.callable(cp) { - p = cp - return false + r := self.callable(val) + glog.V(logger.Detail).Infof("check PO%02d: ", po) + if r == nil { + return i < proxLimit } - return true + p = r + return false }) return false }) @@ -281,8 +347,9 @@ func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) { } // cannot find a candidate, ask for more for this proximity bin specifically o = bpo[i] + want = true } - return p, o, true + return p, o, want } // kademlia table + kaddb table displayed with ascii @@ -291,14 +358,16 @@ func (self *Kademlia) String() string { var rows []string rows = append(rows, "=========================================================================") - rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), 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)) + rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), self.addr.Address.String()[:6])) + rows = append(rows, fmt.Sprintf("population: %d (%d), ProxBinSize: %d, MinBinSize: %d, MaxBinSize: %d", self.conns.Size(), self.peers.Size(), self.MinProxBinSize, self.MinBinSize, self.MaxBinSize)) - liverows := make([]string, self.MaxProx) - peersrows := make([]string, self.MaxProx) + liverows := make([]string, self.MaxProxDisplay) + peersrows := make([]string, self.MaxProxDisplay) var proxLimit int + prev := -1 + var proxLimitSet bool rest := self.conns.Size() - self.conns.EachBin(0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + self.conns.EachBin(self.addr, 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 @@ -307,24 +376,26 @@ func (self *Kademlia) String() string { rowlen++ return rowlen < 4 }) - if rowlen == 0 || rest < self.ProxBinSize { - proxLimit = po - 1 + if !proxLimitSet && (po > prev+1 || rest < self.MinProxBinSize) { + proxLimitSet = true + proxLimit = prev + 1 } - for ; rowlen < 4; rowlen++ { + for ; rowlen <= 5; rowlen++ { row = append(row, " ") } liverows[po] = strings.Join(row, " ") + prev = po return true }) - self.peers.EachBin(0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + self.peers.EachBin(self.addr, 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)} + row := []string{fmt.Sprintf("%2d", size)} f(func(val pot.PotVal, vpo int) bool { kp := val.(*KadPeer) - if kp.Peer != nil { - return true - } + // if kp.Peer != nil { + // return true + // } row = append(row, kp.String()[:6]) rowlen++ return rowlen < 4 @@ -333,20 +404,20 @@ func (self *Kademlia) String() string { return true }) - for i := 0; i < self.MaxProx; i++ { + for i := 0; i < self.MaxProxDisplay; 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 " + left = " 0 " } if len(right) == 0 { - right = "0 " + right = " 0" } rows = append(rows, fmt.Sprintf("%03d %v | %v", i, left, right)) } rows = append(rows, "=========================================================================") - return strings.Join(rows, "\n") + return "\n" + strings.Join(rows, "\n") } diff --git a/swarm/network/kademlia_test.go b/swarm/network/kademlia_test.go new file mode 100644 index 0000000000..a830f2902c --- /dev/null +++ b/swarm/network/kademlia_test.go @@ -0,0 +1,282 @@ +// 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 . +package network + +import ( + "fmt" + "testing" + "time" + + // "github.com/ethereum/go-ethereum/logger" + // "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/pot" +) + +func testKadPeerAddr(s string) *peerAddr { + a := pot.NewHashAddress(s).Bytes() + return &peerAddr{OAddr: a, UAddr: a} +} + +func testKadPeer(s string) *bzzPeer { + return &bzzPeer{peerAddr: testKadPeerAddr(s)} +} + +func testStr(a PeerAddr) string { + s, _ := a.(*KadPeer) + if s == nil { + return "" + } + // return s.String() + // return fmt.Sprintf("%06x", s.OverlayAddr()) + return pot.NewHashAddressFromBytes(a.(*KadPeer).PeerAddr.OverlayAddr()).Bin()[:6] + // return a.(*KadPeer).String()[:6] //.HashAddress.String()[:6] + // return a.(*KadPeer).String() //[:6] //.HashAddress.String()[:6] + // return "wtf" +} + +type testKademlia struct { + *Kademlia +} + +func newTestKademlia(b string) *testKademlia { + params := NewKadParams() + params.MinBinSize = 1 + params.MinProxBinSize = 2 + base := pot.NewHashAddress(b).Bytes() + return &testKademlia{NewKademlia(base, params)} +} + +func (k *testKademlia) On(ons ...string) *testKademlia { + for _, s := range ons { + k.Kademlia.On(testKadPeer(s)) + } + return k +} + +func (k *testKademlia) Off(offs ...string) *testKademlia { + for _, s := range offs { + k.Kademlia.Off(testKadPeer(s)) + } + return k +} + +func (k *testKademlia) Register(regs ...string) *testKademlia { + var ps []PeerAddr + for _, s := range regs { + ps = append(ps, PeerAddr(testKadPeerAddr(s))) + } + k.Kademlia.Register(ps...) + return k +} + +func testSuggestPeer(t *testing.T, k *testKademlia, expAddr string, expPo int, expWant bool) error { + addr, o, want := k.SuggestPeer() + if testStr(addr) != expAddr { + return fmt.Errorf("incorrect peer address suggested. expected %v, got %v", expAddr, testStr(addr)) + } + if o != expPo { + return fmt.Errorf("incorrect prox order suggested. expected %v, got %v", expPo, o) + } + if want != expWant { + return fmt.Errorf("expected SuggestPeer to want peers: %v", expWant) + } + return nil +} + +func TestSuggestPeerFindPeers(t *testing.T) { + // 2 row gap, unsaturated proxbin, no callables -> want PO 0 + k := newTestKademlia("000000").On("001000") + err := testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + + // 2 row gap, saturated proxbin, no callables -> want PO 0 + k.On("000100") + err = testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + + // 1 row gap (1 less), saturated proxbin, no callables -> want PO 1 + k.On("100000") + err = testSuggestPeer(t, k, "", 1, true) + if err != nil { + t.Fatal(err.Error()) + } + + // no gap (1 less), saturated proxbin, no callables -> do not want more + k.On("010000", "001001") + err = testSuggestPeer(t, k, "", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + // oversaturated proxbin, > do not want more + k.On("001001") + err = testSuggestPeer(t, k, "", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + // reintroduce gap, disconnected peer callable + k.Off("010000") + err = testSuggestPeer(t, k, "010000", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + // second time disconnected peer not callable + // with reasonably set Interval + // err = testSuggestPeer(t, k, "010000", 2, true) + err = testSuggestPeer(t, k, "", 1, true) + if err != nil { + t.Fatal(err.Error()) + } + + // on and off again, peer callable again + k.On("010000") + k.Off("010000") + err = testSuggestPeer(t, k, "010000", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + k.On("010000") + k.Off("010000") + // PO1 disconnects + // new closer peer appears, it is immediately wanted + // k.Off("010000") + k.Register("000101") + err = testSuggestPeer(t, k, "000101", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + // second time, gap filling + err = testSuggestPeer(t, k, "010000", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + k.On("010000") + err = testSuggestPeer(t, k, "", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + k.MinBinSize = 2 + err = testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + + k.On("100001") + k.On("010001") + err = testSuggestPeer(t, k, "", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + k.MinBinSize = 3 + k.On("100010") + k.On("010010") + err = testSuggestPeer(t, k, "", 2, true) + if err != nil { + t.Fatal(err.Error()) + } + + k.On("001010") + err = testSuggestPeer(t, k, "", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + +} + +func TestSuggestPeerRetries(t *testing.T) { + // 2 row gap, unsaturated proxbin, no callables -> want PO 0 + k := newTestKademlia("000000") + cycle := 50 * time.Millisecond + k.RetryInterval = int(cycle) + k.MaxRetries = 3 + k.RetryExponent = 3 + k.Register("010000") + k.On("000001", "000010") + err := testSuggestPeer(t, k, "010000", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + err = testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + + // cycle *= time.Duration(k.RetryExponent) + time.Sleep(cycle) + err = testSuggestPeer(t, k, "010000", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + err = testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + + cycle *= time.Duration(k.RetryExponent) + time.Sleep(cycle) + err = testSuggestPeer(t, k, "010000", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + err = testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + + cycle *= time.Duration(k.RetryExponent) + time.Sleep(cycle) + err = testSuggestPeer(t, k, "010000", 0, false) + if err != nil { + t.Fatal(err.Error()) + } + + err = testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + + cycle *= time.Duration(k.RetryExponent) + time.Sleep(cycle) + + err = testSuggestPeer(t, k, "", 0, true) + if err != nil { + t.Fatal(err.Error()) + } + +} + +func TestKademliaHiveString(t *testing.T) { + k := newTestKademlia("000000").On("010000", "001000").Register("100000", "100001") + h := k.String() + expH := "\n=========================================================================\nMon Feb 27 12:10:28 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 000000\npopulation: 2 (4), ProxBinSize: 2, MinBinSize: 1, MaxBinSize: 4\n============ PROX LIMIT: 0 ==========================================\n000 0 | 2 840000 800000\n001 1 400000 | 1 400000\n002 1 200000 | 1 200000\n003 0 | 0\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n=========================================================================" + if expH[100:] != h[100:] { + t.Fatalf("incorrect hive output. expected %v, got %v", expH, h) + } +} diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 6d579739e6..5bdb325027 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -112,7 +112,7 @@ func Bzz(localAddr []byte, hive PeerPool, na adapters.NodeAdapter, ct *protocols return err } - // mount external service models on the peer connection (swap, sync) + // mount external service models on the peer connection (swap, sync, hive) if services != nil { err = services(bee) if err != nil { diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index 67dee2789e..94a83e348d 100644 --- a/swarm/network/simulations/overlay.go +++ b/swarm/network/simulations/overlay.go @@ -6,12 +6,10 @@ package main import ( - "fmt" - "reflect" + "math/rand" "runtime" "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" @@ -61,8 +59,8 @@ func (self *Network) NewSimNode(conf *simulations.NodeConfig) adapters.NodeAdapt id := conf.Id na := adapters.NewSimNode(id, self.Network, adapters.NewSimPipe) addr := network.NewPeerAddrFromNodeId(id) - // to := network.NewKademlia(addr.OverlayAddr(), nil) // overlay topology driver - to := network.NewTestOverlay(addr.OverlayAddr()) // overlay topology driver + 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 @@ -87,60 +85,50 @@ func NewNetwork(network *simulations.Network) *Network { return n } -// NewSessionController sits as the top-most controller for this simulation -// creates an inprocess simulation of basic node running their own bzz+hive -func NewSessionController() (*simulations.ResourceController, chan bool) { - quitc := make(chan bool) - return simulations.NewResourceContoller( - &simulations.ResourceHandlers{ - // POST / - Create: &simulations.ResourceHandler{ - Handle: func(msg interface{}, parent *simulations.ResourceController) (interface{}, error) { - conf := msg.(*simulations.NetworkConfig) - net := simulations.NewNetwork(nil, &event.TypeMux{}) - ppnet := NewNetwork(net) - c := simulations.NewNetworkController(conf, net.Events(), simulations.NewJournal()) - if len(conf.Id) == 0 { - conf.Id = fmt.Sprintf("%d", 0) - } - glog.V(logger.Debug).Infof("new network controller on %v", conf.Id) - if parent != nil { - parent.SetResource(conf.Id, c) - } - ids := p2ptest.RandomNodeIds(50) - for _, id := range ids { - ppnet.NewNode(&simulations.NodeConfig{Id: id}) - ppnet.Start(id) - glog.V(logger.Debug).Infof("node %v starting up", id) - } - // the nodes only know about their 2 neighbours (cyclically) - for i, _ := range ids { - var peerId *adapters.NodeId - if i == 0 { - peerId = ids[len(ids)-1] - } else { - peerId = ids[i-1] - } - err := ppnet.hives[i].Register(network.NewPeerAddrFromNodeId(peerId)) - if err != nil { - panic(err.Error()) - } - } - return struct{}{}, nil - }, - Type: reflect.TypeOf(&simulations.NetworkConfig{}), - }, - // DELETE / - Destroy: &simulations.ResourceHandler{ - Handle: func(msg interface{}, parent *simulations.ResourceController) (interface{}, error) { - glog.V(logger.Debug).Infof("destroy handler called") - // this can quit the entire app (shut down the backend server) - quitc <- true - return struct{}{}, nil - }, - }, - }, - ), quitc +func nethook(conf *simulations.NetworkConfig) (simulations.NetworkControl, *simulations.ResourceController) { + conf.DefaultMockerConfig = simulations.DefaultMockerConfig() + conf.DefaultMockerConfig.SwitchonRate = 100 + // conf.DefaultMockerConfig.NodesTarget = 15 + conf.DefaultMockerConfig.NewConnCount = 1 + conf.DefaultMockerConfig.DegreeTarget = 0 + conf.Id = "0" + conf.Backend = true + net := NewNetwork(simulations.NewNetwork(conf)) + + ids := p2ptest.RandomNodeIds(5) + + for i, id := range ids { + net.NewNode(&simulations.NodeConfig{Id: id}) + var peerId *adapters.NodeId + if i == 0 { + peerId = ids[len(ids)-1] + } else { + peerId = ids[i-1] + } + err := net.hives[i].Register(network.NewPeerAddrFromNodeId(peerId)) + if err != nil { + panic(err.Error()) + } + } + go func() { + for _, id := range ids { + net.NewNode(&simulations.NodeConfig{Id: id}) + net.Start(id) + glog.V(logger.Debug).Infof("node %v starting up", id) + n := rand.Intn(1000) + time.Sleep(time.Duration(n) * time.Millisecond) + // net.Stop(id) + } + }() + // go func() { + // for _, id := range ids { + // net.Stop(id) + // glog.V(logger.Debug).Infof("node %v shutting down", id) + // n := rand.Intn(500) + // time.Sleep(time.Duration(n) * time.Millisecond) + // } + // }() + return net, nil } // var server @@ -149,7 +137,7 @@ func main() { glog.SetV(logger.Info) glog.SetToStderr(true) - c, quitc := NewSessionController() + c, quitc := simulations.NewSessionController(nethook) simulations.StartRestApiServer("8888", c) // wait until server shuts down diff --git a/swarm/network/test_overlay.go b/swarm/network/test_overlay.go index f1b6008ed2..06f1df3d21 100644 --- a/swarm/network/test_overlay.go +++ b/swarm/network/test_overlay.go @@ -97,14 +97,14 @@ func (self *testOverlay) off(po []*testPeerAddr) (nas []PeerAddr) { return nas } -func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer) bool) { +func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer, int) bool) { if base == nil { base = self.addr } for i := o; i < len(self.pos); i++ { for _, na := range self.pos[i] { if na.Peer != nil { - if !f(na.Peer) { + if !f(na.Peer, o) { return } } @@ -112,13 +112,13 @@ func (self *testOverlay) EachLivePeer(base []byte, o int, f func(Peer) bool) { } } -func (self *testOverlay) EachPeer(base []byte, o int, f func(PeerAddr) bool) { +func (self *testOverlay) EachPeer(base []byte, o int, f func(PeerAddr, int) bool) { if base == nil { base = self.addr } for i := o; i < len(self.pos); i++ { for _, na := range self.pos[i] { - if !f(na) { + if !f(na, i) { return } }