Plugin Framework and PowerMetrics Integration

.gitignore

@@ -1,3 +1,8 @@
+.vagrant
+macos-sonoma.conf
+macos-sonoma/
+Vagrantfile
+Session.vim
 .idea
 .venv
 */__pycache__/*
@@ -13,3 +18,5 @@ tmp/
 _trial_temp/
 pycharm-interpreter.sh
 python
+
+scratch.py

examples/linux-powerjoular-profiling/README.md

@@ -20,9 +20,10 @@ pip install -r requirements.txt
 ## Running
 
 From the root directory of the repo, run the following command:
+NOTE: This program must be run as root, as powerjoular requires this for its use of Intel RAPL.
 
 ```bash
-python experiment-runner/ examples/linux-powerjoular-profiling/RunnerConfig.py
+sudo python experiment-runner/ examples/linux-powerjoular-profiling/RunnerConfig.py
 ```
 
 ## Results

examples/linux-powerjoular-profiling/RunnerConfig.py

@@ -6,16 +6,15 @@
 from ConfigValidator.Config.Models.OperationType import OperationType
 from ProgressManager.Output.OutputProcedure import OutputProcedure as output
 
+from Plugins.Profilers.PowerJoular import PowerJoular
+
 from typing import Dict, List, Any, Optional
 from pathlib import Path
 from os.path import dirname, realpath
 
-import os
-import signal
-import pandas as pd
 import time
 import subprocess
-import shlex
+import numpy as np
 
 class RunnerConfig:
     ROOT_DIR = Path(dirname(realpath(__file__)))
@@ -88,16 +87,18 @@ def start_run(self, context: RunnerContext) -> None:
         )
 
         # Configure the environment based on the current variation
-        subprocess.check_call(shlex.split(f'cpulimit -b -p {self.target.pid} --limit {cpu_limit}'))
+        subprocess.check_call(f'cpulimit -p {self.target.pid} --limit {cpu_limit} &', shell=True)
 
+        time.sleep(1) # allow the process to run a little before measuring
 
     def start_measurement(self, context: RunnerContext) -> None:
         """Perform any activity required for starting measurements."""
-
-        profiler_cmd = f'powerjoular -l -p {self.target.pid} -f {context.run_dir / "powerjoular.csv"}'
-
-        time.sleep(1) # allow the process to run a little before measuring
-        self.profiler = subprocess.Popen(shlex.split(profiler_cmd))
+
+        # Set up the powerjoular object, provide an (optional) target and output file name
+        self.meter = PowerJoular(target_pid=self.target.pid, 
+                                 out_file=context.run_dir / "powerjoular.csv")
+        # Start measuring with powerjoular
+        self.meter.start()
 
     def interact(self, context: RunnerContext) -> None:
         """Perform any interaction with the running target system here, or block here until the target finishes."""
@@ -109,30 +110,32 @@ def interact(self, context: RunnerContext) -> None:
 
     def stop_measurement(self, context: RunnerContext) -> None:
         """Perform any activity here required for stopping measurements."""
-
-        os.kill(self.profiler.pid, signal.SIGINT) # graceful shutdown of powerjoular
-        self.profiler.wait()
+        
+        # Stop the measurements
+        stdout = self.meter.stop()
 
     def stop_run(self, context: RunnerContext) -> None:
         """Perform any activity here required for stopping the run.
         Activities after stopping the run should also be performed here."""
-        
+
         self.target.kill()
         self.target.wait()
 
     def populate_run_data(self, context: RunnerContext) -> Optional[Dict[str, Any]]:
         """Parse and process any measurement data here.
         You can also store the raw measurement data under `context.run_dir`
         Returns a dictionary with keys `self.run_table_model.data_columns` and their values populated"""
-
-        # powerjoular.csv - Power consumption of the whole system
-        # powerjoular.csv-PID.csv - Power consumption of that specific process
-        df = pd.read_csv(context.run_dir / f"powerjoular.csv-{self.target.pid}.csv")
-        run_data = {
-            'avg_cpu': round(df['CPU Utilization'].sum(), 3),
-            'total_energy': round(df['CPU Power'].sum(), 3),
+
+        out_file = context.run_dir / "powerjoular.csv"
+
+        results_global = self.meter.parse_log(out_file)
+        # If you specified a target_pid or used the -p paramter 
+        # a second csv for that target will be generated
+        # results_process = self.meter.parse_log(self.meter.target_logfile)
+        return {
+            'avg_cpu': round(np.mean(list(results_global['CPU Utilization'].values())), 3),
+            'total_energy': round(sum(list(results_global['CPU Power'].values())), 3),
         }
-        return run_data
 
     def after_experiment(self) -> None:
         """Perform any activity required after stopping the experiment here

examples/linux-ps-profiling/RunnerConfig.py

@@ -5,12 +5,13 @@
 from ConfigValidator.Config.Models.RunnerContext import RunnerContext
 from ConfigValidator.Config.Models.OperationType import OperationType
 from ProgressManager.Output.OutputProcedure import OutputProcedure as output
+from Plugins.Profilers.Ps import Ps
 
 from typing import Dict, List, Any, Optional
 from pathlib import Path
 from os.path import dirname, realpath
 
-import pandas as pd
+import numpy as np
 import time
 import subprocess
 import shlex
@@ -64,14 +65,16 @@ def create_run_table_model(self) -> RunTableModel:
             exclude_variations = [
                 {cpu_limit_factor: [70], pin_core_factor: [False]} # all runs having the combination <'70', 'False'> will be excluded
             ],
-            data_columns=['avg_cpu']
+            data_columns=["avg_cpu", "avg_mem"]
         )
         return self.run_table_model
 
     def before_experiment(self) -> None:
         """Perform any activity required before starting the experiment here
         Invoked only once during the lifetime of the program."""
-        subprocess.check_call(['make'], cwd=self.ROOT_DIR) # compile
+
+        # compile the target program
+        subprocess.check_call(['make'], cwd=self.ROOT_DIR)
 
     def before_run(self) -> None:
         """Perform any activity required before starting a run.
@@ -93,27 +96,21 @@ def start_run(self, context: RunnerContext) -> None:
 
         # Configure the environment based on the current variation
         if pin_core:
-            subprocess.check_call(shlex.split(f'taskset -cp 0  {self.target.pid}'))
-        subprocess.check_call(shlex.split(f'cpulimit -b -p {self.target.pid} --limit {cpu_limit}'))
+            subprocess.check_call(shlex.split(f'taskset -cp 0 {self.target.pid}'))
+
+        # Limit the targets cputime
+        subprocess.check_call(f'cpulimit --limit={cpu_limit} -p {self.target.pid} &', shell=True)
 
+        time.sleep(1) # allow the process to run a little before measuring
 
     def start_measurement(self, context: RunnerContext) -> None:
         """Perform any activity required for starting measurements."""
-
-        # man 1 ps
-        # %cpu:
-        #   cpu utilization of the process in "##.#" format.  Currently, it is the CPU time used
-        #   divided by the time the process has been running (cputime/realtime ratio), expressed
-        #   as a percentage.  It will not add up to 100% unless you are lucky.  (alias pcpu).
-        profiler_cmd = f'ps -p {self.target.pid} --noheader -o %cpu'
-        wrapper_script = f'''
-        while true; do {profiler_cmd}; sleep 1; done
-        '''
-
-        time.sleep(1) # allow the process to run a little before measuring
-        self.profiler = subprocess.Popen(['sh', '-c', wrapper_script],
-            stdout=subprocess.PIPE, stderr=subprocess.PIPE
-        )
+
+        # Set up the ps object, provide an (optional) target and output file name
+        self.meter = Ps(out_file=context.run_dir / "ps.csv",
+                        target_pid=[self.target.pid])
+        # Start measuring with ps
+        self.meter.start()
 
     def interact(self, context: RunnerContext) -> None:
         """Perform any interaction with the running target system here, or block here until the target finishes."""
@@ -126,8 +123,8 @@ def interact(self, context: RunnerContext) -> None:
     def stop_measurement(self, context: RunnerContext) -> None:
         """Perform any activity here required for stopping measurements."""
 
-        self.profiler.kill()
-        self.profiler.wait()
+        # Stop the measurements
+        stdout = self.meter.stop()
 
     def stop_run(self, context: RunnerContext) -> None:
         """Perform any activity here required for stopping the run.
@@ -141,17 +138,13 @@ def populate_run_data(self, context: RunnerContext) -> Optional[Dict[str, Any]]:
         You can also store the raw measurement data under `context.run_dir`
         Returns a dictionary with keys `self.run_table_model.data_columns` and their values populated"""
 
-        df = pd.DataFrame(columns=['cpu_usage'])
-        for i, l in enumerate(self.profiler.stdout.readlines()):
-            cpu_usage=float(l.decode('ascii').strip())
-            df.loc[i] = [cpu_usage]
-
-        df.to_csv(context.run_dir / 'raw_data.csv', index=False)
+        results = self.meter.parse_log(context.run_dir / "ps.csv", 
+                                       column_names=["cpu_usage", "memory_usage"])
 
-        run_data = {
-            'avg_cpu': round(df['cpu_usage'].mean(), 3)
+        return {
+            "avg_cpu": round(np.mean(list(results['cpu_usage'].values())), 3),
+            "avg_mem": round(np.mean(list(results['memory_usage'].values())), 3)
         }
-        return run_data
 
     def after_experiment(self) -> None:
         """Perform any activity required after stopping the experiment here

examples/powermetrics-profiling/README.md

@@ -0,0 +1,25 @@
+
+# `powermetrics` profiler
+
+A simple example using the OS X [powermetrics](https://developer.apple.com/library/archive/documentation/Performance/Conceptual/power_efficiency_guidelines_osx/PrioritizeWorkAtTheTaskLevel.html#//apple_ref/doc/uid/TP40013929-CH35-SW10) cli tool to measure the ambient power consumtption of the system.
+
+## Requirements
+
+Install the requirements to run:
+
+```bash
+pip install -r requirements.txt
+```
+
+## Running
+
+From the root directory of the repo, run the following command:
+NOTE: This program must be run as root, as powermetrics requires this
+
+```bash
+sudo python experiment-runner/ examples/powermetrics-profiling/RunnerConfig.py
+```
+
+## Results
+
+The results are generated in the `examples/powermetrics-profiling/experiments` folder.

examples/powermetrics-profiling/RunnerConfig.py

@@ -0,0 +1,128 @@
+from EventManager.Models.RunnerEvents import RunnerEvents
+from EventManager.EventSubscriptionController import EventSubscriptionController
+from ConfigValidator.Config.Models.RunTableModel import RunTableModel
+from ConfigValidator.Config.Models.FactorModel import FactorModel
+from ConfigValidator.Config.Models.RunnerContext import RunnerContext
+from ConfigValidator.Config.Models.OperationType import OperationType
+from ProgressManager.Output.OutputProcedure import OutputProcedure as output
+from Plugins.Profilers.PowerMetrics import PowerMetrics
+
+from typing import Dict, List, Any, Optional
+from pathlib import Path
+from os.path import dirname, realpath
+import time
+import numpy as np
+
+class RunnerConfig:
+    ROOT_DIR = Path(dirname(realpath(__file__)))
+
+    # ================================ USER SPECIFIC CONFIG ================================
+    """The name of the experiment."""
+    name:                       str             = "new_runner_experiment"
+
+    """The path in which Experiment Runner will create a folder with the name `self.name`, in order to store the
+    results from this experiment. (Path does not need to exist - it will be created if necessary.)
+    Output path defaults to the config file's path, inside the folder 'experiments'"""
+    results_output_path:        Path             = ROOT_DIR / 'experiments'
+
+    """Experiment operation type. Unless you manually want to initiate each run, use `OperationType.AUTO`."""
+    operation_type:             OperationType   = OperationType.AUTO
+
+    """The time Experiment Runner will wait after a run completes.
+    This can be essential to accommodate for cooldown periods on some systems."""
+    time_between_runs_in_ms:    int             = 1000
+
+    # Dynamic configurations can be one-time satisfied here before the program takes the config as-is
+    # e.g. Setting some variable based on some criteria
+    def __init__(self):
+        """Executes immediately after program start, on config load"""
+
+        EventSubscriptionController.subscribe_to_multiple_events([
+            (RunnerEvents.BEFORE_EXPERIMENT, self.before_experiment),
+            (RunnerEvents.BEFORE_RUN       , self.before_run       ),
+            (RunnerEvents.START_RUN        , self.start_run        ),
+            (RunnerEvents.START_MEASUREMENT, self.start_measurement),
+            (RunnerEvents.INTERACT         , self.interact         ),
+            (RunnerEvents.STOP_MEASUREMENT , self.stop_measurement ),
+            (RunnerEvents.STOP_RUN         , self.stop_run         ),
+            (RunnerEvents.POPULATE_RUN_DATA, self.populate_run_data),
+            (RunnerEvents.AFTER_EXPERIMENT , self.after_experiment )
+        ])
+        self.run_table_model = None  # Initialized later
+        output.console_log("Custom config loaded")
+
+    def create_run_table_model(self) -> RunTableModel:
+        """Create and return the run_table model here. A run_table is a List (rows) of tuples (columns),
+        representing each run performed"""
+
+        # Create the experiment run table with factors, and desired data columns
+        factor1 = FactorModel("test_factor", [1, 2])
+        self.run_table_model = RunTableModel(
+            factors = [factor1],
+            data_columns=["joules", "avg_cpu", "avg_gpu"])
+
+        return self.run_table_model
+
+    def before_experiment(self) -> None:
+        """Perform any activity required before starting the experiment here
+        Invoked only once during the lifetime of the program."""
+        pass
+
+    def before_run(self) -> None:
+        """Perform any activity required before starting a run.
+        No context is available here as the run is not yet active (BEFORE RUN)"""
+        pass
+
+    def start_run(self, context: RunnerContext) -> None:
+        """Perform any activity required for starting the run here.
+        For example, starting the target system to measure.
+        Activities after starting the run should also be performed here."""
+        pass
+
+    def start_measurement(self, context: RunnerContext) -> None:
+        """Perform any activity required for starting measurements."""
+
+        # Create the powermetrics object we will use to collect data
+        self.meter = PowerMetrics(out_file=context.run_dir / "powermetrics.plist")
+        # Start measuring useing powermetrics
+        self.meter.start()
+
+    def interact(self, context: RunnerContext) -> None:
+        """Perform any interaction with the running target system here, or block here until the target finishes."""
+
+        # Wait (block) for a bit to collect some data
+        time.sleep(20)
+
+    def stop_measurement(self, context: RunnerContext) -> None:
+        """Perform any activity here required for stopping measurements."""
+
+        # Stop measuring at the end of a run
+        stdout = self.meter.stop()
+
+    def stop_run(self, context: RunnerContext) -> None:
+        """Perform any activity here required for stopping the run.
+        Activities after stopping the run should also be performed here."""
+        pass       
+
+    def populate_run_data(self, context: RunnerContext) -> Optional[Dict[str, Any]]:
+        """Parse and process any measurement data here.
+        You can also store the raw measurement data under `context.run_dir`
+        Returns a dictionary with keys `self.run_table_model.data_columns` and their values populated"""
+
+        # Retrieve data from run
+        run_results = self.meter.parse_log(context.run_dir / "powermetrics.plist")
+
+        # Parse it as required for your experiment and add it to the run table
+        return {
+                "joules": sum(map(lambda x: x["processor"]["package_joules"], run_results)),
+                "avg_cpu": np.mean(list(map(lambda x: x["processor"]["packages"][0]["cores_active_ratio"], run_results))),
+                "avg_gpu": np.mean(list(map(lambda x: x["processor"]["packages"][0]["gpu_active_ratio"], run_results))),
+        }
+
+    def after_experiment(self) -> None:
+        """Perform any activity required after stopping the experiment here
+        Invoked only once during the lifetime of the program."""
+        pass
+
+    # ================================ DO NOT ALTER BELOW THIS LINE ================================
+    experiment_path:            Path             = None