Improved the performance by removing many invalidates from FlowNodes (#377)

* Updated the UpDatedConsumer to boolean array

* Updated SimTraceWorkload to not invalidate when the next fragment is started.

* Removed as much invalidates as possible

3 files changed, 104 insertions, 45 deletions

diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/SimCpu.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/SimCpu.java
index 52fc6093..91b5eabf 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/SimCpu.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/SimCpu.java
@@ -50,6 +50,8 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
 
     private final PowerModel cpuPowerModel;
 
+    private double previousPowerDemand = 0.0f;
+
     private double currentCpuDemand = 0.0f; // cpu capacity demanded by the mux
     private double currentCpuUtilization = 0.0f;
     private double currentCpuSupplied = 0.0f; // cpu capacity supplied to the mux
@@ -228,6 +230,10 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
      */
     @Override
     public void handleIncomingDemand(FlowEdge consumerEdge, double newCpuDemand) {
+        if (newCpuDemand == this.currentCpuDemand) {
+            return;
+        }
+
         updateCounters();
         this.currentCpuDemand = newCpuDemand;
 
@@ -236,7 +242,16 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
         // Calculate Power Demand and send to PSU
         this.currentPowerDemand = this.cpuPowerModel.computePower(this.currentCpuUtilization);
 
-        this.invalidate();
+        // TODO: find a better solution for this
+        // If current Power Demand is equal to previous Power Demand, it means the CPU is overloaded and we can
+        // distribute
+        // immediately.
+        if (this.currentPowerDemand == this.previousPowerDemand) {
+            this.pushOutgoingSupply(consumerEdge, this.currentCpuSupplied);
+        } else {
+            this.previousPowerDemand = this.currentPowerDemand;
+            this.pushOutgoingDemand(this.psuEdge, this.currentPowerDemand);
+        }
     }
 
     /**
@@ -247,7 +262,9 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
         updateCounters();
         this.currentPowerSupplied = newPowerSupply;
 
-        this.invalidate();
+        this.currentCpuSupplied = Math.min(this.currentCpuDemand, this.maxCapacity);
+
+        this.pushOutgoingSupply(this.distributorEdge, this.currentCpuSupplied, ResourceType.CPU);
     }
 
     /**
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/gpu/SimGpu.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/gpu/SimGpu.java
index a5fccf6c..e0ed64d0 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/gpu/SimGpu.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/gpu/SimGpu.java
@@ -50,6 +50,8 @@ public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer
 
     private final PowerModel gpuPowerModel;
 
+    private double previousPowerDemand = 0.0f;
+
     private double currentGpuDemand = 0.0f; // gpu capacity demanded by the mux
     private double currentGpuUtilization = 0.0f;
     private double currentGpuSupplied = 0.0f; // gpu capacity supplied to the mux
@@ -239,7 +241,16 @@ public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer
         // Calculate Power Demand and send to PSU
         this.currentPowerDemand = this.gpuPowerModel.computePower(this.currentGpuUtilization);
 
-        this.invalidate();
+        // TODO: find a better solution for this
+        // If current Power Demand is equal to previous Power Demand, it means the CPU is overloaded and we can
+        // distribute
+        // immediately.
+        if (this.currentPowerDemand == this.previousPowerDemand) {
+            this.pushOutgoingSupply(consumerEdge, this.currentGpuSupplied);
+        } else {
+            this.previousPowerDemand = this.currentPowerDemand;
+            this.pushOutgoingDemand(this.psuEdge, this.currentPowerDemand);
+        }
     }
 
     /**
@@ -260,7 +271,16 @@ public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer
         // Calculate Power Demand and send to PSU
         this.currentPowerDemand = this.gpuPowerModel.computePower(this.currentGpuUtilization);
 
-        this.invalidate();
+        // TODO: find a better solution for this
+        // If current Power Demand is equal to previous Power Demand, it means the CPU is overloaded and we can
+        // distribute
+        // immediately.
+        if (this.currentPowerDemand == this.previousPowerDemand) {
+            this.pushOutgoingSupply(consumerEdge, this.currentGpuSupplied);
+        } else {
+            this.previousPowerDemand = this.currentPowerDemand;
+            this.pushOutgoingDemand(this.psuEdge, this.currentPowerDemand);
+        }
     }
 
     /**
@@ -271,7 +291,10 @@ public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer
         updateCounters();
         this.currentPowerSupplied = newPowerSupply;
 
-        this.invalidate();
+        this.currentGpuSupplied = virtualizationOverheadModel.getSupply(
+                Math.min(this.currentGpuDemand, this.maxCapacity), this.consumerCount);
+
+        this.pushOutgoingSupply(this.distributorEdge, this.currentGpuSupplied, ResourceType.CPU);
     }
 
     /**
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/SimTraceWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/SimTraceWorkload.java
index 70fe7e96..ff65fbf2 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/SimTraceWorkload.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/SimTraceWorkload.java
@@ -23,6 +23,7 @@
 package org.opendc.simulator.compute.workload.trace;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
@@ -53,8 +54,6 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
 
     private final double[] resourcesSupplied =
             new double[ResourceType.values().length]; // the currently supplied resources
-    private final double[] newResourcesSupply =
-            new double[ResourceType.values().length]; // The supplied resources with next update
     private final double[] resourcesDemand = new double[ResourceType.values().length]; // The demands per resource type
     private final double[] remainingWork =
             new double[ResourceType.values().length]; // The duration of the fragment at the demanded speeds
@@ -116,9 +115,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
         new FlowEdge(this, supplier);
         if (supplier instanceof VirtualMachine) {
             // instead iterate over the resources in the fragment as required resources not provided by the VM
-            for (ResourceType resourceType : workload.getResourceTypes()) {
-                this.usedResourceTypes.add(resourceType);
-            }
+            this.usedResourceTypes.addAll(Arrays.asList(workload.getResourceTypes()));
         }
     }
 
@@ -157,11 +154,8 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
         return true;
     }
 
-    @Override
-    public long onUpdate(long now) {
-        long passedTime = getPassedTime(now);
-        this.startOfFragment = now;
-
+    // Update the remaining work for all resources based on the time passed since last update
+    private void updateRemainingWork(long passedTime) {
         for (ResourceType resourceType : this.usedResourceTypes) {
             // The amount of work done since last update
             double finishedWork = this.scalingPolicy.getFinishedWork(
@@ -172,34 +166,17 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
             // TODO: maybe remove Math.max, as as we are already checking for <= 0
             this.remainingWork[resourceType.ordinal()] =
                     Math.max(0, this.remainingWork[resourceType.ordinal()] - finishedWork);
+
             this.totalRemainingWork -= finishedWork;
+
             if (this.remainingWork[resourceType.ordinal()] <= 0) {
                 this.workloadFinished[resourceType.ordinal()] = true;
             }
         }
+    }
 
-        // If this.totalRemainingWork <= 0, the fragment has been completed across all resources
-        if ((int) this.totalRemainingWork <= 0 && this.isWorkloadFinished()) {
-            this.startNextFragment();
-
-            this.invalidate();
-            return Long.MAX_VALUE;
-        }
-
-        for (ResourceType resourceType : this.usedResourceTypes) {
-            if (this.machineResourceEdges[resourceType.ordinal()] != null) {
-                this.pushOutgoingDemand(
-                        this.machineResourceEdges[resourceType.ordinal()],
-                        this.resourcesDemand[resourceType.ordinal()],
-                        resourceType);
-            }
-        }
-
-        // Update the supplied resources
-        for (ResourceType resourceType : this.usedResourceTypes) {
-            this.resourcesSupplied[resourceType.ordinal()] = this.newResourcesSupply[resourceType.ordinal()];
-        }
-
+    // Determine the next update time based on the remaining work and supplied resources
+    private long getNextUpdateTime(long now) {
         long timeUntilNextUpdate = Long.MIN_VALUE;
 
         for (ResourceType resourceType : this.usedResourceTypes) {
@@ -226,7 +203,41 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
             }
         }
 
-        long nextUpdate = timeUntilNextUpdate == Long.MAX_VALUE ? Long.MAX_VALUE : now + timeUntilNextUpdate;
+        return timeUntilNextUpdate == Long.MAX_VALUE ? Long.MAX_VALUE : now + timeUntilNextUpdate;
+    }
+
+    private void pushNewDemands() {
+        for (ResourceType resourceType : this.usedResourceTypes) {
+            if (this.machineResourceEdges[resourceType.ordinal()] != null) {
+                this.pushOutgoingDemand(
+                        this.machineResourceEdges[resourceType.ordinal()],
+                        this.resourcesDemand[resourceType.ordinal()],
+                        resourceType);
+            }
+        }
+    }
+
+    @Override
+    public long onUpdate(long now) {
+        long passedTime = getPassedTime(now);
+        this.startOfFragment = now;
+
+        this.updateRemainingWork(passedTime);
+
+        // If this.totalRemainingWork <= 0, the fragment has been completed across all resources
+        if ((int) this.totalRemainingWork <= 0 && this.isWorkloadFinished()) {
+            this.startNextFragment();
+
+            if (this.nodeState == NodeState.CLOSING || this.nodeState == NodeState.CLOSED) {
+                return Long.MAX_VALUE;
+            }
+
+            return getNextUpdateTime(this.startOfFragment);
+        }
+
+        this.pushNewDemands();
+
+        long nextUpdate = getNextUpdateTime(this.startOfFragment);
 
         // if for all resources the remaining work is 0, then invalidate the workload, to reschedule the next fragment
         if (nextUpdate == now + Long.MIN_VALUE) {
@@ -257,7 +268,6 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
         // Reset the remaining work for all resources
         this.totalRemainingWork = 0.0;
 
-        // TODO: only acceleration is considered, not memory
         for (ResourceType resourceType : usedResourceTypes) {
             double demand = nextFragment.getResourceUsage(resourceType);
 
@@ -384,10 +394,10 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
         if (this.resourcesSupplied[suppliedResourceType.ordinal()] == newSupply) {
             return;
         }
-        this.resourcesSupplied[suppliedResourceType.ordinal()] =
-                this.newResourcesSupply[suppliedResourceType.ordinal()];
-        this.newResourcesSupply[suppliedResourceType.ordinal()] = newSupply;
 
+        this.resourcesSupplied[suppliedResourceType.ordinal()] = newSupply;
+
+        // TODO: Change this to just update deadline
         this.invalidate();
     }
 
@@ -406,10 +416,19 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
         if (this.resourcesSupplied[resourceType.ordinal()] == newSupply) {
             return;
         }
-        this.resourcesSupplied[resourceType.ordinal()] = this.newResourcesSupply[resourceType.ordinal()];
-        this.newResourcesSupply[resourceType.ordinal()] = newSupply;
 
-        this.invalidate();
+        this.resourcesSupplied[resourceType.ordinal()] = newSupply;
+
+        long now = this.clock.millis();
+        this.startOfFragment = now;
+        long passedTime = getPassedTime(now);
+
+        this.updateRemainingWork(passedTime);
+        long next_deadline = this.getNextUpdateTime(now);
+
+        // Remove stage from the timer queue
+        this.setDeadline(next_deadline);
+        this.engine.scheduleDelayedInContext(this);
     }
 
     /**