feat: NuProcess-based ProcessWrapper implementation

Author: metacosm

backend/pom.xml

@@ -26,6 +26,11 @@
       <artifactId>quarkus-junit5</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.zaxxer</groupId>
+      <artifactId>nuprocess</artifactId>
+      <version>2.0.6</version>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

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

@@ -0,0 +1,22 @@
+package net.laprun.sustainability.power.sensors.macos.powermetrics;
+
+import java.nio.ByteBuffer;
+
+class GrowableBuffer {
+    private ByteBuffer buffer = ByteBuffer.allocate(20000);
+
+    public void put(ByteBuffer input) {
+        if (buffer.remaining() < input.remaining()) {
+            var newBuffer = ByteBuffer.allocate(buffer.capacity() + input.remaining());
+            buffer.flip();
+            newBuffer.put(buffer);
+            buffer = newBuffer;
+        } else {
+            buffer.put(input);
+        }
+    }
+
+    public byte[] array() {
+        return buffer.array();
+    }
+}

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

@@ -0,0 +1,70 @@
+package net.laprun.sustainability.power.sensors.macos.powermetrics;
+
+import java.io.InputStream;
+import java.util.concurrent.ExecutionException;
+
+import com.zaxxer.nuprocess.NuProcess;
+import com.zaxxer.nuprocess.NuProcessBuilder;
+
+public class NuProcessWrapper implements ProcessWrapper {
+    private final PowermetricsProcessHandler metadataHandler;
+    private PowermetricsProcessHandler measureHandler;
+    private String periodInMilliSecondsAsString;
+    private long periodInMilliSeconds;
+
+    public NuProcessWrapper() {
+        metadataHandler = new PowermetricsProcessHandler("cpu_power", "-i", "10", "-n", "1");
+    }
+
+    private NuProcess exec(PowermetricsProcessHandler handler) {
+        if (handler == null)
+            throw new IllegalArgumentException("Handler cannot be null");
+        return new NuProcessBuilder(handler, handler.comand()).start();
+    }
+
+    @Override
+    public InputStream streamForMetadata() {
+        exec(metadataHandler);
+        try {
+            return metadataHandler.getInputStream().get();
+        } catch (InterruptedException | ExecutionException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public void start(long periodInMilliSeconds) {
+        // todo? check if asked period is the same as the current used one
+        this.periodInMilliSeconds = periodInMilliSeconds > 100 ? periodInMilliSeconds - 50 : periodInMilliSeconds;
+        this.periodInMilliSecondsAsString = Long.toString(this.periodInMilliSeconds);
+    }
+
+    @Override
+    public void stop() {
+        if (measureHandler != null) {
+            measureHandler.stop();
+            measureHandler = null;
+        }
+    }
+
+    @Override
+    public boolean isRunning() {
+        return measureHandler != null && measureHandler.isRunning();
+    }
+
+    @Override
+    public InputStream streamForMeasure() {
+        if (!isRunning()) {
+            measureHandler = new PowermetricsProcessHandler("cpu_power,tasks",
+                    "--show-process-samp-norm", "--show-process-gpu", "-i",
+                    periodInMilliSecondsAsString, "-n", "1");
+            exec(measureHandler);
+            try {
+                return measureHandler.getInputStream().get();
+            } catch (InterruptedException | ExecutionException e) {
+                throw new RuntimeException(e);
+            }
+        }
+        throw new IllegalStateException("Measure is still running");
+    }
+}

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

@@ -4,6 +4,8 @@
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import com.zaxxer.nuprocess.NuAbstractProcessHandler;
@@ -13,7 +15,8 @@ public class PowermetricsProcessHandler extends NuAbstractProcessHandler {
     private String errorMsg;
     private NuProcess process;
     private final String[] command;
-    private ByteArrayInputStream bais;
+    private final GrowableBuffer stdOutBuffer = new GrowableBuffer();
+    private final CompletableFuture<InputStream> output = new CompletableFuture<>();
 
     public PowermetricsProcessHandler(String... command) {
         if (command == null || command.length == 0) {
@@ -63,8 +66,12 @@ public void stop() {
 
     @Override
     public void onStdout(ByteBuffer buffer, boolean closed) {
-        if (!closed) {
-            bais = new ByteArrayInputStream(buffer.array());
+        if (buffer.hasRemaining()) {
+            stdOutBuffer.put(buffer);
+        }
+
+        if (closed) {
+            output.complete(new ByteArrayInputStream(stdOutBuffer.array()));
         }
     }
 
@@ -78,8 +85,8 @@ public void onStderr(ByteBuffer buffer, boolean closed) {
         super.onStderr(buffer, closed);
     }
 
-    public InputStream getInputStream() {
-        return bais;
+    public Future<InputStream> getInputStream() {
+        return output;
     }
 
     public boolean isRunning() {

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

@@ -3,7 +3,7 @@
 import java.io.InputStream;
 
 public class ProcessMacOSPowermetricsSensor extends MacOSPowermetricsSensor {
-    private final ProcessWrapper processWrapper = new JavaProcessWrapper();
+    private final ProcessWrapper processWrapper = new NuProcessWrapper();
 
     public ProcessMacOSPowermetricsSensor() {
         // extract metadata

server/src/main/java/net/laprun/sustainability/power/PowerMeasurer.java

@@ -40,6 +40,7 @@ private RegisteredPID track(long pid) throws Exception {
             sensor.start(samplingPeriod.toMillis());
             periodicSensorCheck = Multi.createFrom().ticks()
                     .every(samplingPeriod)
+                    .log()
                     .map(sensor::update)
                     .broadcast()
                     .withCancellationAfterLastSubscriberDeparture()