Conversation
Signed-off-by: xgui <xgui@anyscale.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a SpillMetricsMonitor actor to compute and report peak and average object store spilling rates, which is a valuable addition for performance monitoring in benchmarks. The implementation uses a detached Ray actor with a background polling thread, which is a suitable design. The integration into RayDataLoaderFactory is clean. My review includes a couple of suggestions to enhance the robustness of the metric calculation and improve code consistency.
| ) | ||
| return memory_info.store_stats.spilled_bytes_total | ||
|
|
||
| def _poll_loop(self): |
There was a problem hiding this comment.
The _poll_loop method is missing a return type hint. For consistency with the rest of the codebase's type annotations, it should be specified. Since this method doesn't return a value, the hint should be -> None.
| def _poll_loop(self): | |
| def _poll_loop(self) -> None: |
| if delta_time > 0: | ||
| rate_gb_s = (delta_bytes / (1024**3)) / delta_time | ||
| with self._lock: | ||
| self._spill_rates_gb_s.append(rate_gb_s) |
There was a problem hiding this comment.
The spilled_bytes_total counter could theoretically reset (e.g., on GCS restart), which would cause delta_bytes to be negative. This would result in a negative spill rate being recorded, skewing the average calculation. It's safer to only calculate the rate for non-negative delta_bytes.
| if delta_time > 0: | |
| rate_gb_s = (delta_bytes / (1024**3)) / delta_time | |
| with self._lock: | |
| self._spill_rates_gb_s.append(rate_gb_s) | |
| if delta_time > 0 and delta_bytes >= 0: | |
| rate_gb_s = (delta_bytes / (1024**3)) / delta_time | |
| with self._lock: | |
| self._spill_rates_gb_s.append(rate_gb_s) |
1 participant
Description
Related issues
Additional information