feat: also send ticks with measures, avoid creating a Map per measure

metadata/src/main/java/io/github/metacosm/power/SensorMeasure.java

@@ -0,0 +1,4 @@
+package io.github.metacosm.power;
+
+public record SensorMeasure(double[] components, long tick) {
+}

server/src/main/java/io/github/metacosm/power/PowerMeasurer.java

@@ -1,5 +1,6 @@
 package io.github.metacosm.power;
 
+import io.github.metacosm.power.sensors.Measures;
 import io.github.metacosm.power.sensors.PowerSensor;
 import io.github.metacosm.power.sensors.RegisteredPID;
 import io.smallrye.mutiny.Multi;
@@ -17,9 +18,9 @@ public class PowerMeasurer {
     @Inject
     PowerSensor sensor;
 
-    private Multi<Map<RegisteredPID, double[]>> periodicSensorCheck;
+    private Multi<Measures> periodicSensorCheck;
 
-    public Multi<double[]> startTracking(String pid) throws Exception {
+    public Multi<SensorMeasure> startTracking(String pid) throws Exception {
         // first make sure that the process with that pid exists
         final var parsedPID = Long.parseLong(pid);
         ProcessHandle.of(parsedPID).orElseThrow(() -> new IllegalArgumentException("Unknown process: " + pid));
@@ -36,7 +37,7 @@ public Multi<double[]> startTracking(String pid) throws Exception {
         // todo: the timing of things could make it so that the pid has been removed before the map operation occurs so
         //  currently return -1 instead of null but this needs to be properly addressed
         return periodicSensorCheck
-                .map(registeredPIDMap -> registeredPIDMap.getOrDefault(registeredPID, new double[]{-1.0}))
+                .map(measures -> measures.getOrDefault(registeredPID))
                 .onCancellation().invoke(() -> sensor.unregister(registeredPID));
     }
 

server/src/main/java/io/github/metacosm/power/PowerResource.java

@@ -4,24 +4,19 @@
 import jakarta.inject.Inject;
 import jakarta.ws.rs.*;
 import jakarta.ws.rs.core.MediaType;
+import org.jboss.resteasy.reactive.RestStreamElementType;
 
 @Path("/power")
 public class PowerResource {
     @Inject
     PowerMeasurer measurer;
 
     @GET
-    @Produces(MediaType.SERVER_SENT_EVENTS)
+    @RestStreamElementType(MediaType.APPLICATION_JSON)
     @Path("{pid}")
-    public Multi<String> powerFor(@PathParam("pid") String pid) throws Exception {
+    public Multi<SensorMeasure> powerFor(@PathParam("pid") String pid) throws Exception {
         try {
-            return measurer.startTracking(pid).map(array -> {
-                String result = "\n";
-                for (double v : array) {
-                    result += v + " ";
-                }
-                return result;
-            });
+            return measurer.startTracking(pid);
         } catch (IllegalArgumentException e) {
             throw new NotFoundException("Unknown process: " + pid);
         }

server/src/main/java/io/github/metacosm/power/sensors/Measures.java

@@ -0,0 +1,20 @@
+package io.github.metacosm.power.sensors;
+
+import io.github.metacosm.power.SensorMeasure;
+
+import java.util.Set;
+
+public interface Measures {
+    SensorMeasure missing = new SensorMeasure(new double[]{-1.0}, -1);
+    RegisteredPID register(long pid);
+
+    void unregister(RegisteredPID registeredPID);
+
+    Set<RegisteredPID> trackedPIDs();
+
+    int numberOfTrackerPIDs();
+
+    void record(RegisteredPID pid, SensorMeasure sensorMeasure);
+
+    SensorMeasure getOrDefault(RegisteredPID pid);
+}

server/src/main/java/io/github/metacosm/power/sensors/PowerSensor.java

@@ -1,5 +1,6 @@
 package io.github.metacosm.power.sensors;
 
+import io.github.metacosm.power.SensorMeasure;
 import io.github.metacosm.power.SensorMetadata;
 
 import java.util.Map;
@@ -17,7 +18,7 @@ default void stop() {
 
     RegisteredPID register(long pid);
 
-    Map<RegisteredPID,double[]> update(Long tick);
+    Measures update(Long tick);
 
     void unregister(RegisteredPID registeredPID);
 }

server/src/main/java/io/github/metacosm/power/sensors/linux/rapl/IntelRAPLSensor.java

@@ -1,6 +1,8 @@
 package io.github.metacosm.power.sensors.linux.rapl;
 
+import io.github.metacosm.power.SensorMeasure;
 import io.github.metacosm.power.SensorMetadata;
+import io.github.metacosm.power.sensors.Measures;
 import io.github.metacosm.power.sensors.PowerSensor;
 import io.github.metacosm.power.sensors.RegisteredPID;
 
@@ -15,7 +17,7 @@ public class IntelRAPLSensor implements PowerSensor {
     private final SensorMetadata metadata;
     private final double[] lastMeasuredSensorValues;
     private long frequency;
-    private final Set<RegisteredPID> trackedPIDs = new HashSet<>();
+    private final SingleMeasureMeasures measures = new SingleMeasureMeasures();
 
     public IntelRAPLSensor() {
         // if we total system energy is not available, read package and DRAM if possible
@@ -89,27 +91,24 @@ public boolean isStarted() {
 
     @Override
     public RegisteredPID register(long pid) {
-        final var registeredPID = new RegisteredPID(pid);
-        trackedPIDs.add(registeredPID);
-        return registeredPID;
+        return measures.register(pid);
     }
 
     @Override
-    public Map<RegisteredPID, double[]> update(Long tick) {
+    public Measures update(Long tick) {
         final var measure = new double[raplFiles.length];
         for (int i = 0; i < raplFiles.length; i++) {
             final var value = raplFiles[i].extractPowerMeasure();
             final var newComponentValue = computeNewComponentValue(i, value);
             measure[i] = newComponentValue;
             lastMeasuredSensorValues[i] = newComponentValue;
         }
-        final var result = new HashMap<RegisteredPID, double[]>(trackedPIDs.size());
-        trackedPIDs.forEach(registeredPID -> result.put(registeredPID, measure));
-        return result;
+        measures.singleMeasure(new SensorMeasure(measure, tick));
+        return measures;
     }
 
     @Override
     public void unregister(RegisteredPID registeredPID) {
-        trackedPIDs.remove(registeredPID);
+        measures.unregister(registeredPID);
     }
 }

server/src/main/java/io/github/metacosm/power/sensors/linux/rapl/SingleMeasureMeasures.java

@@ -0,0 +1,48 @@
+package io.github.metacosm.power.sensors.linux.rapl;
+
+import io.github.metacosm.power.SensorMeasure;
+import io.github.metacosm.power.sensors.Measures;
+import io.github.metacosm.power.sensors.RegisteredPID;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class SingleMeasureMeasures implements Measures {
+    private final Set<RegisteredPID> trackedPIDs = new HashSet<>();
+    private SensorMeasure measure;
+    void singleMeasure(SensorMeasure sensorMeasure) {
+        this.measure = sensorMeasure;
+    }
+
+    @Override
+    public RegisteredPID register(long pid) {
+        final var registeredPID = new RegisteredPID(pid);
+        trackedPIDs.add(registeredPID);
+        return registeredPID;
+    }
+
+    @Override
+    public void unregister(RegisteredPID registeredPID) {
+        trackedPIDs.remove(registeredPID);
+    }
+
+    @Override
+    public Set<RegisteredPID> trackedPIDs() {
+        return trackedPIDs;
+    }
+
+    @Override
+    public int numberOfTrackerPIDs() {
+        return trackedPIDs.size();
+    }
+
+    @Override
+    public void record(RegisteredPID pid, SensorMeasure sensorMeasure) {
+       throw new UnsupportedOperationException("Shouldn't be needed");
+    }
+
+    @Override
+    public SensorMeasure getOrDefault(RegisteredPID pid) {
+        return trackedPIDs.contains(pid) && measure != null ? measure : Measures.missing;
+    }
+}

server/src/main/java/io/github/metacosm/power/sensors/macos/powermetrics/MacOSPowermetricsSensor.java

@@ -1,6 +1,8 @@
 package io.github.metacosm.power.sensors.macos.powermetrics;
 
+import io.github.metacosm.power.SensorMeasure;
 import io.github.metacosm.power.SensorMetadata;
+import io.github.metacosm.power.sensors.Measures;
 import io.github.metacosm.power.sensors.PowerSensor;
 import io.github.metacosm.power.sensors.RegisteredPID;
 
@@ -11,7 +13,6 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
 public class MacOSPowermetricsSensor implements PowerSensor {
@@ -30,7 +31,7 @@ public class MacOSPowermetricsSensor implements PowerSensor {
     private static final SensorMetadata.ComponentMetadata gpu = new SensorMetadata.ComponentMetadata(GPU, 1, "GPU power", true, "mW");
     private static final SensorMetadata.ComponentMetadata ane = new SensorMetadata.ComponentMetadata(ANE, 2, "Apple Neural Engine power", false, "mW");
     private static final SensorMetadata.ComponentMetadata cpuShare = new SensorMetadata.ComponentMetadata(CPU_SHARE, 3, "Computed share of CPU", false, "decimal percentage");
-    private final Map<String, RegisteredPID> trackedPIDs = new ConcurrentHashMap<>();
+    private final Measures measures = new MapMeasures();
 
     private final SensorMetadata metadata;
 
@@ -125,13 +126,10 @@ public ProcessRecord(String line) throws IllegalArgumentException {
 
     @Override
     public RegisteredPID register(long pid) {
-        final var key = RegisteredPID.prepare(pid);
-        final var registeredPID = new RegisteredPID(key);
-        trackedPIDs.put(key, registeredPID);
-        return registeredPID;
+        return measures.register(pid);
     }
 
-    Map<RegisteredPID, double[]> extractPowerMeasure(InputStream powerMeasureInput) {
+    Measures extractPowerMeasure(InputStream powerMeasureInput, Long tick) {
         try {
             // Should not be closed since it closes the process
             BufferedReader input = new BufferedReader(new InputStreamReader(powerMeasureInput));
@@ -141,9 +139,9 @@ Map<RegisteredPID, double[]> extractPowerMeasure(InputStream powerMeasureInput)
             double totalSampledGPU = -1;
             int headerLinesToSkip = 10;
             // copy the pids so that we can remove them as soon as we've processed them
-            final var pidsToProcess = new HashSet<>(trackedPIDs.keySet());
+            final var pidsToProcess = new HashSet<>(measures.trackedPIDs());
             // start measure
-            final var measures = new HashMap<RegisteredPID, ProcessRecord>(trackedPIDs.size());
+            final var pidMeasures = new HashMap<RegisteredPID, ProcessRecord>(measures.numberOfTrackerPIDs());
             final var powerComponents = new HashMap<String, Integer>(metadata.componentCardinality());
             while ((line = input.readLine()) != null) {
                 if (headerLinesToSkip != 0) {
@@ -157,10 +155,12 @@ Map<RegisteredPID, double[]> extractPowerMeasure(InputStream powerMeasureInput)
 
                 // first, look for process line detailing share
                 if (!pidsToProcess.isEmpty()) {
-                    if (pidsToProcess.stream().anyMatch(line::contains)) {
-                        final var procInfo = new ProcessRecord(line);
-                        measures.put(trackedPIDs.get(procInfo.pid), procInfo);
-                        pidsToProcess.remove(procInfo.pid);
+                    for (RegisteredPID pid : pidsToProcess) {
+                        if(line.contains(pid.stringForMatching())) {
+                            pidMeasures.put(pid, new ProcessRecord(line));
+                            pidsToProcess.remove(pid);
+                            break;
+                        }
                     }
                     continue;
                 }
@@ -189,8 +189,7 @@ Map<RegisteredPID, double[]> extractPowerMeasure(InputStream powerMeasureInput)
             final var hasGPU = totalSampledGPU != 0;
             double finalTotalSampledGPU = totalSampledGPU;
             double finalTotalSampledCPU = totalSampledCPU;
-            final var results = new HashMap<RegisteredPID, double[]>(measures.size());
-            measures.forEach((pid, record) -> {
+            pidMeasures.forEach((pid, record) -> {
                 final var cpuShare = record.cpu / finalTotalSampledCPU;
                 final var measure = new double[metadata.componentCardinality()];
 
@@ -211,12 +210,12 @@ Map<RegisteredPID, double[]> extractPowerMeasure(InputStream powerMeasureInput)
                     }
                 });
 
-                results.put(pid, measure);
+                measures.record(pid, new SensorMeasure(measure, tick));
             });
-            return results;
         } catch (Exception exception) {
             throw new RuntimeException(exception);
         }
+        return measures;
     }
 
     private static void extractPowerComponents(String line, HashMap<String, Integer> powerComponents) {
@@ -252,8 +251,8 @@ public boolean isStarted() {
     }
 
     @Override
-    public Map<RegisteredPID, double[]> update(Long tick) {
-        return extractPowerMeasure(powermetrics.getInputStream());
+    public Measures update(Long tick) {
+        return extractPowerMeasure(powermetrics.getInputStream(), tick);
     }
 
     @Override
@@ -263,9 +262,9 @@ public void stop() {
 
     @Override
     public void unregister(RegisteredPID registeredPID) {
-        trackedPIDs.remove(registeredPID.stringForMatching());
+        measures.unregister(registeredPID);
         // if we're not tracking any processes anymore, stop powermetrics as well
-        if (trackedPIDs.isEmpty()) {
+        if (measures.numberOfTrackerPIDs() == 0) {
             stop();
         }
     }

server/src/main/java/io/github/metacosm/power/sensors/macos/powermetrics/MapMeasures.java

@@ -0,0 +1,45 @@
+package io.github.metacosm.power.sensors.macos.powermetrics;
+
+import io.github.metacosm.power.SensorMeasure;
+import io.github.metacosm.power.sensors.Measures;
+import io.github.metacosm.power.sensors.RegisteredPID;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+class MapMeasures implements Measures {
+    private final ConcurrentMap<RegisteredPID, SensorMeasure> measures = new ConcurrentHashMap<>();
+
+    @Override
+    public RegisteredPID register(long pid) {
+        final var key = new RegisteredPID(pid);
+        measures.put(key, missing);
+        return key;
+    }
+
+    @Override
+    public void unregister(RegisteredPID registeredPID) {
+        measures.remove(registeredPID);
+    }
+
+    @Override
+    public Set<RegisteredPID> trackedPIDs() {
+        return measures.keySet();
+    }
+
+    @Override
+    public int numberOfTrackerPIDs() {
+        return measures.size();
+    }
+
+    @Override
+    public void record(RegisteredPID pid, SensorMeasure sensorMeasure) {
+        measures.put(pid, sensorMeasure);
+    }
+
+    @Override
+    public SensorMeasure getOrDefault(RegisteredPID pid) {
+        return measures.getOrDefault(pid, missing);
+    }
+}

server/src/test/java/io/github/metacosm/power/sensors/macos/powermetrics/MacOSPowermetricsSensorTest.java

@@ -58,19 +58,19 @@ void extractPowerMeasure() {
 
         // re-open the stream to read the measure this time
         in = Thread.currentThread().getContextClassLoader().getResourceAsStream("sonoma-m1max.txt");
-        final var measure = sensor.extractPowerMeasure(in);
+        final var measure = sensor.extractPowerMeasure(in, 0L);
         final var cpuIndex = metadata.metadataFor(MacOSPowermetricsSensor.CPU).index();
         final var pid1CPUShare = 23.88 / 1222.65;
-        assertEquals((pid1CPUShare * 211), measure.get(pid1)[cpuIndex]);
+        assertEquals((pid1CPUShare * 211), measure.getOrDefault(pid1).components()[cpuIndex]);
         final var pid2CPUShare = 283.25 / 1222.65;
-        assertEquals((pid2CPUShare * 211), measure.get(pid2)[cpuIndex]);
+        assertEquals((pid2CPUShare * 211), measure.getOrDefault(pid2).components()[cpuIndex]);
         // check cpu share
         final var cpuShareIndex = metadata.metadataFor(MacOSPowermetricsSensor.CPU_SHARE).index();
-        assertEquals(pid1CPUShare, measure.get(pid1)[cpuShareIndex]);
-        assertEquals(pid2CPUShare, measure.get(pid2)[cpuShareIndex]);
+        assertEquals(pid1CPUShare, measure.getOrDefault(pid1).components()[cpuShareIndex]);
+        assertEquals(pid2CPUShare, measure.getOrDefault(pid2).components()[cpuShareIndex]);
         // check that gpu should be 0
         final var gpuIndex = metadata.metadataFor(MacOSPowermetricsSensor.GPU).index();
-        assertEquals(0.0, measure.get(pid1)[gpuIndex]);
-        assertEquals(0.0, measure.get(pid2)[gpuIndex]);
+        assertEquals(0.0, measure.getOrDefault(pid1).components()[gpuIndex]);
+        assertEquals(0.0, measure.getOrDefault(pid2).components()[gpuIndex]);
     }
 }