build/torch-rocm/comms/iris.py

# SPDX-License-Identifier: MIT
# Copyright (C) 2024-2025, Advanced Micro Devices, Inc. All rights reserved.

"""
Minimal Iris context wrapper for AITER communication operations.

This module provides a thin wrapper around the iris library context to
support reduce-scatter and all-gather operations. All core iris functions
(load, store, put, atomic_*, etc.) are provided by the iris library itself.
"""

import logging
import math
from typing import Optional

import torch
import iris

# If we got here, iris is available
IRIS_AVAILABLE = True

logger = logging.getLogger("aiter")


def calculate_heap_size(
    M: int,
    N: int,
    dtype: "torch.dtype",
    world_size: Optional[int] = None,
    quant_mode: str = "none",
    all_gather: bool = True,
    overhead_factor: float = 1.2,
) -> int:
    """
    Calculate required Iris heap size for communication operations.

    This function estimates the total memory needed in the Iris symmetric heap
    for reduce-scatter, RMSNorm, quantization, and all-gather operations.

    Args:
        M (int): Number of rows in input tensor
        N (int): Number of columns in input tensor
        dtype (torch.dtype): Data type of input tensor (e.g., torch.float32, torch.float16)
        world_size (int, optional): Number of GPUs. If None, uses torch.distributed.get_world_size()
        quant_mode (str): Quantization mode - "none", "fp8_per_token", or "fp4_per_token"
        all_gather (bool): Whether all-gather is performed after quantization
        overhead_factor (float): Safety margin multiplier (default: 1.2 for 20% overhead)

    Returns:
        int: Required heap size in bytes

    Example:
        >>> # For 2 GPUs, 4096x4096 float32 tensor with FP8 quant and all-gather
        >>> heap_size = calculate_heap_size(4096, 4096, torch.float32,
        ...                                  world_size=2, quant_mode="fp8_per_token")
        >>> with IrisCommContext(heap_size=heap_size) as ctx:
        ...     # Your operations here
        ...     pass
    """
    if world_size is None:
        if not torch.distributed.is_initialized():
            raise RuntimeError(
                "torch.distributed not initialized and world_size not provided. "
                "Either initialize torch.distributed or pass world_size explicitly."
            )
        world_size = torch.distributed.get_world_size()

    # Calculate element size in bytes (fast path for common types)
    if dtype in (torch.float32, torch.int32):
        elem_bytes = 4
    elif dtype in (torch.float16, torch.bfloat16, torch.int16):
        elem_bytes = 2
    elif dtype in (torch.float64, torch.int64):
        elem_bytes = 8
    elif dtype == torch.int8:
        elem_bytes = 1
    else:
        # Fallback for uncommon types (e.g., float8, complex types, future dtypes)
        elem_bytes = torch.empty(0, dtype=dtype).element_size()

    M_shard = M // world_size

    # Memory for input tensor (M x N)
    mem_input = M * N * elem_bytes

    # Memory for reduce-scatter output (M_shard x N)
    mem_rs = M_shard * N * elem_bytes

    # Memory for quantization output
    mem_quant = 0
    if quant_mode == "fp8_per_token":
        mem_quant = M_shard * N  # 1 byte per element
    elif quant_mode == "fp4_per_token":
        if N % 32 != 0:
            raise ValueError("FP4 quantization requires N divisible by 32")
        mem_quant = M_shard * (N // 2)  # 0.5 bytes per element

    # Memory for all-gather output
    mem_gather = 0
    if all_gather:
        if quant_mode == "fp8_per_token":
            mem_gather = M * N  # 1 byte per element
        elif quant_mode == "fp4_per_token":
            mem_gather = M * (N // 2)  # 0.5 bytes per element
        else:
            mem_gather = M * N * elem_bytes  # Full precision

    # Total with overhead
    total_bytes = mem_input + mem_rs + mem_quant + mem_gather
    total_with_overhead = int(math.ceil(total_bytes * overhead_factor))

    logger.debug(
        f"Heap size calculation: M={M}, N={N}, dtype={dtype}, world_size={world_size}, "
        f"quant_mode={quant_mode}, all_gather={all_gather}\n"
        f"  Input: {mem_input:,} bytes\n"
        f"  RS buffer: {mem_rs:,} bytes\n"
        f"  Quant buffer: {mem_quant:,} bytes\n"
        f"  Gather buffer: {mem_gather:,} bytes\n"
        f"  Total (with {overhead_factor}x overhead): {total_with_overhead:,} bytes "
        f"({total_with_overhead / (1024**3):.2f} GB)"
    )

    return total_with_overhead


class IrisCommContext:
    """
    Minimal context wrapper for Iris-based communication operations.

    This is a thin wrapper around iris.iris() that provides convenient access
    to the iris context for use in reduce-scatter and all-gather operations.

    Example:
        >>> # Manual heap size
        >>> with IrisCommContext(heap_size=2**30) as ctx:
        >>>     shard = ctx.iris_ctx.zeros((1024, 1024), dtype=torch.float32)
        >>>     full = all_gather(shard, ctx)

        >>> # Automatic heap size calculation
        >>> heap_size = calculate_heap_size(4096, 4096, torch.float32,
        ...                                  world_size=2, quant_mode="fp8_per_token")
        >>> with IrisCommContext(heap_size=heap_size) as ctx:
        >>>     # Guaranteed to have enough memory for your operations
        >>>     input_tensor = ctx.iris_ctx.empty((4096, 4096), dtype=torch.float32)
    """

    def __init__(self, heap_size: int = 1 << 30):
        """
        Initialize Iris communication context.

        Args:
            heap_size (int): Size of the symmetric heap in bytes. Default: 1GB
                Use calculate_heap_size() to automatically determine the required size
                based on your tensor dimensions and operations.

        Example:
            >>> # Option 1: Fixed size
            >>> ctx = IrisCommContext(heap_size=2**30)  # 1GB

            >>> # Option 2: Auto-calculated size
            >>> M, N = 4096, 4096
            >>> heap_size = calculate_heap_size(M, N, torch.float32,
            ...                                  world_size=2, quant_mode="fp8_per_token")
            >>> ctx = IrisCommContext(heap_size=heap_size)
        """
        if not IRIS_AVAILABLE:
            raise RuntimeError("Iris library is not available. Please install iris.")

        self.heap_size = heap_size
        self.iris_ctx = None
        self._initialized = False

    def __enter__(self):
        """Initialize Iris context when entering context manager."""
        if not self._initialized:
            self.iris_ctx = iris.iris(heap_size=self.heap_size)
            self._initialized = True
            self.cur_rank = self.iris_ctx.cur_rank
            self.num_ranks = self.iris_ctx.num_ranks

            logger.info(
                f"Iris context initialized: rank {self.cur_rank}/{self.num_ranks}, heap_size={self.heap_size}"
            )
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Clean up when exiting context manager.

        Used by Python's with statement (automatically called).

        Example:
            with IrisCommContext(...) as ctx:
                pass
            # __exit__() is called here automatically
        """
        # Iris context cleanup is handled automatically
        pass

    @property
    def is_initialized(self) -> bool:
        """Check if the Iris context has been initialized."""
        return self._initialized

    def get_heap_bases(self):
        """Get the heap bases tensor for use in Triton kernels."""
        if not self.is_initialized:
            raise RuntimeError("Iris context not initialized. Use as context manager.")
        return self.iris_ctx.heap_bases