diff options
| author | Niels Thiele <noleu66@posteo.net> | 2025-06-22 12:31:21 +0200 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2025-06-22 12:31:21 +0200 |
| commit | 0203254b709614fa732c114aa25916f61b8b3275 (patch) | |
| tree | 63232140a8e60e16e1668a51eb58954d8609fbdc /opendc-simulator | |
| parent | 8f846655347195bf6f22a4a102aa06f0ab127da1 (diff) | |
Implemented Single GPU Support & outline of host-level allocation policies (#342)
* renamed performance counter to distinguish different resource types
* added GPU, modelled similar to CPU
* added GPUs to machine model
* list of GPUs instead of single instance
* renamed memory speed to bandwidth
* enabled parsing of GPU resources
* split powermodel into cpu and GPU powermodel
* added gpu parsing tests
* added idea of host level scheduling
* added tests for multi gpu parsing
* renamed powermodel to cpupowermodel
* clarified naming of cpu and gpu components
* added resource type to flow suplier and edge
* added resourcetype
* added GPU components and resource type to fragments
* added GPU to workload and updated resource usage retrieval
* implemented first version of multi resource
* added name to workload
* renamed perfomance counters
* removed commented out code
* removed deprecated comments
* included demand and supply into calculations
* resolving rebase mismatches
* moved resource type from flowedge class to common package
* added available resources to machinees
* cleaner separation if workload is started of simmachine or vm
* Replaced exception with dedicated enum
* Only looping over resources that are actually used
* using hashmaps to handle resourcetype instead of arrays for readability
* fixed condition
* tracking finished workloads per resource type
* removed resource type from flowedge
* made supply and demand distribution resource specific
* added power model for GPU
* removed unused test setup
* removed depracated comments
* removed unused parameter
* added ID for GPU
* added GPUs and GPU performance counters (naively)
* implemented capturing of GPU statistics
* added reminders for future implementations
* renamed properties for better identification
* added capturing GPU statistics
* implemented first tests for GPUs
* unified access to performance counters
* added interface for general compute resource handling
* implemented multi resource support in simmachine
* added individual edge to VM per resource
* extended compute resource interface
* implemented multi-resource support in PSU
* implemented generic retrieval of computeresources
* implemented mult-resource suppport in vm
* made method use more resource specific
* implemented simple GPU tests
* rolled back frquency and demand use
* made naming independent of used resource
* using workloads resources instead of VMs to determine available resource
* implemented determination of used resources in workload
* removed logging statements
* implemented reading from workload
* fixed naming for host-level allocation
* fixed next deadline calculation
* fixed forwarding supply
* reduced memory footprint
* made GPU powermodel nullable
* maded Gpu powermodel configurable in topology
* implemented tests for basic gpu scheduler
* added gpu properties
* implemented weights, filter and simple cpu-gpu scheduler
* spotless apply
* spotless apply pt. 2
* fixed capitalization
* spotless kotlin run
* implemented coloumn export
* todo update
* removed code comments
* Merged PerformanceCounter classes into one & removed interface
* removed GPU specific powermodel
* Rebase master: kept both versions of TopologyFactories
* renamed CpuPowermodel to resource independent Powermodel
Moved it from Cpu package to power package
* implementated default of getResourceType & removed overrides if possible
* split getResourceType into Consumer and Supplier
* added power as resource type
* reduced supply demand from arrayList to single value
* combining GPUs into one large GPU, until full multi-gpu support
* merged distribution policy enum with corresponding factory
* added comment
* post-rebase fixes
* aligned naming
* Added GPU metrics to task output
* Updates power resource type to uppercase.
Standardizes the `ResourceType.Power` enum to `ResourceType.POWER`
for consistency with other resource types and improved readability.
* Removes deprecated test assertions
Removes commented-out assertions in GPU tests.
These assertions are no longer needed and clutter the test code.
* Renames MaxMinFairnessStrategy to Policy
Renames MaxMinFairnessStrategy to MaxMinFairnessPolicy for
clarity and consistency with naming conventions. This change
affects the factory and distributor to use the updated name.
* applies spotless
* nulls GPUs as it is not used
Diffstat (limited to 'opendc-simulator')
33 files changed, 1944 insertions, 379 deletions
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/ComputeResource.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/ComputeResource.java new file mode 100644 index 00000000..1167cf06 --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/ComputeResource.java @@ -0,0 +1,38 @@ +/* + * Copyright (c) 2025 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.compute; + +import org.opendc.simulator.compute.machine.PerformanceCounters; + +public interface ComputeResource { + + public int getId(); + + public PerformanceCounters getPerformanceCounters(); + + public double getCapacity(); + + public double getDemand(); + + public double getSupply(); +} 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 1a56650e..5669eb16 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 @@ -24,21 +24,29 @@ package org.opendc.simulator.compute.cpu; import java.util.List; import java.util.Map; +import org.opendc.common.ResourceType; +import org.opendc.simulator.compute.ComputeResource; import org.opendc.simulator.compute.machine.PerformanceCounters; import org.opendc.simulator.compute.models.CpuModel; +import org.opendc.simulator.compute.power.PowerModel; import org.opendc.simulator.engine.engine.FlowEngine; 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 SimCpu} of a machine. */ -public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer { +public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer, ComputeResource { + + private static final Logger LOGGER = LoggerFactory.getLogger(SimCpu.class); + private int id; private final CpuModel cpuModel; - private final CpuPowerModel cpuPowerModel; + private final PowerModel cpuPowerModel; private double currentCpuDemand = 0.0f; // cpu capacity demanded by the mux private double currentCpuUtilization = 0.0f; @@ -60,6 +68,10 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer // Basic Getters and Setters //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + public int getId() { + return id; + } + public double getFrequency() { return cpuModel.getTotalCapacity(); } @@ -87,7 +99,7 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer return this.currentCpuDemand; } - public double getSpeed() { + public double getSupply() { return this.currentCpuSupplied; } @@ -104,8 +116,9 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer // Constructors //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - public SimCpu(FlowEngine engine, CpuModel cpuModel, CpuPowerModel powerModel, int id) { + public SimCpu(FlowEngine engine, CpuModel cpuModel, PowerModel powerModel, int id) { super(engine); + this.id = id; this.cpuModel = cpuModel; this.maxCapacity = this.cpuModel.getTotalCapacity(); @@ -135,7 +148,7 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer this.currentCpuSupplied = Math.min(this.currentCpuDemand, this.maxCapacity); - this.pushOutgoingSupply(this.distributorEdge, this.currentCpuSupplied); + this.pushOutgoingSupply(this.distributorEdge, this.currentCpuSupplied, ResourceType.CPU); return Long.MAX_VALUE; } @@ -161,14 +174,14 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer final double factor = this.cpuFrequencyInv * delta; - this.performanceCounters.addCpuActiveTime(Math.round(rate * factor)); - this.performanceCounters.addCpuIdleTime(Math.round((capacity - rate) * factor)); - this.performanceCounters.addCpuStealTime(Math.round((demand - rate) * factor)); + this.performanceCounters.addActiveTime(Math.round(rate * factor)); + this.performanceCounters.addIdleTime(Math.round((capacity - rate) * factor)); + this.performanceCounters.addStealTime(Math.round((demand - rate) * factor)); } - this.performanceCounters.setCpuDemand(this.currentCpuDemand); - this.performanceCounters.setCpuSupply(this.currentCpuSupplied); - this.performanceCounters.setCpuCapacity(this.maxCapacity); + this.performanceCounters.setDemand(this.currentCpuDemand); + this.performanceCounters.setSupply(this.currentCpuSupplied); + this.performanceCounters.setCapacity(this.maxCapacity); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -182,7 +195,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); + this.psuEdge.pushDemand(newPowerDemand, false, ResourceType.CPU); } /** @@ -193,7 +206,15 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer updateCounters(); this.currentCpuSupplied = newCpuSupply; - this.distributorEdge.pushSupply(newCpuSupply, true); + this.distributorEdge.pushSupply(newCpuSupply, true, ResourceType.CPU); + } + + @Override + public void pushOutgoingSupply(FlowEdge consumerEdge, double newCpuSupply, ResourceType resourceType) { + updateCounters(); + this.currentCpuSupplied = newCpuSupply; + + this.distributorEdge.pushSupply(newCpuSupply, true, resourceType); } /** @@ -265,4 +286,14 @@ public final class SimCpu extends FlowNode implements FlowSupplier, FlowConsumer FlowEdge.NodeType.CONSUMING, List.of(this.psuEdge), FlowEdge.NodeType.SUPPLYING, List.of(this.distributorEdge)); } + + @Override + public ResourceType getSupplierResourceType() { + return ResourceType.CPU; + } + + @Override + public ResourceType getConsumerResourceType() { + return 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 new file mode 100644 index 00000000..c5778dc0 --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/gpu/SimGpu.java @@ -0,0 +1,295 @@ +/* + * Copyright (c) 2024 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.compute.gpu; + +import java.util.List; +import java.util.Map; +import org.opendc.common.ResourceType; +import org.opendc.simulator.compute.ComputeResource; +import org.opendc.simulator.compute.machine.PerformanceCounters; +import org.opendc.simulator.compute.models.GpuModel; +import org.opendc.simulator.compute.power.PowerModel; +import org.opendc.simulator.engine.engine.FlowEngine; +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 SimGpu} of a machine. + */ +public final class SimGpu extends FlowNode implements FlowSupplier, FlowConsumer, ComputeResource { + private static final Logger LOGGER = LoggerFactory.getLogger(SimGpu.class); + private final int id; + private final GpuModel gpuModel; + + private final PowerModel gpuPowerModel; + + private double currentGpuDemand = 0.0f; // cpu capacity demanded by the mux + private double currentGpuUtilization = 0.0f; + private double currentGpuSupplied = 0.0f; // cpu capacity supplied to the mux + + private double currentPowerDemand; // power demanded of the psu + private double currentPowerSupplied = 0.0f; // cpu capacity supplied by the psu + + private double maxCapacity; + + private final PerformanceCounters performanceCounters = new PerformanceCounters(); + private long lastCounterUpdate; + private final double gpuFrequencyInv; + + private FlowEdge distributorEdge; + private FlowEdge psuEdge; + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + // Basic Getters and Setters + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + public double getFrequency() { + return gpuModel.getTotalCoreCapacity(); + } + + public int getId() { + return id; + } + + @Override + public double getCapacity() { + return maxCapacity; + } // TODO: take memory into account + + public PerformanceCounters getPerformanceCounters() { + return performanceCounters; + } + + public double getPowerDraw() { + return this.currentPowerSupplied; + } + + public double getDemand() { + return this.currentGpuDemand; + } + + // TODO: take memory into account + public double getSupply() { + return this.currentGpuSupplied; + } // TODO: take memory into account + + public GpuModel getGpuModel() { + return gpuModel; + } + + @Override + public String toString() { + return "SimBareMetalMachine.Gpu[model=" + gpuModel + "]"; + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + // Constructors + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + public SimGpu(FlowEngine engine, GpuModel gpuModel, PowerModel powerModel, int id) { + super(engine); + this.id = id; + this.gpuModel = gpuModel; + this.maxCapacity = this.gpuModel.getTotalCoreCapacity(); + + this.gpuPowerModel = powerModel; + + this.lastCounterUpdate = clock.millis(); + + this.gpuFrequencyInv = 1 / this.maxCapacity; + + this.currentPowerDemand = this.gpuPowerModel.computePower(this.currentGpuUtilization); + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + // FlowNode related functionality + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + @Override + public long onUpdate(long now) { + updateCounters(now); + + // Check if supply == demand + if (this.currentPowerDemand != this.currentPowerSupplied) { + this.pushOutgoingDemand(this.psuEdge, this.currentPowerDemand); + + return Long.MAX_VALUE; + } + + this.currentGpuSupplied = Math.min(this.currentGpuDemand, this.maxCapacity); + this.pushOutgoingSupply(this.distributorEdge, this.currentGpuSupplied); + + return Long.MAX_VALUE; + } + + public void updateCounters() { + this.updateCounters(this.clock.millis()); + } + + /** + * Update the performance counters of the GPU. + * + * @param now The timestamp at which to update the counter. + */ + public void updateCounters(long now) { + long lastUpdate = this.lastCounterUpdate; + this.lastCounterUpdate = now; + long delta = now - lastUpdate; + + if (delta > 0) { + double demand = this.currentGpuDemand; + double rate = this.currentGpuSupplied; + double capacity = this.maxCapacity; + + final double factor = this.gpuFrequencyInv * delta; + + this.performanceCounters.addActiveTime(Math.round(rate * factor)); + this.performanceCounters.addIdleTime(Math.round((capacity - rate) * factor)); + this.performanceCounters.addStealTime(Math.round((demand - rate) * factor)); + } + + this.performanceCounters.setDemand(this.currentGpuDemand); + this.performanceCounters.setSupply(this.currentGpuSupplied); + this.performanceCounters.setCapacity(this.maxCapacity); + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + // FlowGraph Related functionality + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + /** + * Push new demand to the psu + */ + @Override + public void pushOutgoingDemand(FlowEdge supplierEdge, double newPowerDemand) { + updateCounters(); + this.currentPowerDemand = newPowerDemand; + this.psuEdge.pushDemand(newPowerDemand, false, ResourceType.GPU); + } + + /** + * Push updated supply to the mux + */ + @Override + public void pushOutgoingSupply(FlowEdge consumerEdge, double newGpuSupply) { + updateCounters(); + this.currentGpuSupplied = newGpuSupply; + + this.distributorEdge.pushSupply(newGpuSupply, true, ResourceType.GPU); + } + + /** + * Push updated supply to the mux + */ + @Override + public void pushOutgoingSupply(FlowEdge consumerEdge, double newGpuSupply, ResourceType resourceType) { + updateCounters(); + this.currentGpuSupplied = newGpuSupply; + + this.distributorEdge.pushSupply(newGpuSupply, true, resourceType); + } + + /** + * Handle new demand coming in from the mux + */ + @Override + public void handleIncomingDemand(FlowEdge consumerEdge, double newGpuDemand) { + updateCounters(); + this.currentGpuDemand = newGpuDemand; + + this.currentGpuUtilization = Math.min(this.currentGpuDemand / this.maxCapacity, 1.0); + + // Calculate Power Demand and send to PSU + this.currentPowerDemand = this.gpuPowerModel.computePower(this.currentGpuUtilization); + + this.invalidate(); + } + + /** + * Handle updated supply from the psu + */ + @Override + public void handleIncomingSupply(FlowEdge supplierEdge, double newPowerSupply) { + updateCounters(); + this.currentPowerSupplied = newPowerSupply; + + this.invalidate(); + } + + /** + * Add a connection to the mux + */ + @Override + public void addConsumerEdge(FlowEdge consumerEdge) { + this.distributorEdge = consumerEdge; + } + + /** + * Add a connection to the psu + */ + @Override + public void addSupplierEdge(FlowEdge supplierEdge) { + this.psuEdge = supplierEdge; + + this.invalidate(); + } + + /** + * Remove the connection to the mux + */ + @Override + public void removeConsumerEdge(FlowEdge consumerEdge) { + this.distributorEdge = null; + this.invalidate(); + } + + /** + * Remove the connection to the psu + */ + @Override + public void removeSupplierEdge(FlowEdge supplierEdge) { + this.psuEdge = null; + this.invalidate(); + } + + @Override + public Map<FlowEdge.NodeType, List<FlowEdge>> getConnectedEdges() { + return Map.of( + FlowEdge.NodeType.CONSUMING, List.of(this.psuEdge), + FlowEdge.NodeType.SUPPLYING, List.of(this.distributorEdge)); + } + + @Override + public ResourceType getSupplierResourceType() { + return ResourceType.GPU; + } + + @Override + public ResourceType getConsumerResourceType() { + return ResourceType.GPU; + } +} 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 f5b8d27d..93033bc0 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 @@ -23,80 +23,97 @@ package org.opendc.simulator.compute.machine; public class PerformanceCounters { - private long cpuActiveTime = 0; - private long cpuIdleTime = 0; - private long cpuStealTime = 0; - private long cpuLostTime = 0; - private double cpuCapacity = 0.0f; - private double cpuDemand = 0.0f; - private double cpuSupply = 0.0f; + private long activeTime = 0; + private long idleTime = 0; + private long stealTime = 0; + private long lostTime = 0; - public long getCpuActiveTime() { - return cpuActiveTime; + private double capacity = 0.0f; + private double demand = 0.0f; + private double supply = 0.0f; + + public long getActiveTime() { + return this.activeTime; + } + + public long getIdleTime() { + return this.idleTime; + } + + public long getStealTime() { + return this.stealTime; + } + + public long getLostTime() { + return this.lostTime; + } + + public double getCapacity() { + return this.capacity; } - public void setCpuActiveTime(long cpuActiveTime) { - this.cpuActiveTime = cpuActiveTime; + public double getDemand() { + return this.demand; } - public void addCpuActiveTime(long cpuActiveTime) { - this.cpuActiveTime += cpuActiveTime; + public double getSupply() { + return this.supply; } - public long getCpuIdleTime() { - return cpuIdleTime; + public void setActiveTime(long activeTime) { + this.activeTime = activeTime; } - public void setCpuIdleTime(long cpuIdleTime) { - this.cpuIdleTime = cpuIdleTime; + public void setIdleTime(long idleTime) { + this.idleTime = idleTime; } - public void addCpuIdleTime(long cpuIdleTime) { - this.cpuIdleTime += cpuIdleTime; + public void setStealTime(long stealTime) { + this.stealTime = stealTime; } - public long getCpuStealTime() { - return cpuStealTime; + public void setLostTime(long lostTime) { + this.lostTime = lostTime; } - public void setCpuStealTime(long cpuStealTime) { - this.cpuStealTime = cpuStealTime; + public void setCapacity(double capacity) { + this.capacity = capacity; } - public void addCpuStealTime(long cpuStealTime) { - this.cpuStealTime += cpuStealTime; + public void setDemand(double demand) { + this.demand = demand; } - public long getCpuLostTime() { - return cpuLostTime; + public void setSupply(double supply) { + this.supply = supply; } - public void setCpuLostTime(long cpuLostTime) { - this.cpuLostTime = cpuLostTime; + public void addActiveTime(long activeTime) { + this.activeTime += activeTime; } - public double getCpuCapacity() { - return cpuCapacity; + public void addIdleTime(long idleTime) { + this.idleTime += idleTime; } - public void setCpuCapacity(double cpuCapacity) { - this.cpuCapacity = cpuCapacity; + public void addStealTime(long stealTime) { + this.stealTime += stealTime; } - public double getCpuDemand() { - return cpuDemand; + public void addLostTime(long lostTime) { + this.lostTime += lostTime; } - public void setCpuDemand(double cpuDemand) { - this.cpuDemand = cpuDemand; + public void addCapacity(double capacity) { + this.capacity += capacity; } - public double getCpuSupply() { - return cpuSupply; + public void addDemand(double demand) { + this.demand += demand; } - public void setCpuSupply(double cpuSupply) { - this.cpuSupply = cpuSupply; + public void addSupply(double supply) { + this.supply += supply; } } 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 8baa7f34..8792552e 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 @@ -23,18 +23,29 @@ package org.opendc.simulator.compute.machine; import java.time.InstantSource; +import java.util.ArrayList; +import java.util.Hashtable; +import java.util.List; import java.util.function.Consumer; -import org.opendc.simulator.compute.cpu.CpuPowerModel; +import org.jetbrains.annotations.Nullable; +import org.opendc.common.ResourceType; +import org.opendc.simulator.compute.ComputeResource; import org.opendc.simulator.compute.cpu.SimCpu; +import org.opendc.simulator.compute.gpu.SimGpu; import org.opendc.simulator.compute.memory.Memory; +import org.opendc.simulator.compute.models.GpuModel; import org.opendc.simulator.compute.models.MachineModel; +import org.opendc.simulator.compute.power.PowerModel; import org.opendc.simulator.compute.power.SimPsu; import org.opendc.simulator.compute.workload.ChainWorkload; import org.opendc.simulator.compute.workload.SimWorkload; import org.opendc.simulator.compute.workload.VirtualMachine; import org.opendc.simulator.engine.engine.FlowEngine; +import org.opendc.simulator.engine.graph.FlowConsumer; import org.opendc.simulator.engine.graph.FlowDistributor; import org.opendc.simulator.engine.graph.FlowEdge; +import org.opendc.simulator.engine.graph.FlowNode; +import org.opendc.simulator.engine.graph.FlowSupplier; /** * A machine that is able to execute {@link SimWorkload} objects. @@ -45,19 +56,63 @@ public class SimMachine { private final InstantSource clock; - private SimCpu cpu; - private FlowDistributor cpuDistributor; private SimPsu psu; private Memory memory; + private final Hashtable<ResourceType, FlowDistributor> distributors = new Hashtable<>(); + + private final Hashtable<ResourceType, ArrayList<ComputeResource>> computeResources = new Hashtable<>(); + private final List<ResourceType> availableResources; + private final Consumer<Exception> completion; //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Basic Getters and Setters //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + public ComputeResource getResource(ResourceType resourceType, int id) { + if (!this.computeResources.containsKey(resourceType)) { + throw new RuntimeException("No such resource type: " + resourceType); + } + for (ComputeResource resource : this.computeResources.get(resourceType)) { + if (resource.getId() == id) { + return resource; + } + } + throw new RuntimeException("No such resource with id: " + id + " of type: " + resourceType); + } + + public ArrayList<ComputeResource> getResources(ResourceType resourceType) { + if (!this.computeResources.containsKey(resourceType)) { + throw new RuntimeException("No such resource type: " + resourceType); + } + return this.computeResources.get(resourceType); + } + public PerformanceCounters getPerformanceCounters() { - return this.cpu.getPerformanceCounters(); + + return this.computeResources.get(ResourceType.CPU).getFirst().getPerformanceCounters(); + } + + public List<PerformanceCounters> getGpuPerformanceCounters() { + List<PerformanceCounters> counters = new ArrayList<>(); + List<ComputeResource> gpus = this.computeResources.get(ResourceType.GPU) == null + ? new ArrayList<>() + : this.computeResources.get(ResourceType.GPU); + + for (ComputeResource gpu : gpus) { + counters.add(gpu.getPerformanceCounters()); + } + return counters; + } + + public PerformanceCounters getGpuPerformanceCounters(int GpuId) { + for (ComputeResource gpu : this.computeResources.get(ResourceType.GPU)) { + if (gpu.getId() == GpuId) { + return gpu.getPerformanceCounters(); + } + } + throw new RuntimeException("No such gpu id: " + GpuId); } public MachineModel getMachineModel() { @@ -73,7 +128,7 @@ public class SimMachine { } public SimCpu getCpu() { - return cpu; + return (SimCpu) this.computeResources.get(ResourceType.CPU).getFirst(); } public Memory getMemory() { @@ -84,6 +139,28 @@ public class SimMachine { return psu; } + public ArrayList<SimGpu> getGpus() { + ArrayList<SimGpu> gpus = new ArrayList<>(); + if (!this.computeResources.containsKey(ResourceType.GPU)) { + return gpus; + } + for (ComputeResource gpu : this.computeResources.get(ResourceType.GPU)) { + if (gpu instanceof SimGpu) { + gpus.add((SimGpu) gpu); + } + } + return gpus; + } + + public SimGpu getGpu(int gpuId) { + for (ComputeResource gpu : this.computeResources.get(ResourceType.GPU)) { + if (gpu.getId() == gpuId) { + return (SimGpu) gpu; + } + } + throw new RuntimeException("No such gpu id: " + gpuId); + } + /** * Return the CPU capacity of the hypervisor in MHz. */ @@ -105,6 +182,10 @@ public class SimMachine { return 0.0; } + public List<ResourceType> getAvailableResources() { + return availableResources; + } + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Constructors //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -113,27 +194,47 @@ public class SimMachine { FlowEngine engine, MachineModel machineModel, FlowDistributor powerDistributor, - CpuPowerModel cpuPowerModel, + PowerModel cpuPowerModel, + @Nullable PowerModel gpuPowerModel, Consumer<Exception> completion) { this.engine = engine; this.machineModel = machineModel; this.clock = engine.getClock(); + this.availableResources = this.machineModel.getUsedResources(); + // Create the psu and cpu and connect them this.psu = new SimPsu(engine); - new FlowEdge(this.psu, powerDistributor); - this.cpu = new SimCpu(engine, this.machineModel.getCpuModel(), cpuPowerModel, 0); - - new FlowEdge(this.cpu, this.psu); + this.computeResources.put( + ResourceType.CPU, + new ArrayList<>(List.of(new SimCpu(engine, this.machineModel.getCpuModel(), cpuPowerModel, 0)))); - this.memory = new Memory(engine, this.machineModel.getMemory()); + new FlowEdge((FlowConsumer) this.computeResources.get(ResourceType.CPU).getFirst(), this.psu); // Create a FlowDistributor and add the cpu as supplier - this.cpuDistributor = new FlowDistributor(engine); + this.distributors.put(ResourceType.CPU, new FlowDistributor(engine)); + new FlowEdge(this.distributors.get(ResourceType.CPU), (FlowSupplier) + this.computeResources.get(ResourceType.CPU).getFirst()); - new FlowEdge(this.cpuDistributor, this.cpu); + // TODO: include memory as flow node + this.memory = new Memory(engine, this.machineModel.getMemory()); + + if (this.availableResources.contains(ResourceType.GPU)) { + this.distributors.put(ResourceType.GPU, new FlowDistributor(engine)); + short i = 0; + ArrayList<ComputeResource> gpus = new ArrayList<>(); + + for (GpuModel gpuModel : machineModel.getGpuModels()) { + SimGpu gpu = new SimGpu(engine, gpuModel, gpuPowerModel, i); + gpus.add(gpu); + // suspends here without the distributor + new FlowEdge(this.distributors.get(ResourceType.GPU), gpu); + new FlowEdge(gpu, this.psu); + } + this.computeResources.put(ResourceType.GPU, gpus); + } this.completion = completion; } @@ -149,14 +250,20 @@ public class SimMachine { this.psu.closeNode(); this.psu = null; - this.cpu.closeNode(); - this.cpu = null; - - this.cpuDistributor.closeNode(); - this.cpuDistributor = null; - + // Close resource Flow Nodes + for (List<ComputeResource> resources : this.computeResources.values()) { + for (ComputeResource resource : resources) { + ((FlowNode) resource).closeNode(); + } + resources.clear(); + } this.memory = null; + for (ResourceType resourceType : this.distributors.keySet()) { + this.distributors.get(resourceType).closeNode(); + } + this.distributors.clear(); + this.completion.accept(cause); } @@ -180,6 +287,12 @@ public class SimMachine { * @param completion The completion callback that needs to be called when the workload is done */ public VirtualMachine startWorkload(ChainWorkload workload, Consumer<Exception> completion) { - return (VirtualMachine) workload.startWorkload(this.cpuDistributor, this, completion); + + ArrayList<FlowSupplier> distributors = new ArrayList<>(); + for (ResourceType resourceType : this.availableResources) { + distributors.add(this.distributors.get(resourceType)); + } + + return (VirtualMachine) workload.startWorkload(distributors, this, completion); } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/CpuModel.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/CpuModel.java index ab829bc4..903a985e 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/CpuModel.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/CpuModel.java @@ -130,7 +130,7 @@ public final class CpuModel { @Override public String toString() { - return "ProcessingUnit[" + "id= " + id + ", coreCount= " + coreCount + ", coreSpeed= " + coreSpeed + return "ProcessingUnit[" + "id= " + id + ", cpuCoreCount= " + coreCount + ", coreSpeed= " + coreSpeed + ", frequency= " + totalCapacity + ", vendor= " + vendor + ", modelName= " + modelName + ", arch= " + arch + "]"; } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/GpuModel.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/GpuModel.java new file mode 100644 index 00000000..b804b061 --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/GpuModel.java @@ -0,0 +1,179 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.compute.models; + +import java.util.Objects; + +/** + * A single logical compute unit of processor node, either virtual or physical. + */ +public final class GpuModel { + private final int id; + private final int coreCount; + private final double coreSpeed; + private final double totalCoreCapacity; + private final double memoryBandwidth; + private final long memorySize; + private final String vendor; + private final String modelName; + private final String arch; + + /** + * Construct a {@link GpuModel} instance. + * + * @param id The identifier of the GPU core within the processing node. + * @param coreCount The number of cores present in the GPU + * @param coreSpeed The speed of a single core + * @param memoryBandwidth The speed of the memory in MHz + * @param memorySize The memory size of the GPU + * @param vendor The vendor of the GPU + * @param modelName The name of the GPU + * @param arch The architecture of the GPU + */ + public GpuModel( + int id, + int coreCount, + double coreSpeed, + double memoryBandwidth, + long memorySize, + String vendor, + String modelName, + String arch) { + this.id = id; + this.coreCount = coreCount; + this.coreSpeed = coreSpeed; + this.memoryBandwidth = memoryBandwidth; + this.memorySize = memorySize; + this.totalCoreCapacity = coreSpeed * coreCount; + this.vendor = vendor; + this.modelName = modelName; + this.arch = arch; + } + + /** + * Construct a {@link GpuModel} instance. Purely as a processing unit + * + * @param id The identifier of the GPU core within the processing node. + * @param coreCount The number of cores present in the GPU + * @param coreSpeed The speed of a single core + */ + public GpuModel(int id, int coreCount, double coreSpeed) { + this(id, coreCount, coreSpeed, 0, 0, "unkown", "unkown", "unkown"); + } + + public GpuModel(int id, int coreCount, double coreSpeed, double memoryBandwidth, long memorySize) { + this(id, coreCount, coreSpeed, memoryBandwidth, memorySize, "unkown", "unkown", "unkown"); + } + + /** + * Return the identifier of the GPU core within the processing node. + */ + public int getId() { + return id; + } + + /** + * Return the number of logical GPUs in the processor node. + */ + public int getCoreCount() { + return coreCount; + } + + /** + * Return the clock rate of a single core of the GPU MHz. + */ + public double getCoreSpeed() { + return coreSpeed; + } + + /** + * Return the clock rate of the GPU in MHz. + */ + public double getTotalCoreCapacity() { + return totalCoreCapacity; + } + + /** + * Return the speed of the memory in Mhz. + */ + public double getMemoryBandwidth() { + return memoryBandwidth; + } + + /** + * Return the size of the memory in MB. + */ + public long getMemorySize() { + return memorySize; + } + + /** + * Return the vendor of the storage device. + */ + public String getVendor() { + return vendor; + } + + /** + * Return the model name of the device. + */ + public String getModelName() { + return modelName; + } + + /** + * Return the micro-architecture of the processor node. + */ + public String getArchitecture() { + return arch; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + GpuModel that = (GpuModel) o; + return id == that.id + && Double.compare(that.totalCoreCapacity, totalCoreCapacity) == 0 + && Double.compare(that.coreSpeed, coreSpeed) == 0 + && Double.compare(that.memoryBandwidth, memoryBandwidth) == 0 + && Double.compare(that.memorySize, memorySize) == 0 + && Objects.equals(vendor, that.vendor) + && Objects.equals(modelName, that.modelName) + && Objects.equals(arch, that.arch); + } + + @Override + public int hashCode() { + return Objects.hash( + id, coreCount, coreSpeed, totalCoreCapacity, memoryBandwidth, memorySize, vendor, modelName, arch); + } + + @Override + public String toString() { + return "ProcessingUnit[" + "id= " + id + ", gpuCoreCount= " + coreCount + ", gpuCoreSpeed= " + coreSpeed + + ", frequency= " + totalCoreCapacity + ", gpuMemoryBandwidth" + memoryBandwidth + ", gpuMemorySize" + + memorySize + ", vendor= " + vendor + ", modelName= " + modelName + ", arch= " + + arch + "]"; + } +} diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/MachineModel.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/MachineModel.java index 6c47fbe6..874194f6 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/MachineModel.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/models/MachineModel.java @@ -22,8 +22,12 @@ package org.opendc.simulator.compute.models; +import java.util.ArrayList; import java.util.List; import java.util.Objects; +import org.jetbrains.annotations.Nullable; +import org.opendc.common.ResourceType; +import org.opendc.simulator.engine.graph.distributionPolicies.DistributionPolicy; /** * A description of the physical or virtual machine on which a bootable image runs. @@ -31,16 +35,46 @@ import java.util.Objects; public final class MachineModel { private final CpuModel cpuModel; private final MemoryUnit memory; - + // private final List<GpuModel> gpuModels = new ArrayList<>(); // TODO: Implement multi GPU support + private final List<GpuModel> gpuModels; + private final DistributionPolicy cpuDistributionStrategy; + private final DistributionPolicy gpuDistributionPolicy; + private final List<ResourceType> availableResources = new ArrayList<>(); /** * Construct a {@link MachineModel} instance. * * @param cpuModel The cpu available to the image. * @param memory The list of memory units available to the image. */ - public MachineModel(CpuModel cpuModel, MemoryUnit memory) { + public MachineModel( + CpuModel cpuModel, + MemoryUnit memory, + @Nullable List<GpuModel> gpuModels, + DistributionPolicy cpuDistributionPolicy, + DistributionPolicy gpuDistributionPolicy) { this.cpuModel = cpuModel; this.memory = memory; + this.cpuDistributionStrategy = cpuDistributionPolicy; + this.gpuDistributionPolicy = gpuDistributionPolicy; + this.availableResources.add(ResourceType.CPU); + // TODO: Add Memory + // this.usedResources.add(ResourceType.Memory); + if (gpuModels != null && !gpuModels.isEmpty()) { + // this.gpuModels = gpuModels; + this.gpuModels = new ArrayList<>(); + this.gpuModels.add(new GpuModel( + 0, + gpuModels.getFirst().getCoreCount() * gpuModels.size(), // merges multiple GPUs into one + gpuModels.getFirst().getCoreSpeed(), + gpuModels.getFirst().getMemoryBandwidth(), + gpuModels.getFirst().getMemorySize() * gpuModels.size(), // merges multiple GPUs into one + gpuModels.getFirst().getVendor(), + gpuModels.getFirst().getModelName(), + gpuModels.getFirst().getArchitecture())); + this.availableResources.add(ResourceType.GPU); + } else { + this.gpuModels = new ArrayList<>(); + } } /** @@ -61,7 +95,40 @@ public final class MachineModel { cpus.get(0).getVendor(), cpus.get(0).getModelName(), cpus.get(0).getArchitecture()), - memory); + memory, + null, + null, + null); + } + + /** + * Construct a {@link MachineModel} instance. + * A list of the same cpus, are automatically converted to a single CPU with the number of cores of + * all cpus in the list combined. + * + * @param cpus The list of processing units available to the image. + * @param memory The list of memory units available to the image. + * @param gpus The list of GPUs available to the image. + */ + public MachineModel( + List<CpuModel> cpus, + MemoryUnit memory, + List<GpuModel> gpus, + DistributionPolicy cpuDistributionPolicy, + DistributionPolicy gpuDistributionPolicy) { + + this( + new CpuModel( + cpus.get(0).getId(), + cpus.get(0).getCoreCount() * cpus.size(), // merges multiple CPUs into one + cpus.get(0).getCoreSpeed(), + cpus.get(0).getVendor(), + cpus.get(0).getModelName(), + cpus.get(0).getArchitecture()), + memory, + gpus != null ? gpus : new ArrayList<>(), + cpuDistributionPolicy, + gpuDistributionPolicy); } /** @@ -78,21 +145,60 @@ public final class MachineModel { return memory; } + public List<GpuModel> getGpuModels() { + return gpuModels; + } + + /** + * Return specific GPU model by id. + * @param modelId The id of the GPU model to return. + * @return The GPU model with the given id, or null if not found. + */ + public GpuModel getGpuModel(int modelId) { + for (GpuModel gpuModel : gpuModels) { + if (gpuModel.getId() == modelId) { + return gpuModel; + } + } + return null; + } + + /** + * Return the distribution strategy for the CPU. + */ + public DistributionPolicy getCpuDistributionStrategy() { + return cpuDistributionStrategy; + } + + /** + * Return the distribution strategy for the GPU. + */ + public DistributionPolicy getGpuDistributionStrategy() { + return gpuDistributionPolicy; + } + + /** + * Return the resources of this machine. + */ + public List<ResourceType> getUsedResources() { + return availableResources; + } + @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; MachineModel that = (MachineModel) o; - return cpuModel.equals(that.cpuModel) && memory.equals(that.memory); + return cpuModel.equals(that.cpuModel) && memory.equals(that.memory) && gpuModels.equals(that.gpuModels); } @Override public int hashCode() { - return Objects.hash(cpuModel, memory); + return Objects.hash(cpuModel, memory, gpuModels); } @Override public String toString() { - return "MachineModel[cpus=" + cpuModel + ",memory=" + memory + "]"; + return "MachineModel[cpus=" + cpuModel + ",memory=" + memory + ",gpus=" + gpuModels + "]"; } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/CpuPowerModel.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/PowerModel.java index 4323294e..597b6fc3 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/CpuPowerModel.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/PowerModel.java @@ -20,16 +20,16 @@ * SOFTWARE. */ -package org.opendc.simulator.compute.cpu; +package org.opendc.simulator.compute.power; import org.opendc.simulator.compute.machine.SimMachine; /** - * A model for estimating the power usage of a {@link SimMachine} based on the CPU usage. + * A model for estimating the power usage of a {@link SimMachine} based on the resource usage. */ -public interface CpuPowerModel { +public interface PowerModel { /** - * Computes CPU power consumption for each host. + * Computes resource power consumption for each host. * * @param utilization The CPU utilization percentage. * @return A double value of CPU power consumption (in W). diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/CpuPowerModels.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/PowerModels.java index b91bd7e2..af532908 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/CpuPowerModels.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/PowerModels.java @@ -20,118 +20,118 @@ * SOFTWARE. */ -package org.opendc.simulator.compute.cpu; +package org.opendc.simulator.compute.power; import java.util.Arrays; /** * A collection {@link CpuPowerModel} implementations. */ -public class CpuPowerModels { - private CpuPowerModels() {} +public class PowerModels { + private PowerModels() {} /** - * Construct a constant {@link CpuPowerModel}. + * Construct a constant {@link PowerModel}. * * @param power The power consumption of the host at all times (in W). */ - public static CpuPowerModel constant(double power) { + public static PowerModel constant(double power) { return new ConstantPowerModel(power); } /** - * Construct a square root {@link CpuPowerModel} that is adapted from CloudSim. + * Construct a square root {@link PowerModel} that is adapted from CloudSim. * * @param maxPower The maximum power draw of the host in W. - * @param idlePower The power draw of the host at its lowest utilization level in W. + * @param idlePower The power draw of the host at its lowest resource utilization level in W. */ - public static CpuPowerModel sqrt(double maxPower, double idlePower) { + public static PowerModel sqrt(double maxPower, double idlePower) { return new SqrtPowerModel(maxPower, idlePower); } /** - * Construct a linear {@link CpuPowerModel} that is adapted from CloudSim. + * Construct a linear {@link PowerModel} that is adapted from CloudSim. * * @param maxPower The maximum power draw of the host in W. - * @param idlePower The power draw of the host at its lowest utilization level in W. + * @param idlePower The power draw of the host at its lowest resource utilization level in W. */ - public static CpuPowerModel linear(double maxPower, double idlePower) { + public static PowerModel linear(double maxPower, double idlePower) { return new LinearPowerModel(maxPower, idlePower); } /** - * Construct a square {@link CpuPowerModel} that is adapted from CloudSim. + * Construct a square {@link PowerModel} that is adapted from CloudSim. * * @param maxPower The maximum power draw of the host in W. - * @param idlePower The power draw of the host at its lowest utilization level in W. + * @param idlePower The power draw of the host at its lowest resource utilization level in W. */ - public static CpuPowerModel square(double maxPower, double idlePower) { + public static PowerModel square(double maxPower, double idlePower) { return new SquarePowerModel(maxPower, idlePower); } /** - * Construct a cubic {@link CpuPowerModel} that is adapted from CloudSim. + * Construct a cubic {@link PowerModel} that is adapted from CloudSim. * * @param maxPower The maximum power draw of the host in W. - * @param idlePower The power draw of the host at its lowest utilization level in W. + * @param idlePower The power draw of the host at its lowest resource utilization level in W. */ - public static CpuPowerModel cubic(double maxPower, double idlePower) { + public static PowerModel cubic(double maxPower, double idlePower) { return new CubicPowerModel(maxPower, idlePower); } /** - * Construct a {@link CpuPowerModel} that minimizes the mean squared error (MSE) + * Construct a {@link PowerModel} that minimizes the mean squared error (MSE) * to the actual power measurement by tuning the calibration parameter. * * @param maxPower The maximum power draw of the host in W. - * @param idlePower The power draw of the host at its lowest utilization level in W. + * @param idlePower The power draw of the host at its lowest resource utilization level in W. * @param calibrationFactor The parameter set to minimize the MSE. * @see <a href="https://dl.acm.org/doi/abs/10.1145/1273440.1250665"> * Fan et al., Power provisioning for a warehouse-sized computer, ACM SIGARCH'07</a> */ - public static CpuPowerModel mse(double maxPower, double idlePower, double calibrationFactor) { + public static PowerModel mse(double maxPower, double idlePower, double calibrationFactor) { return new MsePowerModel(maxPower, idlePower, calibrationFactor); } /** - * Construct an asymptotic {@link CpuPowerModel} adapted from GreenCloud. + * Construct an asymptotic {@link PowerModel} adapted from GreenCloud. * * @param maxPower The maximum power draw of the host in W. - * @param idlePower The power draw of the host at its lowest utilization level in W. + * @param idlePower The power draw of the host at its lowest resource utilization level in W. * @param asymUtil A utilization level at which the host attains asymptotic, * i.e., close to linear power consumption versus the offered load. - * For most of the CPUs,a is in [0.2, 0.5]. + * For most of the s,a is in [0.2, 0.5]. * @param dvfs A flag indicates whether DVFS is enabled. */ - public static CpuPowerModel asymptotic(double maxPower, double idlePower, double asymUtil, boolean dvfs) { + public static PowerModel asymptotic(double maxPower, double idlePower, double asymUtil, boolean dvfs) { return new AsymptoticPowerModel(maxPower, idlePower, asymUtil, dvfs); } /** - * Construct a linear interpolation model {@link CpuPowerModel} that is adapted from CloudSim. + * Construct a linear interpolation model {@link PowerModel} that is adapted from CloudSim. * * <p> * The power consumption is linearly interpolated over the given power levels. In case of two values, the first * represents 0% utilization, while the last value represent 100% utilization. * - * @param powerLevels An array of power consumption steps (in W) for a specific CPU utilization. + * @param powerLevels An array of power consumption steps (in W) for a specific utilization. * @see <a href="http://www.spec.org/power_ssj2008/results/res2011q1/">Machines used in the SPEC benchmark</a> */ - public static CpuPowerModel interpolate(double... powerLevels) { + public static PowerModel interpolate(double... powerLevels) { return new InterpolationPowerModel(powerLevels.clone()); } /** - * Decorate an existing {@link CpuPowerModel} to ensure that zero power consumption is reported when there is no + * Decorate an existing {@link PowerModel} to ensure that zero power consumption is reported when there is no * utilization. * - * @param delegate The existing {@link CpuPowerModel} to decorate. + * @param delegate The existing {@link PowerModel} to decorate. */ - public static CpuPowerModel zeroIdle(CpuPowerModel delegate) { + public static PowerModel zeroIdle(PowerModel delegate) { return new ZeroIdlePowerDecorator(delegate); } - private static final class ConstantPowerModel implements CpuPowerModel { + private static final class ConstantPowerModel implements PowerModel { private final double power; ConstantPowerModel(double power) { @@ -154,7 +154,7 @@ public class CpuPowerModels { } } - private abstract static class MaxIdlePowerModel implements CpuPowerModel { + private abstract static class MaxIdlePowerModel implements PowerModel { protected final double maxPower; protected final double idlePower; @@ -344,7 +344,7 @@ public class CpuPowerModels { } } - private static final class InterpolationPowerModel implements CpuPowerModel { + private static final class InterpolationPowerModel implements PowerModel { private final double[] powerLevels; InterpolationPowerModel(double[] powerLevels) { @@ -380,10 +380,10 @@ public class CpuPowerModels { } } - private static final class ZeroIdlePowerDecorator implements CpuPowerModel { - private final CpuPowerModel delegate; + private static final class ZeroIdlePowerDecorator implements PowerModel { + private final PowerModel delegate; - ZeroIdlePowerDecorator(CpuPowerModel delegate) { + ZeroIdlePowerDecorator(PowerModel delegate) { this.delegate = delegate; } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/CpuPowerModelsFactory.kt b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/PowerModelsFactory.kt index 56610136..53107d19 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/cpu/CpuPowerModelsFactory.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/PowerModelsFactory.kt @@ -20,10 +20,10 @@ * SOFTWARE. */ -package org.opendc.simulator.compute.cpu +package org.opendc.simulator.compute.power // TODO: couple this correctly -public enum class CPUPowerModelEnum { +public enum class PowerModelEnum { Constant, Sqrt, Linear, @@ -41,30 +41,28 @@ public fun getPowerModel( calibrationFactor: Double = 1.0, asymUtil: Double = 0.0, dvfs: Boolean = true, -): CpuPowerModel { +): PowerModel { return when (modelType) { - "constant" -> CpuPowerModels.constant(power) - "sqrt" -> CpuPowerModels.sqrt(maxPower, idlePower) - "linear" -> CpuPowerModels.linear(maxPower, idlePower) - "square" -> CpuPowerModels.square(maxPower, idlePower) - "cubic" -> CpuPowerModels.cubic(maxPower, idlePower) - "mse" -> CpuPowerModels.mse(maxPower, idlePower, calibrationFactor) - "asymptotic" -> CpuPowerModels.asymptotic(maxPower, idlePower, asymUtil, dvfs) - + "constant" -> PowerModels.constant(power) + "sqrt" -> PowerModels.sqrt(maxPower, idlePower) + "linear" -> PowerModels.linear(maxPower, idlePower) + "square" -> PowerModels.square(maxPower, idlePower) + "cubic" -> PowerModels.cubic(maxPower, idlePower) + "mse" -> PowerModels.mse(maxPower, idlePower, calibrationFactor) + "asymptotic" -> PowerModels.asymptotic(maxPower, idlePower, asymUtil, dvfs) else -> throw IllegalArgumentException("Unknown power modelType $modelType") } } -public fun getPowerModel(modelType: String): CpuPowerModel { +public fun getPowerModel(modelType: String): PowerModel { return when (modelType) { - "constant" -> CpuPowerModels.constant(200.0) - "sqrt" -> CpuPowerModels.sqrt(350.0, 200.0) - "linear" -> CpuPowerModels.linear(350.0, 200.0) - "square" -> CpuPowerModels.square(350.0, 200.0) - "cubic" -> CpuPowerModels.cubic(350.0, 200.0) - "mse" -> CpuPowerModels.mse(350.0, 200.0, 1.0) - "asymptotic" -> CpuPowerModels.asymptotic(350.0, 200.0, 0.0, true) - + "constant" -> PowerModels.constant(200.0) + "sqrt" -> PowerModels.sqrt(350.0, 200.0) + "linear" -> PowerModels.linear(350.0, 200.0) + "square" -> PowerModels.square(350.0, 200.0) + "cubic" -> PowerModels.cubic(350.0, 200.0) + "mse" -> PowerModels.mse(350.0, 200.0, 1.0) + "asymptotic" -> PowerModels.asymptotic(350.0, 200.0, 0.0, true) else -> throw IllegalArgumentException("Unknown power modelType $modelType") } } 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 34804230..b00bb468 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 @@ -24,6 +24,7 @@ package org.opendc.simulator.compute.power; import java.util.List; import java.util.Map; +import org.opendc.common.ResourceType; import org.opendc.simulator.compute.cpu.SimCpu; import org.opendc.simulator.engine.engine.FlowEngine; import org.opendc.simulator.engine.graph.FlowEdge; @@ -217,4 +218,9 @@ public final class SimPowerSource extends FlowNode implements FlowSupplier, Carb return Map.of(FlowEdge.NodeType.SUPPLYING, supplierEdges); } + + @Override + public ResourceType getSupplierResourceType() { + return ResourceType.POWER; + } } 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 87a4e791..1ea7c570 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,29 +22,35 @@ 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; import org.opendc.simulator.compute.cpu.SimCpu; import org.opendc.simulator.engine.engine.FlowEngine; 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 double powerDemand = 0.0; - private double powerSupplied = 0.0; + private final HashMap<ResourceType, ArrayList<Double>> powerDemandsPerResource = new HashMap<>(); + private final HashMap<ResourceType, ArrayList<Double>> powerSuppliedPerResource = new HashMap<>(); private double totalEnergyUsage = 0.0; - private FlowEdge cpuEdge; + private final HashMap<ResourceType, ArrayList<FlowEdge>> resourceEdges = new HashMap<>(); private FlowEdge powerSupplyEdge; - private double capacity = Long.MAX_VALUE; + private final double capacity = Long.MAX_VALUE; //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Basic Getters and Setters @@ -56,7 +62,8 @@ 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 cpuEdge != null; + return !this.resourceEdges.isEmpty() + && this.resourceEdges.values().stream().anyMatch(list -> !list.isEmpty()); } /** @@ -65,14 +72,28 @@ 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.powerDemand; + return this.powerDemandsPerResource.values().stream() + .flatMap(List::stream) + .findFirst() + .orElse(0.0); + } + + public double getPowerDemand(ResourceType resourceType) { + return this.powerDemandsPerResource.get(resourceType).getFirst(); } /** * Return the instantaneous power usage of the machine (in W) measured at the InPort of the power supply. */ public double getPowerDraw() { - return this.powerSupplied; + return this.powerSuppliedPerResource.values().stream() + .flatMap(List::stream) + .findFirst() + .orElse(0.0); + } + + public double getPowerDraw(ResourceType resourceType) { + return this.powerSuppliedPerResource.get(resourceType).getFirst(); } /** @@ -105,10 +126,20 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer @Override public long onUpdate(long now) { updateCounters(); - double powerSupply = this.powerDemand; - - if (powerSupply != this.powerSupplied) { - this.pushOutgoingSupply(this.cpuEdge, powerSupply); + for (ResourceType resourceType : this.resourceEdges.keySet()) { + ArrayList<FlowEdge> edges = this.resourceEdges.get(resourceType); + if (edges != null && !edges.isEmpty()) { + double powerSupply = + this.powerDemandsPerResource.get(resourceType).getFirst(); + double powerSupplied = + this.powerSuppliedPerResource.get(resourceType).getFirst(); + + if (powerSupply != powerSupplied) { + for (FlowEdge edge : edges) { + edge.pushSupply(powerSupply); + } + } + } } return Long.MAX_VALUE; @@ -127,8 +158,11 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer long duration = now - lastUpdate; if (duration > 0) { - // Compute the energy usage of the psu - this.totalEnergyUsage += (this.powerSupplied * duration * 0.001); + for (ResourceType resourceType : this.powerSuppliedPerResource.keySet()) { + for (double powerSupplied : this.powerSuppliedPerResource.get(resourceType)) { + this.totalEnergyUsage += (powerSupplied * duration * 0.001); + } + } } } @@ -137,38 +171,61 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @Override - public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) { - this.powerDemand = newDemand; + public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand, ResourceType resourceType) { + this.powerDemandsPerResource.put(resourceType, new ArrayList<>(List.of(newDemand))); powerSupplyEdge.pushDemand(newDemand); } @Override + public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) { + double totalDemand = this.powerDemandsPerResource.values().stream() + .flatMap(List::stream) + .reduce(0.0, Double::sum); + this.powerSupplyEdge.pushDemand(totalDemand); + } + + @Override public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply) { - this.powerSupplied = newSupply; - cpuEdge.pushSupply(newSupply); + this.pushOutgoingSupply(consumerEdge, newSupply, consumerEdge.getConsumerResourceType()); + } + + @Override + public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply, ResourceType resourceType) { + this.powerSuppliedPerResource.put(resourceType, new ArrayList<>(List.of(newSupply))); + consumerEdge.pushSupply(newSupply, false, resourceType); } @Override - public void handleIncomingDemand(FlowEdge consumerEdge, double newPowerDemand) { + public void handleIncomingDemand(FlowEdge consumerEdge, double newDemand) { + handleIncomingDemand(consumerEdge, newDemand, consumerEdge.getConsumerResourceType()); + } + @Override + public void handleIncomingDemand(FlowEdge consumerEdge, double newPowerDemand, ResourceType resourceType) { updateCounters(); - this.powerDemand = newPowerDemand; + this.powerDemandsPerResource.put(resourceType, new ArrayList<>(List.of(newPowerDemand))); pushOutgoingDemand(this.powerSupplyEdge, newPowerDemand); } @Override - public void handleIncomingSupply(FlowEdge supplierEdge, double newPowerSupply) { - + public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply) { updateCounters(); - this.powerSupplied = newPowerSupply; - - pushOutgoingSupply(this.cpuEdge, newPowerSupply); + for (ResourceType resourceType : this.resourceEdges.keySet()) { + for (FlowEdge edge : this.resourceEdges.get(resourceType)) { + double outgoingSupply = + Math.min(this.powerDemandsPerResource.get(resourceType).getFirst(), newSupply); + pushOutgoingSupply(edge, outgoingSupply, resourceType); + } + } } @Override public void addConsumerEdge(FlowEdge consumerEdge) { - this.cpuEdge = consumerEdge; + ResourceType consumerResourceType = consumerEdge.getConsumerResourceType(); + this.resourceEdges.put(consumerResourceType, new ArrayList<>(List.of(consumerEdge))); + this.powerDemandsPerResource.put(consumerResourceType, new ArrayList<>(List.of(0.0))); + this.powerSuppliedPerResource.put(consumerResourceType, new ArrayList<>(List.of(0.0))); } @Override @@ -178,7 +235,12 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer @Override public void removeConsumerEdge(FlowEdge consumerEdge) { - this.cpuEdge = null; + ResourceType resourceType = consumerEdge.getConsumerResourceType(); + if (this.resourceEdges.containsKey(resourceType)) { + this.resourceEdges.remove(resourceType); + this.powerDemandsPerResource.remove(resourceType); + this.powerSuppliedPerResource.remove(resourceType); + } } @Override @@ -188,11 +250,27 @@ public final class SimPsu extends FlowNode implements FlowSupplier, FlowConsumer @Override public Map<FlowEdge.NodeType, List<FlowEdge>> getConnectedEdges() { - List<FlowEdge> supplyingEdges = cpuEdge != null ? List.of(cpuEdge) : List.of(); + List<FlowEdge> supplyingEdges = new ArrayList<>(); + for (ResourceType resourceType : this.resourceEdges.keySet()) { + List<FlowEdge> edges = this.resourceEdges.get(resourceType); + if (edges != null && !edges.isEmpty()) { + supplyingEdges.addAll(edges); + } + } List<FlowEdge> consumingEdges = powerSupplyEdge != null ? List.of(powerSupplyEdge) : List.of(); return Map.of( FlowEdge.NodeType.SUPPLYING, supplyingEdges, FlowEdge.NodeType.CONSUMING, consumingEdges); } + + @Override + public ResourceType getSupplierResourceType() { + return ResourceType.POWER; + } + + @Override + public ResourceType getConsumerResourceType() { + return ResourceType.POWER; + } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/batteries/BatteryAggregator.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/batteries/BatteryAggregator.java index 9a05f2b3..13674369 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/batteries/BatteryAggregator.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/batteries/BatteryAggregator.java @@ -26,6 +26,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; +import org.opendc.common.ResourceType; import org.opendc.simulator.engine.engine.FlowEngine; import org.opendc.simulator.engine.graph.FlowConsumer; import org.opendc.simulator.engine.graph.FlowDistributor; @@ -190,4 +191,14 @@ public class BatteryAggregator extends FlowNode implements FlowConsumer, FlowSup FlowEdge.NodeType.CONSUMING, consumingEdges, FlowEdge.NodeType.SUPPLYING, supplyingEdges); } + + @Override + public ResourceType getSupplierResourceType() { + return ResourceType.POWER; + } + + @Override + public ResourceType getConsumerResourceType() { + return ResourceType.POWER; + } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/batteries/SimBattery.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/batteries/SimBattery.java index d749af72..d6377ef6 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/batteries/SimBattery.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/power/batteries/SimBattery.java @@ -24,6 +24,7 @@ package org.opendc.simulator.compute.power.batteries; import java.util.List; import java.util.Map; +import org.opendc.common.ResourceType; import org.opendc.simulator.compute.power.batteries.policy.BatteryPolicy; import org.opendc.simulator.engine.engine.FlowEngine; import org.opendc.simulator.engine.graph.FlowConsumer; @@ -331,4 +332,14 @@ public class SimBattery extends FlowNode implements FlowConsumer, FlowSupplier { FlowEdge.NodeType.CONSUMING, consumingEdges, FlowEdge.NodeType.SUPPLYING, supplyingEdges); } + + @Override + public ResourceType getSupplierResourceType() { + return ResourceType.POWER; + } + + @Override + public ResourceType getConsumerResourceType() { + return ResourceType.POWER; + } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/ChainWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/ChainWorkload.java index 3cdde40a..56e6093b 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/ChainWorkload.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/ChainWorkload.java @@ -23,6 +23,7 @@ package org.opendc.simulator.compute.workload; import java.util.ArrayList; +import java.util.List; import java.util.function.Consumer; import org.opendc.simulator.compute.machine.SimMachine; import org.opendc.simulator.engine.graph.FlowSupplier; @@ -47,7 +48,7 @@ public record ChainWorkload( } @Override - public SimWorkload startWorkload(FlowSupplier supplier, SimMachine machine, Consumer<Exception> completion) { + public SimWorkload startWorkload(List<FlowSupplier> supplier, SimMachine machine, Consumer<Exception> completion) { return new VirtualMachine(supplier, this, machine, completion); } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/VirtualMachine.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/VirtualMachine.java index 7632b503..622d2b89 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/VirtualMachine.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/VirtualMachine.java @@ -22,32 +22,41 @@ package org.opendc.simulator.compute.workload; +import java.util.ArrayList; +import java.util.Hashtable; import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.function.Consumer; +import org.opendc.common.ResourceType; +import org.opendc.simulator.compute.ComputeResource; import org.opendc.simulator.compute.machine.PerformanceCounters; import org.opendc.simulator.compute.machine.SimMachine; 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 VirtualMachine} that composes multiple {@link SimWorkload}s. */ public final class VirtualMachine extends SimWorkload implements FlowSupplier { + private static final Logger LOGGER = LoggerFactory.getLogger(VirtualMachine.class); private final LinkedList<Workload> workloads; private int workloadIndex; private SimWorkload activeWorkload; - private double cpuDemand = 0.0f; - private double cpuSupply = 0.0f; - private double d = 0.0f; private FlowEdge workloadEdge; - private FlowEdge machineEdge; - private double capacity = 0; + private final Hashtable<ResourceType, Double> resourceDemands = new Hashtable<>(); + private final Hashtable<ResourceType, Double> resourceSupplies = new Hashtable<>(); + private final Hashtable<ResourceType, Double> resourceCapacities = new Hashtable<>(); + private final Hashtable<ResourceType, Double> resourceTimeScalingFactor = new Hashtable<>(); // formerly known as d + private final Hashtable<ResourceType, FlowEdge> distributorEdges = new Hashtable<>(); + private final Hashtable<ResourceType, List<PerformanceCounters>> resourcePerformanceCounters = new Hashtable<>(); private final long checkpointInterval; private final long checkpointDuration; @@ -57,16 +66,25 @@ public final class VirtualMachine extends SimWorkload implements FlowSupplier { private final ChainWorkload snapshot; private long lastUpdate; - private final PerformanceCounters performanceCounters = new PerformanceCounters(); private Consumer<Exception> completion; + private final List<ResourceType> availableResources = new ArrayList<>(); + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Basic Getters and Setters //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @Override public double getCapacity() { - return this.capacity; + throw new UnsupportedOperationException("getCapacity() is not supported for VirtualMachine"); + } + + @Override + public double getCapacity(ResourceType resourceType) { + if (resourceType == ResourceType.AUXILIARY) { + return 0.0; + } + return this.resourceCapacities.get(resourceType); } @Override @@ -89,8 +107,22 @@ public final class VirtualMachine extends SimWorkload implements FlowSupplier { return checkpointIntervalScaling; } - public PerformanceCounters getPerformanceCounters() { - return performanceCounters; + public PerformanceCounters getCpuPerformanceCounters() { + return this.resourcePerformanceCounters.get(ResourceType.CPU).getFirst(); + } + + public List<PerformanceCounters> getGpuPerformanceCounters() { + return this.resourcePerformanceCounters.get(ResourceType.GPU) != null + ? this.resourcePerformanceCounters.get(ResourceType.GPU) + : new ArrayList<>(); + } + + public PerformanceCounters getGpuPerformanceCounters(int gpuId) { + List<PerformanceCounters> gpuPerformanceCounters = this.resourcePerformanceCounters.get(ResourceType.GPU); + if (gpuId < 0 || gpuId >= gpuPerformanceCounters.size()) { + throw new IndexOutOfBoundsException("No such GPU id: " + gpuId); + } + return gpuPerformanceCounters.get(gpuId); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -116,16 +148,54 @@ public final class VirtualMachine extends SimWorkload implements FlowSupplier { } this.workloadIndex = -1; - + this.availableResources.add(supplier.getSupplierResourceType()); this.onStart(); } - VirtualMachine(FlowSupplier supplier, ChainWorkload workload, SimMachine machine, Consumer<Exception> completion) { - this(supplier, workload); + VirtualMachine( + List<FlowSupplier> suppliers, ChainWorkload workload, SimMachine machine, Consumer<Exception> completion) { + super(((FlowNode) suppliers.getFirst()).getEngine()); + + this.snapshot = workload; + + for (FlowSupplier supplier : suppliers) { + new FlowEdge(this, supplier); + ResourceType resourceType = supplier.getSupplierResourceType(); + + this.availableResources.add(resourceType); + + ArrayList<ComputeResource> resources = machine.getResources(resourceType); + if (resources.isEmpty()) { + throw new IllegalArgumentException("No resources of type " + resourceType + " found in machine "); + } + + this.resourceCapacities.put(resourceType, resources.getFirst().getCapacity()); + + ArrayList<PerformanceCounters> performanceCounters = new ArrayList<>(); + + for (ComputeResource resource : resources) { + performanceCounters.add(new PerformanceCounters()); + this.resourceTimeScalingFactor.put(resourceType, 1.0 / resource.getCapacity()); + } + this.resourcePerformanceCounters.put(resourceType, performanceCounters); + this.resourceDemands.put(resourceType, 0.0); + this.resourceSupplies.put(resourceType, 0.0); + } + + this.workloads = new LinkedList<>(workload.workloads()); + this.checkpointInterval = workload.checkpointInterval(); + this.checkpointDuration = workload.checkpointDuration(); + this.checkpointIntervalScaling = workload.checkpointIntervalScaling(); + + this.lastUpdate = clock.millis(); + + if (checkpointInterval > 0) { + this.createCheckpointModel(); + } - this.capacity = machine.getCpu().getFrequency(); - this.d = 1 / machine.getCpu().getFrequency(); + this.workloadIndex = -1; this.completion = completion; + this.onStart(); } public Workload getNextWorkload() { @@ -152,18 +222,25 @@ public final class VirtualMachine extends SimWorkload implements FlowSupplier { this.lastUpdate = now; long delta = now - lastUpdate; - double cpuCapacity = 0.0f; - if (delta > 0) { - final double factor = this.d * delta; - - this.performanceCounters.addCpuActiveTime(Math.round(this.cpuSupply * factor)); - this.performanceCounters.setCpuIdleTime(Math.round((cpuCapacity - this.cpuSupply) * factor)); - this.performanceCounters.addCpuStealTime(Math.round((this.cpuDemand - this.cpuSupply) * factor)); + for (ResourceType resourceType : this.availableResources) { + int i = 0; + final double factor = this.resourceTimeScalingFactor.get(resourceType) * delta; + for (PerformanceCounters performanceCounter : this.resourcePerformanceCounters.get(resourceType)) { + if (delta > 0) { + performanceCounter.addActiveTime(Math.round(this.resourceSupplies.get(resourceType) * factor)); + performanceCounter.setIdleTime(Math.round( + (this.resourceCapacities.get(resourceType) - this.resourceSupplies.get(resourceType)) + * factor)); + performanceCounter.addStealTime(Math.round( + (this.resourceDemands.get(resourceType) - this.resourceSupplies.get(resourceType)) + * factor)); + } + performanceCounter.setDemand(this.resourceDemands.get(resourceType)); + performanceCounter.setSupply(this.resourceSupplies.get(resourceType)); + performanceCounter.setCapacity(this.resourceCapacities.get(resourceType)); + i++; + } } - - this.performanceCounters.setCpuDemand(this.cpuDemand); - this.performanceCounters.setCpuSupply(this.cpuSupply); - this.performanceCounters.setCpuCapacity(cpuCapacity); } @Override @@ -233,40 +310,66 @@ public final class VirtualMachine extends SimWorkload implements FlowSupplier { } /** - * Add Connection to the cpuMux + * Add Connection to the resource flow distributor * - * @param supplierEdge The edge to the cpuMux + * @param supplierEdge The edge to the resource flow distributor */ @Override public void addSupplierEdge(FlowEdge supplierEdge) { - this.machineEdge = supplierEdge; - this.capacity = supplierEdge.getCapacity(); + ResourceType resourceType = supplierEdge.getSupplierResourceType(); + this.resourceCapacities.put(resourceType, supplierEdge.getCapacity()); + this.distributorEdges.put(resourceType, supplierEdge); } /** - * Push demand to the cpuMux + * Push demand to the resource flow distributor * - * @param supplierEdge The edge to the cpuMux - * @param newDemand new demand to sent to the cpu + * @param supplierEdge The edge to the resource flow distributor + * @param newDemand new demand to sent to the resource flow distributor */ @Override public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) { + // FIXME: Needs to be assigned to specific resource if multiple exist -> add resource Id as parameter + this.pushOutgoingDemand(supplierEdge, newDemand, supplierEdge.getSupplierResourceType()); + } - this.cpuDemand = newDemand; - this.machineEdge.pushDemand(newDemand); + /** + * Push demand to the resource flow distributor + * + * @param supplierEdge The edge to the resource flow distributor + * @param newDemand new demand to sent to the resource flow distributor + */ + @Override + public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand, ResourceType resourceType) { + // FIXME: Needs to be assigned to specific resource if multiple exist -> add resource Id as parameter + this.resourceDemands.put(resourceType, newDemand); + this.distributorEdges.get(resourceType).pushDemand(newDemand, false, resourceType); } /** * Push supply to the workload * - * @param consumerEdge The edge to the cpuMux + * @param consumerEdge The edge to the resource flow distributor * @param newSupply new supply to sent to the workload */ @Override public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply) { + this.resourceSupplies.put(consumerEdge.getConsumerResourceType(), newSupply); + this.distributorEdges + .get(consumerEdge.getConsumerResourceType()) + .pushSupply(newSupply, false, consumerEdge.getConsumerResourceType()); + } - this.cpuSupply = newSupply; - this.workloadEdge.pushSupply(newSupply); + /** + * Push supply to the workload + * + * @param consumerEdge The edge to the resource flow distributor + * @param newSupply new supply to sent to the workload + */ + @Override + public void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply, ResourceType resourceType) { + this.resourceSupplies.put(resourceType, newSupply); + this.workloadEdge.pushSupply(newSupply, false, resourceType); } /** @@ -278,21 +381,42 @@ public final class VirtualMachine extends SimWorkload implements FlowSupplier { @Override public void handleIncomingDemand(FlowEdge consumerEdge, double newDemand) { updateCounters(this.clock.millis()); + this.pushOutgoingDemand(this.distributorEdges.get(consumerEdge.getConsumerResourceType()), newDemand); + } - this.pushOutgoingDemand(this.machineEdge, newDemand); + @Override + public void handleIncomingDemand(FlowEdge consumerEdge, double newDemand, ResourceType resourceType) { + updateCounters(this.clock.millis()); + this.pushOutgoingDemand(this.distributorEdges.get(resourceType), newDemand, resourceType); } /** - * Handle new supply coming from the cpuMux + * Handle new supply coming from the resource flow distributor * - * @param supplierEdge The edge to the cpuMux + * @param supplierEdge The edge to the resource flow distributor * @param newSupply The new supply that is sent to the workload */ @Override public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply) { updateCounters(this.clock.millis()); - this.pushOutgoingSupply(this.machineEdge, newSupply); + this.pushOutgoingSupply( + this.distributorEdges.get(supplierEdge.getSupplierResourceType()), + newSupply, + supplierEdge.getSupplierResourceType()); + } + + /** + * Handle new supply coming from the resource flow distributor + * + * @param supplierEdge The edge to the resource flow distributor + * @param newSupply The new supply that is sent to the workload + */ + @Override + public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply, ResourceType resourceType) { + updateCounters(this.clock.millis()); + + this.pushOutgoingSupply(this.distributorEdges.get(resourceType), newSupply, resourceType); } /** @@ -322,14 +446,14 @@ public final class VirtualMachine extends SimWorkload implements FlowSupplier { } /** - * Handle the removal of the connection to the cpuMux + * Handle the removal of the connection to the resource flow distributor * When this happens, close the SimChainWorkload * - * @param supplierEdge The edge to the cpuMux + * @param supplierEdge The edge to the resource flow distributor */ @Override public void removeSupplierEdge(FlowEdge supplierEdge) { - if (this.machineEdge == null) { + if (!this.distributorEdges.contains(supplierEdge.getSupplierResourceType())) { return; } @@ -338,11 +462,16 @@ public final class VirtualMachine extends SimWorkload implements FlowSupplier { @Override public Map<FlowEdge.NodeType, List<FlowEdge>> getConnectedEdges() { - List<FlowEdge> consumerEdges = (this.machineEdge != null) ? List.of(this.machineEdge) : List.of(); + List<FlowEdge> consumerEdges = + this.distributorEdges.values().stream().filter(Objects::nonNull).toList(); List<FlowEdge> supplierEdges = (this.workloadEdge != null) ? List.of(this.workloadEdge) : List.of(); return Map.of( FlowEdge.NodeType.CONSUMING, consumerEdges, FlowEdge.NodeType.SUPPLYING, supplierEdges); } + + public List<ResourceType> getAvailableResources() { + return this.availableResources; + } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/Workload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/Workload.java index 3ad7597d..5edacb3b 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/Workload.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/Workload.java @@ -22,6 +22,7 @@ package org.opendc.simulator.compute.workload; +import java.util.List; import java.util.function.Consumer; import org.opendc.simulator.compute.machine.SimMachine; import org.opendc.simulator.engine.graph.FlowSupplier; @@ -36,5 +37,5 @@ public interface Workload { SimWorkload startWorkload(FlowSupplier supplier); - SimWorkload startWorkload(FlowSupplier supplier, SimMachine machine, Consumer<Exception> completion); + SimWorkload startWorkload(List<FlowSupplier> supplier, SimMachine machine, Consumer<Exception> completion); } 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 d5635439..8b3a7188 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 @@ -22,36 +22,50 @@ package org.opendc.simulator.compute.workload.trace; +import java.util.ArrayList; +import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; +import org.opendc.common.ResourceType; import org.opendc.simulator.compute.workload.SimWorkload; +import org.opendc.simulator.compute.workload.VirtualMachine; import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy; 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; public class SimTraceWorkload extends SimWorkload implements FlowConsumer { + private static final Logger LOGGER = LoggerFactory.getLogger(SimTraceWorkload.class); private LinkedList<TraceFragment> remainingFragments; private int fragmentIndex; private TraceFragment currentFragment; private long startOfFragment; - private FlowEdge machineEdge; - - private double cpuFreqDemand = 0.0; // The Cpu demanded by fragment - private double cpuFreqSupplied = 0.0; // The Cpu speed supplied - private double newCpuFreqSupplied = 0.0; // The Cpu speed supplied - private double remainingWork = 0.0; // The duration of the fragment at the demanded speed + private final Map<ResourceType, FlowEdge> machineResourceEdges = new HashMap<>(); + + // TODO: Currently GPU memory is not considered and can not be used + private final ArrayList<ResourceType> usedResourceTypes = new ArrayList<>(); + private final Map<ResourceType, Double> resourcesSupplied = new HashMap<>(); // the currently supplied resources + private final Map<ResourceType, Double> newResourcesSupply = + new HashMap<>(); // The supplied resources with next update + private final Map<ResourceType, Double> resourcesDemand = new HashMap<>(); // The demands per resource type + private final Map<ResourceType, Double> remainingWork = + new HashMap<>(); // The duration of the fragment at the demanded speeds + private double totalRemainingWork = + 0.0; // The total remaining work of the fragment across all resources, used to determine the end of the + // fragment + private final Map<ResourceType, Boolean> workloadFinished = + new HashMap<>(); // The workload finished for each resource type private final long checkpointDuration; - private final TraceWorkload snapshot; private final ScalingPolicy scalingPolicy; - private final String taskName; //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -98,6 +112,44 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { this.startOfFragment = this.clock.millis(); 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.resourcesSupplied.put(resourceType, 0.0); + this.newResourcesSupply.put(resourceType, 0.0); + this.resourcesDemand.put(resourceType, 0.0); + this.remainingWork.put(resourceType, 0.0); + this.workloadFinished.put(resourceType, false); + } + } + } + + // Needed if workload not started by VM + public SimTraceWorkload(List<FlowSupplier> resourceSuppliers, TraceWorkload workload) { + // same engine for all suppliers + super(((FlowNode) resourceSuppliers.getFirst()).getEngine()); + + this.snapshot = workload; + this.checkpointDuration = workload.checkpointDuration(); + this.scalingPolicy = workload.getScalingPolicy(); + this.remainingFragments = new LinkedList<>(workload.getFragments()); + this.fragmentIndex = 0; + this.taskName = workload.getTaskName(); + + this.startOfFragment = this.clock.millis(); + + for (FlowSupplier supplier : resourceSuppliers) { + if (supplier.getSupplierResourceType() != ResourceType.AUXILIARY) { + new FlowEdge(this, supplier); + this.usedResourceTypes.add(supplier.getSupplierResourceType()); + this.resourcesSupplied.put(supplier.getSupplierResourceType(), 0.0); + this.newResourcesSupply.put(supplier.getSupplierResourceType(), 0.0); + this.resourcesDemand.put(supplier.getSupplierResourceType(), 0.0); + this.remainingWork.put(supplier.getSupplierResourceType(), 0.0); + this.workloadFinished.put(supplier.getSupplierResourceType(), false); + } + } } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -109,30 +161,64 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { long passedTime = getPassedTime(now); this.startOfFragment = now; - // The amount of work done since last update - double finishedWork = this.scalingPolicy.getFinishedWork(this.cpuFreqDemand, this.cpuFreqSupplied, passedTime); - - this.remainingWork -= finishedWork; + for (ResourceType resourceType : this.usedResourceTypes) { + // 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.totalRemainingWork -= finishedWork; + if (this.remainingWork.get(resourceType) <= 0) { + this.workloadFinished.put(resourceType, true); + } + } - // If this.remainingWork <= 0, the fragment has been completed - if (this.remainingWork <= 0) { + // If this.totalRemainingWork <= 0, the fragment has been completed across all resources + if (this.totalRemainingWork <= 0 && !this.workloadFinished.containsValue(false)) { this.startNextFragment(); this.invalidate(); return Long.MAX_VALUE; } - this.cpuFreqSupplied = this.newCpuFreqSupplied; + for (ResourceType resourceType : this.usedResourceTypes) { + if (this.machineResourceEdges.get(resourceType) != null) { + this.pushOutgoingDemand( + this.machineResourceEdges.get(resourceType), + this.resourcesDemand.get(resourceType), + resourceType); + } + } - // The amount of time required to finish the fragment at this speed - long remainingDuration = this.scalingPolicy.getRemainingDuration( - this.cpuFreqDemand, this.newCpuFreqSupplied, this.remainingWork); + // Update the supplied resources + for (ResourceType resourceType : this.usedResourceTypes) { + this.resourcesSupplied.put(resourceType, this.newResourcesSupply.get(resourceType)); + } - if (remainingDuration == 0.0) { - this.remainingWork = 0.0; + long timeUntilNextUpdate = Long.MIN_VALUE; + + for (ResourceType resourceType : this.usedResourceTypes) { + // The amount of time required to finish the fragment at this speed + long remainingDuration = this.scalingPolicy.getRemainingDuration( + this.resourcesDemand.get(resourceType), + this.resourcesSupplied.get(resourceType), + this.remainingWork.get(resourceType)); + + if (remainingDuration == 0.0) { + // if resource not initialized, then nothing happens + this.totalRemainingWork -= this.remainingWork.get(resourceType); + this.remainingWork.put(resourceType, 0.0); + this.workloadFinished.put(resourceType, true); + } + + // The next update should happen when the fastest resource is done, so that it is no longer tracked when + // unused + if (remainingDuration > 0 + && (timeUntilNextUpdate == Long.MIN_VALUE || remainingDuration < timeUntilNextUpdate)) { + timeUntilNextUpdate = remainingDuration; + } } - return now + remainingDuration; + return timeUntilNextUpdate == Long.MIN_VALUE ? now : now + timeUntilNextUpdate; } public TraceFragment getNextFragment() { @@ -152,14 +238,27 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { this.stopWorkload(); return; } - double demand = nextFragment.cpuUsage(); - this.remainingWork = this.scalingPolicy.getRemainingWork(demand, nextFragment.duration()); - this.pushOutgoingDemand(this.machineEdge, demand); + + // 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); + + this.remainingWork.put(resourceType, this.scalingPolicy.getRemainingWork(demand, nextFragment.duration())); + this.totalRemainingWork += this.remainingWork.get(resourceType); + this.workloadFinished.put(resourceType, false); + + if (this.machineResourceEdges.get(resourceType) != null) { + this.pushOutgoingDemand(this.machineResourceEdges.get(resourceType), demand, resourceType); + } + } } @Override public void stopWorkload() { - if (this.machineEdge == null) { + if (areAllEdgesNull()) { return; } @@ -167,7 +266,10 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { // Currently stopWorkload is called twice this.closeNode(); - this.machineEdge = null; + for (ResourceType resourceType : this.usedResourceTypes) { + this.machineResourceEdges.put(resourceType, null); + this.workloadFinished.put(resourceType, true); + } this.remainingFragments = null; this.currentFragment = null; } @@ -195,22 +297,38 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { long passedTime = getPassedTime(now); // The amount of work done since last update - double finishedWork = this.scalingPolicy.getFinishedWork(this.cpuFreqDemand, this.cpuFreqSupplied, passedTime); + for (ResourceType resourceType : this.usedResourceTypes) { + double finishedWork = this.scalingPolicy.getFinishedWork( + this.resourcesDemand.get(resourceType), this.resourcesSupplied.get(resourceType), passedTime); + this.remainingWork.put(resourceType, this.remainingWork.get(resourceType) - finishedWork); + this.totalRemainingWork -= finishedWork; + } - this.remainingWork -= finishedWork; + long remainingDuration = 0; + for (ResourceType resourceType : this.usedResourceTypes) { - // The amount of time required to finish the fragment at this speed - long remainingTime = - this.scalingPolicy.getRemainingDuration(this.cpuFreqDemand, this.cpuFreqDemand, this.remainingWork); + // The amount of time required to finish the fragment at this speed + remainingDuration = Math.max( + remainingDuration, + this.scalingPolicy.getRemainingDuration( + this.resourcesDemand.get(resourceType), + this.resourcesSupplied.get(resourceType), + this.remainingWork.get(resourceType))); + } // If this is the end of the Task, don't make a snapshot - if (this.currentFragment == null || (remainingTime <= 0 && remainingFragments.isEmpty())) { + if (this.currentFragment == null || (remainingDuration <= 0 && remainingFragments.isEmpty())) { return; } // Create a new fragment based on the current fragment and remaining duration - TraceFragment newFragment = - new TraceFragment(remainingTime, currentFragment.cpuUsage(), currentFragment.coreCount()); + TraceFragment newFragment = new TraceFragment( + remainingDuration, + currentFragment.cpuUsage(), + currentFragment.cpuCoreCount(), + currentFragment.gpuUsage(), + currentFragment.gpuCoreCount(), + currentFragment.gpuMemoryUsage()); // Alter the snapshot by removing finished fragments this.snapshot.removeFragments(this.fragmentIndex); @@ -220,7 +338,12 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { // Create and add a fragment for processing the snapshot process TraceFragment snapshotFragment = new TraceFragment( - this.checkpointDuration, this.snapshot.getMaxCpuDemand(), this.snapshot.getMaxCoreCount()); + this.checkpointDuration, + this.snapshot.getMaxCpuDemand(), + this.snapshot.getMaxCoreCount(), + this.snapshot.getMaxGpuDemand(), + this.snapshot.getMaxGpuCoreCount(), + this.snapshot.getMaxGpuMemoryDemand()); this.remainingFragments.addFirst(snapshotFragment); this.fragmentIndex = -1; @@ -243,12 +366,29 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { */ @Override public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply) { - if (newSupply == this.cpuFreqSupplied) { + ResourceType suppliedResourceType = ResourceType.CPU; + if (this.resourcesSupplied.get(suppliedResourceType) == newSupply) { return; } + this.resourcesSupplied.put(suppliedResourceType, this.newResourcesSupply.get(suppliedResourceType)); + this.newResourcesSupply.put(suppliedResourceType, newSupply); - this.cpuFreqSupplied = this.newCpuFreqSupplied; - this.newCpuFreqSupplied = newSupply; + this.invalidate(); + } + + /** + * Handle updates in supply from the Virtual Machine + * + * @param supplierEdge edge to the VM on which this is running + * @param newSupply The new demand that needs to be sent to the VM + */ + @Override + public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply, ResourceType resourceType) { + if (this.resourcesSupplied.get(resourceType) == newSupply) { + return; + } + this.resourcesSupplied.put(resourceType, this.newResourcesSupply.get(resourceType)); + this.newResourcesSupply.put(resourceType, newSupply); this.invalidate(); } @@ -261,12 +401,28 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { */ @Override public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) { - if (newDemand == this.cpuFreqDemand) { + ResourceType demandedResourceType = ResourceType.CPU; + if (this.resourcesDemand.get(demandedResourceType) == newDemand) { return; } - this.cpuFreqDemand = newDemand; - this.machineEdge.pushDemand(newDemand); + this.resourcesDemand.put(demandedResourceType, newDemand); + this.machineResourceEdges.get(demandedResourceType).pushDemand(newDemand); + } + /** + * Push a new demand to the Virtual Machine + * + * @param supplierEdge edge to the VM on which this is running + * @param newDemand The new demand that needs to be sent to the VM + */ + @Override + public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand, ResourceType resourceType) { + if (this.resourcesDemand.get(resourceType) == newDemand) { + return; + } + + this.resourcesDemand.put(resourceType, newDemand); + this.machineResourceEdges.get(resourceType).pushDemand(newDemand, false, resourceType); } /** @@ -276,7 +432,24 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { */ @Override public void addSupplierEdge(FlowEdge supplierEdge) { - this.machineEdge = supplierEdge; + ResourceType incommingResourceType = supplierEdge.getResourceType(); + + if (machineResourceEdges.containsValue(supplierEdge)) { + return; // Skip if this exact edge is already registered + } + + this.machineResourceEdges.put(incommingResourceType, supplierEdge); + if (supplierEdge.getSupplier() instanceof VirtualMachine vm) { + for (ResourceType resourceType : vm.getAvailableResources()) { + if (resourceType == incommingResourceType || resourceType == ResourceType.AUXILIARY) { + continue; + } + + if (!this.machineResourceEdges.containsKey(resourceType)) { + new FlowEdge(this, vm, resourceType); + } + } + } } /** @@ -287,7 +460,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { */ @Override public void removeSupplierEdge(FlowEdge supplierEdge) { - if (this.machineEdge == null) { + if (areAllEdgesNull()) { return; } @@ -296,6 +469,24 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { @Override public Map<FlowEdge.NodeType, List<FlowEdge>> getConnectedEdges() { - return Map.of(FlowEdge.NodeType.CONSUMING, (this.machineEdge != null) ? List.of(this.machineEdge) : List.of()); + Map<FlowEdge.NodeType, List<FlowEdge>> connectedEdges = new HashMap<>(); + for (ResourceType resourceType : ResourceType.values()) { + if (this.machineResourceEdges.get(resourceType) != null) { + connectedEdges.put(FlowEdge.NodeType.CONSUMING, List.of(this.machineResourceEdges.get(resourceType))); + } + } + return connectedEdges; + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + // Util Methods + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + private boolean areAllEdgesNull() { + for (FlowEdge edge : this.machineResourceEdges.values()) { + if (edge != null) { + return false; + } + } + return true; } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceFragment.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceFragment.java index a09206a1..bc3685a3 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceFragment.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceFragment.java @@ -22,9 +22,49 @@ package org.opendc.simulator.compute.workload.trace; -public record TraceFragment(long duration, double cpuUsage, int coreCount) { +import org.opendc.common.ResourceType; - public TraceFragment(long start, long duration, double cpuUsage, int coreCount) { - this(duration, cpuUsage, coreCount); +public record TraceFragment( + long duration, double cpuUsage, int cpuCoreCount, double gpuUsage, int gpuCoreCount, Long gpuMemoryUsage) { + + public TraceFragment(long start, long duration, double cpuUsage, int cpuCoreCount) { + this(duration, cpuUsage, cpuCoreCount, 0.0, 0, 0L); + } + + public TraceFragment(long duration, double cpuUsage, int cpuCoreCount) { + this(duration, cpuUsage, cpuCoreCount, 0.0, 0, 0L); + } + + public TraceFragment(long duration, double cpuUsage, int cpuCoreCount, double gpuUsage, int gpuCoreCount) { + this(duration, cpuUsage, cpuCoreCount, gpuUsage, gpuCoreCount, 0L); + } + + /** + * Returns the resource usage for the specified resource type. + * + * @param resourceType the type of resource + * @return the usage value for the specified resource type + */ + public double getResourceUsage(ResourceType resourceType) throws IllegalArgumentException { + return switch (resourceType) { + case CPU -> cpuUsage; + case GPU -> gpuUsage; + // case GPU_MEMORY -> gpuMemoryUsage; + default -> throw new IllegalArgumentException("Invalid resource type: " + resourceType); + }; + } + + /** + * Returns the core count for the specified resource type. + * + * @param resourceType the type of resource + * @return the core count for the specified resource type + */ + public int getCoreCount(ResourceType resourceType) throws IllegalArgumentException { + return switch (resourceType) { + case CPU -> cpuCoreCount; + case GPU -> gpuCoreCount; + default -> throw new IllegalArgumentException("Invalid resource type: " + resourceType); + }; } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceWorkload.java index 9c31a833..d698a48d 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceWorkload.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceWorkload.java @@ -23,8 +23,12 @@ package org.opendc.simulator.compute.workload.trace; import java.util.ArrayList; +import java.util.Arrays; import java.util.Comparator; +import java.util.List; +import java.util.Objects; import java.util.function.Consumer; +import org.opendc.common.ResourceType; import org.opendc.simulator.compute.machine.SimMachine; import org.opendc.simulator.compute.workload.SimWorkload; import org.opendc.simulator.compute.workload.Workload; @@ -37,13 +41,12 @@ public class TraceWorkload implements Workload { private final long checkpointDuration; private final double checkpointIntervalScaling; private final double maxCpuDemand; - private final int maxCoreCount; - - public String getTaskName() { - return taskName; - } - + private final int maxCpuCoreCount; + private final double maxGpuDemand; + private final int maxGpuCoreCount; + private final long maxGpuMemoryDemand; private final String taskName; + private ResourceType[] resourceTypes = new ResourceType[ResourceType.values().length]; public ScalingPolicy getScalingPolicy() { return scalingPolicy; @@ -57,7 +60,8 @@ public class TraceWorkload implements Workload { long checkpointDuration, double checkpointIntervalScaling, ScalingPolicy scalingPolicy, - String taskName) { + String taskName, + ResourceType[] resourceTypes) { this.fragments = fragments; this.checkpointInterval = checkpointInterval; this.checkpointDuration = checkpointDuration; @@ -69,11 +73,25 @@ public class TraceWorkload implements Workload { this.maxCpuDemand = fragments.stream() .max(Comparator.comparing(TraceFragment::cpuUsage)) .get() - .cpuUsage(); - this.maxCoreCount = fragments.stream() - .max(Comparator.comparing(TraceFragment::coreCount)) + // .cpuUsage(); + .getResourceUsage(ResourceType.CPU); + this.maxCpuCoreCount = fragments.stream() + .max(Comparator.comparing(TraceFragment::cpuCoreCount)) .get() - .coreCount(); + // .cpuCoreCount(); + .getCoreCount(ResourceType.CPU); + + this.maxGpuDemand = fragments.stream() + .max(Comparator.comparing(TraceFragment::gpuUsage)) + .get() + .getResourceUsage(ResourceType.GPU); + this.maxGpuCoreCount = fragments.stream() + .max(Comparator.comparing(TraceFragment::gpuCoreCount)) + .get() + .getCoreCount(ResourceType.GPU); + this.maxGpuMemoryDemand = 0L; // TODO: add GPU memory demand to the trace fragments + + this.resourceTypes = resourceTypes; } public ArrayList<TraceFragment> getFragments() { @@ -96,13 +114,29 @@ public class TraceWorkload implements Workload { } public int getMaxCoreCount() { - return maxCoreCount; + return maxCpuCoreCount; } public double getMaxCpuDemand() { return maxCpuDemand; } + public double getMaxGpuDemand() { + return maxGpuDemand; + } + + public int getMaxGpuCoreCount() { + return maxGpuCoreCount; + } + + public long getMaxGpuMemoryDemand() { + return maxGpuMemoryDemand; + } + + public String getTaskName() { + return taskName; + } + public void removeFragments(int numberOfFragments) { if (numberOfFragments <= 0) { return; @@ -114,14 +148,22 @@ public class TraceWorkload implements Workload { this.fragments.addFirst(fragment); } + public ResourceType[] getResourceTypes() { + return Arrays.stream(resourceTypes).filter(Objects::nonNull).toArray(ResourceType[]::new); + } + @Override public SimWorkload startWorkload(FlowSupplier supplier) { return new SimTraceWorkload(supplier, this); + // ArrayList<FlowSupplier> flowSuppliers = new ArrayList<>(); + // flowSuppliers.add(supplier); + // return new SimTraceWorkload(flowSuppliers, this); } @Override - public SimWorkload startWorkload(FlowSupplier supplier, SimMachine machine, Consumer<Exception> completion) { - return this.startWorkload(supplier); + public SimWorkload startWorkload(List<FlowSupplier> supplier, SimMachine machine, Consumer<Exception> completion) { + // return this.startWorkload(supplier); + return new SimTraceWorkload(supplier, this); } public static Builder builder( @@ -140,6 +182,7 @@ public class TraceWorkload implements Workload { private final double checkpointIntervalScaling; private final ScalingPolicy scalingPolicy; private final String taskName; + private final ResourceType[] resourceTypes = new ResourceType[ResourceType.values().length]; /** * Construct a new {@link Builder} instance. @@ -162,11 +205,23 @@ public class TraceWorkload implements Workload { * Add a fragment to the trace. * * @param duration The timestamp at which the fragment ends (in epoch millis). - * @param usage The CPU usage at this fragment. - * @param cores The number of cores used during this fragment. + * @param cpuUsage The CPU usage at this fragment. + * @param cpuCores The number of cores used during this fragment. + * @param gpuUsage The GPU usage at this fragment. + * @param gpuCores The number of GPU cores used during this fragment. + * @param gpuMemoryUsage The GPU memory usage at this fragment. */ - public void add(long duration, double usage, int cores) { - fragments.add(fragments.size(), new TraceFragment(duration, usage, cores)); + public void add( + long duration, double cpuUsage, int cpuCores, double gpuUsage, int gpuCores, long gpuMemoryUsage) { + if (cpuUsage > 0.0) { + this.resourceTypes[ResourceType.CPU.ordinal()] = ResourceType.CPU; + } + if (gpuUsage > 0.0) { + this.resourceTypes[ResourceType.GPU.ordinal()] = ResourceType.GPU; + } + fragments.add( + fragments.size(), + new TraceFragment(duration, cpuUsage, cpuCores, gpuUsage, gpuCores, gpuMemoryUsage)); } /** @@ -179,7 +234,8 @@ public class TraceWorkload implements Workload { this.checkpointDuration, this.checkpointIntervalScaling, this.scalingPolicy, - this.taskName); + this.taskName, + this.resourceTypes); } } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/NoDelayScaling.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/NoDelayScaling.java index d0c1cc2e..91538c85 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/NoDelayScaling.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/NoDelayScaling.java @@ -31,17 +31,17 @@ package org.opendc.simulator.compute.workload.trace.scaling; */ public class NoDelayScaling implements ScalingPolicy { @Override - public double getFinishedWork(double cpuFreqDemand, double cpuFreqSupplied, long passedTime) { + public double getFinishedWork(double demand, double supplied, long passedTime) { return passedTime; } @Override - public long getRemainingDuration(double cpuFreqDemand, double cpuFreqSupplied, double remainingWork) { + public long getRemainingDuration(double demand, double supplied, double remainingWork) { return (long) remainingWork; } @Override - public double getRemainingWork(double cpuFreqDemand, long duration) { + public double getRemainingWork(double demand, long duration) { return duration; } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/PerfectScaling.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/PerfectScaling.java index 7eae70e6..c4cfba66 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/PerfectScaling.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/PerfectScaling.java @@ -31,17 +31,17 @@ package org.opendc.simulator.compute.workload.trace.scaling; */ public class PerfectScaling implements ScalingPolicy { @Override - public double getFinishedWork(double cpuFreqDemand, double cpuFreqSupplied, long passedTime) { - return cpuFreqSupplied * passedTime; + public double getFinishedWork(double demand, double supplied, long passedTime) { + return supplied * passedTime; } @Override - public long getRemainingDuration(double cpuFreqDemand, double cpuFreqSupplied, double remainingWork) { - return (long) (remainingWork / cpuFreqSupplied); + public long getRemainingDuration(double demand, double supplied, double remainingWork) { + return (long) (remainingWork / supplied); } @Override - public double getRemainingWork(double cpuFreqDemand, long duration) { - return cpuFreqDemand * duration; + public double getRemainingWork(double demand, long duration) { + return demand * duration; } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/ScalingPolicy.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/ScalingPolicy.java index a0f473ba..f0676103 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/ScalingPolicy.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/ScalingPolicy.java @@ -31,29 +31,29 @@ public interface ScalingPolicy { /** * Calculate how much work was finished based on the demanded and supplied cpu * - * @param cpuFreqDemand - * @param cpuFreqSupplied - * @param passedTime - * @return + * @param demand demand of the resource by the workload + * @param supplied resource supplied for the workload + * @param passedTime time that has passed since the start + * @return the amount of work that was finished */ - double getFinishedWork(double cpuFreqDemand, double cpuFreqSupplied, long passedTime); + double getFinishedWork(double demand, double supplied, long passedTime); /** * Calculate the remaining duration of this fragment based on the demanded and supplied cpu * - * @param cpuFreqDemand - * @param cpuFreqSupplied - * @param remainingWork - * @return + * @param demand of the resource by the workload + * @param supplied resource supplied for the workload + * @param remainingWork the remaining work that needs to be done + * @return the remaining duration of the fragment */ - long getRemainingDuration(double cpuFreqDemand, double cpuFreqSupplied, double remainingWork); + long getRemainingDuration(double demand, double supplied, double remainingWork); /** * Calculate how much work is remaining based on the demanded and supplied cpu * - * @param cpuFreqDemand - * @param duration - * @return + * @param demand of the resource by the workload + * @param duration the duration of the fragment + * @return the amount of work that is remaining */ - double getRemainingWork(double cpuFreqDemand, long duration); + double getRemainingWork(double demand, long duration); } diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt index 173c60e7..eb3d3377 100644 --- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt +++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt @@ -22,32 +22,31 @@ package org.opendc.simulator.compute -import org.junit.jupiter.api.BeforeEach -import org.opendc.simulator.compute.models.CpuModel import org.opendc.simulator.compute.models.MachineModel -import org.opendc.simulator.compute.models.MemoryUnit /** * Test suite for the [SimBareMetalMachine] class. */ + class SimMachineTest { private lateinit var machineModel: MachineModel - - @BeforeEach - fun setUp() { - machineModel = - MachineModel( - CpuModel( - 0, - 2, - 1000.0, - "Intel", - "Xeon", - "amd64", - ), - MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000 * 4), - ) - } +// +// @BeforeEach +// fun setUp() { +// machineModel = +// MachineModel( +// CpuModel( +// 0, +// 2, +// 1000.0, +// "Intel", +// "Xeon", +// "amd64", +// ), +// MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000 * 4), +// null +// ) +// } // @Test // fun testFlopsWorkload() = @@ -104,10 +103,10 @@ class SimMachineTest { // val cpuNode = machineModel.cpu // val machineModel = // MachineModel( -// List(cpuNode.coreCount * 2) { +// List(cpuNode.cpuCoreCount * 2) { // CpuModel( // it, -// cpuNode.coreCount, +// cpuNode.cpuCoreCount, // 1000.0, // ) // }, diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowConsumer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowConsumer.java index a9da6f5d..ac6ba8da 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowConsumer.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowConsumer.java @@ -22,13 +22,28 @@ package org.opendc.simulator.engine.graph; +import org.opendc.common.ResourceType; + public interface FlowConsumer { void handleIncomingSupply(FlowEdge supplierEdge, double newSupply); + default void handleIncomingSupply(FlowEdge supplierEdge, double newSupply, ResourceType resourceType) { + handleIncomingSupply(supplierEdge, newSupply); + } + void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand); + default void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand, ResourceType resourceType) { + pushOutgoingDemand(supplierEdge, newDemand); + } + void addSupplierEdge(FlowEdge supplierEdge); void removeSupplierEdge(FlowEdge supplierEdge); + + // needed for flow nodes with multiple edges to same other flow node (PSU, VM) + default ResourceType getConsumerResourceType() { + return ResourceType.AUXILIARY; + } } 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 09cd73f6..674db8ca 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 @@ -23,13 +23,15 @@ package org.opendc.simulator.engine.graph; import java.util.ArrayList; -import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; +import org.opendc.common.ResourceType; import org.opendc.simulator.engine.engine.FlowEngine; +import org.opendc.simulator.engine.graph.distributionPolicies.DistributionPolicy; +import org.opendc.simulator.engine.graph.distributionPolicies.MaxMinFairnessPolicy; public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsumer { private final ArrayList<FlowEdge> consumerEdges = new ArrayList<>(); @@ -47,9 +49,16 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu private boolean overloaded = false; private double capacity; // What is the max capacity. Can probably be removed + private DistributionPolicy distributionPolicy; public FlowDistributor(FlowEngine engine) { super(engine); + this.distributionPolicy = new MaxMinFairnessPolicy(); + } + + public FlowDistributor(FlowEngine engine, DistributionPolicy distributionPolicy) { + super(engine); + this.distributionPolicy = distributionPolicy; } public double getTotalIncomingDemand() { @@ -88,6 +97,7 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu this.invalidate(); } + // TODO: This should probably be moved to the distribution strategy private void updateOutgoingSupplies() { // If the demand is higher than the current supply, the system is overloaded. @@ -95,10 +105,11 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu if (this.totalIncomingDemand > this.currentIncomingSupply) { this.overloaded = true; - double[] supplies = distributeSupply(this.incomingDemands, this.currentIncomingSupply); + double[] supplies = + this.distributionPolicy.distributeSupply(this.incomingDemands, this.currentIncomingSupply); for (int idx = 0; idx < this.consumerEdges.size(); idx++) { - this.pushOutgoingSupply(this.consumerEdges.get(idx), supplies[idx]); + this.pushOutgoingSupply(this.consumerEdges.get(idx), supplies[idx], this.getConsumerResourceType()); } } else { @@ -108,7 +119,10 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu if (this.overloaded) { for (int idx = 0; idx < this.consumerEdges.size(); idx++) { if (!Objects.equals(this.outgoingSupplies.get(idx), this.incomingDemands.get(idx))) { - this.pushOutgoingSupply(this.consumerEdges.get(idx), this.incomingDemands.get(idx)); + this.pushOutgoingSupply( + this.consumerEdges.get(idx), + this.incomingDemands.get(idx), + this.getConsumerResourceType()); } } this.overloaded = false; @@ -117,7 +131,8 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu // Update the supplies of the consumers that changed their demand in the current cycle else { for (int idx : this.updatedDemands) { - this.pushOutgoingSupply(this.consumerEdges.get(idx), this.incomingDemands.get(idx)); + this.pushOutgoingSupply( + this.consumerEdges.get(idx), this.incomingDemands.get(idx), this.getConsumerResourceType()); } } } @@ -125,48 +140,6 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu this.updatedDemands.clear(); } - private record Demand(int idx, double value) {} - - /** - * Distributed the available supply over the different demands. - * The supply is distributed using MaxMin Fairness. - */ - private static double[] distributeSupply(ArrayList<Double> demands, double currentSupply) { - int inputSize = demands.size(); - - final double[] supplies = new double[inputSize]; - final Demand[] tempDemands = new Demand[inputSize]; - - for (int i = 0; i < inputSize; i++) { - tempDemands[i] = new Demand(i, demands.get(i)); - } - - Arrays.sort(tempDemands, (o1, o2) -> { - Double i1 = o1.value; - Double i2 = o2.value; - return i1.compareTo(i2); - }); - - double availableCapacity = currentSupply; // totalSupply - - for (int i = 0; i < inputSize; i++) { - double d = tempDemands[i].value; - - if (d == 0.0) { - continue; - } - - double availableShare = availableCapacity / (inputSize - i); - double r = Math.min(d, availableShare); - - int idx = tempDemands[i].idx; - supplies[idx] = r; // Update the rates - availableCapacity -= r; - } - - return supplies; - } - /** * Add a new consumer. * Set its demand and supply to 0.0 @@ -260,6 +233,15 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu } @Override + public void handleIncomingDemand(FlowEdge consumerEdge, double newDemand, ResourceType resourceType) { + if (resourceType != this.getSupplierResourceType()) { + throw new IllegalArgumentException("Resource type " + resourceType + + " does not match distributor resource type " + this.getSupplierResourceType()); + } + this.handleIncomingDemand(consumerEdge, newDemand); + } + + @Override public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply) { this.currentIncomingSupply = newSupply; @@ -268,7 +250,7 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu @Override public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) { - this.supplierEdge.pushDemand(newDemand); + this.supplierEdge.pushDemand(newDemand, false, this.getSupplierResourceType()); } @Override @@ -284,7 +266,8 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu } outgoingSupplies.set(idx, newSupply); - consumerEdge.pushSupply(newSupply); + consumerEdge.pushSupply(newSupply, false, this.getSupplierResourceType()); + consumerEdge.pushSupply(newSupply, false, this.getSupplierResourceType()); } @Override @@ -293,4 +276,14 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu return Map.of(FlowEdge.NodeType.CONSUMING, supplyingEdges, FlowEdge.NodeType.SUPPLYING, this.consumerEdges); } + + @Override + public ResourceType getSupplierResourceType() { + return this.supplierEdge.getSupplierResourceType(); + } + + @Override + public ResourceType getConsumerResourceType() { + return this.consumerEdges.getFirst().getConsumerResourceType(); + } } diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowEdge.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowEdge.java index 95eac20b..aa3894c1 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowEdge.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowEdge.java @@ -22,6 +22,8 @@ package org.opendc.simulator.engine.graph; +import org.opendc.common.ResourceType; + /** * An edge that connects two FlowStages. * A connection between FlowStages always consist of a FlowStage that demands @@ -38,7 +40,9 @@ public class FlowEdge { private double demand = 0.0; private double supply = 0.0; - private double capacity; + private final double capacity; + + private final ResourceType resourceType; public enum NodeType { CONSUMING, @@ -46,6 +50,10 @@ public class FlowEdge { } public FlowEdge(FlowConsumer consumer, FlowSupplier supplier) { + this(consumer, supplier, ResourceType.AUXILIARY); + } + + public FlowEdge(FlowConsumer consumer, FlowSupplier supplier, ResourceType resourceType) { if (!(consumer instanceof FlowNode)) { throw new IllegalArgumentException("Flow consumer is not a FlowNode"); } @@ -55,8 +63,9 @@ public class FlowEdge { this.consumer = consumer; this.supplier = supplier; + this.resourceType = resourceType; - this.capacity = supplier.getCapacity(); + this.capacity = supplier.getCapacity(resourceType); this.consumer.addSupplierEdge(this); this.supplier.addConsumerEdge(this); @@ -112,6 +121,33 @@ public class FlowEdge { return this.supply; } + /** + * Get the resource type of this edge. + * + * @return The resource type of this edge. + */ + public ResourceType getResourceType() { + return this.resourceType; + } + + /** + * Get the resource type of the supplier of this edge. + * + * @return The resource type of the supplier. + */ + public ResourceType getSupplierResourceType() { + return this.supplier.getSupplierResourceType(); + } + + /** + * Get the resource type of the consumer of this edge. + * + * @return The resource type of the consumer. + */ + public ResourceType getConsumerResourceType() { + return this.consumer.getConsumerResourceType(); + } + public int getConsumerIndex() { return consumerIndex; } @@ -128,6 +164,16 @@ public class FlowEdge { this.supplierIndex = supplierIndex; } + public void pushDemand(double newDemand, boolean forceThrough, ResourceType resourceType) { + // or store last resource type in the edge + if ((newDemand == this.demand) && !forceThrough) { + return; + } + + this.demand = newDemand; + this.supplier.handleIncomingDemand(this, newDemand, resourceType); + } + /** * Push new demand from the Consumer to the Supplier */ @@ -150,19 +196,19 @@ public class FlowEdge { /** * Push new supply from the Supplier to the Consumer */ - public void pushSupply(double newSupply, boolean forceThrough) { + public void pushSupply(double newSupply, boolean forceThrough, ResourceType resourceType) { if ((newSupply == this.supply) && !forceThrough) { return; } this.supply = newSupply; - this.consumer.handleIncomingSupply(this, newSupply); + this.consumer.handleIncomingSupply(this, newSupply, resourceType); } /** * Push new supply from the Supplier to the Consumer */ public void pushSupply(double newSupply) { - this.pushSupply(newSupply, false); + this.pushSupply(newSupply, false, this.supplier.getSupplierResourceType()); } } diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowSupplier.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowSupplier.java index da65392b..eb665b8c 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowSupplier.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowSupplier.java @@ -22,15 +22,35 @@ package org.opendc.simulator.engine.graph; +import org.opendc.common.ResourceType; + public interface FlowSupplier { void handleIncomingDemand(FlowEdge consumerEdge, double newDemand); + default void handleIncomingDemand(FlowEdge consumerEdge, double newDemand, ResourceType resourceType) { + handleIncomingDemand(consumerEdge, newDemand); + } + void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply); + default void pushOutgoingSupply(FlowEdge consumerEdge, double newSupply, ResourceType resourceType) { + pushOutgoingSupply(consumerEdge, newSupply); + } + ; + void addConsumerEdge(FlowEdge consumerEdge); void removeConsumerEdge(FlowEdge consumerEdge); double getCapacity(); + + default double getCapacity(ResourceType resourceType) { + return getCapacity(); + } + + default ResourceType getSupplierResourceType() { + return ResourceType.AUXILIARY; + } + ; } diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/DistributionPolicy.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/DistributionPolicy.java new file mode 100644 index 00000000..9d2246cd --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/DistributionPolicy.java @@ -0,0 +1,29 @@ +/* + * Copyright (c) 2025 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.engine.graph.distributionPolicies; + +import java.util.ArrayList; + +public interface DistributionPolicy { + double[] distributeSupply(ArrayList<Double> supply, double currentSupply); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/DistributionPolicyFactory.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/DistributionPolicyFactory.java new file mode 100644 index 00000000..53cded87 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/DistributionPolicyFactory.java @@ -0,0 +1,42 @@ +/* + * Copyright (c) 2025 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.engine.graph.distributionPolicies; + +public class DistributionPolicyFactory { + + public enum DistributionPolicyType { + MaxMinFairness, + FixedShare; + } + + public static DistributionPolicy getDistributionStrategy(DistributionPolicyType distributionPolicyType) { + + return switch (distributionPolicyType) { + case MaxMinFairness -> new MaxMinFairnessPolicy(); + case FixedShare -> new FixedShare(1); + // actively misspelling + default -> throw new IllegalArgumentException( + "Unknown distribution strategy type: " + distributionPolicyType); + }; + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FixedShare.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FixedShare.java new file mode 100644 index 00000000..40d70b5e --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/FixedShare.java @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2025 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.engine.graph.distributionPolicies; + +import java.util.ArrayList; + +/** + * A distribution policy that distributes supply equally among all nodes. + * The share can be set to a fixed value, defaulting to 1. + * This policy not implemented yet and is used as a placeholder. + */ +public class FixedShare implements DistributionPolicy { + + private int share; + + public FixedShare() { + this.share = 1; + } + + public FixedShare(int share) { + this.share = share; + } + + @Override + public double[] distributeSupply(ArrayList<Double> supply, double currentSupply) { + return new double[0]; + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/MaxMinFairnessPolicy.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/MaxMinFairnessPolicy.java new file mode 100644 index 00000000..1d387349 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/distributionPolicies/MaxMinFairnessPolicy.java @@ -0,0 +1,72 @@ +/* + * Copyright (c) 2025 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.engine.graph.distributionPolicies; + +import java.util.ArrayList; +import java.util.Arrays; + +/** + * A distribution policy that implements the Max-Min Fairness algorithm. + * This policy distributes supply to demands in a way that maximizes the minimum + * allocation across all demands, ensuring fairness. + */ +public class MaxMinFairnessPolicy implements DistributionPolicy { + private record Demand(int idx, double value) {} + + @Override + public double[] distributeSupply(ArrayList<Double> demands, double currentSupply) { + int inputSize = demands.size(); + + final double[] supplies = new double[inputSize]; + final Demand[] tempDemands = new Demand[inputSize]; + + for (int i = 0; i < inputSize; i++) { + tempDemands[i] = new Demand(i, demands.get(i)); + } + + Arrays.sort(tempDemands, (o1, o2) -> { + Double i1 = o1.value; + Double i2 = o2.value; + return i1.compareTo(i2); + }); + + double availableCapacity = currentSupply; // totalSupply + + for (int i = 0; i < inputSize; i++) { + double d = tempDemands[i].value; + + if (d == 0.0) { + continue; + } + + double availableShare = availableCapacity / (inputSize - i); + double r = Math.min(d, availableShare); + + int idx = tempDemands[i].idx; + supplies[idx] = r; // Update the rates + availableCapacity -= r; + } + + return supplies; + } +} |