import torch import torch.nn as nn import torch.nn.functional as F import math import importlib try: xm = importlib.import_module('torch_xla.core.xla_model') xs = importlib.import_module('torch_xla.distributed.spmd.xla_sharding') except ImportError: xm = None xs = None class Rotary3D(nn.Module): def __init__(self, dim, base=100): super().__init__() assert dim % 16 == 0, "Embedding dim must be divisible by 16" # Embedding dimensions must align precisely with dim // num_heads self.x_dim = (6 * dim) // 16 self.y_dim = (6 * dim) // 16 self.t_dim = dim - self.x_dim - self.y_dim # Precompute inverse frequencies self.register_buffer('inv_freq_x', 1.0 / (base ** (torch.arange(0, self.x_dim, 2).float() / self.x_dim))) self.register_buffer('inv_freq_y', 1.0 / (base ** (torch.arange(0, self.y_dim, 2).float() / self.y_dim))) self.register_buffer('inv_freq_t', 1.0 / (base ** (torch.arange(0, self.t_dim, 2).float() / self.t_dim))) def forward(self, x, pos): """ x: [batch, nh, seq_len, head_dim] pos: [batch, seq_len, 3] integer positions along (x, y, t) """ B, nh, T, hs = x.shape assert pos.shape[-1] == 3, "Position tensor must have shape [batch, seq_len, 3]" # Compute embeddings directly to match `hs` dim_total = hs assert dim_total % 2 == 0, "head_dim (hs) must be divisible by 2 for rotary embedding." # Positional dimensions expanded explicitly dtype = self.inv_freq_x.dtype pos_x = pos[..., 0].to(dtype) # [B, T] pos_y = pos[..., 1].to(dtype) # [B, T] pos_t = pos[..., 2].to(dtype) # [B, T] # Generate embeddings for x, y, t and combine freqs_x = torch.einsum('bt,f -> btf', pos_x, self.inv_freq_x) freqs_y = torch.einsum('bt,f -> btf', pos_y, self.inv_freq_y) freqs_t = torch.einsum('bt,f -> btf', pos_t, self.inv_freq_t) # Concatenate embeddings and match dimensions exactly freq_combined = torch.cat([freqs_x, freqs_y, freqs_t], dim=-1) # Cos and Sin embedding, reshape to match x exactly cos_emb = freq_combined.cos().unsqueeze(1) # [B, 1, T, hs/2] sin_emb = freq_combined.sin().unsqueeze(1) # [B, 1, T, hs/2] # Split embedding dimension for rotation x1, x2 = x[..., :hs//2], x[..., hs//2:] # Ensure exact dimensional matching x_rotated = torch.cat([ x1 * cos_emb - x2 * sin_emb, x1 * sin_emb + x2 * cos_emb ], dim=-1) return x_rotated class PSIAttentionLayer(nn.Module): def __init__(self, config): super().__init__() assert config.n_embd % config.n_head == 0 # key, query, value projections for all heads, but in a batch self.c_attn = nn.Linear(config.n_embd, 3 * config.n_embd, bias=config.bias) # output projection self.c_proj = nn.Linear(config.n_embd, config.n_embd, bias=config.bias) # regularization self.attn_dropout = nn.Dropout(config.dropout) self.resid_dropout = nn.Dropout(config.dropout) self.n_head = config.n_head self.n_embd = config.n_embd self.dropout = config.dropout # positional embedding self.rope = Rotary3D(config.n_embd // config.n_head) # check if we are using causal attention if config.attention_mask == "causal": self.is_causal = True else: self.is_causal = False # check if GPU Flash Attention is available self.flash = hasattr(torch.nn.functional, 'scaled_dot_product_attention') # check if we are running on TPU try: # Use local import to avoid conflict if global xm is None and to check TPU specifically for this flag xm_local = importlib.import_module('torch_xla.core.xla_model') self.tpu = True except ImportError: self.tpu = False # Apply XLA sharding for model parallelism xla_device_available = False if xm is not None: try: device_kind = xm.xla_device_kind() if device_kind is not None: xla_device_available = True except RuntimeError: pass @torch.compiler.disable def emplace_kv(self, T, k_cache, v_cache, k, v): # torch.compile doesn't play well with this op (5x slowdown) # so we insert a graph break and copy eagerly k_cache[:,:,-T:].copy_(k) v_cache[:,:,-T:].copy_(v) return k_cache, v_cache def forward(self, x, pos, k_cache=None, v_cache=None, return_kv=False, inplace_kv=False, mask=None): B, T, C = x.size() # batch size, sequence length, embedding dimensionality (n_embd) # calculate query, key, values for all heads in batch and move head forward to be the batch dim q, k, v = self.c_attn(x).split(self.n_embd, dim=2) k = k.view(B, T, self.n_head, C // self.n_head).transpose(1, 2) # (B, nh, T, hs) q = q.view(B, T, self.n_head, C // self.n_head).transpose(1, 2) # (B, nh, T, hs) v = v.view(B, T, self.n_head, C // self.n_head).transpose(1, 2) # (B, nh, T, hs) # Apply rotary positional embedding k = self.rope(k, pos) q = self.rope(q, pos) if inplace_kv and k_cache is not None and v_cache is not None: # assign into kv cache in-place k, v = self.emplace_kv(T, k_cache, v_cache, k, v) else: # append cached keys and values with new keys and values if k_cache is not None: k = torch.cat((k_cache, k), dim=2) if v_cache is not None: v = torch.cat((v_cache, v), dim=2) # Apply attention if self.tpu: # (1) flash_attention = importlib.import_module('torch_xla.experimental.custom_kernel.flash_attention') q_norm = q / math.sqrt(k.size(-1)) y = flash_attention( q_norm, k, v, causal=True, partition_spec=('fsdp', None, None, None)) # (2) # y = torch.nn.functional.scaled_dot_product_attention( # q, k, v, # # dropout_p=self.dropout if self.training else 0, # # attn_mask=None if mask is None else mask.to(q.dtype), # is_causal=True # ) elif self.flash: # efficient attention using Flash Attention CUDA kernels L, S = q.size(-2), k.size(-2) is_causal = self.is_causal and mask is None # is_causal doesn't work when not square, so replace with a manual mask if needed if is_causal and L < S: if L > 1: # if L=1, just use no mask mask = torch.ones(L, S, dtype=q.dtype, device=q.device) mask.masked_fill_(mask.to(torch.bool).triu(S-L+1), float('-inf')) is_causal = False y = torch.nn.functional.scaled_dot_product_attention( q, k, v, dropout_p=self.dropout if self.training else 0, attn_mask=None if mask is None else mask.to(q.dtype), is_causal=is_causal ) else: # manual implementation of attention att = torch.einsum('bnsh,bnkh->bnsk', q, k) * (1.0 / math.sqrt(k.size(-1))) # apply mask, or use causal if default if mask is not None: att = att + mask elif self.is_causal: L, S = q.size(-2), k.size(-2) mask = torch.ones(1, 1, L, S).triu(S-L+1).to(dtype=torch.bool).to(x.device) att.masked_fill_(mask, float('-inf')) # upcast to float32 for numerical stability, as per llama implementation att = F.softmax(att, dim=-1, dtype=torch.float32).to(q.dtype) att = self.attn_dropout(att) # multiply attention weights with values to get output y = torch.einsum('bnsk,bnkh->bnsh', att, v) y = y.transpose(1, 2).contiguous().view(B, T, C) # re-assemble all head outputs side by side # output projection y = self.resid_dropout(self.c_proj(y)) # return key and value caches if requested if return_kv: return y, k, v return y def kv_cache_forward(self, x, pos, k_cache=None, v_cache=None): B, T, C = x.size() # batch size, sequence length, embedding dimensionality (n_embd) # calculate query, key, values for all heads in batch and move head forward to be the batch dim q, k, v = self.c_attn(x).split(self.n_embd, dim=2) k = k.view(B, T, self.n_head, C // self.n_head).transpose(1, 2) # (B, nh, T, hs) q = q.view(B, T, self.n_head, C // self.n_head).transpose(1, 2) # (B, nh, T, hs) v = v.view(B, T, self.n_head, C // self.n_head).transpose(1, 2) # (B, nh, T, hs) # Apply rotary positional embedding (before concat) k = self.rope(k, pos) q = self.rope(q, pos) # append cached keys and values with new keys and values if k_cache is not None: k = torch.cat((k_cache, k), dim=2) if v_cache is not None: v = torch.cat((v_cache, v), dim=2) # manual implementation of attention att = (q @ k.transpose(-2, -1)) * (1.0 / math.sqrt(k.size(-1))) att = F.softmax(att, dim=-1) att = self.attn_dropout(att) y = att @ v # (B, nh, T, T) x (B, nh, T, hs) -> (B, nh, T, hs) y = y.transpose(1, 2).contiguous().view(B, T, C) # re-assemble all head outputs side by side # output projection y = self.resid_dropout(self.c_proj(y)) return y, k, v class MLP(nn.Module): def __init__(self, config): super().__init__() self.c_fc = nn.Linear(config.n_embd, 4 * config.n_embd, bias=config.bias) self.gelu = nn.GELU() self.c_proj = nn.Linear(4 * config.n_embd, config.n_embd, bias=config.bias) self.dropout = nn.Dropout(config.dropout) # Apply XLA sharding for model parallelism xla_device_available = False if xm is not None: try: device_kind = xm.xla_device_kind() if device_kind is not None: xla_device_available = True except RuntimeError: pass if xla_device_available and xs is not None and xs.global_mesh() is not None: mesh = xs.global_mesh() if mesh.mesh_shape[1] > 1: # If the 'model' axis has size > 1 xs.mark_sharding(self.c_fc.weight, mesh, (1, 0)) if self.c_fc.bias is not None: xs.mark_sharding(self.c_fc.bias, mesh, (1,)) print(f"MLP: Applied MP sharding to c_fc {mesh.mesh_shape} spec weight(1,0), bias(1,)") xs.mark_sharding(self.c_proj.weight, mesh, (0, 1)) if self.c_proj.bias is not None: xs.mark_sharding(self.c_proj.bias, mesh, (0,)) print(f"MLP: Applied MP sharding to c_proj {mesh.mesh_shape} spec weight(0,1), bias(0,)") def forward(self, x, spmd_mesh=None): x = self.c_fc(x) x = self.gelu(x) if spmd_mesh is not None: xs.mark_sharding(x, spmd_mesh, (('dcn', 'data'), None, 'model')) x = self.c_proj(x) x = self.dropout(x) if spmd_mesh is not None: xs.mark_sharding(x, spmd_mesh, (('dcn', 'data'), None, 'model')) return x class RMSNorm(nn.Module): """ Root Mean Square Normalization """ def __init__(self, dim: int, weight: bool = True, bias: bool = False, eps: float = 1e-5): # whl super().__init__() self.eps = eps self.weight = nn.Parameter(torch.ones(dim)) if weight else None def _norm(self, x): return x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps) def forward(self, x): output = self._norm(x.float()).type_as(x) if self.weight is not None: return output * self.weight return output class PSIBlock(nn.Module): def __init__(self, config): super().__init__() self.ln_1 = RMSNorm(config.n_embd, bias=config.bias) self.attn = PSIAttentionLayer(config) self.ln_2 = RMSNorm(config.n_embd, bias=config.bias) self.mlp = MLP(config) def forward(self, x, pos, k_cache=None, v_cache=None, return_kv=False, inplace_kv=False, spmd_mesh=None, mask=None): # If we are given a key and value cache, we will use the pre-computed values to minimize # the computation cost if return_kv: # Pass the key and value cache to the attention layer, obtain new key and value caches x_attn, k, v = self.attn(self.ln_1(x), pos, k_cache=k_cache, v_cache=v_cache, return_kv=True, inplace_kv=inplace_kv, mask=mask) x = x + x_attn x = x + self.mlp(self.ln_2(x)) return x, k, v # Else we proceed with the regular forward pass x = x + self.attn(self.ln_1(x), pos, k_cache=k_cache, v_cache=v_cache, inplace_kv=inplace_kv, mask=mask) x = x + self.mlp(self.ln_2(x)) return x class PartitionedEmbedding(nn.Module): def __init__(self, num_embeddings, embedding_dim, partition_size=65536): super().__init__() self.num_embeddings = num_embeddings self.embedding_dim = embedding_dim self.partition_size = partition_size self.num_partitions = (num_embeddings + partition_size - 1) // partition_size self.embedding_layers = nn.ModuleList() for i in range(self.num_partitions): start_idx = i * self.partition_size end_idx = min(start_idx + self.partition_size, num_embeddings) vocab_size = end_idx - start_idx self.embedding_layers.append(nn.Embedding(vocab_size, embedding_dim)) def forward(self, input_ids): partition_ids = input_ids // self.partition_size relative_ids = input_ids % self.partition_size output = torch.zeros(*input_ids.shape, self.embedding_dim, device=input_ids.device, dtype=self.embedding_layers[0].weight.dtype) for i in range(self.num_partitions): mask = (partition_ids == i) if mask.any(): partition_input_ids = relative_ids[mask] embedded = self.embedding_layers[i](partition_input_ids) output[mask] = embedded return output class PartitionedLinear(nn.Module): def __init__(self, in_features, out_features, partition_size=65536, bias=False): super().__init__() self.in_features = in_features self.out_features = out_features self.partition_size = partition_size self.num_partitions = (out_features + partition_size - 1) // partition_size self.linear_layers = nn.ModuleList() for i in range(self.num_partitions): start_idx = i * self.partition_size end_idx = min(start_idx + self.partition_size, out_features) output_partition_size = end_idx - start_idx self.linear_layers.append(nn.Linear(in_features, output_partition_size, bias=bias)) def forward(self, input): outputs = [layer(input) for layer in self.linear_layers] return torch.cat(outputs, dim=-1)