Refactor DAG Optimizer (skypilot-org#628)

* Refactor optimizer

* Remove unnecessary import

* yapf

* Minor fix

* Add NotImplementedError

* Minor

* Rename vars & Annotate types

* Minor fix

* Minor

* Minor fix

* Fix type annotation

* yapf

* [Minor] Address comment

* Add type alias & enhance comments

* yapf

* Fix minor error in dag_lib.Dag

* Add is_chain to Dag

* Address comments

* yapf

* yapf

* Address comments

* Add total in optimizer msg

* Add a comment in is_chain

* Address reviews & Fix egress msg

* yapf

* Minor fix

* Fix egress msg

* yapf

* obj -> objective

* pass yapf

* cost -> cost/time

Author: WoosukKwon

examples/example_app.py

@@ -76,4 +76,4 @@ def make_application():
 
 dag = make_application()
 sky.optimize(dag, minimize=sky.OptimizeTarget.COST)
-# sky.optimize(dag, minimize=OptimizeTarget.TIME)
+# sky.optimize(dag, minimize=sky.OptimizeTarget.TIME)

sky/dag.py

@@ -71,3 +71,20 @@ def __repr__(self):
 
     def get_graph(self):
         return self.graph
+
+    def is_chain(self) -> bool:
+        # NOTE: this method assumes that the graph has no cycle.
+        is_chain = True
+        visited_zero_out_degree = False
+        for node in self.graph.nodes:
+            out_degree = self.graph.out_degree(node)
+            if out_degree > 1:
+                is_chain = False
+                break
+            elif out_degree == 0:
+                if visited_zero_out_degree:
+                    is_chain = False
+                    break
+                else:
+                    visited_zero_out_degree = True
+        return is_chain