Fixes, updates and refactor.

Author: Strabox

Gopkg.lock

@@ -9,10 +9,10 @@ SchedulingPolicy = "binpack"
  SpreadOffersInterval = "2m30s"
  RefreshingInterval = "7m30s"
  # Debug performance flags
- SpreadOffers = true
- SpreadPartitionsState = true
+ SpreadOffers = false
+ SpreadPartitionsState = false
  PartitionsStateBufferSize = 10
- PseudoSuperPeers = true
+ PseudoSuperPeers = false
  GUIDEstimatedNetworkSize = 20000
  GUIDScaleFactor = 5
  [Caravela.DiscoveryBackend.RandomChordBackend]

configuration.toml

@@ -21,6 +21,8 @@ import (
 // engineLogTag log's tag for the simulator engine.
 const engineLogTag = "ENGINE"
 
+const numOfRandomBagsOfNode = 20
+
 // Engine represents an instance of a Caravela's simulator engine.
 // It holds all the structures to control, feed and analyse a engine during a simulation.
 type Engine struct {
@@ -54,7 +56,7 @@ func NewEngine(metricsCollector *metrics.Collector, simConfig *configuration.Con
 nodes: make([]*caravelaNode.Node, simConfig.TotalNumberOfNodes()),
 overlayMock: nil,
 feeder: feeder.Create(simConfig, caravelaConfigurations, baseRngSeed),
-nodesBags: make([][]*caravelaNode.Node, 10),
+nodesBags: make([][]*caravelaNode.Node, numOfRandomBagsOfNode),
 
 metricsCollector: metricsCollector,
 workersPool: grpool.NewPool(maxWorkers, maxWorkers*30),
@@ -107,7 +109,8 @@ func (e *Engine) Init() {
 // Initialize metric's collector.
 maxNodesResources := make([]types.Resources, e.simulatorConfigs.TotalNumberOfNodes())
 for i := range maxNodesResources {
-maxNodesResources[i] = e.nodes[i].MaximumResourcesSim()
+_, nodeMaxResources, _ := e.nodes[i].NodeInformationSim()
+maxNodesResources[i] = nodeMaxResources
 }
 e.metricsCollector.InitNewSimulation(e.caravelaConfigs.DiscoveryBackend(), maxNodesResources)
 
@@ -122,7 +125,7 @@ func (e *Engine) Init() {
 allNodes[i] = i
 }
 for i := range e.nodesBags {
-e.nodesBags[i] = make([]*caravelaNode.Node, e.simulatorConfigs.TotalNumberOfNodes()/10)
+e.nodesBags[i] = make([]*caravelaNode.Node, e.simulatorConfigs.TotalNumberOfNodes()/numOfRandomBagsOfNode)
 for j := range e.nodesBags[i] {
 randIndex := util.RandomInteger(0, len(allNodes)-1)
 e.nodesBags[i][j] = e.nodes[allNodes[randIndex]]
@@ -136,7 +139,7 @@ func (e *Engine) Init() {
 
 // Start starts the simulator engine.
 func (e *Engine) Start() {
-const ticksPerPersist = 5
+const ticksPerSnapshot = 5
 
 if !e.isInit {
 panic(errors.New("simulator is not initialized"))
@@ -175,11 +178,9 @@ func (e *Engine) Start() {
 break
 }
 
-if numTicks != 0 && (numTicks%ticksPerPersist) == 0 {
+if numTicks != 0 && (numTicks%ticksPerSnapshot) == 0 {
 e.metricsCollector.Persist(simCurrentTime)
-continue
 }
-e.metricsCollector.CreateNewGlobalSnapshot(simCurrentTime)
 }
 
 close(ticksChan) // Alert feeder that the engine has ended.
@@ -266,7 +267,7 @@ func (e *Engine) fireTimerActions(currentTime time.Duration, lastTimeRefreshes,
 return lastTimeRefreshes, lastTimeSpreadOffers
 }
 
-// updateMetrics updates all the collector metrics.
+// updateMetrics updates all the collector's metrics.
 func (e *Engine) updateMetrics() {
 defer e.workersPool.WaitAll()
 
@@ -277,10 +278,9 @@ func (e *Engine) updateMetrics() {
 e.workersPool.WaitCount(1)
 e.workersPool.JobQueue <- func() {
 defer e.workersPool.JobDone()
-nodeFreeResources := tempNode.AvailableResourcesSim()
-nodeMaxResources := tempNode.MaximumResourcesSim()
+nodeFreeResources, nodeMaxResources, numActiveOffers := tempNode.NodeInformationSim()
 e.assertNodeState(nodeFreeResources, nodeMaxResources)
-e.metricsCollector.SetAvailableNodeResources(tempI, nodeFreeResources)
+e.metricsCollector.SetNodeInformation(tempI, nodeFreeResources, numActiveOffers)
 }
 }
 }
@@ -336,20 +336,16 @@ func (e *Engine) nextRngSeed() int64 {
 
 func (e *Engine) assertNodeState(freeResources, maximumResources types.Resources) {
 if freeResources.CPUs < 0 {
-util.Log.Error("ASDDDDDDDDDDDDDD")
 panic(errors.New("negative free CPUs"))
 }
 if freeResources.Memory < 0 {
-util.Log.Error("ASDDDDasdasdDDDDDDDDDD")
 panic(errors.New("negative free Memory"))
 }
 
 if freeResources.CPUs > maximumResources.CPUs {
-util.Log.Error("ASDDDDDDDDDDDDDDasdasdadasd")
 panic(errors.New("over free CPUs"))
 }
 if freeResources.Memory > maximumResources.Memory {
-util.Log.Error("ASDDDDDDDDDDDDDDjkadlkasdkadslk")
 panic(errors.New("over free Memory"))
 }
 }

engine/engine.go

@@ -84,6 +84,12 @@ func (coll *Collector) InitNewSimulation(simLabel string, nodesMaxRes []types.Re
 
 // ================================= Metrics Collector Methods ====================================
 
+func (coll *Collector) IncrChordMessages(amount int64) {
+if activeGlobal, err := coll.activeGlobal(); err == nil {
+activeGlobal.IncrChordMessages(amount)
+}
+}
+
 // GetOfferRelayed increment the number of messages traded from type GetOffersRelayed.
 func (coll *Collector) GetOfferRelayed(amount int64) {
 if activeGlobal, err := coll.activeGlobal(); err == nil {
@@ -111,10 +117,10 @@ func (coll *Collector) APIRequestReceived(nodeIndex int) {
 }
 }
 
-// SetAvailableNodeResources sets the available resources of a node.
-func (coll *Collector) SetAvailableNodeResources(nodeIndex int, res types.Resources) {
+// SetNodeInformation sets the available resources of a node.
+func (coll *Collector) SetNodeInformation(nodeIndex int, freeResources types.Resources, numActiveOffers int) {
 if activeGlobal, err := coll.activeGlobal(); err == nil {
-activeGlobal.SetAvailableNodeResources(nodeIndex, res)
+activeGlobal.SetAvailableNodeResources(nodeIndex, freeResources)
 }
 }
 
@@ -288,7 +294,7 @@ func (coll *Collector) plotGraphics() {
 
 goroutinePool.WaitCount(1)
 goroutinePool.JobQueue <- func() {
-coll.plotMessagesTraderByRequestBoxPlots()
+coll.plotMessagesExchangedByRequestBoxPlots()
 goroutinePool.JobDone()
 }
 

engine/metrics/collector.go

@@ -4,6 +4,7 @@ import (
 "errors"
 "fmt"
 "github.com/strabox/caravela-sim/engine/metrics/graphics"
+"github.com/strabox/caravela-sim/util"
 "gonum.org/v1/gonum/mat"
 "gonum.org/v1/plot/palette"
 "gonum.org/v1/plot/plotter"
@@ -96,24 +97,24 @@ func (coll *Collector) plotSystemFreeResourcesVSRequestSuccess() {
 }
 }
 
-func (coll *Collector) plotMessagesTraderByRequestBoxPlots() {
-plotRes := graphics.NewPlot("Lookup Messages Traded Distribution", "", "#Messages", false)
-
-dataPoints := make([]interface{}, 0)
+func (coll *Collector) plotMessagesExchangedByRequestBoxPlots() {
 for _, simData := range coll.simulations {
-boxPoints := make(plotter.Values, 0)
+plotRes := graphics.NewPlot(fmt.Sprintf("Lookup Messages Traded Distribution (%s)", simData.label), "", "#Messages", false)
+
+dataPoints := make([]interface{}, 0)
+boxPlotPoints := make(plotter.Values, 0)
 for _, snapshot := range simData.snapshots {
-boxPoints = append(boxPoints, snapshot.MessagesExchangedByRequest()...)
+boxPlotPoints = append(boxPlotPoints, snapshot.MessagesExchangedByRequest()...)
+dataPoints = append(dataPoints, util.FmtDuration(snapshot.EndTime()), boxPlotPoints)
 }
-dataPoints = append(dataPoints, simData.label, boxPoints)
-}
 
-err := plotutil.AddBoxPlots(plotRes, vg.Points(55), dataPoints...)
-if err != nil {
-panic(errors.New("Problem with plots, error: " + err.Error()))
-}
+err := plotutil.AddBoxPlots(plotRes, vg.Points(20), dataPoints...)
+if err != nil {
+panic(errors.New("Problem with plots, error: " + err.Error()))
+}
 
-graphics.Save(plotRes, 15*vg.Centimeter, 15*vg.Centimeter, generatePNGFileName(coll.outputDirPath, "MessagesDistribution"))
+graphics.Save(plotRes, 35*vg.Centimeter, 12*vg.Centimeter, generatePNGFileName(coll.outputDirPath, "MessagesDistribution_"+simData.label))
+}
 }
 
 func (coll *Collector) plotResourcesUsedDistributionByNodesOverTime() {
@@ -134,8 +135,9 @@ func (coll *Collector) plotResourcesUsedDistributionByNodesOverTime() {
 
 dataGrid := &graphics.UnitGrid{Data: mat.NewDense(len(simData.snapshots), coll.numNodes, data)}
 
-graphics.NewHeatMap(generatePNGFileName("ResourcesUsedDistribution_"+simData.label), fmt.Sprintf("Resources Used Distribution (%s)", simData.label),
-"Nodes", "Time (Seconds)", coll.outputDirPath, yTicks, dataGrid, palette.Heat(64, 1))
+graphics.NewHeatMap(generatePNGFileName(coll.outputDirPath, "ResourcesUsedDistribution_"+simData.label),
+fmt.Sprintf("Resources Used Distribution (%s)", simData.label),
+"Nodes", "Time (Seconds)", yTicks, dataGrid, palette.Heat(64, 1))
 }
 }
 
@@ -157,8 +159,9 @@ func (coll *Collector) plotResourcesUnreachableDistributionByNodesOverTime() {
 
 dataGrid := &graphics.UnitGrid{Data: mat.NewDense(len(simData.snapshots), coll.numNodes, data)}
 
-graphics.NewHeatMap(generatePNGFileName("ResourcesUnreachableDistribution_"+simData.label), fmt.Sprintf("Resources Unreachable Distribution (%s)", simData.label),
-"Nodes", "Time (Seconds)", coll.outputDirPath, yTicks, dataGrid, palette.Heat(64, 1))
+graphics.NewHeatMap(generatePNGFileName(coll.outputDirPath, "ResourcesUnreachableDistribution_"+simData.label),
+fmt.Sprintf("Resources Unreachable Distribution (%s)", simData.label),
+"Nodes", "Time (Seconds)", yTicks, dataGrid, palette.Heat(64, 1))
 }
 }
 
@@ -180,8 +183,9 @@ func (coll *Collector) plotMessagesAPIDistributionByNodesOverTime() {
 
 dataGrid := &graphics.UnitGrid{Data: mat.NewDense(len(simData.snapshots), coll.numNodes, data)}
 
-graphics.NewHeatMap(generatePNGFileName("MessagesAPIDistribution_"+simData.label), fmt.Sprintf("Messages Distribution (%s)", simData.label),
-"Nodes", "Time (Seconds)", coll.outputDirPath, yTicks, dataGrid, palette.Heat(64, 1))
+graphics.NewHeatMap(generatePNGFileName(coll.outputDirPath, "MessagesAPIDistribution_"+simData.label),
+fmt.Sprintf("Messages Distribution (%s)", simData.label),
+"Nodes", "Time (Seconds)", yTicks, dataGrid, palette.Heat(64, 1))
 }
 }
 
@@ -195,7 +199,7 @@ func (coll *Collector) plotTotalMessagesTradedInSystem() {
 for i, simData := range coll.simulations {
 messagesTradedAcc := float64(0)
 for _, snapshot := range simData.snapshots {
-messagesTradedAcc += snapshot.TotalAPIMessagesReceivedByAllNodes()
+messagesTradedAcc += snapshot.TotalMessagesReceivedByAllNodes()
 }
 
 barChart, err := plotter.NewBarChart(plotter.Values{messagesTradedAcc}, barWidth)
@@ -217,7 +221,7 @@ func (coll *Collector) plotTotalMessagesTradedInSystem() {
 plotRes.Legend.Top = true
 plotRes.NominalX("Main Simulation")
 
-graphics.Save(plotRes, 15*vg.Centimeter, 15*vg.Centimeter, generatePNGFileName(coll.outputDirPath, "MessagesTradedInSystem"))
+graphics.Save(plotRes, 10*vg.Centimeter, 12*vg.Centimeter, generatePNGFileName(coll.outputDirPath, "MessagesTradedInSystem"))
 }
 
 // ======================================= Debug Performance Plots ===============================

engine/metrics/collector_plots.go

@@ -1,7 +1,6 @@
 package metrics
 
 import (
-"github.com/strabox/caravela-sim/util"
 "github.com/strabox/caravela/api/types"
 "sync"
 "sync/atomic"
@@ -15,6 +14,7 @@ type Global struct {
 
 RunRequestsSucceeded int64 `json:"RunRequestsSucceeded"` // Number of run requests that were successful deployed.
 NodesMetrics []Node `json:"NodesMetrics"` // Metrics collected for each system's node.
+ChordMessagesTraded int64 `json:"ChordMessagesTraded"`
 
 RunRequestsAggregator sync.Map `json:"-"`
 RunRequestsCompleted []RunRequest `json:"RunRequestsCompleted"`
@@ -69,6 +69,10 @@ func NewGlobalNext(numNodes int, prevGlobal *Global) *Global {
 
 // ========================= Metrics Collector Methods ====================================
 
+func (g *Global) IncrChordMessages(amount int64) {
+atomic.AddInt64(&g.ChordMessagesTraded, amount)
+}
+
 func (g *Global) GetOfferRelayed(amount int64) {
 atomic.AddInt64(&g.GetOffersRelayed, amount)
 }
@@ -166,10 +170,6 @@ func (g *Global) ResourcesUsedNodeRatio() []float64 {
 res := make([]float64, len(g.NodesMetrics))
 for i, nodeMetric := range g.NodesMetrics {
 res[i] = nodeMetric.UsedResourcesRatio()
-if res[i] < 0 {
-util.Log.Errorf("NodeIndex: %d, Max: <%d;%d>, Free: <%d;%d>, Used: <%d;%d>", i, nodeMetric.MaxResources.CPUs, nodeMetric.MaxResources.Memory,
-nodeMetric.FreeRes.CPUs, nodeMetric.FreeRes.Memory, nodeMetric.UsedResources().CPUs, nodeMetric.UsedResources().Memory)
-}
 }
 return res
 }
@@ -190,6 +190,15 @@ func (g *Global) TotalAPIMessagesReceivedByAllNodes() float64 {
 return float64(acc)
 }
 
+func (g *Global) TotalMessagesReceivedByAllNodes() float64 {
+acc := int64(0)
+for _, nodeMetric := range g.NodesMetrics {
+acc += nodeMetric.TotalAPIRequestsReceived()
+}
+acc += g.ChordMessagesTraded
+return float64(acc)
+}
+
 // ================================= Getters and Setters =================================
 
 func (g *Global) StartTime() time.Duration {
@@ -204,6 +213,10 @@ func (g *Global) SetEndTime(endTime time.Duration) {
 g.End = endTime
 }
 
+func (g *Global) TotalChordMessages() int64 {
+return g.ChordMessagesTraded
+}
+
 func (g *Global) TotalGetOffersRelayed() int64 {
 return g.GetOffersRelayed
 }

engine/metrics/global.go

@@ -13,10 +13,9 @@ import (
 "log"
 "math"
 "os"
-"path/filepath"
 )
 
-func NewHeatMap(fileName, title, XLabel, YLabel, outputDirPath string, yTicks []int, grid *UnitGrid, colorPalette palette.Palette) {
+func NewHeatMap(fileName, title, XLabel, YLabel string, yTicks []int, grid *UnitGrid, colorPalette palette.Palette) {
 heatMap := plotter.NewHeatMap(grid, colorPalette)
 
 plotRes := NewPlot(title, XLabel, YLabel, false)
@@ -50,7 +49,7 @@ func NewHeatMap(fileName, title, XLabel, YLabel, outputDirPath string, yTicks []
 plotRes.X.Max = 1.5
 plotRes.Y.Max = 1.5
 
-img := vgimg.New(2550, 1640)
+img := vgimg.New(1850, 940)
 dc := draw.New(img)
 
 legend.Top = true
@@ -63,7 +62,7 @@ func NewHeatMap(fileName, title, XLabel, YLabel, outputDirPath string, yTicks []
 dc = draw.Crop(dc, 0, -legendWidth-vg.Millimeter, 0, 0) // Make space for the legend.
 plotRes.Draw(dc)
 
-w, err := os.Create(filepath.Join(outputDirPath, fileName))
+w, err := os.Create(fileName)
 if err != nil {
 log.Panic(err)
 }