feat: decouple UI from power measurement (#16)

deployment/src/main/java/io/quarkiverse/power/deployment/devui/commands/StartCommand.java

@@ -1,18 +1,19 @@
 package io.quarkiverse.power.deployment.devui.commands;
 
 import org.aesh.command.CommandDefinition;
-import org.aesh.command.CommandException;
 import org.aesh.command.CommandResult;
 import org.aesh.command.invocation.CommandInvocation;
 import org.aesh.command.option.Option;
 
+import io.quarkiverse.power.runtime.PowerMeasure;
 import io.quarkiverse.power.runtime.PowerMeasurer;
 import io.quarkiverse.power.runtime.SensorMeasure;
 import io.quarkus.deployment.console.QuarkusCommand;
 
 @CommandDefinition(name = "start", description = "Starts measuring power consumption of the current application")
 public class StartCommand extends QuarkusCommand {
     private final PowerMeasurer<? extends SensorMeasure> sensor;
+    private PowerMeasure<?> baseline;
 
     @Option(name = "stopAfter", shortName = 's', description = "Automatically stop the measures after the specified duration in seconds", defaultValue = "-1")
     private long duration;
@@ -25,7 +26,7 @@ public StartCommand(PowerMeasurer<? extends SensorMeasure> sensor) {
     }
 
     @Override
-    public CommandResult doExecute(CommandInvocation commandInvocation) throws CommandException, InterruptedException {
+    public CommandResult doExecute(CommandInvocation commandInvocation) {
         try {
             if (!sensor.isRunning()) {
                 if (duration > 0) {
@@ -35,7 +36,30 @@ public CommandResult doExecute(CommandInvocation commandInvocation) throws Comma
                     commandInvocation.println("Measuring power every " + frequency
                             + " milliseconds. Execute 'power stop' to stop measurements and get the results.");
                 }
-                sensor.start(duration, frequency, commandInvocation::println);
+
+                if (baseline == null) {
+                    commandInvocation.println("Establishing baseline for 30s, please do not use your application until done.");
+                    commandInvocation.println("Power measurement will start as configured after this initial measure is done.");
+                    sensor.start(30, 1000);
+                    sensor.onError(e -> commandInvocation.println("An error occurred: " + e.getMessage()));
+                    sensor.onCompleted((m) -> {
+                        baseline = m;
+                        outputConsumptionSinceStarted(baseline, commandInvocation, true);
+                        commandInvocation.println("Baseline established! You can now interact with your application normally.");
+
+                        try {
+                            sensor.start(duration, frequency);
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
+                        }
+                        sensor.onCompleted(
+                                (finished) -> outputConsumptionSinceStarted(finished, commandInvocation, false));
+                    });
+                } else {
+                    sensor.start(duration, frequency);
+                    sensor.onCompleted((m) -> outputConsumptionSinceStarted(m, commandInvocation, false));
+                }
+
             } else {
                 commandInvocation.println("Power measurement is already ongoing. Execute 'power stop' to stop it now.");
             }
@@ -45,4 +69,20 @@ public CommandResult doExecute(CommandInvocation commandInvocation) throws Comma
         }
         return CommandResult.SUCCESS;
     }
+
+    private void outputConsumptionSinceStarted(PowerMeasure<?> measure, CommandInvocation out, boolean isBaseline) {
+        final var durationInSeconds = measure.duration() / 1000;
+        final var title = isBaseline ? "Baseline power: " : "Measured power: ";
+        out.println(title + getReadablePower(measure) + " over " + durationInSeconds
+                + " seconds (" + measure.numberOfSamples() + " samples)");
+        if (!isBaseline) {
+            sensor.additionalSensorInfo().ifPresent(out::println);
+            out.println("Baseline power was " + getReadablePower(baseline));
+        }
+    }
+
+    private static String getReadablePower(PowerMeasure<?> measure) {
+        final var measuredMilliWatts = measure.total();
+        return measuredMilliWatts >= 1000 ? (measuredMilliWatts / 1000) + " W" : measuredMilliWatts + "mW";
+    }
 }

deployment/src/main/java/io/quarkiverse/power/deployment/devui/commands/StopCommand.java

@@ -1,7 +1,6 @@
 package io.quarkiverse.power.deployment.devui.commands;
 
 import org.aesh.command.CommandDefinition;
-import org.aesh.command.CommandException;
 import org.aesh.command.CommandResult;
 import org.aesh.command.invocation.CommandInvocation;
 
@@ -19,9 +18,10 @@ public StopCommand(PowerMeasurer<? extends SensorMeasure> sensor) {
     }
 
     @Override
-    public CommandResult doExecute(CommandInvocation commandInvocation) throws CommandException, InterruptedException {
+    public CommandResult doExecute(CommandInvocation commandInvocation) {
+        sensor.onError(e -> commandInvocation.println("An error occurred: " + e.getMessage()));
         if (sensor.isRunning()) {
-            sensor.stop(commandInvocation::println);
+            sensor.stop();
         } else {
             commandInvocation.println("Power measurement hasn't started. Execute 'power start' to start it first.");
         }

runtime/src/main/java/io/quarkiverse/power/runtime/PowerMeasurer.java

@@ -1,10 +1,10 @@
 package io.quarkiverse.power.runtime;
 
 import java.lang.management.ManagementFactory;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
+import java.util.Optional;
+import java.util.concurrent.*;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 
 import com.sun.management.OperatingSystemMXBean;
 
@@ -26,8 +26,10 @@ public class PowerMeasurer<M extends IncrementableMeasure> {
     private ScheduledFuture<?> scheduled;
     private final PowerSensor<M> sensor;
     private OngoingPowerMeasure<M> measure;
-    private StoppedPowerMeasure<M> lastMeasure;
-    private StoppedPowerMeasure<M> baseline;
+
+    private Consumer<PowerMeasure<M>> completed;
+    private BiConsumer<Integer, PowerMeasure<M>> sampled;
+    private Consumer<Exception> errorHandler;
 
     private final static PowerMeasurer<? extends SensorMeasure> instance = new PowerMeasurer<>(
             PowerSensorProducer.determinePowerSensor());
@@ -46,103 +48,83 @@ public double cpuShareOfJVMProcess() {
         return (processCpuLoad < 0 || cpuLoad <= 0) ? 0 : processCpuLoad / cpuLoad;
     }
 
-    PowerSensor<M> sensor() {
-        return sensor;
+    public void onCompleted(Consumer<PowerMeasure<M>> completed) {
+        this.completed = completed;
     }
 
-    public boolean isRunning() {
-        return measure != null;
+    public void onSampled(BiConsumer<Integer, PowerMeasure<M>> sampled) {
+        this.sampled = sampled;
     }
 
-    public void start(long durationInSeconds, long frequencyInMilliseconds, PowerSensor.Writer out) throws Exception {
-        start(durationInSeconds, frequencyInMilliseconds, false, out);
+    public void onError(Consumer<Exception> errorHandler) {
+        this.errorHandler = errorHandler != null ? errorHandler : (exception) -> {
+            throw new RuntimeException(exception);
+        };
     }
 
-    void start(long durationInSeconds, long frequencyInMilliseconds, boolean skipBaseline, PowerSensor.Writer out)
-            throws Exception {
-        if (!isRunning()) {
-            if (!skipBaseline && baseline == null) {
-                out.println("Establishing baseline for 30s, please do not use your application until done.");
-                out.println("Power measurement will start as configured after this initial measure is done.");
-                doStart(30, 1000, out, () -> baselineDone(durationInSeconds, frequencyInMilliseconds, out));
-            } else {
-                doStart(durationInSeconds, frequencyInMilliseconds, out, () -> stop(out));
-            }
-
-        }
+    public Optional<String> additionalSensorInfo() {
+        return Optional.ofNullable(measure).flatMap(sensor::additionalInfo);
     }
 
-    private void doStart(long duration, long frequency, PowerSensor.Writer out, Runnable doneAction) throws Exception {
-        measure = sensor.start(duration, frequency, out);
-
-        if (duration > 0) {
-            executor.schedule(doneAction, duration, TimeUnit.SECONDS);
-        }
-
-        scheduled = executor.scheduleAtFixedRate(() -> update(out),
-                0, frequency,
-                TimeUnit.MILLISECONDS);
+    public boolean isRunning() {
+        return measure != null;
     }
 
-    private void update(PowerSensor.Writer out) {
-        sensor.update(measure, out);
-        measure.incrementSamples();
-    }
+    public void start(long durationInSeconds, long frequencyInMilliseconds)
+            throws Exception {
+        try {
+            measure = sensor.start(durationInSeconds, frequencyInMilliseconds);
+
+            if (durationInSeconds > 0) {
+                executor.schedule(this::stop, durationInSeconds, TimeUnit.SECONDS);
+            }
 
-    public void stop(PowerSensor.Writer out) {
-        if (isRunning()) {
-            sensor.stop();
-            scheduled.cancel(true);
-            outputConsumptionSinceStarted(out, false);
-            lastMeasure = new StoppedPowerMeasure<>(measure);
-            measure = null;
+            scheduled = executor.scheduleAtFixedRate(this::update, 0, frequencyInMilliseconds, TimeUnit.MILLISECONDS);
+        } catch (Exception e) {
+            handleError(e);
         }
     }
 
-    private void baselineDone(long durationInSeconds, long frequencyInMilliseconds, PowerSensor.Writer out) {
-        if (isRunning()) {
-            sensor.stop();
-            scheduled.cancel(true);
-            outputConsumptionSinceStarted(out, true);
-            baseline = new StoppedPowerMeasure<>(measure);
-            out.println("Baseline established! You can now interact with your application normally.");
-            measure = null;
-            try {
-                doStart(durationInSeconds, frequencyInMilliseconds, out, () -> stop(out));
-            } catch (Exception e) {
-                throw new RuntimeException(e);
+    private void update() {
+        try {
+            sensor.update(measure);
+            measure.incrementSamples();
+            if (this.sampled != null) {
+                sampled.accept(measure.numberOfSamples(), measure);
             }
+        } catch (Exception e) {
+            handleError(e);
         }
     }
 
-    public PowerMeasure<M> current() {
-        // use the ongoing power measure if it exists
-        if (measure == null) {
-            // or use the last recorded measure if we have one
-            if (lastMeasure != null) {
-                return lastMeasure;
-            } else {
-                throw new IllegalStateException("No power measure found. Please start it first.");
+    private void handleError(Exception e) {
+        errorHandler.accept(e);
+        try {
+            if (scheduled != null) {
+                scheduled.cancel(true);
+            }
+            if (sensor != null) {
+                sensor.stop();
             }
-        } else {
-            return measure;
+        } catch (Exception ex) {
+            // ignore shutting down exceptions
         }
     }
 
-    private void outputConsumptionSinceStarted(PowerSensor.Writer out, boolean isBaseline) {
-        out = out == null ? System.out::println : out;
-        final var durationInSeconds = measure.duration() / 1000;
-        final var title = isBaseline ? "Baseline power: " : "Measured power: ";
-        out.println(title + getReadablePower(measure) + " over " + durationInSeconds
-                + " seconds (" + measure.numberOfSamples() + " samples)");
-        if (!isBaseline) {
-            sensor.additionalInfo(measure, out);
-            out.println("Baseline power was " + getReadablePower(baseline));
+    public void stop() {
+        try {
+            if (isRunning()) {
+                sensor.stop();
+                scheduled.cancel(true);
+                // record the result
+                final var measured = new StoppedPowerMeasure<>(measure);
+                // then set the measure to null to mark that we're ready for a new measure
+                measure = null;
+                // and finally, but only then, run the completion handler
+                completed.accept(measured);
+            }
+        } catch (Exception e) {
+            handleError(e);
         }
     }
-
-    private static String getReadablePower(PowerMeasure<?> measure) {
-        final var measuredMilliWatts = measure.total();
-        return measuredMilliWatts >= 1000 ? (measuredMilliWatts / 1000) + " W" : measuredMilliWatts + "mW";
-    }
 }

runtime/src/main/java/io/quarkiverse/power/runtime/sensors/PowerSensor.java

@@ -1,20 +1,19 @@
 package io.quarkiverse.power.runtime.sensors;
 
+import java.util.Optional;
+
 import io.quarkiverse.power.runtime.PowerMeasure;
 
 public interface PowerSensor<T extends IncrementableMeasure> {
 
-    OngoingPowerMeasure<T> start(long duration, long frequency, Writer out) throws Exception;
+    OngoingPowerMeasure<T> start(long duration, long frequency) throws Exception;
 
     default void stop() {
     }
 
-    void update(OngoingPowerMeasure<T> ongoingMeasure, Writer out);
-
-    default void additionalInfo(PowerMeasure<T> measure, Writer out) {
-    }
+    void update(OngoingPowerMeasure<T> ongoingMeasure);
 
-    interface Writer {
-        void println(String message);
+    default Optional<String> additionalInfo(PowerMeasure<T> measure) {
+        return Optional.empty();
     }
 }

runtime/src/main/java/io/quarkiverse/power/runtime/sensors/linux/rapl/IntelRAPLSensor.java

@@ -26,23 +26,6 @@ static RAPLFile createFrom(Path file) {
         }
     }
 
-    private static class NaiveRAPLFile implements RAPLFile {
-        private final Path path;
-
-        private NaiveRAPLFile(Path path) {
-            this.path = path;
-        }
-
-        @Override
-        public long extractPowerMeasure() {
-            try {
-                return Long.parseLong(Files.readString(path).trim());
-            } catch (IOException e) {
-                throw new RuntimeException(e);
-            }
-        }
-    }
-
     static class ByteBufferRAPLFile implements RAPLFile {
         private static final int CAPACITY = 64;
         private final ByteBuffer buffer;
@@ -124,7 +107,7 @@ private static boolean isReadable(Path file) {
     }
 
     @Override
-    public OngoingPowerMeasure<IntelRAPLMeasure> start(long duration, long frequency, Writer out) throws Exception {
+    public OngoingPowerMeasure<IntelRAPLMeasure> start(long duration, long frequency) throws Exception {
         this.frequency = frequency;
         IntelRAPLMeasure measure = new IntelRAPLMeasure();
         update(measure);
@@ -137,7 +120,7 @@ private void update(IntelRAPLMeasure measure) {
     }
 
     @Override
-    public void update(OngoingPowerMeasure<IntelRAPLMeasure> ongoingMeasure, Writer out) {
+    public void update(OngoingPowerMeasure<IntelRAPLMeasure> ongoingMeasure) {
         update(ongoingMeasure.sensorMeasure());
     }
 }

runtime/src/main/java/io/quarkiverse/power/runtime/sensors/macos/jmx/JMXCPUSensor.java

@@ -14,14 +14,14 @@ public class JMXCPUSensor implements PowerSensor<AppleSiliconMeasure> {
     private Process powermetrics;
 
     @Override
-    public OngoingPowerMeasure<AppleSiliconMeasure> start(long duration, long frequency, Writer out)
+    public OngoingPowerMeasure<AppleSiliconMeasure> start(long duration, long frequency)
             throws Exception {
         final var freq = Long.toString(Math.round(frequency));
         powermetrics = Runtime.getRuntime().exec("sudo powermetrics --samplers cpu_power -i " + freq);
         return new OngoingPowerMeasure<>(new AppleSiliconMeasure());
     }
 
-    public void update(OngoingPowerMeasure<AppleSiliconMeasure> ongoingMeasure, Writer out) {
+    public void update(OngoingPowerMeasure<AppleSiliconMeasure> ongoingMeasure) {
         try {
             // Should not be closed since it closes the process
             BufferedReader input = new BufferedReader(new InputStreamReader(powermetrics.getInputStream()));