Emit Quent events for cudf-polars memory reservation. - #24038
Emit Quent events for cudf-polars memory reservation.#24038TomAugspurger wants to merge 1 commit into
Conversation
|
Auto-sync is disabled for draft pull requests in this repository. Workflows must be run manually. Contributors can view more details about this message here. |
github-actions
Bot
added
Python
| They're all built after the fact from recorded timestamps, so emitting | ||
| them doesn't inflate the wait we're trying to measure. |
| A reservation that induces spilling ought to pass through the Task's | ||
| ``Spilling`` state, but rapidsmpf doesn't report that back to us yet. |
There was a problem hiding this comment.
Remove this too. I'll have a longer issue about tracking spilled operations.
| This emits the following events: | ||
|
|
||
| - queueing | ||
| - allocating (with the Quent Processor for the current thread) |
There was a problem hiding this comment.
Just a note: making a memory reservation is async in python, so this will to be running on the asyncio event loop, concurrently with many other things. I think this is the right way to model this in Quent, but we'll want to double check that. We aren't really consuming CPU resources to run this task.
| ``Spilling`` state, but rapidsmpf doesn't report that back to us yet. | ||
| """ | ||
| quent_processor = quent_ir_execution_context.get_or_declare_processor( | ||
| thread_ident=threading.get_ident(), |
There was a problem hiding this comment.
Maybe this should be a parameter? We're doing this at emit time, which in theory could be running on a different thread than the one that requested the memory reservation. But currently that's not an issue since we always create the reservation and then emit the events.
| def _format_bytes(nbytes: int) -> str: | ||
| """Format a byte count compactly (e.g. ``256.0MiB``) for display.""" | ||
| value = float(nbytes) | ||
| for unit in ("B", "KiB", "MiB", "GiB", "TiB"): | ||
| if abs(value) < 1024.0: | ||
| return f"{value:.0f}B" if unit == "B" else f"{value:.1f}{unit}" | ||
| value /= 1024.0 | ||
| return f"{value:.1f}PiB" |
There was a problem hiding this comment.
This probably isn't necessary. I don't think the size of the reservation should appear in the label...
| purpose | ||
| What the memory is reserved for (e.g. ``"scan"``). Distinguishes | ||
| reservations made by a single operator. |
There was a problem hiding this comment.
I'd like to understand this better: what operators make multiple reservations?
Requiring this might not be smart. Then we'll end up with redundant info in the telemetry like "The Scan operator reserved memory for a scan."
And perhaps this could be an enum rather than an arbitrary string, to make things easier for consumers of the output.
|
|
||
| Quent's data processing domain doesn't declare attributes on the | ||
| ``Allocating`` state yet, so an analyzer built against the current model | ||
| reads the timing but ignores the attributes below. They're still written | ||
| to the event stream, and will show up once the model declares them. |
| def label(self) -> str: | ||
| """A compact description, e.g. ``scan-256.0MiB-device``.""" | ||
| return ( | ||
| f"{self.purpose}-{_format_bytes(self.size_bytes)}-{self.mem_type.lower()}" |
There was a problem hiding this comment.
Remove the size from the label.
1 participant
Description
This updates our emitted Quent telemetry to better track memory reservations. I've added a wrapper around
rapidsmpf.streaming.core.memory_reserve_or_wait.reserve_memorythat includesLike any other Quent event, this can be tied back to an individual Operator (cudf-polars Actor).