fla / build /torch-cuda /modules /activations.py
kernels-bot's picture
Uploaded using `kernel-builder`.
e19323e verified
Raw
History Blame
35.1 kB
# Copyright (c) 2023-2026, Songlin Yang, Yu Zhang, Zhiyuan Li
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.
# For a list of all contributors, visit:
# https://github.com/fla-org/flash-linear-attention/graphs/contributors
"""Fused activation kernels.
The kernels address their inputs through the row stride instead of assuming a fully contiguous buffer.
An inner-contiguous input — such as one half of ``x.chunk(2, dim=-1)`` — is therefore read in place, sparing the extra
``.contiguous()`` copy (and its memory traffic) that a plain flat element-wise kernel would force on every call.
"""
import torch
import torch.nn.functional as F
import triton
import triton.language as tl
from ..modules.backends import dispatch
from ..ops.utils.op import exp, log
from ..utils import IS_AMD, autocast_custom_bwd, autocast_custom_fwd, autotune_cache_kwargs, input_guard
NUM_WARPS_AUTOTUNE = [1, 2, 4, 8, 16] if IS_AMD else [1, 2, 4, 8, 16, 32]
def _get_stride(x: torch.Tensor) -> int:
"""Get the row stride for viewing a tensor as 2D (num_rows, D) where D = shape[-1].
Returns stride(-2) if the tensor is at least 2D, or 0 for 1D tensors.
The caller must ensure the tensor is "inner-contiguous" (stride(-1) == 1 and
higher dims are contiguous relative to dim -2) before using this value.
"""
if x.ndim < 2:
return 0
return x.stride(-2)
def _is_inner_contiguous(x: torch.Tensor) -> bool:
"""Check if a tensor can be safely viewed as 2D (num_rows, D) with row stride = stride(-2).
This holds when stride(-1) == 1 and all dimensions above -2 are contiguous
with respect to the dimension below them.
"""
ndim = x.ndim
if ndim < 2:
return True
if x.stride(-1) != 1:
return False
if ndim == 2:
# 2D: any layout with stride(-1)==1 is valid (can view as (T, D))
return True
if ndim == 3:
# 3D (B, T, D): stride should be (T*D, D, 1)
return x.stride(0) == x.stride(-2) * x.shape[-2]
if ndim == 4:
# 4D (B, H, T, D): stride should be (H*T*D, T*D, D, 1)
if x.stride(1) != x.stride(-2) * x.shape[-2]:
return False
return x.stride(0) == x.stride(1) * x.shape[1]
# 5D+ fallback to loop
expected = x.stride(-2) * x.shape[-2]
for d in range(ndim - 3, -1, -1):
if x.stride(d) != expected:
return False
expected *= x.shape[d]
return True
def _ensure_inner_contiguous(x: torch.Tensor) -> torch.Tensor:
"""Make the tensor inner-contiguous if it isn't already."""
if _is_inner_contiguous(x):
return x
return x.contiguous()
def _alloc_output(x: torch.Tensor, contiguous: bool = False) -> torch.Tensor:
"""Allocate the output: a fresh contiguous buffer, or ``empty_like`` otherwise.
``empty_like`` keeps the input's memory format only when it is dense; a non-dense
strided view (e.g. a ``chunk`` slice) falls back to contiguous, not the input stride.
"""
if contiguous:
return x.new_empty(x.shape)
return torch.empty_like(x)
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def sigmoid_fwd_kernel(
x, y,
stride_x_row,
stride_y_row,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_y = tl.sigmoid(b_x)
tl.store(y + row * stride_y_row + col, b_y.to(y.dtype.element_ty), mask=mask)
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def sigmoid_bwd_kernel(
x, dy, dx,
stride_x_row,
stride_dy_row,
stride_dx_row,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_dy = tl.load(dy + row * stride_dy_row + col, mask=mask, other=0.).to(tl.float32)
b_s = tl.sigmoid(b_x)
b_dx = b_dy * b_s * (1.0 - b_s)
tl.store(dx + row * stride_dx_row + col, b_dx.to(dx.dtype.element_ty), mask=mask)
@dispatch('modules')
def sigmoid_fwd(x: torch.Tensor, output_contiguous: bool = False) -> torch.Tensor:
x = _ensure_inner_contiguous(x)
T, D = x.numel(), x.shape[-1]
y = _alloc_output(x, output_contiguous)
sigmoid_fwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
T=T,
D=D,
)
return y
@dispatch('modules')
def sigmoid_bwd(x: torch.Tensor, dy: torch.Tensor, output_contiguous: bool = False) -> torch.Tensor:
x = _ensure_inner_contiguous(x)
dy = _ensure_inner_contiguous(dy)
T, D = x.numel(), x.shape[-1]
dx = _alloc_output(x, output_contiguous)
sigmoid_bwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
dy=dy,
dx=dx,
stride_x_row=_get_stride(x),
stride_dy_row=_get_stride(dy),
stride_dx_row=_get_stride(dx),
T=T,
D=D,
)
return dx
class SigmoidFunction(torch.autograd.Function):
@staticmethod
@input_guard(no_guard_contiguous=True)
def forward(ctx, x):
ctx.save_for_backward(x)
return sigmoid_fwd(x)
@staticmethod
@input_guard(no_guard_contiguous=True)
def backward(ctx, dout):
x, = ctx.saved_tensors
return sigmoid_bwd(x, dout)
sigmoid = SigmoidFunction.apply
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def logsigmoid_fwd_kernel(
x,
y,
stride_x_row,
stride_y_row,
temperature,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_m = tl.minimum(0., b_x)
b_z = 1. + exp(-tl.abs(b_x))
b_y = (b_m - log(b_z)) / temperature
tl.store(y + row * stride_y_row + col, b_y.to(y.dtype.element_ty), mask=mask)
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def logsigmoid_bwd_kernel(
x,
dy,
dx,
stride_x_row,
stride_dy_row,
stride_dx_row,
temperature,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_dy = tl.load(dy + row * stride_dy_row + col, mask=mask, other=0.).to(tl.float32)
b_dx = b_dy * ((1. - tl.sigmoid(b_x)) / temperature)
tl.store(dx + row * stride_dx_row + col, b_dx.to(dx.dtype.element_ty), mask=mask)
@dispatch('modules')
def logsigmoid_fwd(x: torch.Tensor, temperature: float = 1., output_contiguous: bool = False) -> torch.Tensor:
x = _ensure_inner_contiguous(x)
T, D = x.numel(), x.shape[-1]
y = _alloc_output(x, output_contiguous)
logsigmoid_fwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
temperature=temperature,
T=T,
D=D,
)
return y
@dispatch('modules')
def logsigmoid_bwd(
x: torch.Tensor,
dy: torch.Tensor,
temperature: float = 1.,
output_contiguous: bool = False,
) -> torch.Tensor:
x = _ensure_inner_contiguous(x)
dy = _ensure_inner_contiguous(dy)
T, D = x.numel(), x.shape[-1]
dx = _alloc_output(x, output_contiguous)
logsigmoid_bwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
dy=dy,
dx=dx,
stride_x_row=_get_stride(x),
stride_dy_row=_get_stride(dy),
stride_dx_row=_get_stride(dx),
temperature=temperature,
T=T,
D=D,
)
return dx
class LogSigmoidFunction(torch.autograd.Function):
@staticmethod
@input_guard(no_guard_contiguous=True)
def forward(ctx, x, temperature):
ctx.save_for_backward(x)
ctx.temperature = temperature
return logsigmoid_fwd(x, temperature)
@staticmethod
@input_guard(no_guard_contiguous=True)
def backward(ctx, dy):
x, = ctx.saved_tensors
return logsigmoid_bwd(x, dy, ctx.temperature), None
def logsigmoid(x: torch.Tensor, temperature: float = 1.) -> torch.Tensor:
return LogSigmoidFunction.apply(x, temperature)
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def swish_fwd_kernel(
x, y,
stride_x_row,
stride_y_row,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_y = b_x * tl.sigmoid(b_x)
tl.store(y + row * stride_y_row + col, b_y.to(y.dtype.element_ty), mask=mask)
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def swish_bwd_kernel(
x, dy, dx,
stride_x_row,
stride_dy_row,
stride_dx_row,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_dy = tl.load(dy + row * stride_dy_row + col, mask=mask, other=0.).to(tl.float32)
b_s = tl.sigmoid(b_x)
b_dx = b_dy * b_s * (1.0 + b_x * (1.0 - b_s))
tl.store(dx + row * stride_dx_row + col, b_dx.to(dx.dtype.element_ty), mask=mask)
@dispatch('modules')
def swish_fwd(x: torch.Tensor, output_contiguous: bool = False) -> torch.Tensor:
x = _ensure_inner_contiguous(x)
T, D = x.numel(), x.shape[-1]
y = _alloc_output(x, output_contiguous)
swish_fwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
T=T,
D=D,
)
return y
@dispatch('modules')
def swish_bwd(x: torch.Tensor, dy: torch.Tensor, output_contiguous: bool = False) -> torch.Tensor:
x = _ensure_inner_contiguous(x)
dy = _ensure_inner_contiguous(dy)
T, D = x.numel(), x.shape[-1]
dx = _alloc_output(x, output_contiguous)
swish_bwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
dy=dy,
dx=dx,
stride_x_row=_get_stride(x),
stride_dy_row=_get_stride(dy),
stride_dx_row=_get_stride(dx),
T=T,
D=D,
)
return dx
class SwishFunction(torch.autograd.Function):
@staticmethod
@input_guard(no_guard_contiguous=True)
def forward(ctx, x):
ctx.save_for_backward(x)
return swish_fwd(x)
@staticmethod
@input_guard(no_guard_contiguous=True)
def backward(ctx, dout):
x, = ctx.saved_tensors
return swish_bwd(x, dout)
swish = SwishFunction.apply
# 1/sqrt(2*pi)-> 0.3989423
# 1/sqrt(2) -> 0.70710678
# sqrt(2/pi) -> 0.79788456
# this function is tanh approximation of gelu
# actual gelu is:
# x * 0.5 * (1.0 + torch.erf(x * 0.70710678))
@torch.compile
def bias_gelu(y, bias):
x = bias + y
return (x * 0.5 * (1.0 + torch.tanh(0.79788456 * x * (1 + 0.044715 * x * x)))).to(dtype=y.dtype)
# gradient of tanh approximation of gelu
# gradient of actual gelu is:
# 0.5 * (1. + torch.erf(x * 0.70710678)) + 0.3989423 * x * torch.exp(-0.5 * x * x)
@torch.compile
def bias_gelu_bwd(g, y, bias):
"""Assume that y has shape (B, D=D) and bias has shape (D)"""
x = bias + y
tanh_out = torch.tanh(0.79788456 * x * (1 + 0.044715 * x * x))
# sqrt(2/pi) * 3 * 0.044715 -> 0.1070322243
ff = 0.5 * x * ((1 - tanh_out * tanh_out) * (0.79788456 + 0.1070322243 * x * x)) + 0.5 * (
1 + tanh_out
)
grad_y = ff * g
return grad_y.to(dtype=y.dtype), grad_y.sum(dim=(0), dtype=bias.dtype)
class GeLUFunction(torch.autograd.Function):
@staticmethod
# bias is an optional argument
def forward(ctx, input, bias):
ctx.save_for_backward(input, bias)
return bias_gelu(input, bias)
@staticmethod
def backward(ctx, grad_output):
input, bias = ctx.saved_tensors
return bias_gelu_bwd(grad_output, input, bias)
bias_gelu_impl = GeLUFunction.apply
# this function is tanh approximation of gelu
# actual gelu is:
# x * 0.5 * (1.0 + torch.erf(x * 0.70710678))
@dispatch('modules')
@torch.compile
def gelu_fwd(x):
return (x * 0.5 * (1.0 + torch.tanh(0.79788456 * x * (1 + 0.044715 * x * x)))).to(dtype=x.dtype)
# gradient of tanh approximation of gelu
# gradient of actual gelu is:
# 0.5 * (1. + torch.erf(x * 0.70710678)) + 0.3989423 * x * torch.exp(-0.5 * x * x)
@dispatch('modules')
@torch.compile
def gelu_bwd(g, x):
tanh_out = torch.tanh(0.79788456 * x * (1 + 0.044715 * x * x))
# sqrt(2/pi) * 3 * 0.044715 -> 0.1070322243
ff = 0.5 * x * ((1 - tanh_out * tanh_out) * (0.79788456 + 0.1070322243 * x * x)) + 0.5 * (
1 + tanh_out
)
return (ff * g).to(dtype=x.dtype)
class FastGeLUFunction(torch.autograd.Function):
@staticmethod
# bias is an optional argument
def forward(ctx, input):
ctx.save_for_backward(input)
return gelu_fwd(input)
@staticmethod
def backward(ctx, grad_output):
(input,) = ctx.saved_tensors
tmp = gelu_bwd(grad_output, input)
return tmp
fast_gelu_impl = FastGeLUFunction.apply
@torch.compile
def relu_bwd(g, x):
return torch.where(x >= 0, g, 0.0).to(dtype=x.dtype)
@dispatch('modules')
@torch.compile
def sqrelu_fwd(x):
r = F.relu(x.float())
return (r * r).to(dtype=x.dtype)
@dispatch('modules')
@torch.compile
def sqrelu_bwd(g, x):
return (2.0 * g * F.relu(x.float())).to(dtype=x.dtype)
class SquaredReLUFunction(torch.autograd.Function):
@staticmethod
def forward(ctx, input):
ctx.save_for_backward(input)
return sqrelu_fwd(input)
@staticmethod
def backward(ctx, grad_output):
input, = ctx.saved_tensors
return sqrelu_bwd(grad_output, input)
sqrelu = SquaredReLUFunction.apply
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def swiglu_fwd_kernel(
x, y, z,
stride_x_row,
stride_y_row,
stride_z_row,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_y = tl.load(y + row * stride_y_row + col, mask=mask, other=0.).to(tl.float32)
b_z = b_x * tl.sigmoid(b_x) * b_y
tl.store(z + row * stride_z_row + col, b_z.to(z.dtype.element_ty), mask=mask)
@triton.heuristics({
'HAS_WEIGHT': lambda args: args['z'] is not None,
})
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def swiglu_fwdbwd_kernel(
x, y, g, dx, dy, z,
stride_x_row,
stride_y_row,
stride_g_row,
stride_dx_row,
stride_dy_row,
stride_z_row,
T,
D: tl.constexpr,
B: tl.constexpr,
HAS_WEIGHT: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_y = tl.load(y + row * stride_y_row + col, mask=mask, other=0.).to(tl.float32)
b_g = tl.load(g + row * stride_g_row + col, mask=mask, other=0.).to(tl.float32)
b_s = tl.sigmoid(b_x)
b_xs = b_x * b_s
b_dx = b_g * b_s * (1.0 + b_x * (1.0 - b_s)) * b_y
b_dy = b_g * b_xs
tl.store(dx + row * stride_dx_row + col, b_dx.to(dx.dtype.element_ty), mask=mask)
tl.store(dy + row * stride_dy_row + col, b_dy.to(dy.dtype.element_ty), mask=mask)
if HAS_WEIGHT:
b_z = b_xs * b_y
tl.store(z + row * stride_z_row + col, b_z.to(z.dtype.element_ty), mask=mask)
@dispatch('modules')
def swiglu_fwd(x: torch.Tensor, y: torch.Tensor, output_contiguous: bool = False) -> torch.Tensor:
assert x.shape == y.shape, f"swiglu_fwd: shape mismatch x={x.shape} y={y.shape}"
x = _ensure_inner_contiguous(x)
y = _ensure_inner_contiguous(y)
T, D = x.numel(), x.shape[-1]
z = _alloc_output(x, output_contiguous)
swiglu_fwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
z=z,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
stride_z_row=_get_stride(z),
T=T,
D=D,
)
return z
@dispatch('modules')
def swiglu_fwdbwd(
x: torch.Tensor,
y: torch.Tensor,
g: torch.Tensor,
use_weight: bool = False,
output_contiguous: bool = False,
):
assert x.shape == y.shape == g.shape, f"swiglu_fwdbwd: shape mismatch x={x.shape} y={y.shape} g={g.shape}"
x = _ensure_inner_contiguous(x)
y = _ensure_inner_contiguous(y)
g = _ensure_inner_contiguous(g)
T, D = x.numel(), x.shape[-1]
dx = _alloc_output(x, output_contiguous)
dy = _alloc_output(y, output_contiguous)
if use_weight:
z = _alloc_output(x, output_contiguous)
else:
z = None
swiglu_fwdbwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
g=g,
dx=dx,
dy=dy,
z=z,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
stride_g_row=_get_stride(g),
stride_dx_row=_get_stride(dx),
stride_dy_row=_get_stride(dy),
stride_z_row=_get_stride(z) if z is not None else 0,
T=T,
D=D,
)
if use_weight:
return dx, dy, z
return dx, dy
class SwiGLUFunction(torch.autograd.Function):
r"""
Swish-Gated Linear Unit (SwiGLU) function.
.. math::
\text{SwiGLU}(x, y) = swish(x) * y = \frac{x}{1 + \exp(-x)} * y
"""
@staticmethod
@input_guard(no_guard_contiguous=True)
def forward(ctx, x, y):
ctx.save_for_backward(x, y)
return swiglu_fwd(x, y)
@staticmethod
@input_guard(no_guard_contiguous=True)
def backward(ctx, dout):
x, y = ctx.saved_tensors
return swiglu_fwdbwd(x, y, dout)
class SwiGLULinearFunction(torch.autograd.Function):
r"""
Swish-Gated Linear Unit (SwiGLU) function followed by a linear transformation.
.. math::
\text{SwiGLULinear}(x, y, W, b) = (swish(x) * y) W + b
This simple wrap discards the intermediate results of SwiGLU(x, y) to save memory.
"""
@staticmethod
@input_guard(no_guard_contiguous=True)
@autocast_custom_fwd
def forward(ctx, x, y, weight, bias):
z = swiglu_fwd(x, y, output_contiguous=True)
out = F.linear(z, weight, bias)
ctx.save_for_backward(x, y, weight)
ctx.linear_bias_is_none = bias is None
return out
@staticmethod
@input_guard(no_guard_contiguous=True)
@autocast_custom_bwd
def backward(ctx, dout, *args):
x, y, weight = ctx.saved_tensors
dout = dout.reshape(-1, dout.shape[-1])
dz = F.linear(dout, weight.t()).view_as(x)
dx, dy, z = swiglu_fwdbwd(x, y, dz, use_weight=True, output_contiguous=True)
dlinear_weight = torch.einsum("bo,bi->oi", dout, z.reshape(-1, z.shape[-1]))
dlinear_bias = None if ctx.linear_bias_is_none else dout.sum(0)
return dx, dy, dlinear_weight, dlinear_bias
swiglu = SwiGLUFunction.apply
@dispatch('modules')
def swiglu_linear(x, y, weight, bias):
return SwiGLULinearFunction.apply(x, y, weight, bias)
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def sigmoidglu_fwd_kernel(
x, y, z,
stride_x_row,
stride_y_row,
stride_z_row,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_y = tl.load(y + row * stride_y_row + col, mask=mask, other=0.).to(tl.float32)
b_z = tl.sigmoid(b_x) * b_y
tl.store(z + row * stride_z_row + col, b_z.to(z.dtype.element_ty), mask=mask)
@triton.heuristics({
'HAS_WEIGHT': lambda args: args['z'] is not None,
})
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def sigmoidglu_fwdbwd_kernel(
x, y, g, dx, dy, z,
stride_x_row,
stride_y_row,
stride_g_row,
stride_dx_row,
stride_dy_row,
stride_z_row,
T,
D: tl.constexpr,
B: tl.constexpr,
HAS_WEIGHT: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_y = tl.load(y + row * stride_y_row + col, mask=mask, other=0.).to(tl.float32)
b_g = tl.load(g + row * stride_g_row + col, mask=mask, other=0.).to(tl.float32)
b_s = tl.sigmoid(b_x)
b_dx = b_g * b_s * (1.0 - b_s) * b_y
b_dy = b_g * b_s
tl.store(dx + row * stride_dx_row + col, b_dx.to(dx.dtype.element_ty), mask=mask)
tl.store(dy + row * stride_dy_row + col, b_dy.to(dy.dtype.element_ty), mask=mask)
if HAS_WEIGHT:
b_z = b_s * b_y
tl.store(z + row * stride_z_row + col, b_z.to(z.dtype.element_ty), mask=mask)
@torch.compiler.disable
def sigmoidglu_fwd(x: torch.Tensor, y: torch.Tensor, output_contiguous: bool = False) -> torch.Tensor:
assert x.shape == y.shape, f"sigmoidglu_fwd: shape mismatch x={x.shape} y={y.shape}"
x = _ensure_inner_contiguous(x)
y = _ensure_inner_contiguous(y)
T, D = x.numel(), x.shape[-1]
z = _alloc_output(x, output_contiguous)
sigmoidglu_fwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
z=z,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
stride_z_row=_get_stride(z),
T=T,
D=D,
)
return z
@torch.compiler.disable
def sigmoidglu_fwdbwd(
x: torch.Tensor,
y: torch.Tensor,
g: torch.Tensor,
use_weight: bool = False,
output_contiguous: bool = False,
):
assert x.shape == y.shape == g.shape, f"sigmoidglu_fwdbwd: shape mismatch x={x.shape} y={y.shape} g={g.shape}"
x = _ensure_inner_contiguous(x)
y = _ensure_inner_contiguous(y)
g = _ensure_inner_contiguous(g)
T, D = x.numel(), x.shape[-1]
dx = _alloc_output(x, output_contiguous)
dy = _alloc_output(y, output_contiguous)
if use_weight:
z = _alloc_output(x, output_contiguous)
else:
z = None
sigmoidglu_fwdbwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
g=g,
dx=dx,
dy=dy,
z=z,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
stride_g_row=_get_stride(g),
stride_dx_row=_get_stride(dx),
stride_dy_row=_get_stride(dy),
stride_z_row=_get_stride(z) if z is not None else 0,
T=T,
D=D,
)
if use_weight:
return dx, dy, z
return dx, dy
class SigmoidGLUFunction(torch.autograd.Function):
r"""
Sigmoid-Gated Linear Unit (SigmoidGLU) function.
.. math::
\text{SigmoidGLU}(x, y) = sigmoid(x) * y = \frac{1}{1 + \exp(-x)} * y
"""
@staticmethod
@input_guard(no_guard_contiguous=True)
def forward(ctx, x, y):
ctx.save_for_backward(x, y)
return sigmoidglu_fwd(x, y)
@staticmethod
@input_guard(no_guard_contiguous=True)
def backward(ctx, dout):
x, y = ctx.saved_tensors
return sigmoidglu_fwdbwd(x, y, dout)
class SigmoidGLULinearFunction(torch.autograd.Function):
r"""
Sigmoid-Gated Linear Unit (SigmoidGLU) function followed by a linear transformation.
.. math::
\text{SigmoidGLULinear}(x, y, W, b) = (sigmoid(x) * y) W + b
This simple wrap discards the intermediate results of SigmoidGLU(x, y) to save memory.
"""
@staticmethod
@input_guard(no_guard_contiguous=True)
@autocast_custom_fwd
def forward(ctx, x, y, weight, bias):
z = sigmoidglu_fwd(x, y, output_contiguous=True)
out = F.linear(z, weight, bias)
ctx.save_for_backward(x, y, weight)
ctx.linear_bias_is_none = bias is None
return out
@staticmethod
@input_guard(no_guard_contiguous=True)
@autocast_custom_bwd
def backward(ctx, dout, *args):
x, y, weight = ctx.saved_tensors
dout = dout.reshape(-1, dout.shape[-1])
dz = F.linear(dout, weight.t()).view_as(x)
dx, dy, z = sigmoidglu_fwdbwd(x, y, dz, use_weight=True, output_contiguous=True)
dlinear_weight = torch.einsum("bo,bi->oi", dout, z.reshape(-1, z.shape[-1]))
dlinear_bias = None if ctx.linear_bias_is_none else dout.sum(0)
return dx, dy, dlinear_weight, dlinear_bias
sigmoidglu = SigmoidGLUFunction.apply
sigmoidglu_linear = SigmoidGLULinearFunction.apply
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def powglu_fwd_kernel(
x, y, z,
stride_x_row,
stride_y_row,
stride_z_row,
m,
T,
D: tl.constexpr,
B: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_y = tl.load(y + row * stride_y_row + col, mask=mask, other=0.).to(tl.float32)
b_s = tl.sigmoid(b_x)
b_pos = b_x > 0
# feed only positive lanes to log/sqrt; masked lanes give x**p = 1 and are dropped by the where
b_xp = tl.where(b_pos, b_x, 1.0)
b_sqrt = tl.sqrt(b_xp)
b_p = m / (b_sqrt + 1.0)
b_pow = exp(b_p * log(b_xp))
b_g = tl.where(b_pos, b_pow * b_s, b_x * b_s)
b_z = b_g * b_y
tl.store(z + row * stride_z_row + col, b_z.to(z.dtype.element_ty), mask=mask)
@triton.heuristics({
'HAS_WEIGHT': lambda args: args['z'] is not None,
})
@triton.autotune(
configs=[
triton.Config({'B': bs}, num_warps=num_warps)
for bs in [512, 1024, 2048, 4096, 8192]
for num_warps in NUM_WARPS_AUTOTUNE
],
key=['D'],
**autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def powglu_fwdbwd_kernel(
x, y, g, dx, dy, z,
stride_x_row,
stride_y_row,
stride_g_row,
stride_dx_row,
stride_dy_row,
stride_z_row,
m,
T,
D: tl.constexpr,
B: tl.constexpr,
HAS_WEIGHT: tl.constexpr,
):
i_n = tl.program_id(0).to(tl.int64)
offs = i_n * B + tl.arange(0, B)
mask = offs < T
row = offs // D
col = offs % D
b_x = tl.load(x + row * stride_x_row + col, mask=mask, other=0.).to(tl.float32)
b_y = tl.load(y + row * stride_y_row + col, mask=mask, other=0.).to(tl.float32)
b_g = tl.load(g + row * stride_g_row + col, mask=mask, other=0.).to(tl.float32)
b_s = tl.sigmoid(b_x)
b_pos = b_x > 0
b_xp = tl.where(b_pos, b_x, 1.0)
b_sqrt = tl.sqrt(b_xp)
b_ln = log(b_xp)
b_p = m / (b_sqrt + 1.0)
b_pow = exp(b_p * b_ln)
b_gate_pos = b_pow * b_s
# d/dx of the exponent term: p' = -m / (2*sqrt(x)*(sqrt(x)+1)**2)
b_pprime = -m / (2.0 * b_sqrt * (b_sqrt + 1.0) * (b_sqrt + 1.0))
b_dgate_pos = b_gate_pos * (b_pprime * b_ln + b_p / b_xp + 1.0 - b_s)
b_gate_neg = b_x * b_s
b_dgate_neg = b_s * (1.0 + b_x * (1.0 - b_s))
b_gate = tl.where(b_pos, b_gate_pos, b_gate_neg)
b_dgate = tl.where(b_pos, b_dgate_pos, b_dgate_neg)
b_dx = b_g * b_y * b_dgate
b_dy = b_g * b_gate
tl.store(dx + row * stride_dx_row + col, b_dx.to(dx.dtype.element_ty), mask=mask)
tl.store(dy + row * stride_dy_row + col, b_dy.to(dy.dtype.element_ty), mask=mask)
if HAS_WEIGHT:
b_z = b_gate * b_y
tl.store(z + row * stride_z_row + col, b_z.to(z.dtype.element_ty), mask=mask)
@dispatch('modules')
def powglu_fwd(x: torch.Tensor, y: torch.Tensor, power: float = 3.0, output_contiguous: bool = False) -> torch.Tensor:
assert x.shape == y.shape, f"powglu_fwd: shape mismatch x={x.shape} y={y.shape}"
x = _ensure_inner_contiguous(x)
y = _ensure_inner_contiguous(y)
T, D = x.numel(), x.shape[-1]
z = _alloc_output(x, output_contiguous)
powglu_fwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
z=z,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
stride_z_row=_get_stride(z),
m=power,
T=T,
D=D,
)
return z
@dispatch('modules')
def powglu_fwdbwd(
x: torch.Tensor,
y: torch.Tensor,
g: torch.Tensor,
power: float = 3.0,
use_weight: bool = False,
output_contiguous: bool = False,
):
assert x.shape == y.shape == g.shape, f"powglu_fwdbwd: shape mismatch x={x.shape} y={y.shape} g={g.shape}"
x = _ensure_inner_contiguous(x)
y = _ensure_inner_contiguous(y)
g = _ensure_inner_contiguous(g)
T, D = x.numel(), x.shape[-1]
dx = _alloc_output(x, output_contiguous)
dy = _alloc_output(y, output_contiguous)
if use_weight:
z = _alloc_output(x, output_contiguous)
else:
z = None
powglu_fwdbwd_kernel[lambda meta: (triton.cdiv(T, meta['B']),)](
x=x,
y=y,
g=g,
dx=dx,
dy=dy,
z=z,
stride_x_row=_get_stride(x),
stride_y_row=_get_stride(y),
stride_g_row=_get_stride(g),
stride_dx_row=_get_stride(dx),
stride_dy_row=_get_stride(dy),
stride_z_row=_get_stride(z) if z is not None else 0,
m=power,
T=T,
D=D,
)
if use_weight:
return dx, dy, z
return dx, dy
class PowGLUFunction(torch.autograd.Function):
r"""
Power-Gated Linear Unit (PowGLU) function.
.. math::
\text{PowGLU}(x, y) = g(x) * y,\quad
g(x) = \begin{cases} x^{power/(\sqrt{x}+1)}\,\sigma(x) & x > 0 \\ x\,\sigma(x) & x \le 0 \end{cases}
For ``x <= 0`` the gate reduces to swish, matching SwiGLU; for large ``x > 0`` it saturates instead of
growing, replacing SwiGLU's quadratic amplification with bounded growth (Power Linear Unit, arXiv:2605.25704).
"""
@staticmethod
@input_guard(no_guard_contiguous=True)
def forward(ctx, x, y, power):
ctx.save_for_backward(x, y)
ctx.power = power
return powglu_fwd(x, y, power)
@staticmethod
@input_guard(no_guard_contiguous=True)
def backward(ctx, dout):
x, y = ctx.saved_tensors
dx, dy = powglu_fwdbwd(x, y, dout, ctx.power)
return dx, dy, None
class PowGLULinearFunction(torch.autograd.Function):
r"""
Power-Gated Linear Unit (PowGLU) function followed by a linear transformation.
.. math::
\text{PowGLULinear}(x, y, W, b) = (g(x) * y) W + b
This simple wrap discards the intermediate results of PowGLU(x, y) to save memory.
"""
@staticmethod
@input_guard(no_guard_contiguous=True)
@autocast_custom_fwd
def forward(ctx, x, y, weight, bias, power):
z = powglu_fwd(x, y, power, output_contiguous=True)
out = F.linear(z, weight, bias)
ctx.save_for_backward(x, y, weight)
ctx.linear_bias_is_none = bias is None
ctx.power = power
return out
@staticmethod
@input_guard(no_guard_contiguous=True)
@autocast_custom_bwd
def backward(ctx, dout, *args):
x, y, weight = ctx.saved_tensors
dout = dout.reshape(-1, dout.shape[-1])
dz = F.linear(dout, weight.t()).view_as(x)
dx, dy, z = powglu_fwdbwd(x, y, dz, ctx.power, use_weight=True, output_contiguous=True)
dlinear_weight = torch.einsum("bo,bi->oi", dout, z.reshape(-1, z.shape[-1]))
dlinear_bias = None if ctx.linear_bias_is_none else dout.sum(0)
return dx, dy, dlinear_weight, dlinear_bias, None
def powglu(x: torch.Tensor, y: torch.Tensor, power: float = 3.0) -> torch.Tensor:
return PowGLUFunction.apply(x, y, power)
@dispatch('modules')
def powglu_linear(
x: torch.Tensor,
y: torch.Tensor,
weight: torch.Tensor,
bias: torch.Tensor,
power: float = 3.0,
) -> torch.Tensor:
return PowGLULinearFunction.apply(x, y, weight, bias, power)
ACT2FN = {
'relu': F.relu,
'sigmoid': sigmoid,
'logsigmoid': logsigmoid,
'silu': swish,
'swish': swish,
'sqrelu': sqrelu,
'gelu': fast_gelu_impl,
'bias_gelu': bias_gelu_impl,
}