Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 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 older supported PyTorch versions.
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