timeout check in naive and TODO for adj latency stats

Author: tylerharter

eudoxia/scheduler/naive.py

@@ -10,6 +10,7 @@
 def naive_pipeline_init(s):
     s.waiting_queue: List[Pipeline] = []
     s.multi_operator_containers = s.params["multi_operator_containers"]
+    s.tick_number = 0
 
 @register_scheduler(key="naive")
 def naive_pipeline(s, results: List[ExecutionResult],
@@ -27,6 +28,8 @@ def naive_pipeline(s, results: List[ExecutionResult],
             - List of new assignments to provide to Executor
     """
 
+    s.tick_number += 1
+
     # this scheduler doesn't make any new decisions unless there are
     # new pipelines arriving or containers exiting, so exit fast in
     # this case
@@ -54,8 +57,9 @@ def naive_pipeline(s, results: List[ExecutionResult],
         # find a pipeline with ops we can assign
         while s.waiting_queue:
             pipeline = s.waiting_queue.pop(0)
-            has_failures = pipeline.runtime_status().state_counts[OperatorState.FAILED] > 0
-            if pipeline.runtime_status().is_pipeline_successful() or has_failures:
+            status = pipeline.runtime_status()
+            has_failures = status.state_counts[OperatorState.FAILED] > 0
+            if status.is_pipeline_successful() or has_failures or status.has_timed_out(s.tick_number):
                 # we don't retry, so anything complete or with failures
                 # will be permanently removed from the queue
                 continue

eudoxia/simulator.py

@@ -88,6 +88,7 @@ class SimulatorStats(NamedTuple):
     pipelines_interactive: PipelineStats
     pipelines_batch: PipelineStats
 
+    # TODO: add penalty option to give latency for unfinished jobs
     def adjusted_latency(
         self,
         weights: Dict[Priority, float] = None,