Skip to content

Commit 44a6e2f

Browse files
committed
[Feature] Add prepared TensorDict copy_at writer
1 parent 02e0c6a commit 44a6e2f

4 files changed

Lines changed: 180 additions & 0 deletions

File tree

tensordict/_tensorcollection.pyi

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -992,6 +992,7 @@ class TensorCollection:
992992
*,
993993
fast: bool | None = None,
994994
) -> Self: ...
995+
def prepare_copy_at_(self, dim: int, source: T | None = None) -> Any: ...
995996
def is_empty(self) -> bool: ...
996997
def setdefault(
997998
self, key: NestedKey, default: CompatibleType, inplace: bool = False

tensordict/base.py

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -207,6 +207,114 @@ class _NoDefault(enum.IntEnum):
207207
Self = Any
208208

209209

210+
class _TensorDictCopyAtWriter:
211+
"""Reusable tensor-only writer created by :meth:`prepare_copy_at_`."""
212+
213+
def __init__(
214+
self,
215+
destination: TensorDictBase,
216+
dim: int,
217+
source: TensorDictBase | None = None,
218+
):
219+
self.destination = destination
220+
self.dim = _maybe_correct_neg_dim(dim, destination.batch_size)
221+
self.keys, self.dest_values = destination._items_list(True, True)
222+
self._indexed_dest_values: dict[Any, list[Tensor]] = {}
223+
if any(not isinstance(dest, Tensor) for dest in self.dest_values):
224+
raise TypeError(
225+
"prepare_copy_at_ only supports tensor leaves in the destination "
226+
"tensordict."
227+
)
228+
if source is not None:
229+
self._source_values(source)
230+
231+
def _source_values(self, source: TensorDictBase) -> list[Tensor]:
232+
if not _is_tensor_collection(type(source)):
233+
raise TypeError(
234+
"prepare_copy_at_.copy_ expected a TensorDictBase source, "
235+
f"got {type(source)}."
236+
)
237+
source_keys, source_values = source._items_list(True, True)
238+
if source_keys != self.keys:
239+
_, source_values = source._items_list(
240+
True, True, sorting_keys=self.keys, default=None
241+
)
242+
if len(source_values) != len(self.dest_values):
243+
raise RuntimeError(
244+
"The source and destination tensordicts must have matching leaves."
245+
)
246+
if any(not isinstance(source, Tensor) for source in source_values):
247+
raise TypeError(
248+
"prepare_copy_at_ only supports tensor leaves in the source "
249+
"tensordict."
250+
)
251+
return source_values
252+
253+
def _index(self, index: IndexType) -> tuple:
254+
idx = [slice(None)] * self.destination.batch_dims
255+
idx[self.dim] = index
256+
return tuple(idx)
257+
258+
def _cache_key(self, index: IndexType):
259+
try:
260+
hash(index)
261+
except TypeError:
262+
return None
263+
return index
264+
265+
def _get_indexed_dest_values(self, index: IndexType) -> list[Tensor]:
266+
cache_key = self._cache_key(index)
267+
if cache_key is not None:
268+
cached = self._indexed_dest_values.get(cache_key)
269+
if cached is not None:
270+
return cached
271+
idx = self._index(index)
272+
indexed_dest_values = []
273+
for key, dest in zip(self.keys, self.dest_values):
274+
try:
275+
dest_indexed = dest[idx]
276+
except (IndexError, RuntimeError, TypeError) as err:
277+
raise IndexError(
278+
f"Could not index destination leaf {key} with index {idx}."
279+
) from err
280+
if (
281+
dest_indexed is not dest
282+
and getattr(dest_indexed, "_base", None) is None
283+
):
284+
raise IndexError(
285+
f"Index {idx} does not produce a writable view for key {key}."
286+
)
287+
indexed_dest_values.append(dest_indexed)
288+
if cache_key is not None:
289+
self._indexed_dest_values[cache_key] = indexed_dest_values
290+
return indexed_dest_values
291+
292+
def copy_(
293+
self,
294+
source: TensorDictBase,
295+
index: IndexType,
296+
*,
297+
non_blocking: bool = False,
298+
) -> TensorDictBase:
299+
source_values = self._source_values(source)
300+
indexed_dest_values = self._get_indexed_dest_values(index)
301+
for key, dest_indexed, source_value in zip(
302+
self.keys, indexed_dest_values, source_values
303+
):
304+
if dest_indexed.shape != source_value.shape:
305+
raise RuntimeError(
306+
f"Shape mismatch for key {key}: indexed destination has shape "
307+
f"{dest_indexed.shape}, source has shape {source_value.shape}."
308+
)
309+
if _foreach_copy_ is not None:
310+
copy_fn = _foreach_copy_compiled if is_compiling() else _foreach_copy_
311+
copy_fn(indexed_dest_values, source_values, non_blocking=non_blocking)
312+
else:
313+
for dest, source_value in zip(indexed_dest_values, source_values):
314+
dest.copy_(source_value, non_blocking=non_blocking)
315+
return self.destination
316+
317+
210318
class _BEST_ATTEMPT_INPLACE:
211319
def __bool__(self):
212320
# we use an exception to exit when running `inplace = BEST_ATTEMPT_INPLACE if inplace else False`
@@ -8809,6 +8917,44 @@ def copy_at_(
88098917
)
88108918
return self.update_at_(tensordict, idx, non_blocking=non_blocking)
88118919

8920+
def prepare_copy_at_(
8921+
self, dim: int, source: T | None = None
8922+
) -> _TensorDictCopyAtWriter:
8923+
"""Creates a reusable writer for repeated indexed tensor copies.
8924+
8925+
``prepare_copy_at_`` is intended for hot paths that repeatedly copy
8926+
TensorDicts with the same leaf structure into different indices of the
8927+
same destination, for example collector rollouts. It precomputes the
8928+
destination leaves and, for repeated scalar indices, caches the indexed
8929+
destination tensor views used by the writer's ``copy_(source, index)``
8930+
method.
8931+
8932+
The writer is stricter than :meth:`~tensordict.TensorDictBase.update_at_`:
8933+
it only supports tensor leaves and requires the source and destination
8934+
to have matching leaves. Use ``copy_at_`` for one-off optimized copies,
8935+
or ``update_at_`` when the general update semantics are required.
8936+
8937+
Args:
8938+
dim (int): batch dimension of ``self`` along which ``index`` will
8939+
select the destination slice for each copy.
8940+
source (TensorDictBase, optional): if provided, validates at
8941+
preparation time that the source structure is compatible with
8942+
``self``.
8943+
8944+
Returns:
8945+
A lightweight writer with a ``copy_(source, index)`` method that
8946+
copies ``source`` tensor leaves into ``self`` at ``index`` along
8947+
``dim``.
8948+
8949+
Examples:
8950+
>>> dest = TensorDict({"x": torch.zeros(3, 2)}, batch_size=[3, 2])
8951+
>>> src = TensorDict({"x": torch.ones(3)}, batch_size=[3])
8952+
>>> writer = dest.prepare_copy_at_(dim=1, source=src)
8953+
>>> writer.copy_(src, index=0)
8954+
>>> assert (dest[:, 0] == src).all()
8955+
"""
8956+
return _TensorDictCopyAtWriter(self, dim=dim, source=source)
8957+
88128958
def is_empty(self) -> bool:
88138959
"""Checks if the tensordict contains any leaf."""
88148960
for _ in self.keys(True, True):

tensordict/tensorclass.pyi

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1037,6 +1037,7 @@ class TensorClass:
10371037
*,
10381038
fast: bool | None = None,
10391039
) -> Self: ...
1040+
def prepare_copy_at_(self, dim: int, source: T | None = None) -> Any: ...
10401041
def is_empty(self) -> bool: ...
10411042
def setdefault(
10421043
self, key: NestedKey, default: CompatibleType, inplace: bool = False

test/test_tensordict.py

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9315,6 +9315,38 @@ def test_copy_at_fast_transition(self, td_name, device):
93159315
td.copy_at_(newdata, slice(1, None, 2), fast=True)
93169316
assert td.get("val").tolist() == [0] * 10
93179317

9318+
def test_prepare_copy_at_(self, td_name, device):
9319+
td = TensorDict(
9320+
{
9321+
"a": torch.zeros(4, 3, 5, device=device),
9322+
"b": TensorDict(
9323+
{"c": torch.zeros(4, 3, 2, device=device)},
9324+
batch_size=[4, 3],
9325+
device=device,
9326+
),
9327+
},
9328+
batch_size=[4, 3],
9329+
device=device,
9330+
)
9331+
td0 = TensorDict(
9332+
{
9333+
"a": torch.ones(4, 5, device=device),
9334+
"b": TensorDict(
9335+
{"c": torch.ones(4, 2, device=device)},
9336+
batch_size=[4],
9337+
device=device,
9338+
),
9339+
},
9340+
batch_size=[4],
9341+
device=device,
9342+
)
9343+
writer = td.prepare_copy_at_(dim=1, source=td0)
9344+
writer.copy_(td0, index=0)
9345+
writer.copy_(td0.clone().zero_(), index=2)
9346+
assert (td[:, 0] == td0).all()
9347+
assert (td[:, 1] == 0).all()
9348+
assert (td[:, 2] == 0).all()
9349+
93189350
# This is needed because update in lazy permute/view etc does not behave correctly when
93199351
# legacy is False. When these classes will be deprecated, we can just remove the decorator
93209352
@set_lazy_legacy(True)

0 commit comments

Comments
 (0)