From faba395041210ca6bbaf7257497cc2a6e6c7e908 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 | 98 +++++ p2p/simulations/journal.go | 426 +++++++++++++++++++++ p2p/simulations/journal_test.go | 72 ++++ p2p/simulations/network.go | 330 +++------------- p2p/simulations/session_controller.go | 23 +- p2p/simulations/session_controller_test.go | 44 ++- 6 files changed, 693 insertions(+), 300 deletions(-) create mode 100644 p2p/simulations/cytoscape.go create mode 100644 p2p/simulations/journal.go create mode 100644 p2p/simulations/journal_test.go diff --git a/p2p/simulations/cytoscape.go b/p2p/simulations/cytoscape.go new file mode 100644 index 0000000000..cdbb087e38 --- /dev/null +++ b/p2p/simulations/cytoscape.go @@ -0,0 +1,98 @@ +package simulations + +import ( + "bytes" + "fmt" + + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/logger/glog" +) + +// TODO: to implement cytoscape global behav +type CyConfig struct { +} + +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) +} + +type CyUpdate struct { + Add []*CyElement `json:"add"` + Remove []string `json:"remove"` +} + +func UpdateCy(conf *CyConfig, j *Journal) (*CyUpdate, error) { + 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, + }, nil +} diff --git a/p2p/simulations/journal.go b/p2p/simulations/journal.go new file mode 100644 index 0000000000..8bf54343c7 --- /dev/null +++ b/p2p/simulations/journal.go @@ -0,0 +1,426 @@ +package simulations + +import ( + "bytes" + "encoding/json" + "fmt" + "math/rand" + "reflect" + "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 { + Id string + lock sync.Mutex + counter int + cursor int + quitc chan bool + Events []*event.Event +} + +// NewJournal constructor +// Journal can get input events from subscriptions, add event logs +// or scheduled replay of events from another journal +// +// see the Read and TimedRead iterators for use +// the Journal is safe for concurrent reads and writes +func NewJournal() *Journal { + return &Journal{quitc: make(chan bool)} +} + +// Subscribe takes an event.TypeMux and subscibes to types +// and launches a gorourine that appends any new event to the event log +// used for journalling history of a network +// the goroutine terminates when the journal is closed +func (self *Journal) Subscribe(eventer *event.TypeMux, types ...interface{}) { + glog.V(6).Infof("subscribe") + sub := eventer.Subscribe(types...) + go func() { + defer sub.Unsubscribe() + for { + select { + case ev := <-sub.Chan(): + glog.V(6).Infof("appebd ev %v", ev) + self.append(ev) + case <-self.quitc: + return + } + } + }() +} + +// AddJournal appends the event log of another journal to the receiver's one +func (self *Journal) AddJournal(j *Journal) { + self.append(j.Events...) +} + +// NewJournalFromJSON decodes a JSON serialised events log +// into a journal struct +// used to replay recorded history +func NewJournalFromJSON(b []byte) (*Journal, error) { + self := NewJournal() + err := json.Unmarshal(b, self) + if err != nil { + return nil, err + } + return self, nil +} + +// Replay replays the events of another journal preserving (relative) timing of events +// params: +// * acc: using acceleration factor acc +// * journal: journal to use +// * eventer: where to post the replayed events +func Replay(acc float64, j *Journal, eventer *event.TypeMux) { + f := func(d interface{}) bool { + // reposts the data with the eventer (the data receives a new timestamp) + eventer.Post(d) + return true + } + j.TimedRead(acc, f) +} + +// Snapshot creates a snapshot out of the journal +// this is simply done by reading the event log backwards and mark the last action +// on a node/connection ignoring all earlier mentions +// TODO: implmented +func Snapshot(conf *SnapshotConfig, j *Journal) (*Journal, error) { + return nil, fmt.Errorf("snapshot not implemented") +} + +func (self *Journal) Close() { + close(self.quitc) +} + +func (self *Journal) append(evs ...*event.Event) { + self.lock.Lock() + defer self.lock.Unlock() + self.Events = append(self.Events, evs...) + self.counter++ +} + +func (self *Journal) NewEntries() int { + self.lock.Lock() + defer self.lock.Unlock() + return self.counter - self.cursor +} + +func (self *Journal) WaitEntries(n int) { + for self.NewEntries() < n { + glog.V(6).Infof(".") + } +} + +func (self *Journal) Read(f func(*event.Event) bool) (read int, err error) { + self.lock.Lock() + defer self.lock.Unlock() + glog.V(6).Infof("read out of %v", len(self.Events)) + ok := true + for self.cursor < len(self.Events) && ok { + glog.V(6).Infof("read %v", read) + read++ + ok = f(self.Events[self.cursor]) + self.cursor++ + select { + case <-self.quitc: + break + default: + } + } + self.reset(self.cursor) + return read, nil +} + +// TimedRead reads the events but blocks for intervals that correspond to +// the original time intervals, +// NOTE: the events' timestamps are supposed to be strictly ordered otherwise +// the call panics. +// acc is an acceleration factor +func (self *Journal) TimedRead(acc float64, f func(interface{}) bool) (read int, err error) { + var lastEvent time.Time + timer := time.NewTimer(0) + var data interface{} + h := func(ev *event.Event) bool { + // wait for the interval time passes event time + if ev.Time.Before(lastEvent) { + panic("events not ordered") + } + interval := ev.Time.Sub(lastEvent) + glog.V(6).Infof("reset timer to interval %v", interval) + timer.Reset(time.Duration(acc) * interval) + lastEvent = ev.Time + data = ev.Data + return false + } + var n int + for { + // Read blocks for the iteration. need to read one event at a time so that + // waiting for the timer to go off does not block concurrent access to the journal + n, err = self.Read(h) + if read > 0 { + select { + case <-self.quitc: + break + case <-timer.C: + } + } + read += n + if err != nil || n == 0 || !f(data) { + break + } + } + return read, err +} + +func (self *Journal) Reset(n int) { + self.lock.Lock() + defer self.lock.Unlock() + self.reset(n) +} + +func (self *Journal) reset(n int) { + length := len(self.Events) + if length == 0 { + return + } + if n >= length-1 { + n = length - 1 + } + glog.V(6).Infof("cursor reset from %v to %v/%v (%v)", self.cursor, n, len(self.Events), 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 +} + +type SnapshotConfig struct { + Id string +} + +type JournalPlayConfig struct { + Id string + SpeedUp float64 + Journal *Journal + Events []string +} + +func NewJournalPlayersController(eventer *event.TypeMux) Controller { + self := NewResourceContoller( + &ResourceHandlers{ + // POST /o/players/ + Create: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + conf := msg.(*JournalPlayConfig) + go Replay(conf.SpeedUp, conf.Journal, eventer) + c := NewJournalPlayerController(conf) + parent.SetResource(conf.Id, c) + return nil, nil + }, + Type: reflect.TypeOf(&JournalPlayConfig{}), + }, + }) + return self +} + +func NewJournalPlayerController(conf *JournalPlayConfig) Controller { + self := NewResourceContoller( + &ResourceHandlers{ + // GET /0/players/ + Retrieve: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + return nil, fmt.Errorf("info about journal player not implemented") + }, + }, + // DELETE /0/players/ + Destroy: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + conf.Journal.Close() // terminate Replay-> TimedRead routine + parent.DeleteResource(conf.Id) + return nil, nil + }, + }, + }) + return self +} + +type MockerConfig struct { + // TODO: frequency/volume etc. + Id string + 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) + ticker := time.NewTicker(conf.UpdateInterval) + go MockEvents(eventer, ticker.C) + c := NewMockerController(conf, ticker) + parent.SetResource(conf.Id, c) + return nil, 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 nil, 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 + +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)}, + } +} + +// 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, 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/journal_test.go b/p2p/simulations/journal_test.go new file mode 100644 index 0000000000..4e62ddb88d --- /dev/null +++ b/p2p/simulations/journal_test.go @@ -0,0 +1,72 @@ +package simulations + +import ( + // "encoding/json" + "testing" + "time" + + "github.com/ethereum/go-ethereum/event" + // "github.com/ethereum/go-ethereum/p2p/discover" + // "github.com/ethereum/go-ethereum/logger/glog" +) + +func testEvents(intervals ...int) (events []*event.Event) { + t := time.Now() + for i, interval := range intervals { + t = t.Add(time.Duration(interval) * time.Millisecond) + events = append(events, &event.Event{ + Time: t, + Data: interface{}(&Entry{ + Type: "node", + Action: "Off", + Object: interface{}(i), + }), + }) + } + return events +} + +func TestTimedRead(t *testing.T) { + // keys := []string{ + // "aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80", + // "f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3", + // } + // var ids []*discover.NodeID + // for _, key := range keys { + // id := discover.MustHexID(key) + // ids = append(ids, &id) + // } + + j := NewJournal() + intervals := []int{100, 200, 300, 300, 100, 200} + j.Events = testEvents(intervals...) + var newTimes []time.Time + var i int + acc := 0.5 + length := 4 + f := func(data interface{}) bool { + _ = data.(*Entry) + newTimes = append(newTimes, time.Now()) + i++ + return i <= length + } + start := time.Now() + read, err := j.TimedRead(acc, f) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if read != 5 { + t.Fatalf("incorrect number of events read: expected 5, got %v", read) + } + for i, ti := range newTimes { + expInt := time.Duration(acc*float64(intervals[i])) * time.Millisecond + gotInt := ti.Sub(start) + if gotInt-expInt > 1*time.Millisecond { + t.Fatalf("journal timed read incorrect interval: expected %v ,got %v", expInt, gotInt) + } + start = ti + } +} + +// func TestReplay(t *testing.T) { +// } diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 3836117c0c..de163029d6 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -1,12 +1,11 @@ package simulations import ( - "bytes" "fmt" - "math/rand" + "reflect" "sync" - "time" + "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" @@ -14,22 +13,48 @@ import ( const lablen = 4 -func NewNetworkController(n *Network, parent *ResourceController) Controller { +type NetworkConfig struct { + // Type NetworkType + // Config json.RawMessage // type-specific configs + // type + // Events []string + Id string +} + +func NewNetworkController(conf *NetworkConfig, eventer *event.TypeMux, journal *Journal) Controller { self := NewResourceContoller( &ResourceHandlers{ - // Destroy: n.Shutdown, nil - // Create: n.StartNode, NodeConfig - // Update: n.Setup, NodeConfig - // Retrieve: n.Retrieve, + // GET // Retrieve: &ResourceHandler{ - Handle: n.Query, + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + glog.V(6).Infof("msg: %v", msg) + cyConfig, ok := msg.(*CyConfig) + if ok { + return UpdateCy(cyConfig, journal) + } + snapshotConfig, ok := msg.(*SnapshotConfig) + if ok { + return Snapshot(snapshotConfig, journal) + } + return nil, fmt.Errorf("invalid json body: must be CyConfig or SnapshotConfig") + }, + Type: reflect.TypeOf(&CyConfig{}), + }, + // DELETE // + Destroy: &ResourceHandler{ + Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) { + parent.DeleteResource(conf.Id) + return nil, nil + }, }, }, ) - if parent != nil { - parent.SetResource(fmt.Sprintf("%d", parent.id), self) - } - // self.SetResource("nodes", NewNodesController()) + // subscribe to all event entries (generated) + journal.Subscribe(eventer, &Entry{}) + // self.SetResource("nodes", NewNodesController(eventer)) + // self.SetResource("connections", NewConnectionsController(eventer)) + self.SetResource("mockevents", NewMockersController(eventer)) + self.SetResource("journals", NewJournalPlayersController(eventer)) return Controller(self) } @@ -40,7 +65,6 @@ type Network struct { lock sync.RWMutex NodeMap map[discover.NodeID]int Nodes []*SimNode - Journal []*Entry } func NewNetwork(m adapters.Messenger) *Network { @@ -64,15 +88,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,42 +121,9 @@ 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...) - n.lock.Unlock() -} - 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 +132,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() @@ -372,7 +146,7 @@ func (self *Network) StartNode(conf *NodeConfig) error { simnet.Run = conf.Run(simnet, self.Messenger) } self.NodeMap[*id] = len(self.Nodes) - self.Nodes = append(self.Nodes, &SimNode{id, conf, adapters.NetAdapter(simnet)}) + self.Nodes = append(self.Nodes, &SimNode{id, conf, simnet}) return nil } diff --git a/p2p/simulations/session_controller.go b/p2p/simulations/session_controller.go index 754110744c..5825864d1f 100644 --- a/p2p/simulations/session_controller.go +++ b/p2p/simulations/session_controller.go @@ -8,11 +8,10 @@ import ( "io/ioutil" "reflect" "sync" - "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" ) @@ -73,12 +72,18 @@ 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{}) - ticker := time.NewTicker(1000 * time.Millisecond) - go mockJournal(network, &ticker.C) - return NewNetworkController(network, parent), nil + conf := msg.(*NetworkConfig) + m := NewNetworkController(conf, &event.TypeMux{}, NewJournal()) + if len(conf.Id) == 0 { + conf.Id = fmt.Sprintf("%d", parent.id) + } + if parent != nil { + parent.SetResource(conf.Id, m) + } + parent.id++ + return m, nil }, + Type: reflect.TypeOf(&NetworkConfig{}), }, Destroy: &ResourceHandler{ @@ -149,6 +154,10 @@ func (self *ResourceController) SetResource(id string, c Controller) { } } +func (self *ResourceController) DeleteResource(id string) { + delete(self.controllers, id) +} + func RandomNodeID() *discover.NodeID { key, err := crypto.GenerateKey() if err != nil { diff --git a/p2p/simulations/session_controller_test.go b/p2p/simulations/session_controller_test.go index 7cc16570ea..0f9fe09a74 100644 --- a/p2p/simulations/session_controller_test.go +++ b/p2p/simulations/session_controller_test.go @@ -1,12 +1,14 @@ package simulations import ( + "bytes" "fmt" "io/ioutil" "net/http" "testing" "time" + "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p/discover" ) @@ -32,7 +34,6 @@ func url(port, path string) string { func TestQuit(t *testing.T) { req, err := http.NewRequest("DELETE", url(port, ""), nil) - // req, err := http.NewRequest("PUT", url(""), nil) if err != nil { t.Fatalf("unexpected error") } @@ -53,10 +54,7 @@ func TestQuit(t *testing.T) { } func TestUpdate(t *testing.T) { - req, err := http.NewRequest("GET", url(port, "0"), nil) - if err != nil { - t.Fatalf("unexpected error") - } + keys := []string{ "aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80", "f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3", @@ -66,12 +64,25 @@ 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() + journal.Subscribe(eventer, &Entry{}) + mockNewNodes(eventer, ids) + conf := &NetworkConfig{ + Id: "0", + } + mc := NewNetworkController(conf, eventer, journal) + controller.SetResource(conf.Id, mc) + journal.WaitEntries(len(ids)) + + req, err := http.NewRequest("GET", url(port, "0"), bytes.NewReader([]byte("{}"))) + if err != nil { + t.Fatalf("unexpected error creating request: %v", err) + } r, err := (&http.Client{}).Do(req) if err != nil { - t.Fatalf("unexpected error") + t.Fatalf("unexpected error on http.Client request: %v", err) } resp, err := ioutil.ReadAll(r.Body) if err != nil { @@ -107,15 +118,18 @@ 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) { + glog.V(6).Infof("mock starting") for _, id := range ids { - e := &Entry{ + glog.V(6).Infof("mock adding node %v", id) + eventer.Post(&Entry{ Action: "Add", Type: "Node", Object: &SimNode{ID: id, config: &NodeConfig{ID: id}}, - } - self.Journal = append(self.Journal, e) + }) } } + +// func TestReplay(t *testing.T) { + +// }