Implements fixes to run m100 traces with GPUs (#362)

* Updated output format to reduce size

* using sum of gpu capacities instead of single max

* passing provisioned GPU cores to host view

* fix supply update trigger

* fixing floating point error, leading to negative demand

* fixing double mismatch, due to floating point in precision

* adding additional check if demand can be satisfied in the simple way

* adds workload invalidation if remaining duration for all resources is 0

* invalidating flow distributors after demand update

* spotless apply

* updating tests

* exporting power consumption of compute resources directly from gpu instead of PSU

* using big decimal to avoid floating point in-precision

* rolls back to pass-through version of PSU, before GPU implementation

* places flowdistributor between PSU and compute resources

* adds check to avoid null exception if supply is pushed without demand

* fixing task id type

* Adds memorizing GPU scheduler

* adds boundary for negative remaining work

* implemented tests for GPU scheduler filter

* Revert "Updated output format to reduce size"

This reverts commit 7171de8e0512a863df4962f64560ac7bad1fb48d.

* spotless aply

---------

Co-authored-by: DanteNiewenhuis <d.niewenhuis@hotmail.com>

18 files changed, 233 insertions, 196 deletions

diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java
index fde83ead..8feddf54 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java
+++ b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java
@@ -186,6 +186,7 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver {
                     hv.provisionedCpuCores -= flavor.getCpuCoreCount();
                     hv.instanceCount--;
                     hv.availableMemory += flavor.getMemorySize();
+                    hv.provisionedGpuCores -= flavor.getGpuCoreCount();
                 } else {
                     LOGGER.error("Unknown host {}", host);
                 }
@@ -580,6 +581,7 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver {
                 hv.instanceCount++;
                 hv.provisionedCpuCores += flavor.getCpuCoreCount();
                 hv.availableMemory -= flavor.getMemorySize();
+                hv.provisionedGpuCores += flavor.getGpuCoreCount();
 
                 activeTasks.put(task, host);
             } catch (Exception cause) {
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/host/SimHost.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/host/SimHost.kt
index b7d3b730..1a0cc316 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/host/SimHost.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/host/SimHost.kt
@@ -361,7 +361,6 @@ public class SimHost(
         for (gpu in simMachine!!.gpus) {
             gpu.updateCounters(this.clock.millis())
             val counters = simMachine!!.getGpuPerformanceCounters(gpu.id)
-            val powerDraw = simMachine!!.psu.getPowerDraw(ResourceType.GPU, gpu.id)
 
             gpuStats.add(
                 HostGpuStats(
@@ -373,7 +372,7 @@ public class SimHost(
                     counters.demand,
                     counters.supply,
                     counters.supply / gpu.getCapacity(ResourceType.GPU),
-                    powerDraw,
+                    counters.powerDraw,
                 ),
             )
         }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt
index 0376a492..79af6f62 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt
@@ -52,6 +52,7 @@ public enum class ComputeSchedulerEnum {
     Timeshift,
     ProvisionedCpuGpuCores,
     ProvisionedCpuGpuCoresInv,
+    GpuTaskMemorizing,
 }
 
 public fun createPrefabComputeScheduler(
@@ -159,5 +160,15 @@ public fun createPrefabComputeScheduler(
                         VGpuWeigher(gpuAllocationRatio, multiplier = -1.0),
                     ),
             )
+        ComputeSchedulerEnum.GpuTaskMemorizing ->
+            MemorizingScheduler(
+                filters =
+                    listOf(
+                        ComputeFilter(),
+                        VCpuFilter(cpuAllocationRatio),
+                        VGpuFilter(gpuAllocationRatio),
+                        RamFilter(ramAllocationRatio),
+                    ),
+            )
     }
 }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VGpuCapacityFilter.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VGpuCapacityFilter.kt
index 6dc27327..5f517257 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VGpuCapacityFilter.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VGpuCapacityFilter.kt
@@ -42,7 +42,7 @@ public class VGpuCapacityFilter : HostFilter {
 
         return (
             requiredCapacity == null ||
-                ((availableCapacity / availableCores) >= (requiredCapacity / task.flavor.gpuCoreCount))
+                (availableRatio >= (requiredCapacity / task.flavor.gpuCoreCount))
         )
     }
 }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VGpuFilter.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VGpuFilter.kt
index 9f564776..f47013b1 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VGpuFilter.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VGpuFilter.kt
@@ -26,9 +26,9 @@ import org.opendc.compute.simulator.service.HostView
 import org.opendc.compute.simulator.service.ServiceTask
 
 /**
- * A [HostFilter] that filters hosts based on the vCPU requirements of a [ServiceTask] and the available vCPUs on the host.
+ * A [HostFilter] that filters hosts based on the vGPU requirements of a [ServiceTask] and the available vGPUs on the host.
  *
- * @param allocationRatio Virtual CPU to physical CPU allocation ratio.
+ * @param allocationRatio Virtual GPU to physical GPU allocation ratio.
  */
 public class VGpuFilter(private val allocationRatio: Double) : HostFilter {
     override fun test(
@@ -36,7 +36,7 @@ public class VGpuFilter(private val allocationRatio: Double) : HostFilter {
         task: ServiceTask,
     ): Boolean {
         val requested = task.flavor.gpuCoreCount
-        val totalCores = host.host.getModel().gpuHostModels().maxOfOrNull { it.gpuCoreCount() } ?: 0
+        val totalCores = host.host.getModel().gpuHostModels()?.sumOf { it.gpuCoreCount() } ?: 0
         val limit = totalCores * allocationRatio
 
         // Do not allow an instance to overcommit against itself, only against other instances
diff --git a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/scheduler/FilterSchedulerTest.kt b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/scheduler/FilterSchedulerTest.kt
index fe5cea70..65fbfb38 100644
--- a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/scheduler/FilterSchedulerTest.kt
+++ b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/scheduler/FilterSchedulerTest.kt
@@ -28,6 +28,7 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertAll
 import org.junit.jupiter.api.assertThrows
+import org.opendc.compute.simulator.host.GpuHostModel
 import org.opendc.compute.simulator.host.HostModel
 import org.opendc.compute.simulator.host.HostState
 import org.opendc.compute.simulator.scheduler.filters.ComputeFilter
@@ -37,6 +38,8 @@ import org.opendc.compute.simulator.scheduler.filters.RamFilter
 import org.opendc.compute.simulator.scheduler.filters.SameHostFilter
 import org.opendc.compute.simulator.scheduler.filters.VCpuCapacityFilter
 import org.opendc.compute.simulator.scheduler.filters.VCpuFilter
+import org.opendc.compute.simulator.scheduler.filters.VGpuCapacityFilter
+import org.opendc.compute.simulator.scheduler.filters.VGpuFilter
 import org.opendc.compute.simulator.scheduler.weights.CoreRamWeigher
 import org.opendc.compute.simulator.scheduler.weights.InstanceCountWeigher
 import org.opendc.compute.simulator.scheduler.weights.RamWeigher
@@ -437,6 +440,98 @@ internal class FilterSchedulerTest {
     }
 
     @Test
+    fun testVGPUFilter() {
+        val scheduler =
+            FilterScheduler(
+                filters = listOf(VGpuFilter(1.0)),
+                weighers = emptyList(),
+            )
+
+        val hostA = mockk<HostView>()
+        every { hostA.host.getState() } returns HostState.UP
+        every { hostA.host.getModel() } returns
+            HostModel(
+                0.0,
+                0,
+                2048,
+                listOf(
+                    GpuHostModel(8 * 2600.0, 8, 0L, 0.0),
+                ),
+            )
+        every { hostA.provisionedGpuCores } returns 0
+        scheduler.addHost(hostA)
+
+        val hostB = mockk<HostView>()
+        every { hostB.host.getState() } returns HostState.UP
+        every { hostB.host.getModel() } returns
+            HostModel(
+                0.0,
+                0,
+                2048,
+                listOf(
+                    GpuHostModel(8 * 3200.0, 8, 0L, 0.0),
+                    GpuHostModel(8 * 3200.0, 8, 0L, 0.0),
+                ),
+            )
+        every { hostB.provisionedGpuCores } returns 0
+        scheduler.addHost(hostB)
+
+        val req = mockk<SchedulingRequest>()
+        every { req.task.flavor.gpuCoreCount } returns 9
+        every { req.task.flavor.meta } returns mapOf("gpu-capacity" to 9 * 3200.0)
+        every { req.isCancelled } returns false
+
+        // filter selects hostB because hostA does not have enough GPU capacity
+        assertEquals(hostB, scheduler.select(mutableListOf(req).iterator()).host)
+    }
+
+    @Test
+    fun testVGPUCapacityFilter() {
+        val scheduler =
+            FilterScheduler(
+                filters = listOf(VGpuCapacityFilter()),
+                weighers = emptyList(),
+            )
+
+        val hostA = mockk<HostView>()
+        every { hostA.host.getState() } returns HostState.UP
+        every { hostA.host.getModel() } returns
+            HostModel(
+                0.0,
+                0,
+                2048,
+                listOf(
+                    GpuHostModel(8 * 2600.0, 8, 0L, 0.0),
+                ),
+            )
+        every { hostA.availableMemory } returns 512
+        scheduler.addHost(hostA)
+
+        val hostB = mockk<HostView>()
+        every { hostB.host.getState() } returns HostState.UP
+        every { hostB.host.getModel() } returns
+            HostModel(
+                0.0,
+                0,
+                2048,
+                listOf(
+                    GpuHostModel(8 * 3200.0, 8, 0L, 0.0),
+                    GpuHostModel(8 * 3200.0, 8, 0L, 0.0),
+                ),
+            )
+        every { hostB.availableMemory } returns 512
+        scheduler.addHost(hostB)
+
+        val req = mockk<SchedulingRequest>()
+        every { req.task.flavor.gpuCoreCount } returns 8
+        every { req.task.flavor.meta } returns mapOf("gpu-capacity" to 8 * 3200.0)
+        every { req.isCancelled } returns false
+
+        // filter selects hostB because hostA does not have enough GPU capacity
+        assertEquals(hostB, scheduler.select(mutableListOf(req).iterator()).host)
+    }
+
+    @Test
     fun testRamWeigher() {
         val scheduler =
             FilterScheduler(
diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/DistributionPoliciesTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/DistributionPoliciesTest.kt
index 256c067d..730f9fd0 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/DistributionPoliciesTest.kt
+++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/DistributionPoliciesTest.kt
@@ -537,16 +537,16 @@ class DistributionPoliciesTest {
         // Best effort should distribute proportionally based on demand while using round-robin
         assertAll(
             // Task 0
-            { assertEquals(3000.0, monitor.taskGpuDemands[0]?.get(1), "Task 0 GPU demand should be 3000.0") },
-            { assertEquals(3000.0, monitor.taskGpuSupplied[0]?.get(1), "Task 0 GPU supply should be 1000.0") },
+            { assertEquals(3000.0, monitor.taskGpuDemands[0]?.get(0), "Task 0 GPU demand should be 3000.0") },
+            { assertEquals(3000.0, monitor.taskGpuSupplied[0]?.get(0), "Task 0 GPU supply should be 1000.0") },
             // Task 1
-            { assertEquals(2500.0, monitor.taskGpuDemands[1]?.get(1), "Task 1 GPU demand should be 2500.0") },
-            { assertEquals(1000.0, monitor.taskGpuSupplied[1]?.get(1), "Task 0 GPU supply should be 1000.0") },
+            { assertEquals(2500.0, monitor.taskGpuDemands[1]?.get(0), "Task 1 GPU demand should be 2500.0") },
+            { assertEquals(1000.0, monitor.taskGpuSupplied[1]?.get(0), "Task 1 GPU supply should be 1000.0") },
             // Host
-            { assertEquals(2750.0, monitor.hostGpuDemands["DualGpuHost"]?.get(1)?.get(0), "GPU 0 demand at host should be 2000.0") },
-            { assertEquals(2000.0, monitor.hostGpuSupplied["DualGpuHost"]?.get(1)?.get(0), "GPU 0 supplied at host should be 2000.0") },
-            { assertEquals(2750.0, monitor.hostGpuDemands["DualGpuHost"]?.get(1)?.get(1), "GPU 1 demand at host should be 2000.0") },
-            { assertEquals(2000.0, monitor.hostGpuSupplied["DualGpuHost"]?.get(1)?.get(1), "GPU 1 supplied at host should be 2000.0") },
+            { assertEquals(2750.0, monitor.hostGpuDemands["DualGpuHost"]?.get(0)?.get(0), "GPU 0 demand at host should be 2000.0") },
+            { assertEquals(2000.0, monitor.hostGpuSupplied["DualGpuHost"]?.get(0)?.get(0), "GPU 0 supplied at host should be 2000.0") },
+            { assertEquals(2750.0, monitor.hostGpuDemands["DualGpuHost"]?.get(0)?.get(1), "GPU 1 demand at host should be 2000.0") },
+            { assertEquals(2000.0, monitor.hostGpuSupplied["DualGpuHost"]?.get(0)?.get(1), "GPU 1 supplied at host should be 2000.0") },
         )
     }
 
@@ -677,21 +677,21 @@ class DistributionPoliciesTest {
         // Best effort should distribute fairly among all tasks in a round-robin manner
         assertAll(
             // Task Demands at start
-            { assertEquals(2000.0, monitor.taskGpuDemands[0]?.get(1), "Task 0 demand should be 2000.0") },
-            { assertEquals(2000.0, monitor.taskGpuDemands[1]?.get(1), "Task 1 demand should be 2000.0") },
-            { assertEquals(2000.0, monitor.taskGpuDemands[2]?.get(1), "Task 2 demand should be 2000.0") },
+            { assertEquals(2000.0, monitor.taskGpuDemands[0]?.get(0), "Task 0 demand should be 2000.0") },
+            { assertEquals(2000.0, monitor.taskGpuDemands[1]?.get(0), "Task 1 demand should be 2000.0") },
+            { assertEquals(2000.0, monitor.taskGpuDemands[2]?.get(0), "Task 2 demand should be 2000.0") },
             // Task supplies at start
-            { assertEquals(2000.0, monitor.taskGpuSupplied[0]?.get(1), "Task 0 supply at the start should be 2000.0") },
-            { assertEquals(0.0, monitor.taskGpuSupplied[1]?.get(1), "Task 1 supply at the start  should be 2000.0") },
-            { assertEquals(2000.0, monitor.taskGpuSupplied[2]?.get(1), "Task 2 supply at the start  should be 0.0") },
+            { assertEquals(2000.0, monitor.taskGpuSupplied[0]?.get(0), "Task 0 supply at the start should be 2000.0") },
+            { assertEquals(2000.0, monitor.taskGpuSupplied[1]?.get(0), "Task 1 supply at the start  should be 0.0") },
+            { assertEquals(0.0, monitor.taskGpuSupplied[2]?.get(0), "Task 2 supply at the start  should be 2000.0") },
             // Task supplies second step
-            { assertEquals(0.0, monitor.taskGpuSupplied[0]?.get(2), "Task 0 supply at the second step should be 2000.0") },
-            { assertEquals(2000.0, monitor.taskGpuSupplied[1]?.get(2), "Task 1 supply at the second step should be 0.0") },
-            { assertEquals(2000.0, monitor.taskGpuSupplied[2]?.get(2), "Task 2 supply at the second step should be 2000.0") },
+            { assertEquals(0.0, monitor.taskGpuSupplied[0]?.get(1), "Task 0 supply at the second step should be 2000.0") },
+            { assertEquals(2000.0, monitor.taskGpuSupplied[1]?.get(1), "Task 1 supply at the second step should be 0.0") },
+            { assertEquals(2000.0, monitor.taskGpuSupplied[2]?.get(1), "Task 2 supply at the second step should be 2000.0") },
             // Task supplies third step
-            { assertEquals(2000.0, monitor.taskGpuSupplied[0]?.get(3), "Task 0 supply at the third step should be 2000.0") },
-            { assertEquals(2000.0, monitor.taskGpuSupplied[1]?.get(3), "Task 1 supply at the third step should be 2000.0") },
-            { assertEquals(0.0, monitor.taskGpuSupplied[2]?.get(3), "Task 2 supply at the third step should be 0.0") },
+            { assertEquals(2000.0, monitor.taskGpuSupplied[0]?.get(2), "Task 0 supply at the third step should be 2000.0") },
+            { assertEquals(0.0, monitor.taskGpuSupplied[1]?.get(2), "Task 1 supply at the third step should be 0.0") },
+            { assertEquals(2000.0, monitor.taskGpuSupplied[2]?.get(2), "Task 2 supply at the third step should be 2000.0") },
             // Host
             // At start
             { assertEquals(3000.0, monitor.hostGpuDemands["DualGpuHost"]?.get(1)?.get(0), "GPU 0 demand at host should be 2000.0") },
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 5669eb16..52fc6093 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
@@ -22,6 +22,8 @@
 
 package org.opendc.simulator.compute.cpu;
 
+import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.util.List;
 import java.util.Map;
 import org.opendc.common.ResourceType;
@@ -140,7 +142,10 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
         updateCounters(now);
 
         // Check if supply == demand
-        if (this.currentPowerDemand != this.currentPowerSupplied) {
+        // using big decimal to avoid floating point precision issues
+        if (!new BigDecimal(this.currentPowerDemand)
+                .setScale(5, RoundingMode.HALF_UP)
+                .equals(new BigDecimal(this.currentPowerSupplied).setScale(5, RoundingMode.HALF_UP))) {
             this.pushOutgoingDemand(this.psuEdge, this.currentPowerDemand);
 
             return Long.MAX_VALUE;
@@ -182,6 +187,7 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
         this.performanceCounters.setDemand(this.currentCpuDemand);
         this.performanceCounters.setSupply(this.currentCpuSupplied);
         this.performanceCounters.setCapacity(this.maxCapacity);
+        this.performanceCounters.setPowerDraw(this.currentPowerDemand);
     }
 
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -195,7 +201,7 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer
     public void pushOutgoingDemand(FlowEdge supplierEdge, double newPowerDemand) {
         updateCounters();
         this.currentPowerDemand = newPowerDemand;
-        this.psuEdge.pushDemand(newPowerDemand, false, ResourceType.CPU);
+        this.psuEdge.pushDemand(newPowerDemand, false, ResourceType.POWER);
     }
 
     /**
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 99317a08..a5fccf6c 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
@@ -22,6 +22,8 @@
 
 package org.opendc.simulator.compute.gpu;
 
+import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.util.List;
 import java.util.Map;
 import org.opendc.common.ResourceType;
@@ -144,7 +146,10 @@ public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer
         updateCounters(now);
 
         // Check if supply == demand
-        if (this.currentPowerDemand != this.currentPowerSupplied) {
+        // using big decimal to avoid floating point precision issues
+        if (!new BigDecimal(this.currentPowerDemand)
+                .setScale(5, RoundingMode.HALF_UP)
+                .equals(new BigDecimal(this.currentPowerSupplied).setScale(5, RoundingMode.HALF_UP))) {
             this.pushOutgoingDemand(this.psuEdge, this.currentPowerDemand);
 
             return Long.MAX_VALUE;
@@ -185,6 +190,7 @@ public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer
         this.performanceCounters.setDemand(this.currentGpuDemand);
         this.performanceCounters.setSupply(this.currentGpuSupplied);
         this.performanceCounters.setCapacity(this.maxCapacity);
+        this.performanceCounters.setPowerDraw(this.currentPowerSupplied);
     }
 
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -198,7 +204,7 @@ public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer
     public void pushOutgoingDemand(FlowEdge supplierEdge, double newPowerDemand) {
         updateCounters();
         this.currentPowerDemand = newPowerDemand;
-        this.psuEdge.pushDemand(newPowerDemand, false, ResourceType.GPU);
+        this.psuEdge.pushDemand(newPowerDemand, false, ResourceType.POWER);
     }
 
     /**
@@ -209,7 +215,7 @@ public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer
         updateCounters();
         this.currentGpuSupplied = newGpuSupply;
 
-        this.distributorEdge.pushSupply(newGpuSupply, true, ResourceType.GPU);
+        this.distributorEdge.pushSupply(newGpuSupply, true, ResourceType.POWER);
     }
 
     /**
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/PerformanceCounters.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/PerformanceCounters.java
index 93033bc0..013846cd 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/PerformanceCounters.java
+++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/machine/PerformanceCounters.java
@@ -32,6 +32,7 @@ public class PerformanceCounters {
     private double capacity = 0.0f;
     private double demand = 0.0f;
     private double supply = 0.0f;
+    private double powerDraw = 0.0f;
 
     public long getActiveTime() {
         return this.activeTime;
@@ -61,6 +62,10 @@ public class PerformanceCounters {
         return this.supply;
     }
 
+    public double getPowerDraw() {
+        return powerDraw;
+    }
+
     public void setActiveTime(long activeTime) {
         this.activeTime = activeTime;
     }
@@ -89,6 +94,10 @@ public class PerformanceCounters {
         this.supply = supply;
     }
 
+    public void setPowerDraw(double powerDraw) {
+        this.powerDraw = powerDraw;
+    }
+
     public void addActiveTime(long activeTime) {
         this.activeTime += activeTime;
     }
@@ -116,4 +125,8 @@ public class PerformanceCounters {
     public void addSupply(double supply) {
         this.supply += supply;
     }
+
+    public void addPowerDraw(double powerDraw) {
+        this.powerDraw += powerDraw;
+    }
 }
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 7158356a..e56e1b5c 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
@@ -47,6 +47,7 @@ import org.opendc.simulator.engine.graph.FlowEdge;
 import org.opendc.simulator.engine.graph.FlowNode;
 import org.opendc.simulator.engine.graph.FlowSupplier;
 import org.opendc.simulator.engine.graph.distributionPolicies.FlowDistributorFactory;
+import org.opendc.simulator.engine.graph.distributionPolicies.MaxMinFairnessFlowDistributor;
 
 /**
  * A machine that is able to execute {@link SimWorkload} objects.
@@ -207,6 +208,8 @@ public class SimMachine {
         // Create the psu and cpu and connect them
         this.psu = new SimPsu(engine);
         new FlowEdge(this.psu, powerDistributor);
+        this.distributors.put(ResourceType.POWER, new MaxMinFairnessFlowDistributor(engine)); // Maybe First fit
+        new FlowEdge(this.distributors.get(ResourceType.POWER), this.psu);
 
         this.computeResources.put(
                 ResourceType.CPU,
@@ -215,7 +218,7 @@ public class SimMachine {
         // Connect the CPU to the PSU
         new FlowEdge(
                 (FlowConsumer) this.computeResources.get(ResourceType.CPU).getFirst(),
-                this.psu,
+                (FlowSupplier) this.distributors.get(ResourceType.POWER),
                 ResourceType.POWER,
                 0,
                 -1);
@@ -253,7 +256,12 @@ public class SimMachine {
                         gpuModel.getId(),
                         gpuModel.getId());
                 // Connect the GPU to the PSU
-                new FlowEdge(gpu, this.psu, ResourceType.POWER, gpuModel.getId(), gpuModel.getId());
+                new FlowEdge(
+                        gpu,
+                        this.distributors.get(ResourceType.POWER),
+                        ResourceType.POWER,
+                        gpuModel.getId(),
+                        gpuModel.getId());
             }
             this.computeResources.put(ResourceType.GPU, gpus);
         }
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 f40f4fec..ec4089cd 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
@@ -22,8 +22,6 @@
 
 package org.opendc.simulator.compute.power;
 
-import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.opendc.common.ResourceType;
@@ -33,27 +31,21 @@ import org.opendc.simulator.engine.graph.FlowConsumer;
 import org.opendc.simulator.engine.graph.FlowEdge;
 import org.opendc.simulator.engine.graph.FlowNode;
 import org.opendc.simulator.engine.graph.FlowSupplier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * A {@link SimPsu} implementation that estimates the power consumption based on CPU usage.
  */
 public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer {
-    private static final Logger LOGGER = LoggerFactory.getLogger(SimPsu.class);
     private long lastUpdate;
 
-    private final HashMap<ResourceType, HashMap<Integer, Double>> powerDemandsPerResource = new HashMap<>();
-    private final HashMap<ResourceType, HashMap<Integer, Double>> powerSuppliedPerResource = new HashMap<>();
-
-    private double totalPowerDemand = 0.0;
-    private double totalPowerSupplied = 0.0;
+    private double powerDemand = 0.0;
+    private double powerSupplied = 0.0;
     private double totalEnergyUsage = 0.0;
 
-    private final HashMap<ResourceType, HashMap<Integer, FlowEdge>> resourceEdges = new HashMap<>();
+    private FlowEdge cpuEdge;
     private FlowEdge powerSupplyEdge;
 
-    private final double capacity = Long.MAX_VALUE;
+    private double capacity = Long.MAX_VALUE;
 
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
     // Basic Getters and Setters
@@ -65,8 +57,7 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
      * @return <code>true</code> if the InPort is connected to an OutPort, <code>false</code> otherwise.
      */
     public boolean isConnected() {
-        return !this.resourceEdges.isEmpty()
-                && this.resourceEdges.values().stream().anyMatch(list -> !list.isEmpty());
+        return cpuEdge != null;
     }
 
     /**
@@ -75,51 +66,14 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
      * This method provides access to the power consumption of the machine before PSU losses are applied.
      */
     public double getPowerDemand() {
-        return this.totalPowerDemand;
-    }
-
-    /**
-     * Return the power demand of the machine (in W) measured in the PSU for a specific resource type.
-     * <p>
-     * This method provides access to the power consumption of the machine before PSU losses are applied.
-     */
-    public double getPowerDemand(ResourceType resourceType) {
-        //        return this.powerDemandsPerResource.get(resourceType).stream().mapToDouble(Double::doubleValue).sum();
-        return this.powerDemandsPerResource.get(resourceType).values().stream()
-                .mapToDouble(Double::doubleValue)
-                .sum();
-    }
-
-    /**
-     * Return the power demand of the machine (in W) measured in the PSU for a specific resource type for a specific resource.
-     * <p>
-     * This method provides access to the power consumption of the machine before PSU losses are applied.
-     */
-    public double getPowerDemand(ResourceType resourceType, int id) {
-        return this.powerDemandsPerResource.get(resourceType).get(id);
+        return this.powerDemand;
     }
 
     /**
      * Return the instantaneous power usage of the machine (in W) measured at the InPort of the power supply.
      */
     public double getPowerDraw() {
-        return this.totalPowerSupplied;
-    }
-
-    /**
-     * Return the instantaneous power usage of the machine (in W) measured at the InPort of the power supply for a specific resource type.
-     */
-    public double getPowerDraw(ResourceType resourceType) {
-        return this.powerSuppliedPerResource.get(resourceType).values().stream()
-                .mapToDouble(Double::doubleValue)
-                .sum();
-    }
-
-    /**
-     * Return the instantaneous power usage of the machine (in W) measured at the InPort of the power supply for a specific resource type for a specific resource.
-     */
-    public double getPowerDraw(ResourceType resourceType, int id) {
-        return this.powerSuppliedPerResource.get(resourceType).get(id);
+        return this.powerSupplied;
     }
 
     /**
@@ -152,26 +106,10 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
     @Override
     public long onUpdate(long now) {
         updateCounters();
-        for (ResourceType resourceType : this.resourceEdges.keySet()) {
-            HashMap<Integer, FlowEdge> edges = this.resourceEdges.get(resourceType);
-            if (edges != null && !edges.isEmpty()) {
-                for (FlowEdge edge : edges.values()) {
-                    // If the edge is null, it means that the edge has been removed -> no update is needed
-                    if (edge == null) {
-                        continue;
-                    }
-
-                    int consumerIndex = edge.getConsumerIndex() == -1 ? 0 : edge.getConsumerIndex();
-                    double powerDemand =
-                            this.powerDemandsPerResource.get(resourceType).get(consumerIndex);
-                    double powerSupplied =
-                            this.powerSuppliedPerResource.get(resourceType).get(consumerIndex);
-
-                    if (powerDemand != powerSupplied) {
-                        edge.pushSupply(powerDemand);
-                    }
-                }
-            }
+        double powerSupply = this.powerDemand;
+
+        if (powerSupply != this.powerSupplied) {
+            this.pushOutgoingSupply(this.cpuEdge, powerSupply);
         }
 
         return Long.MAX_VALUE;
@@ -190,12 +128,8 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
 
         long duration = now - lastUpdate;
         if (duration > 0) {
-            for (ResourceType resourceType : this.powerSuppliedPerResource.keySet()) {
-                for (double powerSupplied :
-                        this.powerSuppliedPerResource.get(resourceType).values()) {
-                    this.totalEnergyUsage += (powerSupplied * duration * 0.001);
-                }
-            }
+            // Compute the energy usage of the psu
+            this.totalEnergyUsage += (this.powerSupplied * duration * 0.001);
         }
     }
 
@@ -205,77 +139,37 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
 
     @Override
     public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) {
-        this.powerSupplyEdge.pushDemand(newDemand);
+        this.powerDemand = newDemand;
+        powerSupplyEdge.pushDemand(newDemand);
     }
 
     @Override
     public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply) {
-        this.pushOutgoingSupply(consumerEdge, newSupply, consumerEdge.getConsumerResourceType());
+        this.powerSupplied = newSupply;
+        cpuEdge.pushSupply(newSupply);
     }
 
     @Override
-    public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply, ResourceType resourceType) {
-        int consumerIndex = consumerEdge.getConsumerIndex() == -1 ? 0 : consumerEdge.getConsumerIndex();
-
-        double previousSupply = this.powerSuppliedPerResource.get(resourceType).get(consumerIndex);
-        this.totalPowerSupplied += newSupply - previousSupply;
+    public void handleIncomingDemand(FlowEdge consumerEdge, double newPowerDemand) {
 
-        this.powerSuppliedPerResource.get(resourceType).put(consumerIndex, newSupply);
+        updateCounters();
+        this.powerDemand = newPowerDemand;
 
-        consumerEdge.pushSupply(newSupply, false, resourceType);
+        pushOutgoingDemand(this.powerSupplyEdge, newPowerDemand);
     }
 
     @Override
-    public void handleIncomingDemand(FlowEdge consumerEdge, double newDemand) {
-        handleIncomingDemand(consumerEdge, newDemand, consumerEdge.getConsumerResourceType());
-    }
+    public void handleIncomingSupply(FlowEdge supplierEdge, double newPowerSupply) {
 
-    @Override
-    public void handleIncomingDemand(FlowEdge consumerEdge, double newPowerDemand, ResourceType resourceType) {
         updateCounters();
-        int consumerIndex = consumerEdge.getConsumerIndex() == -1 ? 0 : consumerEdge.getConsumerIndex();
-
-        double previousPowerDemand =
-                this.powerDemandsPerResource.get(resourceType).get(consumerIndex);
-        this.totalPowerDemand += newPowerDemand - previousPowerDemand;
+        this.powerSupplied = newPowerSupply;
 
-        this.powerDemandsPerResource.get(resourceType).put(consumerIndex, newPowerDemand);
-
-        pushOutgoingDemand(this.powerSupplyEdge, totalPowerDemand);
-    }
-
-    @Override
-    public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply) {
-        updateCounters();
-        for (ResourceType resourceType : this.resourceEdges.keySet()) {
-            for (FlowEdge edge : this.resourceEdges.get(resourceType).values()) {
-                // If the edge is null, it means that the edge has been removed -> no update is needed
-                if (edge == null) {
-                    continue;
-                }
-                int consumerIndex = edge.getConsumerIndex() == -1 ? 0 : edge.getConsumerIndex();
-                double outgoingSupply =
-                        Math.min(this.powerDemandsPerResource.get(resourceType).get(consumerIndex), newSupply);
-                pushOutgoingSupply(edge, outgoingSupply, resourceType);
-            }
-        }
+        pushOutgoingSupply(this.cpuEdge, newPowerSupply);
     }
 
     @Override
     public void addConsumerEdge(FlowEdge consumerEdge) {
-
-        ResourceType consumerResourceType = consumerEdge.getConsumerResourceType();
-        int consumerIndex = consumerEdge.getConsumerIndex() == -1 ? 0 : consumerEdge.getConsumerIndex();
-
-        if (!this.resourceEdges.containsKey(consumerResourceType)) {
-            this.resourceEdges.put(consumerResourceType, new HashMap<>());
-            this.powerDemandsPerResource.put(consumerResourceType, new HashMap<>());
-            this.powerSuppliedPerResource.put(consumerResourceType, new HashMap<>());
-        }
-
-        this.resourceEdges.get(consumerResourceType).put(consumerIndex, consumerEdge);
-        this.powerDemandsPerResource.get(consumerResourceType).put(consumerIndex, 0.0);
-        this.powerSuppliedPerResource.get(consumerResourceType).put(consumerIndex, 0.0);
+        this.cpuEdge = consumerEdge;
     }
 
     @Override
@@ -285,20 +179,7 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
 
     @Override
     public void removeConsumerEdge(FlowEdge consumerEdge) {
-        ResourceType resourceType = consumerEdge.getConsumerResourceType();
-        int consumerIndex = consumerEdge.getConsumerIndex() == -1 ? 0 : consumerEdge.getConsumerIndex();
-
-        if (this.resourceEdges.containsKey(resourceType)) {
-            this.resourceEdges.get(resourceType).put(consumerIndex, null);
-
-            this.totalPowerDemand -=
-                    this.powerDemandsPerResource.get(resourceType).get(consumerIndex);
-            this.powerDemandsPerResource.get(resourceType).put(consumerIndex, 0.0);
-
-            this.totalPowerSupplied -=
-                    this.powerSuppliedPerResource.get(resourceType).get(consumerIndex);
-            this.powerSuppliedPerResource.get(resourceType).put(consumerIndex, 0.0);
-        }
+        this.cpuEdge = null;
     }
 
     @Override
@@ -308,14 +189,7 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer
 
     @Override
     public Map<FlowEdge.NodeType, List<FlowEdge>> getConnectedEdges() {
-        List<FlowEdge> supplyingEdges = new ArrayList<>();
-        for (ResourceType resourceType : this.resourceEdges.keySet()) {
-            List<FlowEdge> edges =
-                    this.resourceEdges.get(resourceType).values().stream().toList();
-            if (edges != null && !edges.isEmpty()) {
-                supplyingEdges.addAll(edges);
-            }
-        }
+        List<FlowEdge> supplyingEdges = cpuEdge != null ? List.of(cpuEdge) : List.of();
         List<FlowEdge> consumingEdges = powerSupplyEdge != null ? List.of(powerSupplyEdge) : List.of();
 
         return Map.of(
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 95487476..39e17819 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
@@ -165,7 +165,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
             // The amount of work done since last update
             double finishedWork = this.scalingPolicy.getFinishedWork(
                     this.resourcesDemand.get(resourceType), this.resourcesSupplied.get(resourceType), passedTime);
-            this.remainingWork.put(resourceType, this.remainingWork.get(resourceType) - finishedWork);
+            this.remainingWork.put(resourceType, Math.max(0, this.remainingWork.get(resourceType) - finishedWork));
             this.totalRemainingWork -= finishedWork;
             if (this.remainingWork.get(resourceType) <= 0) {
                 this.workloadFinished.put(resourceType, true);
@@ -173,7 +173,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
         }
 
         // If this.totalRemainingWork <= 0, the fragment has been completed across all resources
-        if (this.totalRemainingWork <= 0 && !this.workloadFinished.containsValue(false)) {
+        if ((int) this.totalRemainingWork <= 0 && !this.workloadFinished.containsValue(false)) {
             this.startNextFragment();
 
             this.invalidate();
@@ -203,9 +203,11 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
                     this.resourcesSupplied.get(resourceType),
                     this.remainingWork.get(resourceType));
 
-            if (remainingDuration == 0.0) {
+            if ((int) remainingDuration == 0) {
                 // if resource not initialized, then nothing happens
-                this.totalRemainingWork -= this.remainingWork.get(resourceType);
+                if (this.remainingWork.get(resourceType) >= 0.0) {
+                    this.totalRemainingWork -= this.remainingWork.get(resourceType);
+                }
                 this.remainingWork.put(resourceType, 0.0);
                 this.workloadFinished.put(resourceType, true);
             }
@@ -218,7 +220,14 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
             }
         }
 
-        return timeUntilNextUpdate == Long.MIN_VALUE ? now : now + timeUntilNextUpdate;
+        long nextUpdate = timeUntilNextUpdate == Long.MAX_VALUE ? Long.MAX_VALUE : now + timeUntilNextUpdate;
+
+        // if for all resources the remaining work is 0, then invalidate the workload, to reschedule the next fragment
+        if (nextUpdate == now + Long.MIN_VALUE) {
+            this.invalidate();
+            return Long.MAX_VALUE;
+        }
+        return nextUpdate;
     }
 
     public TraceFragment getNextFragment() {
@@ -380,6 +389,10 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer {
      */
     @Override
     public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply, ResourceType resourceType) {
+        // for cases where equal share or fixed share is used and the resource is provided despite not being used
+        if (!this.usedResourceTypes.contains(resourceType)) {
+            return;
+        }
         if (this.resourcesSupplied.get(resourceType) == newSupply) {
             return;
         }
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowDistributor.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowDistributor.java
index 501bbf10..5cfd16ba 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowDistributor.java
+++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowDistributor.java
@@ -60,6 +60,7 @@ public abstract class FlowDistributor extends FlowNode implements FlowSupplier,
     protected Double totalIncomingSupply = 0.0; // The total supply provided by the suppliers
 
     protected boolean outgoingDemandUpdateNeeded = false;
+    protected boolean outgoingSupplyUpdateNeeded = false;
     protected Set<Integer> updatedDemands =
             new HashSet<>(); // Array of consumers that updated their demand in this cycle
 
@@ -142,6 +143,9 @@ public abstract class FlowDistributor extends FlowNode implements FlowSupplier,
         }
 
         this.totalIncomingDemand -= consumerEdge.getDemand();
+        if (this.totalIncomingDemand < 0) {
+            this.totalIncomingDemand = 0.0;
+        }
 
         // Remove idx from consumers that updated their demands
         this.updatedDemands.remove(idx);
@@ -204,6 +208,9 @@ public abstract class FlowDistributor extends FlowNode implements FlowSupplier,
         incomingDemands.set(idx, newDemand);
         // only update the total supply if the new supply is different from the previous one
         this.totalIncomingDemand += (newDemand - prevDemand);
+        if (totalIncomingDemand < 0) {
+            this.totalIncomingDemand = 0.0;
+        }
 
         this.updatedDemands.add(idx);
 
@@ -230,6 +237,7 @@ public abstract class FlowDistributor extends FlowNode implements FlowSupplier,
         // only update the total supply if the new supply is different from the previous one
         this.totalIncomingSupply += (newSupply - prevSupply);
 
+        this.outgoingSupplyUpdateNeeded = true;
         this.invalidate();
     }
 
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/BestEffortFlowDistributor.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/BestEffortFlowDistributor.java
index 4a13beb2..703aca35 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/BestEffortFlowDistributor.java
+++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/BestEffortFlowDistributor.java
@@ -148,6 +148,7 @@ public class BestEffortFlowDistributor extends FlowDistributor {
             }
         }
 
+        this.outgoingSupplyUpdateNeeded = false;
         this.updatedDemands.clear();
     }
 
@@ -264,7 +265,7 @@ public class BestEffortFlowDistributor extends FlowDistributor {
         }
 
         // Update supplies if needed
-        if (!this.outgoingSupplies.isEmpty() || updateNeeded) {
+        if (this.outgoingSupplyUpdateNeeded || updateNeeded) {
             this.updateOutgoingSupplies();
         }
 
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/EqualShareFlowDistributor.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/EqualShareFlowDistributor.java
index f58164cf..87ed7ca2 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/EqualShareFlowDistributor.java
+++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/EqualShareFlowDistributor.java
@@ -54,6 +54,8 @@ public class EqualShareFlowDistributor extends FlowDistributor {
         }
 
         this.outgoingDemandUpdateNeeded = false;
+        this.updatedDemands.clear();
+        this.invalidate();
     }
 
     /**
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FirstFitPolicyFlowDistributor.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FirstFitPolicyFlowDistributor.java
index c0a8cd13..9ab24d9f 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FirstFitPolicyFlowDistributor.java
+++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FirstFitPolicyFlowDistributor.java
@@ -74,6 +74,7 @@ public class FirstFitPolicyFlowDistributor extends FlowDistributor {
         }
 
         this.outgoingDemandUpdateNeeded = false;
+        this.invalidate();
     }
 
     /**
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FixedShareFlowDistributor.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FixedShareFlowDistributor.java
index 4c0a84d1..f22ea9fb 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FixedShareFlowDistributor.java
+++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FixedShareFlowDistributor.java
@@ -112,9 +112,6 @@ public class FixedShareFlowDistributor extends FlowDistributor {
         } else {
             double[] supplies = distributeSupply(this.incomingDemands, this.outgoingSupplies, this.totalIncomingSupply);
             for (FlowEdge consumerEdge : this.consumerEdges) {
-                if (supplies[consumerIndex] <= 0.0) {
-                    continue;
-                }
                 this.pushOutgoingSupply(consumerEdge, this.fixedShare);
             }
         }
@@ -123,7 +120,8 @@ public class FixedShareFlowDistributor extends FlowDistributor {
     public double[] distributeSupply(ArrayList<Double> demands, ArrayList<Double> currentSupply, double totalSupply) {
         double[] supplies = new double[this.consumerEdges.size()];
 
-        if (this.consumerEdges.size() < this.supplierEdges.size()) {
+        if (this.consumerEdges.size() < this.supplierEdges.size()
+                && this.fixedShare * this.consumerEdges.size() <= totalSupply) {
             for (FlowEdge consumerEdge : this.consumerEdges) {
                 supplies[consumerEdge.getConsumerIndex()] = this.fixedShare;
             }