Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 4 additions & 0 deletions sdm/models/base/icl.py
Original file line number Diff line number Diff line change
Expand Up @@ -354,6 +354,10 @@ def fit(

self._cache = cache.freeze()

# Prepare the stream before compiled predict to avoid recompilation.
if x.is_cuda and x.device not in self._transfer_streams:
self._transfer_streams[x.device] = torch.cuda.Stream(x.device)

def predict(
self,
x: Tensor | TableTensor, # [..., R, D]
Expand Down
54 changes: 44 additions & 10 deletions sdm/models/base/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

import abc
from collections.abc import Iterator, Sequence
from functools import partial
from typing import ClassVar, Generic, Self, TypeVar

import torch
Expand Down Expand Up @@ -91,6 +92,39 @@ def _validate_feature_stype(self, x: TableTensor) -> None:
)


def _copy_to_device(
tensor: Tensor,
device: torch.device,
stream: torch.cuda.Stream,
) -> Tensor:
"""Keep copy allocations separate from the parent graph's buffer reuse."""
with torch.cuda.stream(stream):
return tensor.to(device=device, non_blocking=True)


# Keep eager imports working on PyTorch 2.7; this API was added in 2.8.
if hasattr(torch.compiler, "nested_compile_region"):
_copy_to_device = torch.compiler.nested_compile_region(_copy_to_device)
Comment thread
aw471 marked this conversation as resolved.
Resolved


def _transfer(
item: T,
device: torch.device,
stream: torch.cuda.Stream,
) -> T:
"""Move a tensor or tensor container onto the prefetch stream."""
if not torch.compiler.is_compiling():
with torch.cuda.stream(stream):
return item.to(device=device, non_blocking=True) # type: ignore
if isinstance(item, DeviceMixin):
return item._apply_tensor(
partial(_transfer, device=device, stream=stream)
)
if item.device == device:
return item
return _copy_to_device(item, device, stream) # type: ignore


class _Prefetcher(Generic[T]):
def __init__(
self,
Expand All @@ -112,11 +146,11 @@ def __enter__(self) -> Self:
if self._transfer_stream is not None:
self._compute_stream = torch.cuda.current_stream(self._device)
try:
with torch.cuda.stream(self._transfer_stream):
self._next = self._items[0].to( # type: ignore
device=self._device,
non_blocking=True,
)
self._next = _transfer(
item=self._items[0],
device=self._device,
stream=self._transfer_stream,
)
except BaseException:
self._transfer_stream.synchronize()
raise
Expand Down Expand Up @@ -153,11 +187,11 @@ def __iter__(self) -> Iterator[T]:

if self._next is not None:
if self._transfer_stream is not None:
with torch.cuda.stream(self._transfer_stream):
self._next = self._next.to( # type: ignore
device=self._device,
non_blocking=True,
)
self._next = _transfer(
item=self._next,
device=self._device,
stream=self._transfer_stream,
)
else:
self._next = self._next.to(self._device) # type: ignore

Expand Down
Loading