Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[data] Only dump state once on progress bar close #46928

Merged
merged 6 commits into from
Aug 8, 2024
Merged

[data] Only dump state once on progress bar close #46928

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e88bd35
alter close behavior to only dump_state if the bar is not already closed
omatthew98 Aug 1, 2024
64632bb
rearrange lines
omatthew98 Aug 5, 2024
8a409af
update to hide/unhide bars
omatthew98 Aug 6, 2024
4f6043b
respond to pr comments
omatthew98 Aug 7, 2024
317e959
Merge branch 'master' into mowen/close-tqdm-bar
omatthew98 Aug 7, 2024
a749722
remove undeeded file
omatthew98 Aug 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions python/ray/data/_internal/execution/streaming_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,14 @@ def execute(

logger.debug("Execution config: %s", self._options)

# Note: DAG must be initialized in order to query num_outputs_total.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could you add a comment indicating why this must happen prior to operator bars being initialized, so readers know about this fix?

# Note: Initialize global progress bar before building the streaming
# topology so bars are created in the same order as they should be
# displayed. This is done to ensure correct ordering within notebooks.
self._global_info = ProgressBar(
"Running", dag.num_outputs_total(), unit="bundle"
)

# Setup the streaming DAG topology and start the runner thread.
self._topology, _ = build_streaming_topology(dag, self._options)
self._resource_manager = ResourceManager(
Expand All @@ -126,12 +134,6 @@ def execute(

self._has_op_completed = {op: False for op in self._topology}

if not isinstance(dag, InputDataBuffer):
# Note: DAG must be initialized in order to query num_outputs_total.
self._global_info = ProgressBar(
"Running", dag.num_outputs_total(), unit="bundle"
)

self._output_node: OpState = self._topology[dag]
StatsManager.register_dataset_to_stats_actor(
self._dataset_tag,
Expand Down
4 changes: 4 additions & 0 deletions python/ray/experimental/tqdm_ray.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,8 +241,12 @@ def update_bar(self, state: ProgressBarState) -> None:
def close_bar(self, state: ProgressBarState) -> None:
"""Remove a bar from this group."""
bar = self.bars_by_uuid[state["uuid"]]
# Note: Hide and then unhide bars to prevent flashing of the
# last bar when we are closing multiple bars sequentially.
instance().hide_bars()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same here, let's explain why we are hiding/unhiding the bar

bar.close()
del self.bars_by_uuid[state["uuid"]]
instance().unhide_bars()

def slots_required(self):
"""Return the number of pos slots we need to accomodate bars in this group."""
Expand Down