Rework IR, add automatic parallelization

.gitignore

@@ -8,6 +8,11 @@ __pycache__
 *.egg-info
 build
 
+.vscode/
+
 # Experiment artifacts
 *.png
-*.pkl
+*.pkl
+*.csv
+nohup.out
+*.zip

deathstar_hotel_reservation/demo.py

@@ -268,7 +268,7 @@ def user_login_workload_generator():
 def benchmark_runner(proc_num) -> dict[int, dict]:
     print(f'Generator: {proc_num} starting')
     client = FlinkClientSync("deathstar", "ds-out", "localhost:9092", True)
-    deathstar_generator = user_login_workload_generator()
+    deathstar_generator = deathstar_workload_generator()
     start = timer()
 
     for _ in range(bursts):

deathstar_hotel_reservation/test_demo.py

@@ -1,100 +1,100 @@
 
-import os
-import sys
-
-# import cascade
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
-
-from cascade.runtime.python_runtime import PythonClientSync, PythonRuntime
-from cascade.runtime.flink_runtime import FlinkClientSync, FlinkRuntime
-from deathstar_hotel_reservation.demo import DeathstarDemo, recommend, reserve, search_hotel, user_login
-import time
-import pytest
-
-@pytest.mark.integration
-def test_deathstar_demo():
-    ds = DeathstarDemo()
-    ds.init_runtime(FlinkRuntime("deathstardemo-test", "dsd-out"))
-    ds.runtime.run(run_async=True)
-    print("Populating, press enter to go to the next step when done")
-    ds.populate()
-
-    client = FlinkClientSync("deathstardemo-test", "dsd-out")
-    input()
-    print("testing user login")
-    event = user_login()
-    client.send(event)
-
-    input()
-    print("testing reserve")
-    event = reserve()
-    client.send(event)
-
-    input()
-    print("testing search")
-    event = search_hotel()
-    client.send(event)
-
-    input()
-    print("testing recommend (distance)")
-    time.sleep(0.5)
-    event = recommend(req_param="distance")
-    client.send(event)
-
-    input()
-    print("testing recommend (price)")
-    time.sleep(0.5)
-    event = recommend(req_param="price")
-    client.send(event)
-
-    print(client._futures)
-    input()
-    print("done!")
-    print(client._futures)
-
-def test_deathstar_demo_python():
-    ds = DeathstarDemo()
-    ds.init_runtime(PythonRuntime())
-    ds.runtime.run()
-    print("Populating, press enter to go to the next step when done")
-    ds.populate()
-
-    time.sleep(0.1)
-
-    client = PythonClientSync(ds.runtime)
-    print("testing user login")
-    event = user_login()
-    result = client.send(event)
-    assert result == True
-    event = user_login(succesfull=False)
-    result = client.send(event)
-    assert result == False
-
-    print("testing reserve")
-    event = reserve()
-    result = client.send(event)
-    assert result == True
-
-    return
-    print("testing search")
-    event = search_hotel()
-    result = client.send(event)
-    print(result)
-
-    print("testing recommend (distance)")
-    time.sleep(0.5)
-    event = recommend(req_param="distance")
-    result = client.send(event)
-    print(result)
-
-    print("testing recommend (price)")
-    time.sleep(0.5)
-    event = recommend(req_param="price")
-    result = client.send(event)
-    print(result)
-
-    print("done!")
-
-
-if __name__ == "__main__":
-    test_deathstar_demo()
+# import os
+# import sys
+
+# # import cascade
+# sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
+
+# from cascade.runtime.python_runtime import PythonClientSync, PythonRuntime
+# from cascade.runtime.flink_runtime import FlinkClientSync, FlinkRuntime
+# from deathstar_hotel_reservation.demo import DeathstarDemo, recommend, reserve, search_hotel, user_login
+# import time
+# import pytest
+
+# @pytest.mark.integration
+# def test_deathstar_demo():
+#     ds = DeathstarDemo()
+#     ds.init_runtime(FlinkRuntime("deathstardemo-test", "dsd-out"))
+#     ds.runtime.run(run_async=True)
+#     print("Populating, press enter to go to the next step when done")
+#     ds.populate()
+
+#     client = FlinkClientSync("deathstardemo-test", "dsd-out")
+#     input()
+#     print("testing user login")
+#     event = user_login()
+#     client.send(event)
+
+#     input()
+#     print("testing reserve")
+#     event = reserve()
+#     client.send(event)
+
+#     input()
+#     print("testing search")
+#     event = search_hotel()
+#     client.send(event)
+
+#     input()
+#     print("testing recommend (distance)")
+#     time.sleep(0.5)
+#     event = recommend(req_param="distance")
+#     client.send(event)
+
+#     input()
+#     print("testing recommend (price)")
+#     time.sleep(0.5)
+#     event = recommend(req_param="price")
+#     client.send(event)
+
+#     print(client._futures)
+#     input()
+#     print("done!")
+#     print(client._futures)
+
+# def test_deathstar_demo_python():
+#     ds = DeathstarDemo()
+#     ds.init_runtime(PythonRuntime())
+#     ds.runtime.run()
+#     print("Populating, press enter to go to the next step when done")
+#     ds.populate()
+
+#     time.sleep(0.1)
+
+#     client = PythonClientSync(ds.runtime)
+#     print("testing user login")
+#     event = user_login()
+#     result = client.send(event)
+#     assert result == True
+#     event = user_login(succesfull=False)
+#     result = client.send(event)
+#     assert result == False
+
+#     print("testing reserve")
+#     event = reserve()
+#     result = client.send(event)
+#     assert result == True
+
+#     return
+#     print("testing search")
+#     event = search_hotel()
+#     result = client.send(event)
+#     print(result)
+
+#     print("testing recommend (distance)")
+#     time.sleep(0.5)
+#     event = recommend(req_param="distance")
+#     result = client.send(event)
+#     print(result)
+
+#     print("testing recommend (price)")
+#     time.sleep(0.5)
+#     event = recommend(req_param="price")
+#     result = client.send(event)
+#     print(result)
+
+#     print("done!")
+
+
+# if __name__ == "__main__":
+#     test_deathstar_demo()

deathstar_movie_review/demo.py

@@ -1,14 +1,12 @@
 from typing import Literal
+import cascade
+from cascade.dataflow.dataflow import DataflowRef
 from cascade.dataflow.optimization.dead_node_elim import dead_node_elimination
+from cascade.dataflow.optimization.parallelization import parallelize_until_if
 from cascade.runtime.flink_runtime import FlinkRuntime
-
-from .entities.user import user_op
-from .entities.compose_review import compose_review_op
-from .entities.frontend import frontend_df_parallel, frontend_df_serial, frontend_op, text_op, unique_id_op
-from .entities.movie import movie_id_op, movie_info_op, plot_op
+from tests.integration.flink.utils import create_topics, init_flink_runtime
 
 import os
-from confluent_kafka.admin import AdminClient, NewTopic
 
 KAFKA_BROKER = "localhost:9092"
 KAFKA_FLINK_BROKER = "kafka:9093" # If running a flink cluster and kafka inside docker, the broker url might be different
@@ -17,67 +15,32 @@
 OUT_TOPIC = "ds-movie-out"
 INTERNAL_TOPIC = "ds-movie-internal"
 
-EXPERIMENT: Literal["baseline", "pipelined", "parallel"] = os.getenv("EXPERIMENT", "baseline")
-
-def create_topics(*required_topics):
-    conf = {
-        "bootstrap.servers": KAFKA_BROKER
-    }
-
-    admin_client = AdminClient(conf)
-
-    # Fetch existing topics
-    existing_topics = admin_client.list_topics(timeout=5).topics.keys()
-
-    # Find missing topics
-    missing_topics = [topic for topic in required_topics if topic not in existing_topics]
-
-    if missing_topics:
-        print(f"Creating missing topics: {missing_topics}")
-
-        # Define new topics (default: 1 partition, replication factor 1)
-        new_topics = [NewTopic(topic, num_partitions=1, replication_factor=1) for topic in missing_topics]
-
-        # Create topics
-        futures = admin_client.create_topics(new_topics)
-
-        # Wait for topic creation to complete
-        for topic, future in futures.items():
-            try:
-                future.result()  # Block until the operation is complete
-                print(f"Topic '{topic}' created successfully")
-            except Exception as e:
-                print(f"Failed to create topic '{topic}': {e}")
-    else:
-        print("All required topics exist.")
+EXPERIMENT: Literal["baseline", "parallel"] = os.getenv("EXPERIMENT", "baseline")
 
 
 def main():
     create_topics(IN_TOPIC, OUT_TOPIC, INTERNAL_TOPIC)
 
-    runtime = FlinkRuntime(IN_TOPIC, OUT_TOPIC, internal_topic=INTERNAL_TOPIC)
-    runtime.init(kafka_broker=KAFKA_FLINK_BROKER,bundle_time=5, bundle_size=10)
-
-    if EXPERIMENT == "baseline":
-        frontend_op.dataflow = frontend_df_serial()
-    elif EXPERIMENT == "pipelined":
-        frontend_op.dataflow = frontend_df_serial()
-        dead_node_elimination([], [frontend_op])
-    elif EXPERIMENT == "parallel":
-        frontend_op.dataflow = frontend_df_parallel()
-
-    print(frontend_op.dataflow.to_dot())
+    runtime = init_flink_runtime("deathstar_movie_review.entities.entities", IN_TOPIC, OUT_TOPIC, INTERNAL_TOPIC, kafka_broker=KAFKA_FLINK_BROKER,bundle_time=5, bundle_size=10, thread_mode=True, parallelism=None)
+
     print(f"Creating dataflow [{EXPERIMENT}]")
 
-    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)
-
+    # for parallel experiment
+    df_baseline = cascade.core.dataflows[DataflowRef("Frontend", "compose")]
+    df_parallel, _ = parallelize_until_if(df_baseline)
+    df_parallel.name = "compose_parallel"
+    cascade.core.dataflows[DataflowRef("Frontend", "compose_parallel")] = df_parallel
+    runtime.add_dataflow(df_parallel)
+
+    # for prefetch experiment
+    df_baseline = cascade.core.dataflows[DataflowRef("MovieId", "upload_movie_prefetch")]
+    df_parallel, _ = parallelize_until_if(df_baseline)
+    df_parallel.name = "upload_movie_prefetch_parallel"
+    cascade.core.dataflows[DataflowRef("MovieId", "upload_movie_prefetch_parallel")] = df_parallel
+    runtime.add_dataflow(df_parallel)
+
+    print(cascade.core.dataflows.keys())
+
     runtime.run()
 
 if __name__ == "__main__":