Added energy consumption to Skyline

.gitignore

@@ -316,3 +316,5 @@ cython_debug/
 # Support for Project snippet scope
 
 # End of https://www.toptal.com/developers/gitignore/api/python,osx,pycharm,visualstudiocode
+
+*.sqlite3

protocol/innpv.proto

@@ -77,6 +77,7 @@ message FromServer {
     BreakdownResponse breakdown = 8;
 
     HabitatResponse habitat = 9;
+    EnergyResponse energy = 10;
   }
 
   // Deprecated messages
@@ -97,6 +98,33 @@ message HabitatResponse {
   repeated HabitatDevicePrediction predictions = 1;
 }
 
+// Energy messages
+// =======================================
+
+message EnergyResponse {
+  float total_consumption = 1;
+  repeated EnergyConsumptionComponent components = 2;
+
+  // A list of past energy measurements
+  repeated EnergyResponse past_measurements = 3;
+}
+
+// Reports the energy consumption of one system component (e.g. CPU+DRAM or GPU)
+enum EnergyConsumptionComponentType {
+  ENERGY_UNSPECIFIED = 0;
+  ENERGY_CPU_DRAM = 1;
+  ENERGY_NVIDIA = 2;
+}
+
+message EnergyConsumptionComponent {
+  EnergyConsumptionComponentType component_type = 1;
+  float consumption_joules = 2;
+}
+
+// Records past experiments
+
+
+
 // =======================================
 
 message InitializeResponse {

pyproject.toml

@@ -32,6 +32,7 @@ numpy = "^1.22"
 torch = "*"
 nvidia-ml-py3 = "*"
 toml = "^0.10.2"
+pyRAPL = "^0.2.3"
 
 [tool.poetry.dev-dependencies]
 

skyline/analysis/request_manager.py

@@ -59,7 +59,7 @@ def _handle_analysis_request(self, analysis_request, context):
 
             # Abort early if the connection has been closed
             if not context.state.connected:
-                logger.debug(
+                logger.error(
                     'Aborting request %d from (%s:%d) early '
                     'because the client has disconnected.',
                     context.sequence_number,
@@ -75,7 +75,7 @@ def _handle_analysis_request(self, analysis_request, context):
             )
 
             if not context.state.connected:
-                logger.debug(
+                logger.error(
                     'Aborting request %d from (%s:%d) early '
                     'because the client has disconnected.',
                     context.sequence_number,
@@ -92,7 +92,7 @@ def _handle_analysis_request(self, analysis_request, context):
 
             # send habitat response
             if not context.state.connected:
-                logger.debug(
+                logger.error(
                     'Aborting request %d from (%s:%d) early '
                     'because the client has disconnected.',
                     context.sequence_number,
@@ -107,6 +107,23 @@ def _handle_analysis_request(self, analysis_request, context):
                 context,
             )
 
+            # send energy response
+            if not context.state.connected:
+                logger.error(
+                    'Aborting request %d from (%s:%d) early '
+                    'because the client has disconnected.',
+                    context.sequence_number,
+                    *(context.address),
+                )
+                return
+
+            energy_resp = next(analyzer)
+            self._enqueue_response(
+                self._send_energy_response,
+                energy_resp,
+                context,
+            )
+
             elapsed_time = time.perf_counter() - start_time
             logger.debug(
                 'Processed analysis request %d from (%s:%d) in %.4f seconds.',
@@ -174,3 +191,11 @@ def _send_habitat_response(self, habitat_resp, context):
         except:
             logger.exception(
                 'Exception occurred when sending a habitat response.')
+
+    def _send_energy_response(self, energy_resp, context):
+        # Called from the main executor. Do not call directly!
+        try:
+            self._message_sender.send_energy_response(energy_resp, context)
+        except:
+            logger.exception(
+                'Exception occurred when sending an energy response.')

skyline/analysis/runner.py

@@ -14,6 +14,9 @@ def analyze_project(project_root, entry_point, nvml):
     print("analyze_project: running habitat_predict()")
     yield session.habitat_predict()
 
+    print("analyze_project: running energy_compute()")
+    yield session.energy_compute()
+
 
 def main():
     # This is used for development and debugging purposes

skyline/analysis/session.py

@@ -11,6 +11,8 @@
 
 import skyline.protocol_gen.innpv_pb2 as pm
 from skyline.analysis.static import StaticAnalyzer
+from skyline.db.database import DatabaseInterface, EnergyTableInterface
+from skyline.energy.measurer import EnergyMeasurer
 from skyline.exceptions import AnalysisError, exceptions_as_analysis_errors
 from skyline.profiler.iteration import IterationProfiler
 from skyline.tracking.tracker import Tracker
@@ -67,6 +69,7 @@ def __init__(
         self._memory_usage_percentage = None
         self._batch_size_iteration_run_time_ms = None
         self._batch_size_peak_usage_bytes = None
+        self._energy_table_interface = EnergyTableInterface(DatabaseInterface().connection)
 
     @classmethod
     def new_from(cls, project_root, entry_point):
@@ -131,6 +134,47 @@ def new_from(cls, project_root, entry_point):
             StaticAnalyzer(entry_point_code, entry_point_ast),
         )
 
+    def energy_compute(self) -> pm.EnergyResponse:
+        energy_measurer = EnergyMeasurer()
+
+        model = self._model_provider()
+        inputs = self._input_provider()
+        iteration = self._iteration_provider(model)
+        resp = pm.EnergyResponse()
+
+        try:
+            energy_measurer.begin_measurement()
+            iterations = 20
+            for _ in range(iterations):
+                iteration(*inputs)
+            energy_measurer.end_measurement()
+
+            resp.total_consumption = energy_measurer.total_energy()/float(iterations)
+
+            cpu_component = pm.EnergyConsumptionComponent()
+            cpu_component.component_type = pm.ENERGY_CPU_DRAM
+            cpu_component.consumption_joules = energy_measurer.cpu_energy()/float(iterations)
+
+            gpu_component = pm.EnergyConsumptionComponent()
+            gpu_component.component_type = pm.ENERGY_NVIDIA
+            gpu_component.consumption_joules = energy_measurer.gpu_energy()/float(iterations)
+
+            resp.components.extend([cpu_component, gpu_component])
+
+            # get last 10 runs if they exist
+            path_to_entry_point = os.path.join(self._project_root, self._entry_point)
+            past_runs = self._energy_table_interface.get_latest_n_entries_of_entry_point(10, path_to_entry_point)
+            print(past_runs)
+            resp.past_measurements.extend(_convert_to_energy_responses(past_runs))
+
+            # add current run to database
+            self._energy_table_interface.add_entry([path_to_entry_point, cpu_component.consumption_joules, gpu_component.consumption_joules])
+
+        except PermissionError as err:
+            # Remind user to set their CPU permissions
+            print(err)
+        return resp
+
     def habitat_compute_threshold(self, runnable, context):
         tracker = habitat.OperationTracker(context.origin_device)
         with tracker.track():
@@ -516,3 +560,21 @@ def _fit_linear_model(x, y):
     slope, bias = np.linalg.lstsq(stacked, y_np, rcond=None)[0]
     # Linear model: y = slope * x + bias
     return slope, bias
+
+def _convert_to_energy_responses(entries: list)-> list[pm.EnergyResponse]:
+    energy_response_list = []
+    for entry in entries:
+        if EnergyTableInterface.is_valid_entry_with_timestamp(entry):
+            energy_response = pm.EnergyResponse()
+            cpu_component = pm.EnergyConsumptionComponent()
+            cpu_component.component_type = pm.ENERGY_CPU_DRAM
+            cpu_component.consumption_joules = entry[1]
+
+            gpu_component = pm.EnergyConsumptionComponent()
+            gpu_component.component_type = pm.ENERGY_NVIDIA
+            gpu_component.consumption_joules = entry[2]
+
+            energy_response.total_consumption = gpu_component.consumption_joules+cpu_component.consumption_joules 
+            energy_response.components.extend([cpu_component, gpu_component])
+            energy_response_list.append(energy_response)
+    return energy_response_list

skyline/db/database.py

@@ -0,0 +1,67 @@
+import datetime 
+import sqlite3
+
+class DatabaseInterface:
+    def __init__(self, database_name="skyline.sqlite3") -> None:
+        self.connection = sqlite3.connect(database_name, detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
+        self.create_energy_table()
+
+    def create_energy_table(self) -> None:
+        self.connection.cursor().execute("CREATE TABLE IF NOT EXISTS ENERGY ( \
+            entry_point TEXT, \
+            cpu_component REAL, \
+            gpu_component REAL, \
+            ts TIMESTAMP \
+        );")
+
+
+class EnergyTableInterface:
+    def __init__(self, database_connection: sqlite3.Connection):
+        self.database_connection: sqlite3.Connection = database_connection
+
+    @staticmethod
+    def is_valid_entry(entry: list) -> bool:
+        '''
+        Validates an entry in the Energy table by testing if the length is 3,
+        and the types match the columns. Note that timestamp is not part of the entry.
+        Returns True if it is valid, else False
+        '''
+        return len(entry) == 3 and type(entry[0]) == str and type(entry[1]) == float \
+            and type(entry[2]) == float
+
+    @staticmethod
+    def is_valid_entry_with_timestamp(entry: list) -> bool:
+        '''
+        Validates an entry in the Energy table by testing if the length is 4,
+        and the types match the columns. Returns True if it is valid, else False
+        '''
+        return len(entry) == 4 and type(entry[0]) == str and type(entry[1]) == float \
+            and type(entry[2]) == float and type(entry[3]) == datetime.datetime
+
+    def add_entry(self, entry: list) -> bool:
+        '''
+        Validates an entry and then adds that entry into the Energy table. Note that current timestamp is added 
+        by this function. Returns False if the entry is not a valid format, or if the insertion failed. Else
+        returns True
+        '''
+        if self.is_valid_entry(entry):
+            try:
+                entry.append(datetime.datetime.now())
+                cursor = self.database_connection.cursor()
+                cursor.execute("INSERT INTO ENERGY VALUES(?, ?, ?, ?)", entry)
+                self.database_connection.commit()
+                return True
+            except sqlite3.IntegrityError as e:
+                print(e)
+                return False
+        else:
+            return True
+
+    def get_latest_n_entries_of_entry_point(self, n: int, entry_point: str) -> list:
+        '''
+        Gets the n latest entries of a given entry point
+        '''
+        params = [entry_point, n]
+        cursor = self.database_connection.cursor()
+        results = cursor.execute("SELECT * FROM ENERGY WHERE entry_point=? ORDER BY ts DESC LIMIT ?;", params).fetchall()
+        return results