From 1d152a05b5121712b5cacebcdf3f9a93d94a4170 Mon Sep 17 00:00:00 2001 From: zelig Date: Wed, 2 Nov 2016 13:15:57 +0100 Subject: [PATCH] p2p/simulations: refactor journal and cytoscape bridge --- p2p/simulations/cytoscape.go | 100 +++++++ p2p/simulations/journal.go | 272 +++++++++++++++++++ p2p/simulations/network.go | 296 ++------------------- p2p/simulations/session_controller.go | 15 +- p2p/simulations/session_controller_test.go | 20 +- 5 files changed, 422 insertions(+), 281 deletions(-) create mode 100644 p2p/simulations/cytoscape.go create mode 100644 p2p/simulations/journal.go diff --git a/p2p/simulations/cytoscape.go b/p2p/simulations/cytoscape.go new file mode 100644 index 0000000000..af2184a10e --- /dev/null +++ b/p2p/simulations/cytoscape.go @@ -0,0 +1,100 @@ +package simulations + +import ( + "bytes" + "fmt" + + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/logger/glog" +) + +type CyData struct { + Id string `json:"id"` + Source string `json:"source"` + Target string `json:"target"` + On bool `json:"on"` +} + +type CyElement struct { + Data *CyData `json:"data"` + Classes string `json:"classes"` + Group string `json:"group"` + // selected: false, // whether the element is selected (default false) + // selectable: true, // whether the selection state is mutable (default true) + // locked: false, // when locked a node's position is immutable (default false) + // grabbable: true, // whether the node can be grabbed and moved by the user +} + +// type ConnEntry struct { +// } + +// type MsgEntry struct { +// } + +type Entry struct { + Action string `json:"action"` + Type string `json:"type"` + Object interface{} `json:"object"` +} + +func (self *Entry) Stirng() string { + return fmt.Sprintf("\n", self.Action, self.Type, self.Object) +} + +type CyUpdate struct { + Add []*CyElement `json:"add"` + Remove []string `json:"remove"` +} + +func UpdateCy(j *Journal) *CyUpdate { + added := []*CyElement{} + removed := []string{} + var el *CyElement + update := func(ev *event.Event) bool { + entry := ev.Data.(*Entry) + glog.V(6).Infof("journal entry of %v: %v", ev.Time, entry) + switch entry.Type { + case "Node": + el = &CyElement{Group: "nodes", Data: &CyData{Id: entry.Object.(*SimNode).ID.String()[0:lablen]}} + case "Conn": + // mutually exclusive directed edge (caller -> callee) + source := entry.Object.(*SimConn).Caller.String()[0:lablen] + target := entry.Object.(*SimConn).Callee.String()[0:lablen] + first := source + second := target + if bytes.Compare([]byte(first), []byte(second)) > 1 { + first = target + second = source + } + id := fmt.Sprintf("%v-%v", first, second) + el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}} + case "Know": + // independent directed edge (peer0 registers peer1) + source := entry.Object.(*Know).Subject.String()[0:lablen] + target := entry.Object.(*Know).Object.String()[0:lablen] + id := fmt.Sprintf("%v-%v-%v", source, target, "know") + el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}} + } + switch entry.Action { + case "Add": + added = append(added, el) + case "Remove": + removed = append(removed, el.Data.Id) + case "On": + el.Data.On = true + added = append(added, el) + case "Off": + el.Data.On = false + removed = append(removed, el.Data.Id) + } + return true + } + glog.V(6).Infof("journal read") + j.Read(update) + glog.V(6).Infof("journal read done") + + return &CyUpdate{ + Add: added, + Remove: removed, + } +} diff --git a/p2p/simulations/journal.go b/p2p/simulations/journal.go new file mode 100644 index 0000000000..246813a190 --- /dev/null +++ b/p2p/simulations/journal.go @@ -0,0 +1,272 @@ +package simulations + +import ( + "bytes" + "math/rand" + "sync" + "time" + + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/logger/glog" +) + +// Journal is an instance of a guaranteed no-loss subscription using event.TypeMux +// Network components POST events to the TypeMux, which then is read by the journal +// Each journal belongs to a subscription +type Journal struct { + lock sync.Mutex + counter int + cursor int + sub event.Subscription + events []*event.Event +} + +// func (self *Journal) SnapshotAt(pos int) {} + +// NewJournal constructor takes eventer and types to subscribe to +func NewJournal(eventer *event.TypeMux, types ...interface{}) *Journal { + self := &Journal{} + self.sub = eventer.Subscribe(types...) + go func() { + self.Write() + }() + return self +} + +func (self *Journal) Close() { + self.sub.Unsubscribe() +} + +// Write is a forever loop +func (self *Journal) Write() { + for { + select { + case ev, ok := <-self.sub.Chan(): + if !ok { + return + } + self.append(ev) + } + } +} + +func (self *Journal) append(evs ...*event.Event) { + self.lock.Lock() + defer self.lock.Unlock() + self.events = append(self.events, evs...) +} + +func (self *Journal) WaitEntries(n int) { + for self.NewEntries() < n { + } +} + +func (self *Journal) NewEntries() int { + self.lock.Lock() + defer self.lock.Unlock() + return len(self.events) - self.cursor +} + +func (self *Journal) Read(f func(*event.Event) bool) (read int, err error) { + self.lock.Lock() + defer self.lock.Unlock() + for self.cursor < len(self.events) && f(self.events[self.cursor]) { + read++ + self.cursor++ + } + self.reset(self.cursor) + return read, nil +} + +func (self *Journal) Reset(n int) { + self.lock.Lock() + defer self.lock.Unlock() + self.reset(n) +} + +func (self *Journal) reset(n int) { + if n > self.counter { + n = self.counter + } + self.events = self.events[self.cursor:] + self.cursor = 0 +} + +func (self *Journal) Counter() int { + self.lock.Lock() + defer self.lock.Unlock() + return self.counter +} + +// type History() + +func (self *Journal) Cursor() int { + self.lock.Lock() + defer self.lock.Unlock() + return self.cursor +} + +// 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 + +type Delta struct { + On int + Off int +} + +func oneOutOf(n int) int { + t := rand.Intn(n) + if t == 0 { + return 1 + } + return 0 +} + +func deltas(i int) (d []*Delta) { + if i == 0 { + return []*Delta{ + &Delta{10, 0}, + &Delta{20, 0}, + } + } + return []*Delta{ + &Delta{oneOutOf(10), oneOutOf(10)}, + &Delta{oneOutOf(2), oneOutOf(2)}, + } +} + +func mockJournalTest(nw *Network, ticker *<-chan time.Time) { + + ids := RandomNodeIDs(100) + action := "Off" + for n := 0; ; n++ { + select { + case <-*ticker: + var entries []*Entry + + if n == 0 { + + entries = []*Entry{ + &Entry{ + Type: "Node", + Action: "On", + Object: &SimNode{ID: ids[0]}, + }, + &Entry{ + Type: "Node", + Action: "On", + Object: &SimNode{ID: ids[1]}, + }, + } + } else { + sc := &SimConn{ + Caller: ids[0], + Callee: ids[1], + } + if n%3 == 0 { + if action == "On" { + action = "Off" + } else { + action = "On" + } + entries = append(entries, &Entry{ + Type: "Conn", + Action: action, + Object: sc, + }) + } + } + + glog.V(6).Info("entries: %v", entries) + nw.AppendEntries(entries...) + } + } +} + +// MockNetwork 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 MockNetwork(eventer *event.TypeMux, ticker <-chan time.Time) { + ids := RandomNodeIDs(100) + var onNodes []*SimNode + offNodes := ids + var onConns []*SimConn + + n := 0 + for _ = range ticker { + ds := deltas(n) + for i := 0; len(offNodes) > 0 && i < ds[0].On; i++ { + c := rand.Intn(len(offNodes)) + sn := &SimNode{ID: offNodes[c]} + err := eventer.Post(&Entry{ + Type: "Node", + Action: "On", + Object: sn, + }) + if err != nil { + panic(err.Error()) + } + onNodes = append(onNodes, sn) + offNodes = append(offNodes[0:c], offNodes[c+1:]...) + } + for i := 0; len(onNodes) > 0 && i < ds[0].Off; i++ { + c := rand.Intn(len(onNodes)) + sn := onNodes[c] + err := eventer.Post(&Entry{ + Type: "Node", + Action: "Off", + Object: sn, + }) + if err != nil { + panic(err.Error()) + } + onNodes = append(onNodes[0:c], onNodes[c+1:]...) + offNodes = append(offNodes, sn.ID) + } + for i := 0; len(onNodes) > 1 && i < ds[1].On; i++ { + caller := onNodes[rand.Intn(len(onNodes))].ID + callee := onNodes[rand.Intn(len(onNodes))].ID + if bytes.Compare(caller[:], callee[:]) >= 0 { + i-- + continue + } + sc := &SimConn{ + Caller: caller, + Callee: callee, + } + err := eventer.Post(&Entry{ + Type: "Conn", + Action: "On", + Object: sc, + }) + if err != nil { + panic(err.Error()) + } + onConns = append(onConns, sc) + } + for i := 0; len(onConns) > 0 && i < ds[1].Off; i++ { + c := rand.Intn(len(onConns)) + err := eventer.Post(&Entry{ + Type: "Conn", + Action: "Off", + Object: onConns[c], + }) + if err != nil { + panic(err.Error()) + } + onConns = append(onConns[0:c], onConns[c+1:]...) + } + n++ + } + +} diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 3836117c0c..0ef61eca32 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -1,35 +1,37 @@ package simulations import ( - "bytes" "fmt" - "math/rand" "sync" - "time" - "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p/adapters" "github.com/ethereum/go-ethereum/p2p/discover" ) const lablen = 4 -func NewNetworkController(n *Network, parent *ResourceController) Controller { +// func NewMockNetworkController(n *Network, parent *ResourceController) Controller { +func NewMockNetworkController(journal *Journal, closef func()) Controller { self := NewResourceContoller( &ResourceHandlers{ - // Destroy: n.Shutdown, nil + Destroy: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + journal.Close() + closef() + return nil, nil + }, + }, // Create: n.StartNode, NodeConfig // Update: n.Setup, NodeConfig // Retrieve: n.Retrieve, Retrieve: &ResourceHandler{ - Handle: n.Query, + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + u := UpdateCy(journal) + return u, nil + }, }, }, ) - if parent != nil { - parent.SetResource(fmt.Sprintf("%d", parent.id), self) - } - // self.SetResource("nodes", NewNodesController()) return Controller(self) } @@ -41,6 +43,8 @@ type Network struct { NodeMap map[discover.NodeID]int Nodes []*SimNode Journal []*Entry + // Journals map[string]*Journal + // Events events.Mux } func NewNetwork(m adapters.Messenger) *Network { @@ -64,15 +68,23 @@ func (self *SimConn) String() string { return fmt.Sprintf("SimConn %v->%v", self.Caller.String()[0:lablen], self.Callee.String()[0:lablen]) } +// active connections are represented by the SimNode entry object so that +// you journal updates could filter if passive knowledge about peers is +// irrelevant +type SimConn struct { + Caller *discover.NodeID `json:"caller"` + Callee *discover.NodeID `json:"callee"` + caller, callee *SimNode + // Info + // active connection + // average throughput, recent average throughput +} + type NodeConfig struct { ID *discover.NodeID Run func(adapters.NetAdapter, adapters.Messenger) adapters.ProtoCall } -func Key(id []byte) string { - return string(id) -} - func (self *Network) Protocol(id *discover.NodeID) adapters.ProtoCall { self.lock.Lock() defer self.lock.Unlock() @@ -89,33 +101,6 @@ type QueryConfig struct { Format string // "cy.update", "journal", } -type CyData struct { - Id string `json:"id"` - Source string `json:"source"` - Target string `json:"target"` - On bool `json:"on"` -} - -type CyElement struct { - Data *CyData `json:"data"` - Classes string `json:"classes"` - Group string `json:"group"` - // selected: false, // whether the element is selected (default false) - // selectable: true, // whether the selection state is mutable (default true) - // locked: false, // when locked a node's position is immutable (default false) - // grabbable: true, // whether the node can be grabbed and moved by the user -} - -type Entry struct { - Action string `json:"action"` - Type string `json:"type"` - Object interface{} `json:"object"` -} - -func (self *Entry) Stirng() string { - return fmt.Sprintf("\n", self.Action, self.Type, self.Object) -} - func (n *Network) AppendEntries(entries ...*Entry) { n.lock.Lock() n.Journal = append(n.Journal, entries...) @@ -124,7 +109,7 @@ func (n *Network) AppendEntries(entries ...*Entry) { type Know struct { Subject *discover.NodeID `json:"subject"` - Object *discover.NodeID `json:"objectr"` + Object *discover.NodeID `json:"object"` // Into // number of attempted connections // time of attempted connections @@ -133,231 +118,6 @@ type Know struct { // swap balance } -// active connections are represented by the SimNode entry object so that -// you journal updates could filter if passive knowledge about peers is -// irrelevant -type SimConn struct { - Caller *discover.NodeID `json:"caller"` - Callee *discover.NodeID `json:"callee"` - // Info - // active connection - // average throughput, recent average throughput -} - -func (self *Network) CyUpdate() *CyUpdate { - self.lock.Lock() - defer self.lock.Unlock() - added := []*CyElement{} - removed := []string{} - var el *CyElement - for _, entry := range self.Journal { - glog.V(6).Infof("journal entry: %v", entry) - switch entry.Type { - case "Node": - el = &CyElement{Group: "nodes", Data: &CyData{Id: entry.Object.(*SimNode).ID.String()[0:lablen]}} - case "Conn": - // mutually exclusive directed edge (caller -> callee) - source := entry.Object.(*SimConn).Caller.String()[0:lablen] - target := entry.Object.(*SimConn).Callee.String()[0:lablen] - first := source - second := target - if bytes.Compare([]byte(first), []byte(second)) > 1 { - first = target - second = source - } - id := fmt.Sprintf("%v-%v", first, second) - el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}} - case "Know": - // independent directed edge (peer0 registers peer1) - source := entry.Object.(*Know).Subject.String()[0:lablen] - target := entry.Object.(*Know).Object.String()[0:lablen] - id := fmt.Sprintf("%v-%v-%v", source, target, "know") - el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}} - } - switch entry.Action { - case "Add": - added = append(added, el) - case "Remove": - removed = append(removed, el.Data.Id) - case "On": - el.Data.On = true - added = append(added, el) - case "Off": - el.Data.On = false - removed = append(removed, el.Data.Id) - } - } - self.Journal = nil - return &CyUpdate{ - Add: added, - Remove: removed, - } -} - -type CyUpdate struct { - Add []*CyElement `json:"add"` - Remove []string `json:"remove"` -} - -func (self *Network) Query(conf interface{}, c *ResourceController) (interface{}, error) { - glog.V(6).Infof("query: GET handler ") - // config := conf.(*QueryConfig) - return interface{}(self.CyUpdate()), nil -} - -// 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 - -type Delta struct { - On int - Off int -} - -func oneOutOf(n int) int { - t := rand.Intn(n) - if t == 0 { - return 1 - } - return 0 -} - -func deltas(i int) (d []*Delta) { - if i == 0 { - return []*Delta{ - &Delta{10, 0}, - &Delta{20, 0}, - } - } - return []*Delta{ - &Delta{oneOutOf(10), oneOutOf(10)}, - &Delta{oneOutOf(2), oneOutOf(2)}, - } -} - -func mockJournalTest(nw *Network, ticker *<-chan time.Time) { - - ids := RandomNodeIDs(100) - action := "Off" - for n := 0; ; n++ { - select { - case <-*ticker: - var entries []*Entry - - if n == 0 { - - entries = []*Entry{ - &Entry{ - Type: "Node", - Action: "On", - Object: &SimNode{ID: ids[0]}, - }, - &Entry{ - Type: "Node", - Action: "On", - Object: &SimNode{ID: ids[1]}, - }, - } - } else { - sc := &SimConn{ - Caller: ids[0], - Callee: ids[1], - } - if n%3 == 0 { - if action == "On" { - action = "Off" - } else { - action = "On" - } - entries = append(entries, &Entry{ - Type: "Conn", - Action: action, - Object: sc, - }) - } - } - - glog.V(6).Info("entries: %v", entries) - nw.AppendEntries(entries...) - } - } -} - -func mockJournal(nw *Network, ticker *<-chan time.Time) { - ids := RandomNodeIDs(100) - var onNodes []*SimNode - offNodes := ids - var onConns []*SimConn - - for n := 0; ; n++ { - select { - case <-*ticker: - var entries []*Entry - ds := deltas(n) - for i := 0; len(offNodes) > 0 && i < ds[0].On; i++ { - c := rand.Intn(len(offNodes)) - sn := &SimNode{ID: offNodes[c]} - entries = append(entries, &Entry{ - Type: "Node", - Action: "On", - Object: sn, - }) - onNodes = append(onNodes, sn) - offNodes = append(offNodes[0:c], offNodes[c+1:]...) - } - for i := 0; len(onNodes) > 0 && i < ds[0].Off; i++ { - c := rand.Intn(len(onNodes)) - sn := onNodes[c] - entries = append(entries, &Entry{ - Type: "Node", - Action: "Off", - Object: sn, - }) - onNodes = append(onNodes[0:c], onNodes[c+1:]...) - offNodes = append(offNodes, sn.ID) - } - for i := 0; len(onNodes) > 1 && i < ds[1].On; i++ { - caller := onNodes[rand.Intn(len(onNodes))].ID - callee := onNodes[rand.Intn(len(onNodes))].ID - if caller == callee { - i-- - continue - } - sc := &SimConn{ - Caller: caller, - Callee: callee, - } - entries = append(entries, &Entry{ - Type: "Conn", - Action: "On", - Object: sc, - }) - onConns = append(onConns, sc) - } - for i := 0; len(onConns) > 0 && i < ds[1].Off; i++ { - c := rand.Intn(len(onConns)) - entries = append(entries, &Entry{ - Type: "Conn", - Action: "Off", - Object: onConns[c], - }) - onConns = append(onConns[0:c], onConns[c+1:]...) - } - glog.V(6).Info("entries: %v", entries) - nw.AppendEntries(entries...) - } - } - -} - func (self *Network) StartNode(conf *NodeConfig) error { self.lock.Lock() defer self.lock.Unlock() diff --git a/p2p/simulations/session_controller.go b/p2p/simulations/session_controller.go index 754110744c..6b6c802fc1 100644 --- a/p2p/simulations/session_controller.go +++ b/p2p/simulations/session_controller.go @@ -11,8 +11,8 @@ import ( "time" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger/glog" - "github.com/ethereum/go-ethereum/p2p/adapters" "github.com/ethereum/go-ethereum/p2p/discover" ) @@ -74,10 +74,17 @@ func NewSessionController() (*ResourceController, chan bool) { Create: &ResourceHandler{ Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { // TODO: take config for type of network - network := NewNetwork(&adapters.SimPipe{}) + eventer := &event.TypeMux{} ticker := time.NewTicker(1000 * time.Millisecond) - go mockJournal(network, &ticker.C) - return NewNetworkController(network, parent), nil + journal := NewJournal(eventer, &Entry{}) + + m := NewMockNetworkController(journal, func() { ticker.Stop() }) + if parent != nil { + parent.SetResource(fmt.Sprintf("%d", parent.id), m) + } + go MockNetwork(eventer, ticker.C) + parent.id++ + return m, nil }, }, diff --git a/p2p/simulations/session_controller_test.go b/p2p/simulations/session_controller_test.go index 7cc16570ea..ded566ee8e 100644 --- a/p2p/simulations/session_controller_test.go +++ b/p2p/simulations/session_controller_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p/discover" ) @@ -66,9 +67,13 @@ func TestUpdate(t *testing.T) { id := discover.MustHexID(key) ids = append(ids, &id) } - network := NewNetwork(nil) - NewNetworkController(network, controller) - Update(network, ids) + eventer := &event.TypeMux{} + journal := NewJournal(eventer, &Entry{}) + + mc := NewMockNetworkController(journal, nil) + controller.SetResource("0", mc) + mockNewNodes(eventer, ids) + journal.WaitEntries(len(ids)) r, err := (&http.Client{}).Do(req) if err != nil { t.Fatalf("unexpected error") @@ -107,15 +112,12 @@ func TestUpdate(t *testing.T) { } } -func Update(self *Network, ids []*discover.NodeID) { - self.lock.Lock() - defer self.lock.Unlock() +func mockNewNodes(eventer *event.TypeMux, ids []*discover.NodeID) { for _, id := range ids { - e := &Entry{ + eventer.Post(&Entry{ Action: "Add", Type: "Node", Object: &SimNode{ID: id, config: &NodeConfig{ID: id}}, - } - self.Journal = append(self.Journal, e) + }) } }