Stopping actual stops now, I hope.

Author: stephenfreund

src/flowco/__main__.py

@@ -25,7 +25,6 @@
 from flowco.util.output import Output, error, log, log_timestamp, message, logger
 
 from flowco.util.errors import FlowcoError
-from flowco.util.stopper import Stopper
 from flowco.util.text import md_to_html
 from flowthon.flowthon import FlowthonProgram
 
@@ -394,15 +393,13 @@ def main(page: Optional[Page] = None, argv: List[str] = sys.argv[1:]):
     session.set(
         output=Output(),
         costs=CostTracker(),
-        stopper=Stopper(),
         shells=PythonShells(),
         filesystem=SessionFileSystem(f"file://{os.getcwd()}"),
     )
     log_timestamp()
 
     try:
-        with session.get("stopper", Stopper):
-            main_core(page=page, argv=argv)
+        main_core(page=page, argv=argv)
     except FlowcoError as e:
         error(e)
     finally:

src/flowco/builder/build.py

@@ -9,7 +9,6 @@
 from flowco.dataflow.phase import Phase
 from flowco.session.session import session
 from flowco.util.output import error, log, warn, logger
-from flowco.util.stopper import Stopper
 from typing import (
     Callable,
     Dict,
@@ -194,6 +193,7 @@ def build_with_worklist(
         graph: DataFlowGraph,
         target_phase: Phase,
         node_ids: List[str] | str | None,
+        should_stop: Callable[[], bool] = lambda: False,
     ) -> Iterator[BuildUpdate]:
 
         node_ids = graph.listify_node_ids(node_ids)
@@ -205,11 +205,11 @@ def build_with_worklist(
 
         if config.sequential:
             yield from self.build_with_worklist_sequential(
-                pass_config, graph, target_phase, node_ids
+                pass_config, graph, target_phase, node_ids, should_stop
             )
         else:
             yield from self.build_with_worklist_parallel(
-                pass_config, graph, target_phase, node_ids
+                pass_config, graph, target_phase, node_ids, should_stop
             )
 
     def build_with_worklist_sequential(
@@ -218,6 +218,7 @@ def build_with_worklist_sequential(
         graph: DataFlowGraph,
         target_phase: Phase,
         node_ids: List[str] | str | None,
+        should_stop: Callable[[], bool],
     ) -> Iterator[BuildUpdate]:
 
         @dataclass
@@ -236,7 +237,7 @@ def process_workitem(graph: DataFlowGraph, work_item: WorkItem):
                 node = graph[node_id]
 
                 with logger(f"{node_id}:{node.pill}"):
-                    if session.get("stopper", Stopper).should_stop():
+                    if should_stop():
                         return NodeResult(work_item, node, node, "Stopped")
 
                     node.build_status = phase_to_message.get(
@@ -322,8 +323,7 @@ def submit_items(current_worklist: Dict[WorkItem, List[WorkItem]]):
             current_worklist = submit_items(worklist)
 
             with logger("Running worklist"):
-                stopper = session.get("stopper", Stopper)
-                while len(done) < len(worklist) and not stopper.should_stop():
+                while len(done) < len(worklist) and not should_stop():
                     try:
                         log(f"Building: {len(worklist.keys()) - len(done)} steps")
                         item = in_flight.pop(0)
@@ -372,6 +372,7 @@ def build_with_worklist_parallel(
         graph: DataFlowGraph,
         target_phase: Phase,
         node_ids: List[str] | str | None,
+        should_stop: Callable[[], bool],
     ) -> Iterator[BuildUpdate]:
 
         @dataclass
@@ -396,7 +397,7 @@ def process_workitem(graph: DataFlowGraph, work_item: WorkItem):
 
                 with buffer_output(f"{node.pill}") as buffer:
                     with logger(f"Starting"):
-                        if session.get("stopper", Stopper).should_stop():
+                        if should_stop():
                             result_queue.put(
                                 NodeResult(work_item, node, node, "Stopped")
                             )
@@ -520,8 +521,7 @@ def submit_items(current_worklist: Dict[WorkItem, List[WorkItem]]):
                 current_worklist = submit_items(worklist)
 
                 with logger("Running worklist"):
-                    stopper = session.get("stopper", Stopper)
-                    while len(done) < len(worklist) and not stopper.should_stop():
+                    while len(done) < len(worklist) and not should_stop():
                         try:
                             log(f"Remaining: {len(worklist.keys()) - len(done)} steps")
                             result = result_queue.get()
@@ -553,7 +553,8 @@ def submit_items(current_worklist: Dict[WorkItem, List[WorkItem]]):
                         except Exception as e:
                             error(e)
                             raise e
-                    if stopper.should_stop():
+
+                    if should_stop():
                         executor.shutdown(wait=False, cancel_futures=True)
                         log(f"Stopping!")
 

src/flowco/ui/authenticate.py

@@ -49,29 +49,25 @@ def purge_stale_entries(cache_dict):
 
 
 def sign_in(authorization_url: str):
-    instructions = textwrap.dedent(
-        """\
-        # Flowco!
+    release = os.getenv("RELEASE_VERSION", "unknown")
+    commit_sha = os.getenv("COMMIT_SHA", "unknown")[:7]
+    build_date = os.getenv("BUILD_DATE", "unknown")
 
+    instructions = textwrap.dedent(
+        f"""\
+        # Flowco {release}!
+        
         * Flowco uses Google Sign-In to authenticate users.
         * Your email address will be used to create a unique account.
         * Flowco uses an Amazon S3 bucket to store all user files.
+        * Click the "Report Bug" whenever you see anything fishy!
 
+        [Version Release Notes](https://github.com/stephenfreund/flowco/releases)
         """
     )
     # st.write(f"Please sign in to continue.")
     st.write(instructions)
     st.link_button("Sign In", authorization_url)
-    st.divider()
-    # Read the commit SHA and build date from environment variables
-    release = os.getenv("RELEASE_VERSION", "unknown")
-    commit_sha = os.getenv("COMMIT_SHA", "unknown")[:7]
-    build_date = os.getenv("BUILD_DATE", "unknown")
-
-    st.write(f"Release {release}, {build_date}")
-    st.page_link(
-        "https://github.com/stephenfreund/flowco/releases", label="Release Notes"
-    )
 
 
 # Function to fetch user information from id_token

src/flowco/ui/page_files/build_page.py

@@ -225,6 +225,7 @@ def toggle_building(
             st.session_state.builder_progress = 0
         else:
             builder.stop()
+            self.clear_builder_and_reset_state()
 
     def get_builder_updates(self):
         ui_page: UIPage = st.session_state.ui_page
@@ -284,25 +285,27 @@ def fini(self):
         if builder is not None:
             if not builder.is_alive():
                 self.get_builder_updates()
-                st.session_state.force_update = True
-                st.session_state.builder = None
-                # set all node build_statuses to None
-                ui_page: UIPage = st.session_state.ui_page
-                dfg = ui_page.dfg()
-                dfg = dfg.update(
-                    nodes=[
-                        x.update(
-                            build_status=None,
-                        )
-                        for x in dfg.nodes
-                    ]
-                )
-                ui_page.update_dfg(dfg)
+                self.clear_builder_and_reset_state()
                 st.rerun()
             else:
                 time.sleep(0.25)
                 st.rerun()
 
+    def clear_builder_and_reset_state(self):
+        st.session_state.force_update = True
+        st.session_state.builder = None
+        ui_page: UIPage = st.session_state.ui_page
+        dfg = ui_page.dfg()
+        dfg = dfg.update(
+            nodes=[
+                x.update(
+                    build_status=None,
+                )
+                for x in dfg.nodes
+            ]
+        )
+        ui_page.update_dfg(dfg)
+
     @st.dialog("Edit Description", width="large")
     def edit_description(self):
         ui_page: UIPage = st.session_state.ui_page

src/flowco/ui/ui_builder.py

@@ -1,14 +1,11 @@
-from queue import Empty, Queue
+from queue import Queue
 import threading
-from typing import List
+from typing import Callable, List
 from flowco.builder.build import BuildEngine, BuildUpdate, PassConfig
-from flowco.dataflow.dfg import DataFlowGraph
 from flowco.dataflow.phase import Phase
 from flowco.page.page import Page
-from flowco.pythonshell.shells import PythonShells
 from flowco.session.session import session
-from flowco.util.output import log, logger
-from flowco.util.stopper import Stopper
+from flowco.util.output import log
 
 
 def run(
@@ -18,32 +15,25 @@ def run(
     node_ids: str | List[str] | None,
     target_phase: Phase,
     queue: Queue,
+    should_stop: Callable[[], bool],
 ):
 
-    with session.get("stopper", Stopper):
-        with page:
-            for build_updated in engine.build_with_worklist(
-                build_config, page.dfg, target_phase, node_ids
-            ):
-                # log(
-                #     f"Putting build_updated in queue.  Remaining: {build_updated.steps_remaining}"
-                # )
-                # if session.get("stopper", Stopper).should_stop():
-                #     log("Stopping build early.")
-                #     return
-                queue.put(build_updated)
+    with page:
+        for build_updated in engine.build_with_worklist(
+            build_config, page.dfg, target_phase, node_ids, should_stop
+        ):
+            queue.put(build_updated)
 
-            log("Waiting for no unfinished tasks.")
-            with queue.all_tasks_done:
-                if queue.unfinished_tasks:
-                    if queue.all_tasks_done.wait(3):
-                        log("All tasks done.")
-                    else:
-                        log("Timeout waiting for tasks to finish.")
+        log("Waiting for no unfinished tasks.")
+        with queue.all_tasks_done:
+            if queue.unfinished_tasks:
+                # Just wait a bit for the build page to clear out pending updates...
+                if queue.all_tasks_done.wait(3):
+                    log("All tasks done.")
+                else:
+                    log("Timeout waiting for tasks to finish.")
 
-            # queue.join()
-
-        log("Done building.")
+    log("Done building.")
 
 
 class Builder:
@@ -77,10 +67,20 @@ def __init__(
             name = ", ".join([page.dfg[x].pill for x in node_ids])
         self.message = f"{verb} {name}..."
 
+        self.stopped = False
+
         self.queue = Queue[BuildUpdate]()
         self.thread = threading.Thread(
             target=run,
-            args=(builder, page, build_config, node_ids, target_phase, self.queue),
+            args=(
+                builder,
+                page,
+                build_config,
+                node_ids,
+                target_phase,
+                self.queue,
+                self.should_stop,
+            ),
         )
         setattr(self.thread, "flowco_session", session.get_session())
         self.thread.start()
@@ -100,6 +100,8 @@ def update_done(self) -> None:
     def is_alive(self) -> bool:
         return self.thread.is_alive()
 
+    def should_stop(self) -> bool:
+        return self.stopped
+
     def stop(self) -> None:
-        with logger("Stopping builder"):
-            session.get("stopper", Stopper).stop()
+        self.stopped = True

src/flowco/ui/ui_main.py

@@ -13,7 +13,6 @@
 from flowco.util.output import Output, error, log_timestamp
 
 from flowco.session.session import StreamlitSession, session
-from flowco.util.stopper import Stopper
 
 
 def init_service():
@@ -35,7 +34,6 @@ def init_service():
         session.set(
             output=Output(),
             costs=CostTracker(),
-            stopper=Stopper(),
             shells=PythonShells(),
             filesystem=SessionFileSystem(f"file://{page_path}"),
         )
@@ -73,7 +71,6 @@ def init_service():
     session.set(
         output=Output(),
         costs=CostTracker(),
-        stopper=Stopper(),
         shells=PythonShells(),
         filesystem=SessionFileSystem(f"file://{page_path}"),
     )

src/flowco/util/stopper.py

@@ -1,28 +1,28 @@
-import threading
+# import threading
 
 
-class Stopper:
+# class Stopper:
 
-    def __init__(self):
-        self._stop_flag = False
-        self._nesting_level = 0
-        self._lock = threading.RLock()
+#     def __init__(self):
+#         self._stop_flag = False
+#         self._nesting_level = 0
+#         self._lock = threading.RLock()
 
-    def __enter__(self):
-        with self._lock:
-            if self._nesting_level == 0:
-                self._stop_flag = False
-            self._nesting_level += 1
-        return self
+#     def __enter__(self):
+#         with self._lock:
+#             if self._nesting_level == 0:
+#                 self._stop_flag = False
+#             self._nesting_level += 1
+#         return self
 
-    def __exit__(self, exc_type, exc_value, traceback):
-        with self._lock:
-            self._nesting_level -= 1
+#     def __exit__(self, exc_type, exc_value, traceback):
+#         with self._lock:
+#             self._nesting_level -= 1
 
-    def stop(self):
-        with self._lock:
-            self._stop_flag = True
+#     def stop(self):
+#         with self._lock:
+#             self._stop_flag = True
 
-    def should_stop(self):
-        with self._lock:
-            return self._stop_flag
+#     def should_stop(self):
+#         with self._lock:
+#             return self._stop_flag