feat: move pid management to sensors and out of measures

Author: metacosm

backend/src/main/java/net/laprun/sustainability/power/sensors/AbstractPowerSensor.java

@@ -8,21 +8,14 @@
 import net.laprun.sustainability.power.SensorMetadata;
 
 public abstract class AbstractPowerSensor implements PowerSensor {
-    protected final Measures measures;
+    protected final Measures measures = new MapMeasures();
+    private final PIDRegistry registry = new PIDRegistry();
     private long lastUpdateEpoch;
     private boolean started;
     @ConfigProperty(name = "power-server.enable-cpu-share-sampling", defaultValue = "false")
     protected boolean cpuSharesEnabled;
     private SensorMetadata metadata;
 
-    public AbstractPowerSensor(Measures measures) {
-        this.measures = measures;
-    }
-
-    public AbstractPowerSensor() {
-        this(new MapMeasures());
-    }
-
     @Override
     public SensorMetadata metadata() {
         if (metadata == null) {
@@ -46,15 +39,31 @@ public void enableCPUShareSampling(boolean enable) {
     @Override
     public RegisteredPID register(long pid) {
         Log.debugf("Registered pid: %d", pid);
-        return measures.register(pid);
+        final var key = RegisteredPID.create(pid);
+        registry.register(key);
+        return key;
     }
 
     @Override
     public void unregister(RegisteredPID registeredPID) {
-        measures.unregister(registeredPID);
+        registry.unregister(registeredPID);
         Log.debugf("Unregistered pid: %d", registeredPID.pid());
     }
 
+    protected int numberOfRegisteredPIDs() {
+        return registry.size();
+    }
+
+    @Override
+    public Set<RegisteredPID> registeredPIDs() {
+        return registry.pids();
+    }
+
+    @Override
+    public Set<String> registeredPIDsAsStrings() {
+        return registry.pidsAsStrings();
+    }
+
     @Override
     public void start() throws Exception {
         if (!started) {
@@ -64,6 +73,8 @@ public void start() throws Exception {
         }
     }
 
+    protected abstract void doStart();
+
     @Override
     public boolean isStarted() {
         return started;
@@ -75,14 +86,6 @@ public void stop() {
         started = false;
     }
 
-    @Override
-    public Set<String> getRegisteredPIDs() {
-        return measures.trackedPIDsAsString();
-    }
-
-    protected abstract void doStart();
-
-    @Override
     public Measures update(long tick) {
         final long newUpdateStartEpoch = System.currentTimeMillis();
         Log.debugf("Sensor update last called: %dms ago", newUpdateStartEpoch - lastUpdateEpoch);

backend/src/main/java/net/laprun/sustainability/power/sensors/MapMeasures.java

@@ -1,47 +1,17 @@
 package net.laprun.sustainability.power.sensors;
 
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
 import net.laprun.sustainability.power.SensorMeasure;
 
 public class MapMeasures implements Measures {
     private final ConcurrentMap<RegisteredPID, SensorMeasure> measures = new ConcurrentHashMap<>();
-    private final PIDRegistry registry = new PIDRegistry();
 
     @Override
-    public RegisteredPID register(long pid) {
-        final var key = RegisteredPID.create(pid);
-        measures.put(key, SensorMeasure.missing);
-        registry.register(key);
-        return key;
-    }
-
-    @Override
-    public void unregister(RegisteredPID registeredPID) {
-        measures.remove(registeredPID);
-        registry.unregister(registeredPID);
-    }
-
-    @Override
-    public Set<RegisteredPID> trackedPIDs() {
-        return measures.keySet();
-    }
-
-    @Override
-    public Set<String> trackedPIDsAsString() {
-        return registry.pids();
-    }
-
-    @Override
-    public int numberOfTrackedPIDs() {
-        return measures.size();
-    }
-
-    @Override
-    public void record(RegisteredPID pid, SensorMeasure sensorMeasure) {
+    public Measures record(RegisteredPID pid, SensorMeasure sensorMeasure) {
         measures.put(pid, sensorMeasure);
+        return this;
     }
 
     @Override
@@ -53,4 +23,9 @@ public SensorMeasure getOrDefault(RegisteredPID pid) {
     public String toString() {
         return measures.toString();
     }
+
+    @Override
+    public void clear() {
+        measures.clear();
+    }
 }

backend/src/main/java/net/laprun/sustainability/power/sensors/Measures.java

@@ -1,56 +1,20 @@
 package net.laprun.sustainability.power.sensors;
 
-import java.util.Set;
-
 import net.laprun.sustainability.power.SensorMeasure;
 
 /**
  * A representation of ongoing {@link PowerSensor} measures.
  */
 public interface Measures {
 
-    /**
-     * Tracks the provided process identifier (pid) in the measures. For sensors that only provide system-wide measures, this
-     * probably won't be doing much more than track which processes are of interest to clients of the sensor.
-     *
-     * @param pid the process identifier which power consumption is supposed to be tracked
-     * @return a {@link RegisteredPID} recording the tracking of the specified pid by the sensor
-     */
-    RegisteredPID register(long pid);
-
-    /**
-     * Unregisters the specified {@link RegisteredPID} thus signaling that clients are not interested in tracking the
-     * consumption of the associated process anymore
-     *
-     * @param registeredPID the {@link RegisteredPID} that was returned when the process we want to stop tracking was first
-     *        registered
-     */
-    void unregister(RegisteredPID registeredPID);
-
-    /**
-     * Retrieves the set of tracked process identifiers
-     *
-     * @return the set of tracked process identifiers
-     */
-    Set<RegisteredPID> trackedPIDs();
-
-    Set<String> trackedPIDsAsString();
-
-    /**
-     * Retrieves the number of tracked processes
-     *
-     * @return the number of tracked processes
-     */
-    int numberOfTrackedPIDs();
-
     /**
      * Records the specified measure and associates it to the specified tracked process, normally called once per tick
      *
      * @param pid the {@link RegisteredPID} representing the tracked process with which the recorded measure needs to be
      *        associated
      * @param sensorMeasure the {@link SensorMeasure} to be recorded
      */
-    void record(RegisteredPID pid, SensorMeasure sensorMeasure);
+    Measures record(RegisteredPID pid, SensorMeasure sensorMeasure);
 
     /**
      * Retrieves the last recorded {@link SensorMeasure} associated with the specified {@link RegisteredPID}
@@ -60,10 +24,14 @@ public interface Measures {
      *         it cannot be
      *         retrieved for any reason
      */
-    SensorMeasure getOrDefault(RegisteredPID pid);
+    default SensorMeasure getOrDefault(RegisteredPID pid) {
+        return SensorMeasure.missing;
+    }
 
     @SuppressWarnings("unused")
     default SensorMeasure getSystemTotal() {
         return getOrDefault(RegisteredPID.SYSTEM_TOTAL_REGISTERED_PID);
     }
+
+    void clear();
 }

backend/src/main/java/net/laprun/sustainability/power/sensors/PIDRegistry.java

@@ -2,23 +2,32 @@
 
 import java.util.HashSet;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 public class PIDRegistry {
-    private final Set<String> pids = new HashSet<>();
+    private final Set<RegisteredPID> pids = new HashSet<>();
 
     public void register(RegisteredPID registeredPID) {
         if (!RegisteredPID.SYSTEM_TOTAL_REGISTERED_PID.equals(registeredPID)) {
-            pids.add(registeredPID.pidAsString());
+            pids.add(registeredPID);
         }
     }
 
     public void unregister(RegisteredPID registeredPID) {
         if (!RegisteredPID.SYSTEM_TOTAL_REGISTERED_PID.equals(registeredPID)) {
-            pids.remove(registeredPID.pidAsString());
+            pids.remove(registeredPID);
         }
     }
 
-    public Set<String> pids() {
+    public Set<String> pidsAsStrings() {
+        return pids.stream().map(RegisteredPID::pidAsString).collect(Collectors.toSet());
+    }
+
+    public Set<RegisteredPID> pids() {
         return pids;
     }
+
+    public int size() {
+        return pids.size();
+    }
 }

backend/src/main/java/net/laprun/sustainability/power/sensors/PowerSensor.java

@@ -82,5 +82,7 @@ default void stop() {
      */
     void unregister(RegisteredPID registeredPID);
 
-    Set<String> getRegisteredPIDs();
+    Set<String> registeredPIDsAsStrings();
+
+    Set<RegisteredPID> registeredPIDs();
 }

backend/src/main/java/net/laprun/sustainability/power/sensors/SamplingMeasurer.java

@@ -108,7 +108,7 @@ private void startSamplingIfNeeded() throws Exception {
             if (sensor.wantsCPUShareSamplingEnabled()) {
                 // if enabled, record a cpu share for each tick, group by the over sampling factor and average over these aggregates to produce one value for the power measure interval
                 cpuSharesMulti = cpuSharesTicks
-                        .map(tick -> CPUShare.cpuSharesFor(sensor.getRegisteredPIDs()))
+                        .map(tick -> CPUShare.cpuSharesFor(sensor.registeredPIDsAsStrings()))
                         .group()
                         .intoLists()
                         .of(overSamplingFactor)

backend/src/main/java/net/laprun/sustainability/power/sensors/linux/rapl/IntelRAPLSensor.java

@@ -147,7 +147,7 @@ protected Measures doUpdate(long lastUpdateEpoch, long newUpdateStartEpoch) {
                 newUpdateStartEpoch);
 
         final var single = new NoDurationSensorMeasure(measure, lastUpdateEpoch, newUpdateStartEpoch);
-        measures.trackedPIDs().forEach(pid -> measures.record(pid, single));
+        registeredPIDs().forEach(pid -> measures.record(pid, single));
 
         return measures;
     }

backend/src/main/java/net/laprun/sustainability/power/sensors/macos/powermetrics/MacOSPowermetricsSensor.java

@@ -72,8 +72,8 @@ Measures extractPowerMeasure(InputStream powerMeasureInput, long lastUpdateEpoch
             Log.debugf("powermetrics measure extraction last called %dms ago", (start - lastCalled));
             lastCalled = start;
         }
-        PowerMetricsParser.extractPowerMeasure(powerMeasureInput, measures, lastUpdateEpoch, newUpdateEpoch,
-                measures.trackedPIDs(), metadata(), cpu);
+        PowerMetricsParser.extractPowerMeasure(powerMeasureInput, measures, lastUpdateEpoch, newUpdateEpoch, registeredPIDs(),
+                metadata(), cpu);
         return measures;
     }
 
@@ -88,7 +88,7 @@ protected Measures doUpdate(long lastUpdateEpoch, long newUpdateStartEpoch) {
     public void unregister(RegisteredPID registeredPID) {
         super.unregister(registeredPID);
         // if we're not tracking any processes anymore, stop powermetrics as well
-        if (measures.numberOfTrackedPIDs() == 0) {
+        if (numberOfRegisteredPIDs() == 0) {
             stop();
         }
     }

backend/src/main/java/net/laprun/sustainability/power/sensors/test/TestPowerSensor.java

@@ -7,7 +7,6 @@
 import net.laprun.sustainability.power.SensorMetadata;
 import net.laprun.sustainability.power.measures.NoDurationSensorMeasure;
 import net.laprun.sustainability.power.sensors.AbstractPowerSensor;
-import net.laprun.sustainability.power.sensors.MapMeasures;
 import net.laprun.sustainability.power.sensors.Measures;
 
 @SuppressWarnings("unused")
@@ -23,7 +22,6 @@ public TestPowerSensor() {
     }
 
     public TestPowerSensor(SensorMetadata metadata) {
-        super(new MapMeasures());
         this.metadata = metadata;
     }
 
@@ -39,7 +37,7 @@ public void doStart() {
 
     @Override
     protected Measures doUpdate(long lastUpdateEpoch, long newUpdateStartEpoch) {
-        measures.trackedPIDs().forEach(pid -> measures.record(pid,
+        registeredPIDs().forEach(pid -> measures.record(pid,
                 new NoDurationSensorMeasure(new double[] { Math.random() }, lastUpdateEpoch, newUpdateStartEpoch)));
         return measures;
     }