Kernels:

kernels-community
/

fla

Trusted publisher

fla / build /torch-cuda /ops /utils /pooling.py

Uploaded using `kernel-builder`.

e0d830e verified 5 days ago

6.67 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

import torch
import triton
import triton.language as tl

from ...ops.utils.index import prepare_chunk_indices
from ...ops.utils.op import unflatten_program_id
from ...utils import autocast_custom_bwd, autocast_custom_fwd, autotune_cache_kwargs, input_guard


@triton.heuristics({
    'IS_VARLEN': lambda args: args['cu_seqlens'] is not None,
})
@triton.autotune(
    configs=[
        triton.Config({'BD': BD}, num_warps=num_warps)
        for BD in [16, 32, 64, 128]
        for num_warps in [1, 2, 4, 8]
    ],
    key=['BT'],
    **autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def mean_pooling_fwd_kernel(
    x,
    o,
    cu_seqlens,
    chunk_indices,
    T,
    H: tl.constexpr,
    D: tl.constexpr,
    BT: tl.constexpr,
    BD: tl.constexpr,
    IS_VARLEN: tl.constexpr,
):
    i_d, i_t = unflatten_program_id(tl.cdiv(D, BD))
    i_bh = tl.program_id(1).to(tl.int64)
    i_b, i_h = i_bh // H, i_bh % H
    if IS_VARLEN:
        i_tg = i_t
        i_n, i_t = tl.load(chunk_indices + i_t * 2).to(tl.int32), tl.load(chunk_indices + i_t * 2 + 1).to(tl.int64)
        bos, eos = tl.load(cu_seqlens + i_n).to(tl.int64), tl.load(cu_seqlens + i_n + 1).to(tl.int64)
        T = eos - bos
        NT = tl.cdiv(T, BT)
    else:
        NT = tl.cdiv(T, BT)
        i_tg = i_b * NT + i_t
        bos, eos = i_b * T, i_b * T + T

    o_t = i_t * BT + tl.arange(0, BT)
    o_d = i_d * BD + tl.arange(0, BD)
    m_x = (o_t[:, None] < T) & (o_d[None, :] < D)
    p_x = x + (bos * H + i_h) * D + o_t[:, None] * (H*D) + o_d[None, :]
    p_o = o + (i_tg * H + i_h) * D + o_d
    # [BT, BD]
    b_x = tl.load(p_x, mask=m_x, other=0.0).to(tl.float32)
    # [BD]
    b_o = tl.sum(b_x, axis=0) / min(BT, T - i_t * BT)
    tl.store(p_o, b_o.to(p_o.dtype.element_ty), mask=o_d < D)


@triton.heuristics({
    'IS_VARLEN': lambda args: args['cu_seqlens'] is not None,
})
@triton.autotune(
    configs=[
        triton.Config({'BD': BD}, num_warps=num_warps)
        for BD in [16, 32, 64, 128]
        for num_warps in [1, 2, 4, 8]
    ],
    key=['BT'],
    **autotune_cache_kwargs,
)
@triton.jit(do_not_specialize=['T'])
def mean_pooling_bwd_kernel(
    do,
    dx,
    cu_seqlens,
    chunk_indices,
    T,
    H: tl.constexpr,
    D: tl.constexpr,
    BT: tl.constexpr,
    BD: tl.constexpr,
    IS_VARLEN: tl.constexpr,
):
    i_d, i_t = unflatten_program_id(tl.cdiv(D, BD))
    i_bh = tl.program_id(1).to(tl.int64)
    i_b, i_h = i_bh // H, i_bh % H
    if IS_VARLEN:
        i_tg = i_t
        i_n, i_t = tl.load(chunk_indices + i_t * 2).to(tl.int32), tl.load(chunk_indices + i_t * 2 + 1).to(tl.int64)
        bos, eos = tl.load(cu_seqlens + i_n).to(tl.int64), tl.load(cu_seqlens + i_n + 1).to(tl.int64)
        T = eos - bos
        NT = tl.cdiv(T, BT)
    else:
        NT = tl.cdiv(T, BT)
        i_tg = i_b * NT + i_t
        bos, eos = i_b * T, i_b * T + T

    o_t = i_t * BT + tl.arange(0, BT)
    o_d = i_d * BD + tl.arange(0, BD)
    m_x = (o_t[:, None] < T) & (o_d[None, :] < D)
    p_dx = dx + (bos * H + i_h) * D + o_t[:, None] * (H*D) + o_d[None, :]
    p_do = do + (i_tg * H + i_h) * D + o_d
    # [BD]
    b_do = tl.load(p_do, mask=o_d < D, other=0.0).to(tl.float32)
    # [BT, BD]
    b_dx = b_do / tl.full((BT,), min(BT, T - i_t * BT), dtype=tl.float32)[:, None]
    tl.store(p_dx, b_dx.to(p_dx.dtype.element_ty), mask=m_x)


def mean_pooling_fwd(
    x: torch.Tensor,
    chunk_size: int,
    cu_seqlens: torch.LongTensor | None = None,
    chunk_indices: torch.LongTensor | None = None,
) -> torch.Tensor:
    B, T, H, D = x.shape
    BT = chunk_size
    if chunk_indices is None and cu_seqlens is not None:
        chunk_indices = prepare_chunk_indices(cu_seqlens, chunk_size)
    NT = triton.cdiv(T, BT) if cu_seqlens is None else len(chunk_indices)

    o = x.new_empty(B, NT, H, D)
    def grid(meta): return (triton.cdiv(D, meta['BD']) * NT, B * H)
    mean_pooling_fwd_kernel[grid](
        x,
        o,
        cu_seqlens,
        chunk_indices,
        T=T,
        H=H,
        D=D,
        BT=BT,
    )
    return o


def mean_pooling_bwd(
    do: torch.Tensor,
    batch_size: int,
    seq_len: int,
    chunk_size: int,
    cu_seqlens: torch.LongTensor | None = None,
    chunk_indices: torch.LongTensor | None = None,
) -> torch.Tensor:
    B, T, H, D = batch_size, seq_len, *do.shape[-2:]
    BT = chunk_size
    if chunk_indices is None and cu_seqlens is not None:
        chunk_indices = prepare_chunk_indices(cu_seqlens, chunk_size)
    NT = triton.cdiv(T, BT) if cu_seqlens is None else len(chunk_indices)

    dx = do.new_empty(B, T, H, D)
    def grid(meta): return (triton.cdiv(D, meta['BD']) * NT, B * H)
    mean_pooling_bwd_kernel[grid](
        do,
        dx,
        cu_seqlens,
        chunk_indices,
        T=T,
        H=H,
        D=D,
        BT=BT,
    )
    return dx


class MeanPoolingFunction(torch.autograd.Function):

    @staticmethod
    @input_guard
    @autocast_custom_fwd
    def forward(
        ctx,
        x: torch.Tensor,
        chunk_size: int,
        cu_seqlens: torch.LongTensor | None = None,
    ) -> torch.Tensor:
        o = mean_pooling_fwd(x, chunk_size, cu_seqlens)
        ctx.batch_size = x.shape[0]
        ctx.seq_len = x.shape[1]
        ctx.chunk_size = chunk_size
        ctx.cu_seqlens = cu_seqlens
        return o

    @staticmethod
    @input_guard
    @autocast_custom_bwd
    def backward(
        ctx, do,
    ) -> tuple[torch.Tensor, None, None]:
        batch_size = ctx.batch_size
        seq_len = ctx.seq_len
        chunk_size = ctx.chunk_size
        cu_seqlens = ctx.cu_seqlens
        dx = mean_pooling_bwd(do, batch_size, seq_len, chunk_size, cu_seqlens)
        return dx, None, None


def mean_pooling(
    x: torch.Tensor,
    chunk_size: int,
    cu_seqlens: torch.LongTensor | None = None,
    **kwargs,
) -> torch.Tensor:
    if 'head_first' in kwargs:
        raise DeprecationWarning(
            "head_first has been removed. Inputs must be in `[B, T, H, ...]` format.",
        )
    if cu_seqlens is not None:
        if x.shape[0] != 1:
            raise ValueError(
                f"The batch size is expected to be 1 rather than {x.shape[0]} when using `cu_seqlens`."
                f"Please flatten variable-length inputs before processing.",
            )
    o = MeanPoolingFunction.apply(x, chunk_size, cu_seqlens)
    return o

	# 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

	import torch
	import triton
	import triton.language as tl

	from ...ops.utils.index import prepare_chunk_indices
	from ...ops.utils.op import unflatten_program_id
	from ...utils import autocast_custom_bwd, autocast_custom_fwd, autotune_cache_kwargs, input_guard


	@triton.heuristics({
	'IS_VARLEN': lambda args: args['cu_seqlens'] is not None,
	})
	@triton.autotune(
	configs=[
	triton.Config({'BD': BD}, num_warps=num_warps)
	for BD in [16, 32, 64, 128]
	for num_warps in [1, 2, 4, 8]
	],
	key=['BT'],
	**autotune_cache_kwargs,
	)
	@triton.jit(do_not_specialize=['T'])
	def mean_pooling_fwd_kernel(
	x,
	o,
	cu_seqlens,
	chunk_indices,
	T,
	H: tl.constexpr,
	D: tl.constexpr,
	BT: tl.constexpr,
	BD: tl.constexpr,
	IS_VARLEN: tl.constexpr,
	):
	i_d, i_t = unflatten_program_id(tl.cdiv(D, BD))
	i_bh = tl.program_id(1).to(tl.int64)
	i_b, i_h = i_bh // H, i_bh % H
	if IS_VARLEN:
	i_tg = i_t
	i_n, i_t = tl.load(chunk_indices + i_t * 2).to(tl.int32), tl.load(chunk_indices + i_t * 2 + 1).to(tl.int64)
	bos, eos = tl.load(cu_seqlens + i_n).to(tl.int64), tl.load(cu_seqlens + i_n + 1).to(tl.int64)
	T = eos - bos
	NT = tl.cdiv(T, BT)
	else:
	NT = tl.cdiv(T, BT)
	i_tg = i_b * NT + i_t
	bos, eos = i_b * T, i_b * T + T

	o_t = i_t * BT + tl.arange(0, BT)
	o_d = i_d * BD + tl.arange(0, BD)
	m_x = (o_t[:, None] < T) & (o_d[None, :] < D)
	p_x = x + (bos * H + i_h) * D + o_t[:, None] * (H*D) + o_d[None, :]
	p_o = o + (i_tg * H + i_h) * D + o_d
	# [BT, BD]
	b_x = tl.load(p_x, mask=m_x, other=0.0).to(tl.float32)
	# [BD]
	b_o = tl.sum(b_x, axis=0) / min(BT, T - i_t * BT)
	tl.store(p_o, b_o.to(p_o.dtype.element_ty), mask=o_d < D)


	@triton.heuristics({
	'IS_VARLEN': lambda args: args['cu_seqlens'] is not None,
	})
	@triton.autotune(
	configs=[
	triton.Config({'BD': BD}, num_warps=num_warps)
	for BD in [16, 32, 64, 128]
	for num_warps in [1, 2, 4, 8]
	],
	key=['BT'],
	**autotune_cache_kwargs,
	)
	@triton.jit(do_not_specialize=['T'])
	def mean_pooling_bwd_kernel(
	do,
	dx,
	cu_seqlens,
	chunk_indices,
	T,
	H: tl.constexpr,
	D: tl.constexpr,
	BT: tl.constexpr,
	BD: tl.constexpr,
	IS_VARLEN: tl.constexpr,
	):
	i_d, i_t = unflatten_program_id(tl.cdiv(D, BD))
	i_bh = tl.program_id(1).to(tl.int64)
	i_b, i_h = i_bh // H, i_bh % H
	if IS_VARLEN:
	i_tg = i_t
	i_n, i_t = tl.load(chunk_indices + i_t * 2).to(tl.int32), tl.load(chunk_indices + i_t * 2 + 1).to(tl.int64)
	bos, eos = tl.load(cu_seqlens + i_n).to(tl.int64), tl.load(cu_seqlens + i_n + 1).to(tl.int64)
	T = eos - bos
	NT = tl.cdiv(T, BT)
	else:
	NT = tl.cdiv(T, BT)
	i_tg = i_b * NT + i_t
	bos, eos = i_b * T, i_b * T + T

	o_t = i_t * BT + tl.arange(0, BT)
	o_d = i_d * BD + tl.arange(0, BD)
	m_x = (o_t[:, None] < T) & (o_d[None, :] < D)
	p_dx = dx + (bos * H + i_h) * D + o_t[:, None] * (H*D) + o_d[None, :]
	p_do = do + (i_tg * H + i_h) * D + o_d
	# [BD]
	b_do = tl.load(p_do, mask=o_d < D, other=0.0).to(tl.float32)
	# [BT, BD]
	b_dx = b_do / tl.full((BT,), min(BT, T - i_t * BT), dtype=tl.float32)[:, None]
	tl.store(p_dx, b_dx.to(p_dx.dtype.element_ty), mask=m_x)


	def mean_pooling_fwd(
	x: torch.Tensor,
	chunk_size: int,
	cu_seqlens: torch.LongTensor \| None = None,
	chunk_indices: torch.LongTensor \| None = None,
	) -> torch.Tensor:
	B, T, H, D = x.shape
	BT = chunk_size
	if chunk_indices is None and cu_seqlens is not None:
	chunk_indices = prepare_chunk_indices(cu_seqlens, chunk_size)
	NT = triton.cdiv(T, BT) if cu_seqlens is None else len(chunk_indices)

	o = x.new_empty(B, NT, H, D)
	def grid(meta): return (triton.cdiv(D, meta['BD']) * NT, B * H)
	mean_pooling_fwd_kernel[grid](
	x,
	o,
	cu_seqlens,
	chunk_indices,
	T=T,
	H=H,
	D=D,
	BT=BT,
	)
	return o


	def mean_pooling_bwd(
	do: torch.Tensor,
	batch_size: int,
	seq_len: int,
	chunk_size: int,
	cu_seqlens: torch.LongTensor \| None = None,
	chunk_indices: torch.LongTensor \| None = None,
	) -> torch.Tensor:
	B, T, H, D = batch_size, seq_len, *do.shape[-2:]
	BT = chunk_size
	if chunk_indices is None and cu_seqlens is not None:
	chunk_indices = prepare_chunk_indices(cu_seqlens, chunk_size)
	NT = triton.cdiv(T, BT) if cu_seqlens is None else len(chunk_indices)

	dx = do.new_empty(B, T, H, D)
	def grid(meta): return (triton.cdiv(D, meta['BD']) * NT, B * H)
	mean_pooling_bwd_kernel[grid](
	do,
	dx,
	cu_seqlens,
	chunk_indices,
	T=T,
	H=H,
	D=D,
	BT=BT,
	)
	return dx


	class MeanPoolingFunction(torch.autograd.Function):

	@staticmethod
	@input_guard
	@autocast_custom_fwd
	def forward(
	ctx,
	x: torch.Tensor,
	chunk_size: int,
	cu_seqlens: torch.LongTensor \| None = None,
	) -> torch.Tensor:
	o = mean_pooling_fwd(x, chunk_size, cu_seqlens)
	ctx.batch_size = x.shape[0]
	ctx.seq_len = x.shape[1]
	ctx.chunk_size = chunk_size
	ctx.cu_seqlens = cu_seqlens
	return o

	@staticmethod
	@input_guard
	@autocast_custom_bwd
	def backward(
	ctx, do,
	) -> tuple[torch.Tensor, None, None]:
	batch_size = ctx.batch_size
	seq_len = ctx.seq_len
	chunk_size = ctx.chunk_size
	cu_seqlens = ctx.cu_seqlens
	dx = mean_pooling_bwd(do, batch_size, seq_len, chunk_size, cu_seqlens)
	return dx, None, None


	def mean_pooling(
	x: torch.Tensor,
	chunk_size: int,
	cu_seqlens: torch.LongTensor \| None = None,
	**kwargs,
	) -> torch.Tensor:
	if 'head_first' in kwargs:
	raise DeprecationWarning(
	"head_first has been removed. Inputs must be in `[B, T, H, ...]` format.",
	)
	if cu_seqlens is not None:
	if x.shape[0] != 1:
	raise ValueError(
	f"The batch size is expected to be 1 rather than {x.shape[0]} when using `cu_seqlens`."
	f"Please flatten variable-length inputs before processing.",
	)
	o = MeanPoolingFunction.apply(x, chunk_size, cu_seqlens)
	return o