Multiplexer update (#278)

* Fixed the Multiplexer.java to properly divide the supply over the different consumers.

Fixed a bug where fragments were being loaded in reversed order.

* Optimized the Multiplexer.java, by only updating the supply of the consumer that updated its demand when possible.

1 files changed, 67 insertions, 47 deletions

diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/Multiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/Multiplexer.java
index 8cd2fa6f..ece90c20 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/Multiplexer.java
+++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/Multiplexer.java
@@ -39,6 +39,10 @@ public class Multiplexer extends FlowNode implements FlowSupplier, FlowConsumer
 
     private double totalDemand; // The total demand of all the consumers
     private double totalSupply; // The total supply from the supplier
+
+    private boolean overProvisioned = false;
+    private int currentConsumerIdx = -1;
+
     private double capacity; // What is the max capacity
 
     public Multiplexer(FlowGraph graph) {
@@ -59,52 +63,63 @@ public class Multiplexer extends FlowNode implements FlowSupplier, FlowConsumer
 
     public long onUpdate(long now) {
 
-        if (this.totalDemand > this.capacity) {
-            redistributeSupply(this.consumerEdges, this.supplies, this.capacity);
-        } else {
-            for (int i = 0; i < this.demands.size(); i++) {
-                this.supplies.set(i, this.demands.get(i));
+        return Long.MAX_VALUE;
+    }
+
+    private void distributeSupply() {
+        // if supply >= demand -> push supplies to all tasks
+        // TODO: possible optimization -> Only has to be done for the specific consumer that changed demand
+        if (this.totalSupply >= this.totalDemand) {
+
+            // If this came from a state of over provisioning, provide all consumers with their demand
+            if (this.overProvisioned) {
+                for (int idx = 0; idx < this.consumerEdges.size(); idx++) {
+                    this.pushSupply(this.consumerEdges.get(idx), this.demands.get(idx));
+                }
             }
-        }
 
-        double totalSupply = 0;
-        for (int i = 0; i < this.consumerEdges.size(); i++) {
-            this.pushSupply(this.consumerEdges.get(i), this.supplies.get(i));
-            totalSupply += this.supplies.get(i);
+            if (this.currentConsumerIdx != -1) {
+                this.pushSupply(
+                        this.consumerEdges.get(this.currentConsumerIdx), this.demands.get(this.currentConsumerIdx));
+                this.currentConsumerIdx = -1;
+            }
+
+            this.overProvisioned = false;
         }
 
-        // Only update supplier if supply has changed
-        if (this.totalSupply != totalSupply) {
-            this.totalSupply = totalSupply;
+        // if supply < demand -> distribute the supply over all consumers
+        else {
+            this.overProvisioned = true;
+            double[] supplies = redistributeSupply(this.demands, this.totalSupply);
 
-            pushDemand(this.supplierEdge, this.totalSupply);
+            for (int idx = 0; idx < this.consumerEdges.size(); idx++) {
+                this.pushSupply(this.consumerEdges.get(idx), supplies[idx]);
+            }
         }
-
-        return Long.MAX_VALUE;
     }
 
-    private static double redistributeSupply(
-            ArrayList<FlowEdge> consumerEdges, ArrayList<Double> supplies, double capacity) {
-        final long[] consumers = new long[consumerEdges.size()];
+    private record Demand(int idx, double value) {}
 
-        for (int i = 0; i < consumers.length; i++) {
-            FlowEdge consumer = consumerEdges.get(i);
+    private static double[] redistributeSupply(ArrayList<Double> demands, double totalSupply) {
+        int inputSize = demands.size();
 
-            if (consumer == null) {
-                break;
-            }
+        final double[] supplies = new double[inputSize];
+        final Demand[] tempDemands = new Demand[inputSize];
 
-            consumers[i] = (Double.doubleToRawLongBits(consumer.getDemand()) << 32) | (i & 0xFFFFFFFFL);
+        for (int i = 0; i < inputSize; i++) {
+            tempDemands[i] = new Demand(i, demands.get(i));
         }
-        Arrays.sort(consumers);
 
-        double availableCapacity = capacity;
-        int inputSize = consumers.length;
+        Arrays.sort(tempDemands, (o1, o2) -> {
+            Double i1 = o1.value;
+            Double i2 = o2.value;
+            return i1.compareTo(i2);
+        });
+
+        double availableCapacity = totalSupply; // totalSupply
 
         for (int i = 0; i < inputSize; i++) {
-            long v = consumers[i];
-            int slot = (int) v;
-            double d = Double.longBitsToDouble((int) (v >> 32));
+            double d = tempDemands[i].value;
 
             if (d == 0.0) {
                 continue;
@@ -113,12 +128,13 @@ public class Multiplexer extends FlowNode implements FlowSupplier, FlowConsumer
             double availableShare = availableCapacity / (inputSize - i);
             double r = Math.min(d, availableShare);
 
-            supplies.set(slot, r); // Update the rates
+            int idx = tempDemands[i].idx;
+            supplies[idx] = r; // Update the rates
             availableCapacity -= r;
         }
 
         // Return the used capacity
-        return capacity - availableCapacity;
+        return supplies;
     }
 
     /**
@@ -132,8 +148,6 @@ public class Multiplexer extends FlowNode implements FlowSupplier, FlowConsumer
         this.consumerEdges.add(consumerEdge);
         this.demands.add(0.0);
         this.supplies.add(0.0);
-
-        this.invalidate();
     }
 
     @Override
@@ -141,8 +155,6 @@ public class Multiplexer extends FlowNode implements FlowSupplier, FlowConsumer
         this.supplierEdge = supplierEdge;
         this.capacity = supplierEdge.getCapacity();
         this.totalSupply = 0;
-
-        this.invalidate();
     }
 
     @Override
@@ -164,7 +176,9 @@ public class Multiplexer extends FlowNode implements FlowSupplier, FlowConsumer
             this.consumerEdges.get(i).setConsumerIndex(i);
         }
 
-        this.invalidate();
+        this.currentConsumerIdx = -1;
+
+        this.pushDemand(this.supplierEdge, this.totalDemand);
     }
 
     @Override
@@ -178,28 +192,34 @@ public class Multiplexer extends FlowNode implements FlowSupplier, FlowConsumer
     public void handleDemand(FlowEdge consumerEdge, double newDemand) {
         int idx = consumerEdge.getConsumerIndex();
 
+        this.currentConsumerIdx = idx;
+
         if (idx == -1) {
             System.out.println("Error (Multiplexer): Demand pushed by an unknown consumer");
             return;
         }
 
+        // Update the total demand (This is cheaper than summing over all demands)
         double prevDemand = demands.get(idx);
-        demands.set(idx, newDemand);
 
+        demands.set(idx, newDemand);
         this.totalDemand += (newDemand - prevDemand);
 
-        if (this.totalDemand <= this.capacity) {
-
-            this.totalSupply = this.totalDemand;
-            this.pushDemand(this.supplierEdge, this.totalSupply);
-
-            this.pushSupply(consumerEdge, newDemand);
+        if (overProvisioned) {
+            distributeSupply();
         }
-        // TODO: add behaviour if capacity is reached
+
+        // Send new totalDemand to CPU
+        // TODO: Look at what happens if total demand is not changed (if total demand is higher than totalSupply)
+        this.pushDemand(this.supplierEdge, this.totalDemand);
     }
 
     @Override
-    public void handleSupply(FlowEdge supplierEdge, double newSupply) {}
+    public void handleSupply(FlowEdge supplierEdge, double newSupply) {
+        this.totalSupply = newSupply; // Currently this is from a single supply, might turn into multiple suppliers
+
+        this.distributeSupply();
+    }
 
     @Override
     public void pushDemand(FlowEdge supplierEdge, double newDemand) {