Add deathstar benchmark test for new IR

Author: lucasvanmol

deathstar_movie_review/entities/entities.py

@@ -0,0 +1,91 @@
+import uuid
+from cascade import cascade
+
+@cascade
+class ComposeReview:
+    def __init__(self, req_id: str, *args): # *args is a temporary hack to allow for creation of composereview on the fly
+        self.req_id = req_id
+        self.review_data = {}
+
+    def upload_unique_id(self, review_id: int):
+        self.review_data["review_id"] = review_id
+
+    # could use the User class instead?
+    def upload_user_id(self, user_id: str):
+        self.review_data["userId"] = user_id
+    
+    def upload_movie_id(self, movie_id: str):
+        self.review_data["movieId"] = movie_id
+
+    def upload_rating(self, rating: int):
+        self.review_data["rating"] = rating
+
+    def upload_text(self, text: str):
+        self.review_data["text"] = text
+
+    def get_data(self):
+        x = self.review_data
+        return x
+
+@cascade
+class User:
+    def __init__(self, username: str, user_data: dict):
+        self.username = username
+        self.user_data = user_data
+
+    def upload_user(self, review: ComposeReview):
+        user_id = self.user_data["userId"]
+        review.upload_user_id(user_id)
+
+@cascade
+class MovieId:
+    # key: 'title'
+    def __init__(self, title: str,  movie_id: str):
+        self.title = title
+        self.movie_id = movie_id
+
+    def upload_movie(self, review: ComposeReview, rating: int):
+        # if self.movie_id is not None:
+        #     review.upload_movie_id(self.movie_id)
+        # else:
+        #     review.upload_rating(rating)
+        movie_id = self.movie_id
+        review.upload_movie_id(movie_id)
+
+@cascade
+class Frontend():
+    @staticmethod
+    def compose(review: ComposeReview, user: User, title: MovieId, rating: int, text: str):
+        UniqueId.upload_unique_id_2(review)
+        user.upload_user(review)
+        title.upload_movie(review, rating)
+        # text = text[:CHAR_LIMIT] # an operation like this could be reorderd for better efficiency!
+        Text.upload_text_2(review, text)
+        
+@cascade
+class UniqueId():
+    @staticmethod
+    def upload_unique_id_2(review: ComposeReview):
+        # TODO: support external libraries
+        # review_id = uuid.uuid1().int >> 64
+        review_id = 424242
+        review.upload_unique_id(review_id)
+
+@cascade
+class Text():
+    @staticmethod
+    def upload_text_2(review: ComposeReview, text: str):
+        review.upload_text(text)
+    
+
+@cascade
+class Plot:
+    def __init__(self, movie_id: str, plot: str):
+        self.movie_id = movie_id
+        self.plot = plot
+
+@cascade
+class MovieInfo:
+    def __init__(self, movie_id: str, info: dict):
+        self.movie_id = movie_id
+        self.info = info

deathstar_movie_review/entities/frontend.py

@@ -8,7 +8,6 @@
 from deathstar_movie_review.entities.user import User
 from deathstar_movie_review.entities.text import Text, text_op
 
-
 CHAR_LIMIT = 50
 
 # frontend is made stateless

deathstar_movie_review/entities/unique_id.py

@@ -4,7 +4,6 @@
 from cascade.dataflow.operator import Block, StatelessOperator
 from deathstar_movie_review.entities.compose_review import ComposeReview
 
-
 class UniqueId():
     @staticmethod
     def upload_unique_id_2(review: ComposeReview):

deathstar_movie_review/test_movie_review_demo.py

@@ -1,8 +1,10 @@
 import sys
 import os
 
+
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
 
+from cascade.dataflow.operator import StatefulOperator, StatelessOperator
 from cascade.dataflow.dataflow import Event, InitClass, InvokeMethod, OpNode
 from cascade.dataflow.optimization.dead_node_elim import dead_node_elimination
 from cascade.runtime.python_runtime import PythonClientSync, PythonRuntime
@@ -11,35 +13,36 @@
 from deathstar_movie_review.entities.movie import MovieId, movie_id_op, movie_info_op, plot_op
 from deathstar_movie_review.entities.frontend import frontend_op, text_op, unique_id_op, frontend_df_serial
 
+import cascade
 
-
-def test_deathstar_movie_demo_python():
-    print("starting")
+def init_python_runtime() -> tuple[PythonRuntime, PythonClientSync]:
     runtime = PythonRuntime()
+    for op in cascade.core.operators.values():
+        if isinstance(op, StatefulOperator):
+            runtime.add_operator(op)
+        elif isinstance(op, StatelessOperator):
+            runtime.add_stateless_operator(op)
 
-    # make sure we're running the serial version
-    prev_df = frontend_op.dataflow
-    frontend_op.dataflow = frontend_df_serial()
-
+    runtime.run()
+    return runtime, PythonClientSync(runtime)
 
-    print(frontend_op.dataflow.to_dot())
-    dead_node_elimination([], [frontend_op])
-    print(frontend_op.dataflow.to_dot())
+import time
+def test_deathstar_movie_demo_python():
+    print("starting")
+    cascade.core.clear() 
+    exec(f'import deathstar_movie_review.entities.entities')
+    cascade.core.init()
 
-    runtime.add_operator(compose_review_op)
-    runtime.add_operator(user_op)
-    runtime.add_operator(movie_info_op)
-    runtime.add_operator(movie_id_op)
-    runtime.add_operator(plot_op)
-    runtime.add_stateless_operator(frontend_op)
-    runtime.add_stateless_operator(unique_id_op)
-    runtime.add_stateless_operator(text_op)
+    runtime, client = init_python_runtime()
+    user_op = cascade.core.operators["User"]
+    compose_op = cascade.core.operators["ComposeReview"]
+    movie_op = cascade.core.operators["MovieId"]
+    frontend_op = cascade.core.operators["Frontend"]
 
-    runtime.run()
-    client = PythonClientSync(runtime)
+    for df in cascade.core.dataflows.values():
+        print(df.to_dot())
 
-    init_user = OpNode(User, InitClass(), read_key_from="username")
-    username = "username_1"
+    username = "myUsername"
     user_data = {
             "userId": "user1",
             "FirstName": "firstname",
@@ -48,27 +51,32 @@ def test_deathstar_movie_demo_python():
             "Password": "****",
             "Salt": "salt"
         }
+    
     print("testing user create")
-    event = Event(init_user, {"username": username, "user_data": user_data}, None)
+
+    event = user_op.dataflows["__init__"].generate_event({"username": username, "user_data": user_data}, username)
     result = client.send(event)
-    assert isinstance(result, User) and result.username == username
+    print(result)
+    assert result.username == username
 
     print("testing compose review")
-    req_id = 1
+    req_id = "1"
     movie_title = "Cars 2"
     movie_id = 1
 
     # make the review
-    init_compose_review = OpNode(ComposeReview, InitClass(), read_key_from="req_id")
-    event = Event(init_compose_review, {"req_id": req_id}, None)
+    event = compose_op.dataflows["__init__"].generate_event({"req_id": req_id}, req_id)
     result = client.send(event)
     print("review made")
 
 
-    # make the movie
-    init_movie = OpNode(MovieId, InitClass(), read_key_from="title")
-    event = Event(init_movie, {"title": movie_title, "movie_id": movie_id}, None)
+
+    # # make the movie
+    # init_movie = OpNode(MovieId, InitClass(), read_key_from="title")
+    event = movie_op.dataflows["__init__"].generate_event({"title": movie_title, "movie_id": movie_id}, movie_title)
     result = client.send(event)
+    # event = Event(init_movie, {"title": movie_title, "movie_id": movie_id}, None)
+    # result = client.send(event)
     print("movie made")
 
     # compose the review
@@ -80,11 +88,11 @@ def test_deathstar_movie_demo_python():
         "text": "good movie!"
     }
 
-    event = Event(
-        frontend_op.dataflow.entry, 
-        review_data,
-        frontend_op.dataflow)
-    result = client.send(event)
+    r_data = {r+"_0": v for r, v in review_data.items()}
+
+    event = frontend_op.dataflows["compose"].generate_event(r_data)
+    result = client.send(event, block=False)
+
     print(result)
     print("review composed")
 
@@ -96,19 +104,19 @@ def test_deathstar_movie_demo_python():
         {"req_id": req_id},
         None
     )
+    event = compose_op.dataflows["get_data"].generate_event({"req_id": req_id}, req_id)
     result = client.send(event)
+    print(result)
+    print(runtime.statefuloperators["ComposeReview"].states["1"].review_data)
+    # time.sleep(0.5)
+
+    # result = client.send(event)
     expected = {
         "userId": user_data["userId"],
         "movieId": movie_id,
         "text": review_data["text"]
     }
-    print(result, expected)
+    # print(result, expected)
     assert "review_id" in result
     del result["review_id"] # randomly generated
-    assert result == expected
-
-    print("Success!")
-
-    # put the df back
-    frontend_op.dataflow = prev_df
-    
+    assert result == expected

src/cascade/core.py

@@ -4,14 +4,14 @@
 from klara.core import nodes
 from klara.core.tree_rewriter import AstBuilder
 from klara.core.cfg import Cfg
-
+from klara.core.node_classes import Arguments
 
 from cascade.dataflow.operator import Block, StatefulOperator, StatelessOperator
 from cascade.wrappers import ClassWrapper
 from cascade.descriptors import ClassDescriptor, MethodDescriptor
-from cascade.frontend.generator.generate_split_functions import GenerateSplitFunctions
+from cascade.frontend.generator.generate_split_functions import GenerateSplitFunctions, GroupStatements
 from cascade.frontend.generator.generate_dataflow import GenerateDataflow
-from cascade.dataflow.dataflow import DataFlow, Operator 
+from cascade.dataflow.dataflow import CallLocal, DataFlow, DataflowRef, InitClass, Operator 
 from cascade.frontend.intermediate_representation import StatementDataflowGraph
 from cascade.frontend.generator.build_compiled_method_string import BuildCompiledMethodsString
 from cascade.frontend.ast_visitors import ExtractTypeVisitor
@@ -28,6 +28,7 @@ def setup_cfg(code: str) -> Cfg:
 registered_classes: list[ClassWrapper] = []
 
 operators: dict[str, Operator] = {}
+dataflows: dict[DataflowRef, DataFlow] = {}
 
 def cascade(cls, parse_file=True):
     if not isclass(cls):
@@ -54,12 +55,8 @@ def cascade(cls, parse_file=True):
     registered_classes.append(class_wrapper)
 
 
-
-def build(method) -> tuple[DataFlow, list[Block]]:
-    # TODO: implement
-    pass
-
 def init():
+    # First pass: register operators/classes
     for cls in registered_classes:
         op_name = cls.class_desc.class_name
 
@@ -72,13 +69,35 @@ def init():
 
         # generate split functions
         for method in cls.class_desc.methods_dec:
-            method.build_dataflow()
-            df, blocks = build(method)
+            df_ref = DataflowRef(op_name, method.method_name)
+            # Add version number manually
+            args = [f"{str(arg)}_0" for arg in method.method_node.args.args]
+            # TODO: cleaner solution that checks if the function is stateful or not
+            if args[0] == "self_0":
+                args = args[1:]
+            dataflows[df_ref] = DataFlow(method.method_name, op_name, args)
+
+        operators[op_name] = op
+    
+    # Second pass: build dataflows
+    for cls in registered_classes:
+        op_name = cls.class_desc.class_name
+        op = operators[op_name]
+
+        # generate split functions
+        for method in cls.class_desc.methods_dec:
+            if method.method_name == "__init__":
+                df = DataFlow("__init__", op_name)
+                n0 = CallLocal(InitClass())
+                df.entry = [n0]
+                blocks = []
+            else:
+                df, blocks = GroupStatements(method.method_node).build(dataflows, op_name)
+            
             op.dataflows[df.name] = df
             for b in blocks:
                 op.methods[b.name] = b
 
-        operators[op_name] = op