# Copyright (c) 2026 Bytedance Ltd. and/or its affiliate # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Sequence-parallel tensor ops. Every op is a no-op when Ulysses is disabled (single-GPU) and otherwise delegates to Open-VeOmni, imported lazily so single-GPU inference carries no dependency on it. """ import torch import torch.nn.functional as F from .state import get_parallel_state def gather_seq_scatter_heads(x, seq_dim, head_dim, unpadded_dim_size=0): """All-to-all: gather the sequence dim, scatter the head dim.""" if not get_parallel_state().ulysses_enabled: return x from veomni.distributed.sequence_parallel import gather_seq_scatter_heads as _f return _f(x, seq_dim=seq_dim, head_dim=head_dim, unpadded_dim_size=unpadded_dim_size) def gather_heads_scatter_seq(x, head_dim, seq_dim): """All-to-all: gather the head dim, scatter the sequence dim.""" if not get_parallel_state().ulysses_enabled: return x from veomni.distributed.sequence_parallel import gather_heads_scatter_seq as _f return _f(x, head_dim=head_dim, seq_dim=seq_dim) def slice_input_tensor(x, dim): """Keep only this rank's slice of `x` along `dim`.""" if not get_parallel_state().ulysses_enabled: return x from veomni.distributed.sequence_parallel import slice_input_tensor as _f return _f(x, dim=dim) def slice_input_tensor_scale_grad(x, dim): """`slice_input_tensor` variant used inside autograd-tracked code paths.""" if not get_parallel_state().ulysses_enabled: return x from veomni.distributed.sequence_parallel import slice_input_tensor_scale_grad as _f return _f(x, dim=dim) def gather_outputs(x, gather_dim, padding_dim=None, unpad_dim_size=None, group=None): """Gather a sequence-sharded tensor back to its full length.""" if not get_parallel_state().ulysses_enabled: return x from veomni.distributed.sequence_parallel import gather_outputs as _f return _f(x, gather_dim=gather_dim, padding_dim=padding_dim, unpad_dim_size=unpad_dim_size) def padding_tensor_for_seqeunce_parallel(x, dim): """Pad `x` along `dim` so its size is divisible by the Ulysses world size.""" if not get_parallel_state().ulysses_enabled: return x from veomni.distributed.sequence_parallel.utils import ( padding_tensor_for_seqeunce_parallel as _f, ) return _f(x, dim=dim) def pad_tensor(x, dim, padding_size, padding_value=0): """Append `padding_size` entries along `dim` (F.pad based, low peak memory).""" pad_config = [0, 0] * x.ndim pad_config[(x.ndim - 1 - dim) * 2 + 1] = padding_size return F.pad(x, pad_config, value=padding_value) def unpad_tensor(x, dim, padding_size): """Inverse of `pad_tensor`: drop the last `padding_size` entries along `dim`.""" slc = [slice(None)] * x.ndim slc[dim] = slice(0, -padding_size) return x[slc] def gen_cu_seqlens_for_cross_attn(q_len, batch_seqlens_q, batch_seqlens_k, device="cpu"): """cu_seqlens / max_seqlens for cross-attention under Ulysses sequence parallel. Each rank holds a contiguous ``q_len / sp_world`` slice of the query sequence; this maps the per-sample query/key lengths onto that local slice. """ ps = get_parallel_state() sp_world = ps.ulysses_size rank = ps.ulysses_rank rank_q_len = (q_len + ((sp_world - (q_len % sp_world)) % sp_world)) // sp_world start = rank_q_len * rank end = min(q_len, start + rank_q_len) offset = 0 cu_seqlens_q = [start] index = [] max_seqlen_q = -1 max_seqlen_k = -1 for i, length in enumerate(batch_seqlens_q): offset = min(offset + length, end) if offset <= start: continue cu_seqlens_q.append(offset) index.append(i) max_seqlen_q = max(max_seqlen_q, cu_seqlens_q[-1] - cu_seqlens_q[-2]) max_seqlen_k = max(max_seqlen_k, batch_seqlens_k[i]) if offset >= end: break cu_seqlens_q = torch.tensor(cu_seqlens_q, dtype=torch.int32, device=device) max_seqlen_q = torch.tensor(max_seqlen_q, device=device) cu_seqlens_q -= start cu_seqlens_k = torch.zeros(len(batch_seqlens_k) + 1, dtype=torch.int32, device=device) cu_seqlens_k[1:] = torch.tensor(batch_seqlens_k, dtype=torch.int32, device=device).cumsum(dim=0) cu_seqlens_k = cu_seqlens_k[index[0] : index[-1] + 2] max_seqlen_k = torch.tensor(max_seqlen_k, device=device) return cu_seqlens_k, cu_seqlens_q, max_seqlen_k, max_seqlen_q, rank_q_len