Updated the FlowDistributor (#285)

* Updated the FlowDistributor to work in more cases and be more performant.

* Removed old FlowDistributor

7 files changed, 59 insertions, 73 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 c5b8a9ea..a9edaa97 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
@@ -51,7 +51,7 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
     private long lastCounterUpdate;
     private final double cpuFrequencyInv;
 
-    private FlowEdge muxEdge;
+    private FlowEdge distributorEdge;
     private FlowEdge psuEdge;
 
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -123,21 +123,16 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
     public long onUpdate(long now) {
         updateCounters(now);
 
-        this.currentCpuUtilization = Math.min(this.currentCpuDemand / this.maxCapacity, 1.0);
-
-        // Calculate Power Demand and send to PSU
-        double powerDemand = this.cpuPowerModel.computePower(this.currentCpuUtilization);
+        // Check if supply == demand
+        if (this.currentPowerDemand != this.currentPowerSupplied) {
+            this.pushOutgoingDemand(this.psuEdge, this.currentPowerDemand);
 
-        if (powerDemand != this.currentPowerDemand) {
-            this.pushDemand(this.psuEdge, powerDemand);
+            return Long.MAX_VALUE;
         }
 
-        // Calculate the amount of cpu this can provide
-        double cpuSupply = Math.min(this.currentCpuDemand, this.maxCapacity);
+        this.currentCpuSupplied = Math.min(this.currentCpuDemand, this.maxCapacity);
 
-        if (cpuSupply != this.currentCpuSupplied) {
-            this.pushSupply(this.muxEdge, cpuSupply);
-        }
+        this.pushOutgoingSupply(this.distributorEdge, this.currentCpuSupplied);
 
         return Long.MAX_VALUE;
     }
@@ -181,7 +176,7 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
      * Push new demand to the psu
      */
     @Override
-    public void pushDemand(FlowEdge supplierEdge, double newPowerDemand) {
+    public void pushOutgoingDemand(FlowEdge supplierEdge, double newPowerDemand) {
         updateCounters();
         this.currentPowerDemand = newPowerDemand;
         this.psuEdge.pushDemand(newPowerDemand);
@@ -191,47 +186,38 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
      * Push updated supply to the mux
      */
     @Override
-    public void pushSupply(FlowEdge consumerEdge, double newCpuSupply) {
+    public void pushOutgoingSupply(FlowEdge consumerEdge, double newCpuSupply) {
         updateCounters();
         this.currentCpuSupplied = newCpuSupply;
-        this.muxEdge.pushSupply(newCpuSupply);
+
+        this.distributorEdge.pushSupply(newCpuSupply, true);
     }
 
     /**
      * Handle new demand coming in from the mux
      */
     @Override
-    public void handleDemand(FlowEdge consumerEdge, double newCpuDemand) {
+    public void handleIncomingDemand(FlowEdge consumerEdge, double newCpuDemand) {
         updateCounters();
         this.currentCpuDemand = newCpuDemand;
-        this.currentCpuUtilization = this.currentCpuDemand / this.maxCapacity;
 
         this.currentCpuUtilization = Math.min(this.currentCpuDemand / this.maxCapacity, 1.0);
 
         // Calculate Power Demand and send to PSU
-        double powerDemand = this.cpuPowerModel.computePower(this.currentCpuUtilization);
+        this.currentPowerDemand = this.cpuPowerModel.computePower(this.currentCpuUtilization);
 
-        if (powerDemand != this.currentPowerDemand) {
-            this.pushDemand(this.psuEdge, powerDemand);
-        }
+        this.invalidate();
     }
 
     /**
      * Handle updated supply from the psu
      */
     @Override
-    public void handleSupply(FlowEdge supplierEdge, double newPowerSupply) {
-        // TODO: Implement this
+    public void handleIncomingSupply(FlowEdge supplierEdge, double newPowerSupply) {
         updateCounters();
         this.currentPowerSupplied = newPowerSupply;
 
-        // Calculate the amount of cpu this can provide
-        double cpuSupply = Math.min(this.currentCpuDemand, this.maxCapacity);
-        ;
-
-        if (cpuSupply != this.currentCpuSupplied) {
-            this.pushSupply(this.muxEdge, cpuSupply);
-        }
+        this.invalidate();
     }
 
     /**
@@ -239,7 +225,7 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
      */
     @Override
     public void addConsumerEdge(FlowEdge consumerEdge) {
-        this.muxEdge = consumerEdge;
+        this.distributorEdge = consumerEdge;
     }
 
     /**
@@ -257,7 +243,7 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
      */
     @Override
     public void removeConsumerEdge(FlowEdge consumerEdge) {
-        this.muxEdge = null;
+        this.distributorEdge = null;
         this.invalidate();
     }
 
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/SimMachine.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/SimMachine.java
index 074f0ed8..dab0c421 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/SimMachine.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/SimMachine.java
@@ -44,7 +44,7 @@ public class SimMachine {
     private final InstantSource clock;
 
     private SimCpu cpu;
-    private FlowDistributor cpuMux;
+    private FlowDistributor cpuDistributor;
     private SimPsu psu;
     private Memory memory;
 
@@ -74,8 +74,8 @@ public class SimMachine {
         return cpu;
     }
 
-    public FlowDistributor getCpuMux() {
-        return cpuMux;
+    public FlowDistributor getCpuDistributor() {
+        return cpuDistributor;
     }
 
     public Memory getMemory() {
@@ -114,7 +114,7 @@ public class SimMachine {
     public SimMachine(
             FlowGraph graph,
             MachineModel machineModel,
-            FlowDistributor powerMux,
+            FlowDistributor powerDistributor,
             CpuPowerModel cpuPowerModel,
             Consumer<Exception> completion) {
         this.graph = graph;
@@ -124,7 +124,7 @@ public class SimMachine {
         // Create the psu and cpu and connect them
         this.psu = new SimPsu(graph);
 
-        graph.addEdge(this.psu, powerMux);
+        graph.addEdge(this.psu, powerDistributor);
 
         this.cpu = new SimCpu(graph, this.machineModel.getCpuModel(), cpuPowerModel, 0);
 
@@ -133,8 +133,8 @@ public class SimMachine {
         this.memory = new Memory(graph, this.machineModel.getMemory());
 
         // Create a FlowDistributor and add the cpu as supplier
-        this.cpuMux = new FlowDistributor(this.graph);
-        graph.addEdge(this.cpuMux, this.cpu);
+        this.cpuDistributor = new FlowDistributor(this.graph);
+        graph.addEdge(this.cpuDistributor, this.cpu);
 
         this.completion = completion;
     }
@@ -153,8 +153,8 @@ public class SimMachine {
         this.graph.removeNode(this.cpu);
         this.cpu = null;
 
-        this.graph.removeNode(this.cpuMux);
-        this.cpuMux = null;
+        this.graph.removeNode(this.cpuDistributor);
+        this.cpuDistributor = null;
 
         this.memory = null;
 
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/VirtualMachine.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/VirtualMachine.java
index b8a9c738..1946eecb 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/VirtualMachine.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/VirtualMachine.java
@@ -100,7 +100,7 @@ public class VirtualMachine extends FlowNode implements FlowConsumer, FlowSuppli
         this.clock = this.machine.getClock();
 
         this.parentGraph = machine.getGraph();
-        this.parentGraph.addEdge(this, this.machine.getCpuMux());
+        this.parentGraph.addEdge(this, this.machine.getCpuDistributor());
 
         this.lastUpdate = clock.millis();
         this.lastUpdate = clock.millis();
@@ -185,7 +185,7 @@ public class VirtualMachine extends FlowNode implements FlowConsumer, FlowSuppli
      * Push demand to the cpuMux if the demand has changed
      **/
     @Override
-    public void pushDemand(FlowEdge supplierEdge, double newDemand) {
+    public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) {
         this.cpuEdge.pushDemand(newDemand);
     }
 
@@ -193,7 +193,7 @@ public class VirtualMachine extends FlowNode implements FlowConsumer, FlowSuppli
      * Push supply to the workload if the supply has changed
      **/
     @Override
-    public void pushSupply(FlowEdge consumerEdge, double newSupply) {
+    public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply) {
         this.workloadEdge.pushSupply(newSupply);
     }
 
@@ -201,24 +201,24 @@ public class VirtualMachine extends FlowNode implements FlowConsumer, FlowSuppli
      * Handle new demand from the workload by sending it through to the cpuMux
      **/
     @Override
-    public void handleDemand(FlowEdge consumerEdge, double newDemand) {
+    public void handleIncomingDemand(FlowEdge consumerEdge, double newDemand) {
 
         updateCounters(this.clock.millis());
         this.cpuDemand = newDemand;
 
-        pushDemand(this.cpuEdge, newDemand);
+        pushOutgoingDemand(this.cpuEdge, newDemand);
     }
 
     /**
      * Handle a new supply pushed by the cpuMux by sending it through to the workload
      **/
     @Override
-    public void handleSupply(FlowEdge supplierEdge, double newCpuSupply) {
+    public void handleIncomingSupply(FlowEdge supplierEdge, double newCpuSupply) {
 
         updateCounters(this.clock.millis());
         this.cpuSupply = newCpuSupply;
 
-        pushSupply(this.workloadEdge, newCpuSupply);
+        pushOutgoingSupply(this.workloadEdge, newCpuSupply);
     }
 
     @Override
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/SimPowerSource.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/SimPowerSource.java
index e8626e40..d2270888 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/SimPowerSource.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/SimPowerSource.java
@@ -43,7 +43,7 @@ public final class SimPowerSource extends FlowNode implements FlowSupplier {
     private double totalCarbonEmission = 0.0f;
 
     private CarbonModel carbonModel = null;
-    private FlowEdge muxEdge;
+    private FlowEdge distributorEdge;
 
     private double capacity = Long.MAX_VALUE;
 
@@ -57,7 +57,7 @@ public final class SimPowerSource extends FlowNode implements FlowSupplier {
      * @return <code>true</code> if the InPort is connected to an OutPort, <code>false</code> otherwise.
      */
     public boolean isConnected() {
-        return muxEdge != null;
+        return distributorEdge != null;
     }
 
     /**
@@ -156,30 +156,30 @@ public final class SimPowerSource extends FlowNode implements FlowSupplier {
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
     @Override
-    public void handleDemand(FlowEdge consumerEdge, double newPowerDemand) {
+    public void handleIncomingDemand(FlowEdge consumerEdge, double newPowerDemand) {
         this.powerDemand = newPowerDemand;
 
         double powerSupply = this.powerDemand;
 
         if (powerSupply != this.powerSupplied) {
-            this.pushSupply(this.muxEdge, powerSupply);
+            this.pushOutgoingSupply(this.distributorEdge, powerSupply);
         }
     }
 
     @Override
-    public void pushSupply(FlowEdge consumerEdge, double newSupply) {
+    public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply) {
         this.powerSupplied = newSupply;
         consumerEdge.pushSupply(newSupply);
     }
 
     @Override
     public void addConsumerEdge(FlowEdge consumerEdge) {
-        this.muxEdge = consumerEdge;
+        this.distributorEdge = consumerEdge;
     }
 
     @Override
     public void removeConsumerEdge(FlowEdge consumerEdge) {
-        this.muxEdge = null;
+        this.distributorEdge = null;
     }
 
     // Update the carbon intensity of the power source
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/SimPsu.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/SimPsu.java
index c1e8a1b9..dc5129d6 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/SimPsu.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/SimPsu.java
@@ -106,7 +106,7 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
         double powerSupply = this.powerDemand;
 
         if (powerSupply != this.powerSupplied) {
-            this.pushSupply(this.cpuEdge, powerSupply);
+            this.pushOutgoingSupply(this.cpuEdge, powerSupply);
         }
 
         return Long.MAX_VALUE;
@@ -135,33 +135,33 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
     @Override
-    public void pushDemand(FlowEdge supplierEdge, double newDemand) {
+    public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) {
         this.powerDemand = newDemand;
         powerSupplyEdge.pushDemand(newDemand);
     }
 
     @Override
-    public void pushSupply(FlowEdge consumerEdge, double newSupply) {
+    public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply) {
         this.powerSupplied = newSupply;
         cpuEdge.pushSupply(newSupply);
     }
 
     @Override
-    public void handleDemand(FlowEdge consumerEdge, double newPowerDemand) {
+    public void handleIncomingDemand(FlowEdge consumerEdge, double newPowerDemand) {
 
         updateCounters();
         this.powerDemand = newPowerDemand;
 
-        pushDemand(this.powerSupplyEdge, newPowerDemand);
+        pushOutgoingDemand(this.powerSupplyEdge, newPowerDemand);
     }
 
     @Override
-    public void handleSupply(FlowEdge supplierEdge, double newPowerSupply) {
+    public void handleIncomingSupply(FlowEdge supplierEdge, double newPowerSupply) {
 
         updateCounters();
         this.powerSupplied = newPowerSupply;
 
-        pushSupply(this.cpuEdge, newPowerSupply);
+        pushOutgoingSupply(this.cpuEdge, newPowerSupply);
     }
 
     @Override
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimChainWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimChainWorkload.java
index b612de2c..da6b8334 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimChainWorkload.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimChainWorkload.java
@@ -195,7 +195,7 @@ final class SimChainWorkload extends SimWorkload implements FlowSupplier {
      * @param newDemand new demand to sent to the cpu
      */
     @Override
-    public void pushDemand(FlowEdge supplierEdge, double newDemand) {
+    public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) {
 
         this.demand = newDemand;
         this.machineEdge.pushDemand(newDemand);
@@ -208,7 +208,7 @@ final class SimChainWorkload extends SimWorkload implements FlowSupplier {
      * @param newSupply new supply to sent to the workload
      */
     @Override
-    public void pushSupply(FlowEdge consumerEdge, double newSupply) {
+    public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply) {
 
         this.supply = newSupply;
         this.workloadEdge.pushSupply(newSupply);
@@ -221,8 +221,8 @@ final class SimChainWorkload extends SimWorkload implements FlowSupplier {
      * @param newDemand new demand coming from the workload
      */
     @Override
-    public void handleDemand(FlowEdge consumerEdge, double newDemand) {
-        this.pushDemand(this.machineEdge, newDemand);
+    public void handleIncomingDemand(FlowEdge consumerEdge, double newDemand) {
+        this.pushOutgoingDemand(this.machineEdge, newDemand);
     }
 
     /**
@@ -232,8 +232,8 @@ final class SimChainWorkload extends SimWorkload implements FlowSupplier {
      * @param newSupply The new supply that is sent to the workload
      */
     @Override
-    public void handleSupply(FlowEdge supplierEdge, double newSupply) {
-        this.pushSupply(this.machineEdge, newSupply);
+    public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply) {
+        this.pushOutgoingSupply(this.machineEdge, newSupply);
     }
 
     /**
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimTraceWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimTraceWorkload.java
index 8487fbc2..0735d8ae 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimTraceWorkload.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimTraceWorkload.java
@@ -94,7 +94,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
         graph.addEdge(this, supplier);
 
         this.currentFragment = this.getNextFragment();
-        pushDemand(machineEdge, this.currentFragment.cpuUsage());
+        pushOutgoingDemand(machineEdge, this.currentFragment.cpuUsage());
         this.startOfFragment = now;
     }
 
@@ -131,7 +131,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
         this.startOfFragment = now - passedTime;
 
         // Change the cpu Usage to the new Fragment
-        pushDemand(machineEdge, this.currentFragment.cpuUsage());
+        pushOutgoingDemand(machineEdge, this.currentFragment.cpuUsage());
 
         // Return the time when the current fragment will complete
         return this.startOfFragment + duration;
@@ -190,7 +190,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
 
         this.fragmentIndex = -1;
         this.currentFragment = getNextFragment();
-        pushDemand(this.machineEdge, this.currentFragment.cpuUsage());
+        pushOutgoingDemand(this.machineEdge, this.currentFragment.cpuUsage());
         this.startOfFragment = now;
 
         this.invalidate();
@@ -207,7 +207,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
      * @param newSupply The new demand that needs to be sent to the VM
      */
     @Override
-    public void handleSupply(FlowEdge supplierEdge, double newSupply) {
+    public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply) {
         if (newSupply == this.currentSupply) {
             return;
         }
@@ -222,7 +222,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
      * @param newDemand The new demand that needs to be sent to the VM
      */
     @Override
-    public void pushDemand(FlowEdge supplierEdge, double newDemand) {
+    public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) {
         if (newDemand == this.currentDemand) {
             return;
         }