Adapt code to ar-row autopeering and newest version of hive.go

config.json

@@ -1,7 +1,7 @@
 {
   "NumberNodes": 100,
   "Duration": 100,
-  "SaltLifetime": 3600,
+  "ArrowLifetime": 3600,
   "VisualEnabled": true,
   "DropOnUpdate": false
 }

go.mod

@@ -8,5 +8,7 @@ require (
 	github.com/gorilla/websocket v1.4.2
 	github.com/iotaledger/hive.go v0.0.0-20200424160103-9d9bfc1fe24f
 	github.com/spf13/viper v1.6.3
-	github.com/stretchr/testify v1.5.1
+	github.com/stretchr/testify v1.6.1
 )
+
+replace github.com/iotaledger/hive.go => ../hive.go

main.go

@@ -1,6 +1,8 @@
 package main
 
 import (
+	"github.com/iotaledger/hive.go/configuration"
+	"github.com/iotaledger/hive.go/events"
 	"log"
 	"math/rand"
 	"net"
@@ -13,10 +15,8 @@ import (
 	"github.com/iotaledger/autopeering-sim/simulation/visualizer"
 	"github.com/iotaledger/hive.go/autopeering/peer"
 	"github.com/iotaledger/hive.go/autopeering/selection"
-	"github.com/iotaledger/hive.go/events"
 	"github.com/iotaledger/hive.go/identity"
 	"github.com/iotaledger/hive.go/logger"
-	"github.com/spf13/viper"
 )
 
 var (
@@ -46,46 +46,53 @@ func getAllPeers() []*peer.Peer {
 	return result
 }
 
-func runSim() {
+func runSim(configuration *configuration.Configuration) {
 	log.Println("Run simulation with the following parameters:")
-	config.PrintConfig()
+	configuration.Print()
 
 	selection.SetParameters(selection.Parameters{
-		SaltLifetime:           config.SaltLifetime(),
+		ArRowLifetime:          config.ArrowLifetime(configuration),
 		OutboundUpdateInterval: 200 * time.Millisecond, // use exactly the same update time as previously
 	})
 
-	closure := events.NewClosure(func(ev *selection.PeeringEvent) {
-		if ev.Status {
-			log.Printf("Peering: %s<->%s\n", ev.Self.String(), ev.Peer.ID())
-		}
-	})
-	selection.Events.OutgoingPeering.Attach(closure)
-	defer selection.Events.OutgoingPeering.Detach(closure)
-
-	//lambda := (float64(N) / SaltLifetime.Seconds()) * 10
+	//lambda := (float64(N) / ArrowLifetime.Seconds()) * 10
 	initialSalt := 0.
 
 	log.Println("Creating peers ...")
 	netw := transport.NewNetwork()
-	nodeMap = make(map[identity.ID]simulation.Node, config.NumberNodes())
-	for i := 0; i < config.NumberNodes(); i++ {
-		node := simulation.NewNode(transport.PeerID(i), time.Duration(initialSalt)*time.Second, netw, config.DropOnUpdate(), discover)
+	nodeMap = make(map[identity.ID]simulation.Node, config.NumberNodes(configuration))
+	for i := 0; i < config.NumberNodes(configuration); i++ {
+		node := simulation.NewNode(transport.PeerID(i), time.Duration(initialSalt)*time.Second, netw, config.DropOnUpdate(configuration), discover)
 		nodeMap[node.ID()] = node
 
-		if config.VisEnabled() {
+		if config.VisEnabled(configuration) {
 			visualizer.AddNode(node.ID().String())
 		}
 
 		// initialSalt = initialSalt + (1 / lambda)				 // constant rate
 		// initialSalt = initialSalt + rand.ExpFloat64()/lambda  // poisson process
-		initialSalt = rand.Float64() * config.SaltLifetime().Seconds() // random
+		initialSalt = rand.Float64() * config.ArrowLifetime(configuration).Seconds() // random
+		inClosure := events.NewClosure(func(ev *selection.PeeringEvent) {
+			if ev.Status {
+				log.Printf("Accepted peering: %s<->%s (chan: %d)\n", node.Peer().ID(), ev.Peer.ID(), ev.Channel)
+			} else {
+				log.Printf("Rejected peering: %s<->%s (chan: %d)\n", node.Peer().ID(), ev.Peer.ID(), ev.Channel)
+			}
+		})
+
+		node.Prot.Events().IncomingPeering.Attach(inClosure)
+
+		dropClosure := events.NewClosure(func(ev *selection.DroppedEvent) {
+			log.Printf("Dropping: %s<->%s\n", node.Peer().ID(), ev.DroppedID)
+		})
+		node.Prot.Events().Dropped.Attach(dropClosure)
 	}
+
 	log.Println("Creating peers ... done")
 
-	analysis := simulation.NewLinkAnalysis(nodeMap)
+	analysis := simulation.NewLinkAnalysis(nodeMap, configuration)
 
-	if config.VisEnabled() {
+	if config.VisEnabled(configuration) {
 		statVisualizer()
 	}
 
@@ -97,14 +104,14 @@ func runSim() {
 	log.Println("Starting peering ... done")
 
 	log.Println("Running ...")
-	time.Sleep(config.Duration())
+	time.Sleep(config.Duration(configuration))
 
 	log.Println("Stopping peering  ...")
 	for _, node := range nodeMap {
 		node.Stop()
 	}
 	analysis.Stop()
-	if config.VisEnabled() {
+	if config.VisEnabled(configuration) {
 		stopServer()
 	}
 	log.Println("Stopping peering ... done")
@@ -123,7 +130,7 @@ func runSim() {
 		log.Fatalln("error writing csv:", err)
 	}
 
-	msgAnalysis := simulation.MessagesToString(nodeMap, analysis.Status())
+	msgAnalysis := simulation.MessagesToString(nodeMap, analysis.Status(), configuration)
 	err = simulation.WriteCSV(msgAnalysis, "msgAnalysis", []string{"ID", "OUT", "ACC", "REJ", "IN", "DROP"})
 	if err != nil {
 		log.Fatalln("error writing csv:", err)
@@ -136,14 +143,16 @@ func runSim() {
 }
 
 func main() {
-	config.Load()
-	if err := logger.InitGlobalLogger(viper.GetViper()); err != nil {
+	config := configuration.New()
+	_ = config.LoadFile("config.json")
+
+	if err := logger.InitGlobalLogger(config); err != nil {
 		panic(err)
 	}
-	if config.VisEnabled() {
+	if config.Bool("VisualEnabled") {
 		startServer()
 	}
-	runSim()
+	runSim(config)
 }
 
 func startServer() {

simulation/config/config.go

@@ -1,67 +1,35 @@
 package config
 
 import (
-	"encoding/json"
-	"fmt"
-	"log"
+	"github.com/iotaledger/hive.go/configuration"
 	"time"
-
-	"github.com/spf13/viper"
 )
 
 // Config keys
 const (
-	numberNodes  = "NumberNodes"
-	duration     = "Duration"
-	saltLifetime = "SaltLifetime"
-	vEnabled     = "VisualEnabled"
-	dropOnUpdate = "DropOnUpdate"
+	numberNodes   = "NumberNodes"
+	duration      = "Duration"
+	arrowLifetime = "ArrowLifetime"
+	vEnabled      = "VisualEnabled"
+	dropOnUpdate  = "DropOnUpdate"
 )
 
-func init() {
-	viper.SetDefault(numberNodes, 100)
-	viper.SetDefault(duration, 60)
-	viper.SetDefault(saltLifetime, 60*60)
-	viper.SetDefault(vEnabled, false)
-	viper.SetDefault(dropOnUpdate, false)
-}
-
-func Load() {
-	viper.SetConfigName("config")
-	viper.AddConfigPath(".")
-
-	if err := viper.ReadInConfig(); err != nil {
-		if _, ok := err.(viper.ConfigFileNotFoundError); ok {
-			log.Println("Using default config")
-		} else {
-			log.Fatal(err)
-		}
-	}
-}
-
-func PrintConfig() {
-	settings := viper.AllSettings()
-	if cfg, err := json.MarshalIndent(settings, "", "  "); err == nil {
-		fmt.Println(string(cfg))
-	}
-}
-
-func NumberNodes() int {
-	return viper.GetInt(numberNodes)
+func NumberNodes(config *configuration.Configuration) int {
+	return config.Int(numberNodes)
 }
 
-func Duration() time.Duration {
-	return time.Duration(viper.GetInt(duration)) * time.Second
+func Duration(config *configuration.Configuration) time.Duration {
+	return time.Duration(config.Int(duration)) * time.Second
 }
 
-func SaltLifetime() time.Duration {
-	return time.Duration(viper.GetInt(saltLifetime)) * time.Second
+func ArrowLifetime(config *configuration.Configuration) time.Duration {
+	return time.Duration(config.Int(arrowLifetime)) * time.Second
 }
 
-func DropOnUpdate() bool {
-	return viper.GetBool(dropOnUpdate)
+func DropOnUpdate(config *configuration.Configuration) bool {
+	return config.Bool(dropOnUpdate)
 }
 
-func VisEnabled() bool {
-	return viper.GetBool(vEnabled)
+func VisEnabled(config *configuration.Configuration) bool {
+	return config.Bool(vEnabled)
 }

simulation/linkanalysis.go

@@ -1,6 +1,7 @@
 package simulation
 
 import (
+	"github.com/iotaledger/hive.go/configuration"
 	"sync"
 	"time"
 
@@ -11,28 +12,38 @@ import (
 	"github.com/iotaledger/hive.go/identity"
 )
 
-type linkAnalysis struct {
-	nodeMap map[identity.ID]Node
+type DroppedEventSim struct {
+	selection.DroppedEvent
+	Self identity.ID
+}
+type PeeringEventSim struct {
+	selection.PeeringEvent
+	Self identity.ID
+}
 
-	startTime    time.Time
-	incomingChan chan *selection.PeeringEvent
-	outgoingChan chan *selection.PeeringEvent
-	dropChan     chan *selection.DroppedEvent
-	status       *StatusMap
-	links        []Link
+type linkAnalysis struct {
+	nodeMap       map[identity.ID]Node
+	configuration *configuration.Configuration
+	startTime     time.Time
+	incomingChan  chan *PeeringEventSim
+	outgoingChan  chan *PeeringEventSim
+	dropChan      chan *DroppedEventSim
+	status        *StatusMap
+	links         []Link
 
 	closing chan struct{}
 	wg      sync.WaitGroup
 }
 
-func NewLinkAnalysis(nodeMap map[identity.ID]Node) *linkAnalysis {
+func NewLinkAnalysis(nodeMap map[identity.ID]Node, c *configuration.Configuration) *linkAnalysis {
 	return &linkAnalysis{
-		nodeMap:      nodeMap,
-		incomingChan: make(chan *selection.PeeringEvent, 10),
-		outgoingChan: make(chan *selection.PeeringEvent, 10),
-		dropChan:     make(chan *selection.DroppedEvent, 10),
-		status:       NewStatusMap(),
-		closing:      make(chan struct{}),
+		configuration: c,
+		nodeMap:       nodeMap,
+		incomingChan:  make(chan *PeeringEventSim, 10),
+		outgoingChan:  make(chan *PeeringEventSim, 10),
+		dropChan:      make(chan *DroppedEventSim, 10),
+		status:        NewStatusMap(),
+		closing:       make(chan struct{}),
 	}
 }
 
@@ -70,19 +81,19 @@ func (la *linkAnalysis) loop() {
 
 		// handle incoming peering requests
 		case req := <-la.incomingChan:
-			la.status.Append(req.Peer.ID(), req.Self, INCOMING)
+			la.status.Append(req.PeeringEvent.Peer.ID(), req.Self, INCOMING)
 
 		// handle outgoing peering requests
 		case req := <-la.outgoingChan:
 			from := req.Self
-			to := req.Peer.ID()
+			to := req.PeeringEvent.Peer.ID()
 			la.status.Append(from, to, OUTBOUND)
 
 			// accepted/rejected is only recorded for outgoing requests
 			if req.Status {
 				la.status.Append(from, to, ACCEPTED)
 				la.links = append(la.links, NewLink(from, to, time.Since(la.startTime).Milliseconds()))
-				if config.VisEnabled() {
+				if config.VisEnabled(la.configuration) {
 					visualizer.AddLink(req.Self.String(), req.Peer.ID().String())
 				}
 			} else {
@@ -95,7 +106,7 @@ func (la *linkAnalysis) loop() {
 			to := req.DroppedID
 			la.status.Append(from, to, DROPPED)
 			DropLink(from, to, time.Since(la.startTime).Milliseconds(), la.links)
-			if config.VisEnabled() {
+			if config.VisEnabled(la.configuration) {
 				visualizer.RemoveLink(req.Self.String(), req.DroppedID.String())
 			}
 
@@ -110,18 +121,26 @@ func (la *linkAnalysis) loop() {
 
 // subscribe subscribes to the selection events.
 func (la *linkAnalysis) subscribe() func() {
-	incomingClosure := events.NewClosure(func(e *selection.PeeringEvent) { la.incomingChan <- e })
-	outgoingClosure := events.NewClosure(func(e *selection.PeeringEvent) { la.outgoingChan <- e })
-	dropClosure := events.NewClosure(func(e *selection.DroppedEvent) { la.dropChan <- e })
 
-	selection.Events.IncomingPeering.Attach(incomingClosure)
-	selection.Events.OutgoingPeering.Attach(outgoingClosure)
-	selection.Events.Dropped.Attach(dropClosure)
+	for _, node := range la.nodeMap {
+		id := node.ID()
+		incomingClosure := events.NewClosure(func(e *selection.PeeringEvent) { la.incomingChan <- &PeeringEventSim{Self: id, PeeringEvent: *e} })
+		outgoingClosure := events.NewClosure(func(e *selection.PeeringEvent) { la.outgoingChan <- &PeeringEventSim{Self: id, PeeringEvent: *e} })
+		dropClosure := events.NewClosure(func(e *selection.DroppedEvent) { la.dropChan <- &DroppedEventSim{Self: id, DroppedEvent: *e} })
+
+		node.Prot.Events().IncomingPeering.Attach(incomingClosure)
+		node.Prot.Events().OutgoingPeering.Attach(outgoingClosure)
+		node.Prot.Events().Dropped.Attach(dropClosure)
+
+	}
 
 	return func() {
-		selection.Events.IncomingPeering.Detach(incomingClosure)
-		selection.Events.OutgoingPeering.Detach(outgoingClosure)
-		selection.Events.Dropped.Detach(dropClosure)
+		for _, node := range la.nodeMap {
+			node.Prot.Events().IncomingPeering.DetachAll()
+			node.Prot.Events().OutgoingPeering.DetachAll()
+			node.Prot.Events().Dropped.DetachAll()
+
+		}
 	}
 }
 
@@ -135,7 +154,7 @@ func (la *linkAnalysis) updateConvergence(time time.Duration) {
 		}
 		avgNeighbors += l
 	}
-	c := (float64(counter) / float64(config.NumberNodes())) * 100
-	avg := float64(avgNeighbors) / float64(config.NumberNodes())
+	c := (float64(counter) / float64(config.NumberNodes(la.configuration))) * 100
+	avg := float64(avgNeighbors) / float64(config.NumberNodes(la.configuration))
 	RecordConv.Append(Convergence{time, c, avg})
 }