convert stride_per_key_per_rank to tensor inside KJT (#2959)

Summary:

# context
* this diff is part of the "variable-batch KJT refactoring" project ([doc](https://fburl.com/gdoc/svfysfai))
* previously the `stride_per_key_per_rank` variable is `List[List[int]] | None` which can't be handled correctly in PT2 IR (torch.export)
* this change makes the KJT class variable `_stride_per_key_per_rank` as `torch.IntTensor | None` so it would be compatible with PT2 IR.

# equivalency
* to check if `self._stride_per_key_per_rank` is `None`
this logic is used to differentiate variable_batch case, and should have the same behavior after this diff
* to use `self._stride_per_key_per_rank` as `List[List[int]]`
most of the callsite use the function to get the list: `def stride_per_key_per_rank(self) -> List[List[int]]:`, and this function is modified to covert the `torch.IntTensor` to list as ` _stride_per_key_per_rank.tolist()`, the results should be the same

NOTE: this `self. _stride_per_key_per_rank.tolist()` tensor should always be on CPU since it's effective the meta data of a KJT. For generic torch APIs like `.to(...)`, `record_stream()`, etc. should in general avoid altering this variable.

Differential Revision: D74366343

Author: TroyGarden

torchrec/sparse/jagged_tensor.py

@@ -1096,13 +1096,15 @@ def _maybe_compute_stride_kjt(
     stride: Optional[int],
     lengths: Optional[torch.Tensor],
     offsets: Optional[torch.Tensor],
-    stride_per_key_per_rank: Optional[List[List[int]]],
+    stride_per_key_per_rank: Optional[torch.IntTensor],
 ) -> int:
     if stride is None:
         if len(keys) == 0:
             stride = 0
-        elif stride_per_key_per_rank is not None and len(stride_per_key_per_rank) > 0:
-            stride = max([sum(s) for s in stride_per_key_per_rank])
+        elif (
+            stride_per_key_per_rank is not None and stride_per_key_per_rank.numel() > 0
+        ):
+            stride = int(stride_per_key_per_rank.sum(dim=1).max().item())
         elif offsets is not None and offsets.numel() > 0:
             stride = (offsets.numel() - 1) // len(keys)
         elif lengths is not None:
@@ -1668,14 +1670,15 @@ def _maybe_compute_lengths_offset_per_key(
 
 def _maybe_compute_stride_per_key(
     stride_per_key: Optional[List[int]],
-    stride_per_key_per_rank: Optional[List[List[int]]],
+    stride_per_key_per_rank: Optional[torch.IntTensor],
     stride: Optional[int],
     keys: List[str],
 ) -> Optional[List[int]]:
     if stride_per_key is not None:
         return stride_per_key
     elif stride_per_key_per_rank is not None:
-        return [sum(s) for s in stride_per_key_per_rank]
+        rt: List[int] = stride_per_key_per_rank.sum(dim=1).tolist()
+        return rt
     elif stride is not None:
         return [stride] * len(keys)
     else:
@@ -1766,7 +1769,9 @@ def __init__(
         lengths: Optional[torch.Tensor] = None,
         offsets: Optional[torch.Tensor] = None,
         stride: Optional[int] = None,
-        stride_per_key_per_rank: Optional[List[List[int]]] = None,
+        stride_per_key_per_rank: Optional[
+            Union[torch.IntTensor, List[List[int]]]
+        ] = None,
         # Below exposed to ensure torch.script-able
         stride_per_key: Optional[List[int]] = None,
         length_per_key: Optional[List[int]] = None,
@@ -1788,8 +1793,10 @@ def __init__(
         self._lengths: Optional[torch.Tensor] = lengths
         self._offsets: Optional[torch.Tensor] = offsets
         self._stride: Optional[int] = stride
-        self._stride_per_key_per_rank: Optional[List[List[int]]] = (
-            stride_per_key_per_rank
+        self._stride_per_key_per_rank: Optional[torch.IntTensor] = (
+            torch.IntTensor(stride_per_key_per_rank, device="cpu")
+            if isinstance(stride_per_key_per_rank, list)
+            else stride_per_key_per_rank
         )
         self._stride_per_key: Optional[List[int]] = stride_per_key
         self._length_per_key: Optional[List[int]] = length_per_key
@@ -1816,8 +1823,7 @@ def _init_pt2_checks(self) -> None:
         if self._stride_per_key is not None:
             pt2_checks_all_is_size(self._stride_per_key)
         if self._stride_per_key_per_rank is not None:
-            # pyre-ignore [16]
-            for s in self._stride_per_key_per_rank:
+            for s in self.stride_per_key_per_rank():
                 pt2_checks_all_is_size(s)
 
     @staticmethod
@@ -2028,7 +2034,7 @@ def from_jt_dict(jt_dict: Dict[str, JaggedTensor]) -> "KeyedJaggedTensor":
         kjt_stride, kjt_stride_per_key_per_rank = (
             (stride_per_key[0], None)
             if all(s == stride_per_key[0] for s in stride_per_key)
-            else (None, [[stride] for stride in stride_per_key])
+            else (None, torch.IntTensor(stride_per_key, device="cpu").reshape(-1, 1))
         )
         kjt = KeyedJaggedTensor(
             keys=kjt_keys,
@@ -2193,8 +2199,13 @@ def stride_per_key_per_rank(self) -> List[List[int]]:
         Returns:
             List[List[int]]: stride per key per rank of the KeyedJaggedTensor.
         """
-        stride_per_key_per_rank = self._stride_per_key_per_rank
-        return stride_per_key_per_rank if stride_per_key_per_rank is not None else []
+        # making a local reference to the class variable to make jit.script behave
+        _stride_per_key_per_rank = self._stride_per_key_per_rank
+        return (
+            []
+            if _stride_per_key_per_rank is None
+            else _stride_per_key_per_rank.tolist()
+        )
 
     def variable_stride_per_key(self) -> bool:
         """
@@ -2514,17 +2525,17 @@ def permute(
 
         length_per_key = self.length_per_key()
         permuted_keys: List[str] = []
-        permuted_stride_per_key_per_rank: List[List[int]] = []
         permuted_length_per_key: List[int] = []
         permuted_length_per_key_sum = 0
         for index in indices:
             key = self.keys()[index]
             permuted_keys.append(key)
             permuted_length_per_key.append(length_per_key[index])
-            if self.variable_stride_per_key():
-                permuted_stride_per_key_per_rank.append(
-                    self.stride_per_key_per_rank()[index]
-                )
+        _stride_per_key_per_rank = self._stride_per_key_per_rank
+        if self.variable_stride_per_key() and _stride_per_key_per_rank is not None:
+            permuted_stride_per_key_per_rank = _stride_per_key_per_rank[indices, :]
+        else:
+            permuted_stride_per_key_per_rank = None
 
         permuted_length_per_key_sum = sum(permuted_length_per_key)
         if not torch.jit.is_scripting() and is_non_strict_exporting():
@@ -2576,17 +2587,15 @@ def permute(
                 self.weights_or_none(),
                 permuted_length_per_key_sum,
             )
-        stride_per_key_per_rank = (
-            permuted_stride_per_key_per_rank if self.variable_stride_per_key() else None
-        )
+
         kjt = KeyedJaggedTensor(
             keys=permuted_keys,
             values=permuted_values,
             weights=permuted_weights,
             lengths=permuted_lengths.view(-1),
             offsets=None,
             stride=self._stride,
-            stride_per_key_per_rank=stride_per_key_per_rank,
+            stride_per_key_per_rank=permuted_stride_per_key_per_rank,
             stride_per_key=None,
             length_per_key=permuted_length_per_key if len(permuted_keys) > 0 else None,
             lengths_offset_per_key=None,

torchrec/sparse/tests/keyed_jagged_tensor_benchmark_lib.py

@@ -465,7 +465,7 @@ def bench(
                 keys=kjt.keys(),
                 values=kjt._values,
                 lengths=kjt._lengths,
-                stride_per_key_per_rank=kjt._stride_per_key_per_rank,
+                stride_per_key_per_rank=kjt.stride_per_key_per_rank(),
             )
             vbe_fn_kwargs = fn_kwargs.copy()
             if "kjt" in fn_kwargs:
@@ -490,7 +490,7 @@ def bench(
                 keys=kjt.keys(),
                 values=kjt._values,
                 lengths=kjt._lengths,
-                stride_per_key_per_rank=kjt._stride_per_key_per_rank,
+                stride_per_key_per_rank=kjt.stride_per_key_per_rank(),
             )
             vbe_fn_kwargs = fn_kwargs.copy()
             if "kjt" in fn_kwargs: