Merge pull request #4469 from AI-Hypercomputer:zjiahao/DSA3.2-optimize-reorder-sequence

PiperOrigin-RevId: 949268432

Author: Google-ML-Automation

src/maxtext/utils/max_utils.py

@@ -871,46 +871,43 @@ def reorder_sequence(tensor, cp_size: int, seq_dim: int = 1, to_contiguous: bool
   if seq_len % (cp_size * 2) != 0:
     raise ValueError(f"{tensor.shape=} is not a multiple of {cp_size*2=}")
 
-  # [B, S, H, D]: [B, 2*cp_size, S/2*cp_size, H, D] -> [B, 2, S/2*cp_size, H, D]
-  # [S, B, H, D]: [2*cp_size, S/2*cp_size, B, H, D] -> [2, S/2*cp_size, B, H, D]
+  seq_dim = seq_dim % tensor.ndim
   ori_tensor_shape = tensor.shape
+
+  # Generic transformation: Isolates the target sequence dimension into `2 * cp_size` discrete chunks.
+  # Note: The shape walkthrough below uses [b, s, h, d] with seq_dim=1 as an illustrative example,
+  # but actual dimensions depend on the input tensor (e.g., [s, b, h, d], [b, t, d], etc.):
+  # [b, s, h, d] -> [b, 2*cp_size, group_size, h, d]
   reshaped = tensor.reshape(
       *ori_tensor_shape[:seq_dim],
       2 * cp_size,
       group_size,
       *ori_tensor_shape[seq_dim + 1 :],
   )
 
-  if not to_contiguous:
-    # Create first and second halves
-    first_half = jnp.arange(cp_size)
-    second_half = jnp.arange(2 * cp_size - 1, cp_size - 1, -1)
-
-    # Stack and reshape to interleave
-    src_indices = jnp.stack([first_half, second_half], axis=1).reshape(-1)
+  # Swap target seq_dim with axis 0 to perform slicing/concat easily:
+  # e.g., [b, 2*cp_size, group_size, h, d] -> [2*cp_size, b, group_size, h, d]
+  swapped = jnp.swapaxes(reshaped, 0, seq_dim)
 
+  if not to_contiguous:
+    # Split along axis 0 into halves: each [cp_size, b, group_size, h, d]
+    first_half, second_half = jnp.split(swapped, 2, axis=0)
+    second_half_reversed = second_half[::-1, ...]
+    # Stack along axis 1 to interleave: [cp_size, 2, b, group_size, h, d]
+    stacked = jnp.stack([first_half, second_half_reversed], axis=1)
+    # Reshape squashes the [cp_size, 2] pair dimensions back into [2*cp_size]: [2*cp_size, b, group_size, h, d]
+    permuted = stacked.reshape(2 * cp_size, *swapped.shape[1:])
   else:
-
-    half = cp_size // 2
-
-    # Build the 1st and 2nd groups of contiguous‑pair indices:
-    first_pair = [4 * r for r in range(half)]  # [0, 4, 8, …]
-    second_pair = [4 * r + 2 for r in range(half)]  # [2, 6, 10, …]
-    third_pair = [2 * cp_size - 1 - 4 * r for r in range(half)]  # [2*cp_size-1, 2*cp_size-5, …]
-    fourth_pair = [i - 2 for i in third_pair]  # [2*cp_size-3, 2*cp_size-7, …]
-
-    # Concatenate so each rank’s two indices sit next to each other:
-    # e.g. [0,2, 4,6, …, (2cp‑1),(2cp‑3), …]
-    first_block = first_pair + third_pair
-    second_block = second_pair + fourth_pair
-
-    # Stack into shape (2*cp_size//2, 2) → then flatten → length=2*cp_size
-    src_indices = jnp.stack([jnp.array(first_block), jnp.array(second_block)], axis=1).reshape(-1)
-
-  # One gather and one reshape
-  reordered = jnp.take(reshaped, src_indices, axis=seq_dim)
-
-  # Reshape back to original dimensions
+    # Strided slice extracts every other chunk natively: each [cp_size, b, group_size, h, d]
+    first_half = swapped[0::2, ...]
+    second_half_reversed = swapped[1::2, ...]
+    second_half = second_half_reversed[::-1, ...]
+    # Concatenate along axis 0: [2*cp_size, b, group_size, h, d]
+    permuted = jnp.concatenate([first_half, second_half], axis=0)
+
+  # Swap axis 0 back to seq_dim: e.g., [b, 2*cp_size, group_size, h, d]
+  reordered = jnp.swapaxes(permuted, 0, seq_dim)
+  # Restore original tensor shape: e.g., [b, s, h, d]
   return reordered.reshape(ori_tensor_shape)
 
 
@@ -1245,5 +1242,9 @@ def maybe_pad(inputs, tile_size):
   padding_amount = 0
   if inputs_dim % tile_size:
     padding_amount = tile_size - inputs_dim % tile_size
-    inputs = jax.lax.pad(inputs, jnp.array(0.0, dtype=inputs.dtype), [(0, padding_amount, 0), (0, 0, 0)])
+    inputs = jax.lax.pad(
+        inputs,
+        jnp.array(0.0, dtype=inputs.dtype),
+        [(0, padding_amount, 0), (0, 0, 0)],
+    )
   return inputs, padding_amount

tests/unit/max_utils_test.py

@@ -539,7 +539,9 @@ def test_tpu_checkpointing_no_emergency_calls_jax_init(self, mock_init):
   @mock.patch("maxtext.utils.max_utils.initialize_jax_for_tpu_with_emergency_checkpointing")
   def test_tpu_checkpointing_with_emergency(self, mock_tpu_emergency):
     raw_keys = self._base_keys(
-        enable_checkpointing=True, compile_topology_num_slices=-1, enable_emergency_checkpoint=True
+        enable_checkpointing=True,
+        compile_topology_num_slices=-1,
+        enable_emergency_checkpoint=True,
     )
     max_utils.maybe_initialize_jax_distributed_system(raw_keys)
     mock_tpu_emergency.assert_called_once_with(raw_keys)
@@ -551,5 +553,108 @@ def test_tpu_no_checkpointing_does_not_call_jax_init(self, mock_init):
     mock_init.assert_not_called()
 
 
+class TestReorderSequence(unittest.TestCase):
+  """Tests for reorder_sequence optimization in max_utils.py"""
+
+  def _old_reorder_sequence(self, tensor, cp_size: int, seq_dim: int = 1, to_contiguous: bool = False):
+    """Original gather-based reorder_sequence implementation used as a reference."""
+    if tensor is None:
+      return tensor
+
+    seq_len = tensor.shape[seq_dim]
+    group_size = seq_len // (2 * cp_size)
+
+    if cp_size % 2 != 0:
+      raise ValueError(f"{cp_size=} must be a multiple of 2.")
+
+    if seq_len % (cp_size * 2) != 0:
+      raise ValueError(f"{tensor.shape=} is not a multiple of {cp_size*2=}")
+
+    seq_dim = seq_dim % tensor.ndim
+    ori_tensor_shape = tensor.shape
+    reshaped = tensor.reshape(
+        *ori_tensor_shape[:seq_dim],
+        2 * cp_size,
+        group_size,
+        *ori_tensor_shape[seq_dim + 1 :],
+    )
+
+    if not to_contiguous:
+      first_half = jnp.arange(cp_size)
+      second_half = jnp.arange(2 * cp_size - 1, cp_size - 1, -1)
+      src_indices = jnp.stack([first_half, second_half], axis=1).reshape(-1)
+    else:
+      half = cp_size // 2
+      first_pair = [4 * r for r in range(half)]
+      second_pair = [4 * r + 2 for r in range(half)]
+      third_pair = [2 * cp_size - 1 - 4 * r for r in range(half)]
+      fourth_pair = [i - 2 for i in third_pair]
+      first_block = first_pair + third_pair
+      second_block = second_pair + fourth_pair
+      src_indices = jnp.stack([jnp.array(first_block), jnp.array(second_block)], axis=1).reshape(-1)
+
+    reordered = jnp.take(reshaped, src_indices, axis=seq_dim)
+    return reordered.reshape(ori_tensor_shape)
+
+  def test_reorder_equivalence_sweep(self):
+    key = random.key(42)
+
+    # Sweep configurations
+    cp_sizes = [2, 4, 8]
+    to_contiguous_options = [False, True]
+
+    # Test cases: (shape, seq_dim)
+    test_cases = [
+        ((64,), 0),  # 1D
+        ((2, 128, 4, 8), 1),  # 4D (standard B, S, H, D)
+        ((256, 2, 4, 8), 0),  # 4D (S, B, H, D)
+        ((2, 4, 8, 128), -1),  # Negative seq_dim (last dimension)
+        ((2, 128, 4, 8), -3),  # Negative seq_dim (-3 corresponding to S)
+    ]
+
+    for cp_size in cp_sizes:
+      for to_contiguous in to_contiguous_options:
+        for shape, seq_dim in test_cases:
+          with self.subTest(
+              cp_size=cp_size,
+              to_contiguous=to_contiguous,
+              shape=shape,
+              seq_dim=seq_dim,
+          ):
+            # Generate random input
+            x = random.normal(key, shape)
+
+            # Run old (reference) implementation
+            ref_out = self._old_reorder_sequence(x, cp_size, seq_dim, to_contiguous)
+
+            # Run new (optimized) implementation
+            opt_out = max_utils.reorder_sequence(x, cp_size, seq_dim, to_contiguous)
+
+            # Assert mathematical equivalence
+            self.assertTrue(
+                jnp.allclose(ref_out, opt_out, rtol=1e-5, atol=1e-6),
+                msg=f"Failed for cp_size={cp_size}, to_contiguous={to_contiguous}, shape={shape}, seq_dim={seq_dim}",
+            )
+
+  def test_reorder_roundtrip(self):
+    # If we reorder to load-balanced (to_contiguous=False) and then restore (to_contiguous=True),
+    # we should get back the exact original tensor. Also tests negative seq_dim.
+    key = random.key(123)
+    shape = (2, 128, 4, 8)
+    seq_dim = -3
+    cp_size = 4
+
+    x = random.normal(key, shape)
+
+    # 1. Reorder to load-balanced
+    balanced = max_utils.reorder_sequence(x, cp_size, seq_dim, to_contiguous=False)
+
+    # 2. Restore to contiguous
+    restored = max_utils.reorder_sequence(balanced, cp_size, seq_dim, to_contiguous=True)
+
+    # 3. Assert roundtrip is lossless
+    self.assertTrue(jnp.allclose(x, restored, rtol=1e-5, atol=1e-6))
+
+
 if __name__ == "__main__":
   unittest.main()