[core][aDAG] Clean up shutdown path #47702

rkooo567 · 2024-09-17T06:42:28Z

Why are these changes needed?

Vairous fixes including #47685 to clean up shutdown path.

Make teardown idemoptent
Remove unblocking teardown(wait=False). It is prone to all weird errors because we don't do synchronization properly before shutdown.
Previously, we teardown on __del__. It works well at runtime but not at shutdown time because desetruction order is not guaranteed at shutdown. We should use atexit handler instead. To fix this issue, I keep tracking of all compiled dags created (via weakref) and do teardown inside shutdown API which is called when the interpreter shutdown
Fix asyncio read/write being blocked and joined forever issue. We check read/write every 1 second and check sys.is_finalizing() which sets to True upon interpreting exiting time. We can't rely on atexit handler.teardown because asyncio read/write runs in thread pool, and thread pool is joined "before at exit handler is executed". See https://github.com/python/cpython/blob/8f82d9aa2191db7826bb7a453fe06ce65f966cf8/Lib/concurrent/futures/thread.py#L37 (this atexit handler always is called before python's regular atexit handler).
Change teardown logs to debug so that it won't be printed unless necessary.

Related issue number

Closes #47685 (comment)

Checks

I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
I've run scripts/format.sh to lint the changes in this PR.
I've included any doc changes needed for https://docs.ray.io/en/master/.
- I've added any new APIs to the API Reference. For example, if I added a
  method in Tune, I've added it in doc/source/tune/api/ under the
  corresponding .rst file.
I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
Testing Strategy
- Unit tests
- Release tests
- This PR is not tested :(

stephanie-wang

LGTM. Can you add a description of the changes to the PR text?

python/ray/dag/compiled_dag_node.py

rkooo567 · 2024-09-17T17:53:27Z

python/ray/experimental/channel/common.py

-        return [c.read() for c in self._input_channels]
+        results = []
+        for c in self._input_channels:
+            try:


add while loop

python/ray/dag/compiled_dag_node.py

ruisearch42

Otherwise LGTM

ruisearch42 · 2024-09-19T17:31:15Z

python/ray/dag/compiled_dag_node.py

+# this process. It tracks them as weakref meaning when the compiled dag
+# is GC'ed, it is automatically removed from here. It is used to teardown
+# compiled dags at interpret shutdown time.
+_compiled_dag_queue = weakref.WeakValueDictionary()


nit: this is a dict, not a queue, right?

ruisearch42 · 2024-09-19T17:32:36Z

python/ray/dag/compiled_dag_node.py

@@ -1703,6 +1722,8 @@ def __init__(self):
                # Lock to make sure that we only perform teardown for this DAG
                # once.
                self.in_teardown_lock = threading.Lock()
+                self.name = "MonitorThread"


AdagMonitorThread?

ruisearch42 · 2024-09-19T17:39:16Z

python/ray/dag/compiled_dag_node.py

-            # has already been destructed, so it is not safe to block in
-            # ray.get.
-            monitor.teardown(wait=False)
+            monitor.teardown(wait=True)


Will this wait indefinitely or delay process shutdown?

Also update the comment above

ruisearch42 · 2024-09-19T17:40:29Z

python/ray/dag/compiled_dag_node.py

+# the destructor order is not guaranteed. We call this function
+# upon `ray.worker.shutdown` which is registered to atexit handler
+# so that teardown is properly called before objects are destructed.
+def shutdown_compiled_dag_node():


node is ambiguous here. Can you rename?

ruisearch42 · 2024-09-19T17:47:35Z

python/ray/experimental/channel/common.py

+                    if sys.is_finalizing():
+                        break
+                else:
+                    break


Why is there not an outer check that is similar to L373-374?

ruisearch42 · 2024-09-19T17:49:48Z

python/ray/experimental/channel/common.py

+                else:
+                    break


nit: can we unify with the same code pattern as in L361-L371? i.e., L561-L562 does the same thing as L365.

kevin85421 · 2024-09-19T21:00:54Z

python/ray/dag/compiled_dag_node.py

@@ -1694,7 +1711,9 @@ def _is_same_actor(idx1: int, idx2: int) -> bool:
        return not topological_order_exists

    def _monitor_failures(self):
-        outer = self
+        import weakref


Do we need to import weakref here in case it has already been imported at the top level?

kevin85421 · 2024-09-19T21:12:18Z

python/ray/experimental/channel/common.py

+                    break
+                except ray.exceptions.RayChannelTimeoutError:
+                    # Interpreter exits. We should stop reading
+                    # so that the thread can join.


what does "the thread" refer to?

kevin85421 · 2024-09-19T21:14:13Z

python/ray/experimental/channel/common.py

+                    # so that the thread can join.
+                    if sys.is_finalizing():
+                        break
+                    pass


do we need pass here?

shutdown cleanup.

0f236ba

rkooo567 added the go add ONLY when ready to merge, run all tests label Sep 17, 2024

rkooo567 added 2 commits September 17, 2024 00:43

fixed

106580f

fixed

5685920

rkooo567 mentioned this pull request Sep 17, 2024

[ADAG] Add checking for worker #47227

Open

3 tasks

rkooo567 assigned stephanie-wang Sep 17, 2024

stephanie-wang reviewed Sep 17, 2024

View reviewed changes

python/ray/dag/compiled_dag_node.py Show resolved Hide resolved

python/ray/dag/compiled_dag_node.py Outdated Show resolved Hide resolved

addressed code review.

58feae7

rkooo567 commented Sep 19, 2024

View reviewed changes

rkooo567 changed the title ~~[WIP][core][aDAG] Clean up shutdown path~~ [core][aDAG] Clean up shutdown path Sep 19, 2024

ruisearch42 reviewed Sep 19, 2024

View reviewed changes

kevin85421 reviewed Sep 19, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[core][aDAG] Clean up shutdown path #47702

[core][aDAG] Clean up shutdown path #47702

rkooo567 commented Sep 17, 2024 •

edited

Loading

stephanie-wang left a comment

rkooo567 Sep 17, 2024

ruisearch42 left a comment

ruisearch42 Sep 19, 2024

ruisearch42 Sep 19, 2024

ruisearch42 Sep 19, 2024

ruisearch42 Sep 19, 2024

ruisearch42 Sep 19, 2024

ruisearch42 Sep 19, 2024

kevin85421 Sep 19, 2024

kevin85421 Sep 19, 2024

kevin85421 Sep 19, 2024

[core][aDAG] Clean up shutdown path #47702

Are you sure you want to change the base?

[core][aDAG] Clean up shutdown path #47702

Conversation

rkooo567 commented Sep 17, 2024 • edited Loading

Why are these changes needed?

Related issue number

Checks

stephanie-wang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ruisearch42 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rkooo567 commented Sep 17, 2024 •

edited

Loading