[Data] Prevent unbounded growth of_StatsActor.datasets

[orangeQWJ]

Background
Closes #43924.

The _StatsActor suffers from unbounded memory usage in long-running clusters. Its core metadata dictionaries, datasets and dataset_metadatas, lacked a proper garbage collection (GC) mechanism. This could lead to an Out-of-Memory (OOM) error in the _StatsActor when a large number of datasets are created.

Solution
This patch implements an eviction policy for _StatsActor based on the max_stats configuration to effectively limit the size of the datasets and dataset_metadatas dictionaries, preventing their unbounded growth.

The implementation details are as follows:

1. Optimize Queue Implementation
   The old, unused fifo_queue field has been removed. It is replaced by a new collections.deque, finished_datasets_queue, which serves as a more efficient FIFO queue for storing the tags of completed datasets.

2. Implement Eviction Logic

   • When a dataset's status is updated to FINISHED or FAILED, its tag is appended to the finished_datasets_queue.
   • A check is then immediately performed to see if the total number of entries in the datasets dictionary exceeds max_stats.
   • If the limit is exceeded, the oldest dataset tag is popped from the front of the finished_datasets_queue, and the corresponding entries are synchronously deleted from the datasets and dataset_metadatas dictionaries.

3. Clarify Limitation Strategy
   max_stats is not a strict hard limit. Since the eviction logic is only triggered when a dataset completes (FINISHED or FAILED), it is possible for the number of RUNNING datasets to cause the total entry count to temporarily exceed max_stats. This design ensures that metadata for in-progress tasks is never evicted, while still effectively preventing unbounded memory growth and OOM errors by cleaning up the oldest completed data as soon as a task finishes.

Testing
To verify the correctness of this fix, a new unit test, test_stats_actor_datasets_eviction, has been added. This test sets a low max_stats value and asserts that the oldest finished dataset is correctly evicted when the limit is surpassed.

[gemini-code-assist]

Code Review

This pull request introduces a crucial fix to prevent unbounded memory growth in the _StatsActor by implementing an eviction policy for finished datasets. The changes are well-structured and include a new collections.deque for efficient FIFO operations and clear eviction logic within update_dataset. The addition of the test_stats_actor_datasets_eviction unit test is excellent, as it thoroughly verifies the new functionality.

My review includes a couple of minor suggestions to improve code conciseness and readability in the eviction logic. Overall, this is a solid improvement to the stability of long-running Ray Data clusters.

[goutamvenkat-anyscale]

ooc did you try considering using 1 OrderedDict instead of a dict + deque to limit the memory usage? Basically on each use of the dataset tag you can move the tag to the end of the ordereddict

[goutamvenkat-anyscale]

What was the follow-up here?

[goutamvenkat-anyscale]

Thanks for your contribution. Please ensure to sign the commits.

As far as the change goes, I was wondering if you considered using OrderedDict for dataset instead of a separate deque to keep track of the used tags.