WIP: Service rewrite

analyze.py

@@ -335,7 +335,7 @@ def analyze_resource_utilization(
     # Plotting defaults.
     # hatches = ['//', '--', '**']
     # alphas = np.arange(0.2, 1.2, 0.2)
-    resource_color = {"GPU": "red", "CPU": "green"}
+    resource_color = {"Slot": "green"}
 
     # Worker Pool statistics
     worker_pool_stats = csv_reader.get_worker_pool_utilizations(scheduler_csv_file)
@@ -1246,16 +1246,16 @@ def log_aggregate_stats(
                 / sum(stat.resource_utilizations[resource])
                 for stat in worker_pool_stats
             ]
-            for resource in ("GPU", "CPU")
+            for resource in ("Slot",)
         }
 
         scheduler_invocations = csv_reader.get_scheduler_invocations(csv_file)
         placed_tasks = [
-            scheduler_invocation.placed_tasks
+            scheduler_invocation.num_placed_tasks
             for scheduler_invocation in scheduler_invocations
         ]
         unplaced_tasks = [
-            scheduler_invocation.unplaced_tasks
+            scheduler_invocation.num_unplaced_tasks
             for scheduler_invocation in scheduler_invocations
         ]
 
@@ -1268,8 +1268,7 @@ def log_aggregate_stats(
                 placement_delay,
                 deadline_delay,
                 stat_function(e2e_response_time),
-                stat_function(resource_uses["GPU"]),
-                stat_function(resource_uses["CPU"]),
+                stat_function(resource_uses["Slot"]),
                 stat_function(placed_tasks),
                 stat_function(unplaced_tasks),
                 log_name,
@@ -1288,8 +1287,7 @@ def log_aggregate_stats(
                 "Placement",
                 "Deadline",
                 "JCT",
-                "GPU",
-                "CPU",
+                "Slot",
                 "Placed",
                 "Unplaced",
                 "Log",

configs/tpch_replay_dsched.conf

@@ -0,0 +1,38 @@
+# Output configs.
+--log=./tpch_replay_dsched.log
+--log_level=debug
+--csv=./tpch_replay_dsched.csv
+
+# Task configs.
+--runtime_variance=0
+
+# Scheduler configs.
+
+# DSched
+--scheduler=TetriSched
+--scheduler_runtime=0
+--enforce_deadlines
+--retract_schedules
+--release_taskgraphs
+--drop_skipped_tasks
+--scheduler_time_discretization=1
+
+# Deadline variance
+--min_deadline_variance=10
+--max_deadline_variance=25
+
+# Execution mode configs.
+--execution_mode=replay
+--replay_trace=tpch
+
+# Release time config.
+--override_release_policy=gamma
+--override_gamma_coefficient=1
+--override_poisson_arrival_rate=1
+--override_num_invocation=10
+
+# TPCH flags
+--random_seed=1234
+--tpch_query_dag_spec=profiles/workload/tpch/queries.yaml
+--tpch_dataset_size=50
+--worker_profile_path=profiles/workers/tpch_cluster.yaml

configs/tpch_replay_edf.conf

@@ -0,0 +1,47 @@
+# Output configs.
+# --log=./tpch_replay_dsched.log
+# --log_level=debug
+# --csv=./tpch_replay_dsched.csv
+
+--log=./tpch_replay_edf.log
+--log_level=debug
+--csv=./tpch_replay_edf.csv
+
+# Task configs.
+--runtime_variance=0
+
+# Scheduler configs.
+
+# EDF
+--scheduler=EDF
+--scheduler_runtime=0
+--enforce_deadlines
+
+# DSched
+# --scheduler=TetriSched
+# --scheduler_runtime=0
+# --enforce_deadlines
+# --retract_schedules
+# --release_taskgraphs
+# --drop_skipped_tasks
+# --scheduler_time_discretization=1
+
+# Deadline variance
+--min_deadline_variance=10
+--max_deadline_variance=25
+
+# Execution mode configs.
+--execution_mode=replay
+--replay_trace=tpch
+
+# Release time config.
+--override_release_policy=gamma
+--override_gamma_coefficient=1
+--override_poisson_arrival_rate=1
+--override_num_invocation=10
+
+# TPCH flags
+--random_seed=1234
+--tpch_query_dag_spec=profiles/workload/tpch/queries.yaml
+--tpch_dataset_size=50
+--worker_profile_path=profiles/workers/tpch_cluster.yaml

data/__init__.py

@@ -7,6 +7,7 @@
 from .task_loader_benchmark import TaskLoaderBenchmark
 from .task_loader_pylot import TaskLoaderPylot
 from .task_loader_synthetic import TaskLoaderSynthetic
+from .tpch_loader import TpchWorkloadLoader
 from .worker_loader import WorkerLoader
 from .worker_loader_benchmark import WorkerLoaderBenchmark
 from .workload_loader import WorkloadLoader

data/csv_reader.py

@@ -63,6 +63,11 @@ def parse_events(self, readings: Mapping[str, Sequence[str]]):
                         )
                     elif reading[1] == "UPDATE_WORKLOAD":
                         simulator.total_tasks += int(reading[2])
+                    elif reading[1] == "LOG_STATS":
+                        assert (
+                            simulator is not None
+                        ), "No SIMULATOR_START found for a corresponding SIMULATOR_END."
+                        simulator.update_stats(reading)
                     elif reading[1] == "SIMULATOR_END":
                         assert (
                             simulator is not None

data/csv_types.py

@@ -385,6 +385,18 @@ def __init__(self, csv_path: str, start_time: int, total_tasks: int = 0):
         self.scheduler_invocations: list[Scheduler] = []
         self.task_graphs: dict[str, TaskGraph] = {}
 
+    def update_stats(self, csv_reading: str):
+        assert (
+            csv_reading[1] == "LOG_STATS"
+        ), f"The event {csv_reading[1]} was not of type LOG_STATS."
+        self.finished_tasks = int(csv_reading[2])
+        self.dropped_tasks = int(csv_reading[3])
+        self.missed_deadlines = int(csv_reading[4])
+        self.finished_task_graphs = int(csv_reading[5])
+        self.dropped_taskgraphs = int(csv_reading[6])
+        self.missed_taskgraphs = int(csv_reading[7])
+        self.goodput_taskgraphs = self.finished_task_graphs - self.missed_taskgraphs
+
     def update_finish(self, csv_reading: str):
         """Updates the values of the Simulator based on the SIMULATOR_END event from
         CSV.
@@ -396,10 +408,10 @@ def update_finish(self, csv_reading: str):
             csv_reading[1] == "SIMULATOR_END"
         ), f"The event {csv_reading[1]} was not of type SIMULATOR_END."
         self.end_time = int(csv_reading[0])
-        self.finished_tasks = int(csv_reading[2])
-        self.dropped_tasks = int(csv_reading[3])
-        self.missed_deadlines = int(csv_reading[4])
-        self.finished_task_graphs = int(csv_reading[5])
-        self.dropped_taskgraphs = int(csv_reading[6])
-        self.missed_taskgraphs = int(csv_reading[7])
-        self.goodput_taskgraphs = self.finished_task_graphs - self.missed_taskgraphs
+        # self.finished_tasks = int(csv_reading[2])
+        # self.dropped_tasks = int(csv_reading[3])
+        # self.missed_deadlines = int(csv_reading[4])
+        # self.finished_task_graphs = int(csv_reading[5])
+        # self.dropped_taskgraphs = int(csv_reading[6])
+        # self.missed_taskgraphs = int(csv_reading[7])
+        # self.goodput_taskgraphs = self.finished_task_graphs - self.missed_taskgraphs