def profile_packing(
    dataloader_cfg: dict[str, Any],
    tokenizer: PreTrainedTokenizerBase,
    min_ratio: float,
    max_ratio: float,
    num_packing_ratios: int,
    device_batch_size: int,
    packing_collator_cls: type[BinPackCollator] = BinPackCollator,
    inner_collator: Callable = lambda x: x,
) -> Iterable[tuple[float, Optional[float], Optional[float]]]:
    """Generator function that profiles example packing across packing ratios.

    Args:
        dataloader_cfg (DictConfig): The dataloader configuration for profiling.
        tokenizer (PreTrainedTokenizerBase): The tokenizer for profiling.
        min_ratio (float): Smallest packing_ratio to test. Must be >=1.
        max_ratio (float): Largest packing_ratio to test. Must be larger than `min_ratio`.
        num_packing_ratios (int): Number of packing_ratio values (spaced between `min_ratio` and `max_ratio`) to try.
        device_batch_size (int): The size of the batches (number of examples) per device.
        packing_collator_cls (BinPackCollator): The collator class to use for packing.
        inner_collator (Callable): The inner collator to use by the packing collator.

    Returns:
        An iterable of tuples of packing ratio, padding, and waste, sorted by smallest to largest packing ratio.
    """
    import copy

    from llmfoundry.data.dataloader import build_dataloader

    dataset_cfg = dataloader_cfg['dataset']
    max_seq_len = dataset_cfg.get('max_seq_len')
    max_leftovers_to_keep = dataset_cfg.get('max_leftovers_to_keep', None)

    # Turn off packing and sequence parallelism for the dataloader (we want raw, pre-packed, full-length examples)
    dataloader_cfg = copy.deepcopy(dataloader_cfg)
    dataloader_cfg.update({
        'drop_last': False,
        'num_workers': 0,
        'prefetch_factor': None,
        'persistent_workers': False,
    })
    dataset_cfg = dataloader_cfg['dataset']
    dataset_cfg['packing_ratio'] = 1.0
    seq_parallel_replication = dataset_cfg.get('seq_parallel_replication', 1)
    dataset_cfg['auto_packing_replication'] = seq_parallel_replication or 1
    dataset_cfg['seq_parallel_replication'] = 1
    dataset_cfg['pad_to_longest'] = True

    # If streaming dataset, use a temporary local folder for profiling
    local_rank_zero = dist.get_global_rank() - dist.get_local_rank()
    if dataset_cfg.get(
        'remote',
    ) is not None and dataset_cfg.get('local') is None:
        tmp_path_to_broadcast = tempfile.TemporaryDirectory().name
        gathered_paths = dist.all_gather_object(tmp_path_to_broadcast)
        tmp_path = gathered_paths[local_rank_zero]
        dataset_cfg['local'] = tmp_path

    if dataset_cfg.get('streams') is not None:
        for stream_config in dataset_cfg['streams'].values():
            tmp_path_to_broadcast = tempfile.TemporaryDirectory().name
            gathered_paths = dist.all_gather_object(tmp_path_to_broadcast)
            tmp_path = gathered_paths[local_rank_zero]
            if stream_config.get('local') is None:
                stream_config['local'] = tmp_path

    # Determine the packing_ratio values we'll try
    packing_ratios, raw_batch_sizes = [], []
    for packing_ratio in np.linspace(
        min_ratio,
        max_ratio,
        num_packing_ratios,
        endpoint=True,
    ):
        packing_ratio = np.round(10 * packing_ratio) / 10
        raw_batch_size = int(packing_ratio * device_batch_size)
        if raw_batch_size not in raw_batch_sizes:
            packing_ratios.append(packing_ratio)
            raw_batch_sizes.append(raw_batch_size)

    n_profile_examples = max(raw_batch_sizes) * 100

    train_dataspec = build_dataloader(
        dataloader_cfg,
        tokenizer,
        n_profile_examples,
    )
    train_dataloader = train_dataspec.dataloader

    # Get a bunch of raw examples
    big_batch = next(iter(train_dataloader))

    # Cut everything down to size
    if 'total_tokens' in big_batch:
        del big_batch['total_tokens']
    if 'loss_generating_tokens' in big_batch:
        del big_batch['loss_generating_tokens']

    sizes, trimmed_examples = packing_collator_cls._trim_batch(big_batch)

    def profile(raw_batch_size: int) -> tuple[Optional[float], Optional[float]]:
        # Copy trimmed examples so that the dicts are not shared between profiling runs.
        trimmed_examples_copy = [te.copy() for te in trimmed_examples]

        # Create the packing collator.
        packer = packing_collator_cls(
            collator=inner_collator,
            target_batch_size=device_batch_size,
            max_seq_len=max_seq_len,
            pad_token_id=0,  # <-- Doesn't need to be correct for profiling
            padding_side='left',  # <-- Doesn't need to be correct for profiling
            max_leftover_bins_to_keep=max_leftovers_to_keep,
            is_profiling=True,
        )

        # Simulate feeding the packing collator a bunch of data
        for idx in range(0, len(trimmed_examples_copy), raw_batch_size):
            batch = trimmed_examples_copy[idx:idx + raw_batch_size]
            if len(batch) < device_batch_size:
                continue
            packer._pack_trimmed_examples(
                batch,
                sizes[idx:idx + raw_batch_size],
            )

        if packer.n_packed_examples == 0:
            log.debug(
                'No examples packed during profiling. Dataset is smaller than device batch size.',
            )
            return None, None

        # Return the padding and waste stats over that bunch of data
        padding_percent = 100 * (1 - packer.efficiency)
        waste_percent = 100 * packer.waste
        return padding_percent, waste_percent

    log.debug('Profiling packing ratios')
    total_packing_ratios = min(len(packing_ratios), len(raw_batch_sizes))
    for i, (packing_ratio,
            raw_batch_size) in enumerate(zip(packing_ratios, raw_batch_sizes)):
        log.debug(
            f'Progress [{i}/{total_packing_ratios}]: Profiling packing ratio {packing_ratio}',
        )
        padding, waste = profile(raw_batch_size)
        yield (packing_ratio, padding, waste)
Generator function that profiles example packing across packing ratios.

    Args:
        dataloader_cfg (DictConfig): The dataloader configuration for profiling.
        tokenizer (PreTrainedTokenizerBase): The tokenizer for profiling.
        min_ratio (float): Smallest packing_ratio to test. Must be >=1.
        max_ratio (float): Largest packing_ratio to test. Must be larger than `min_ratio`.
        num_packing_ratios (int): Number of packing_ratio values (spaced between `min_ratio` and `max_ratio`) to try.
        device_batch_size (int): The size of the batches (number of examples) per device.
        packing_collator_cls (BinPackCollator): The collator class to use for packing.
        inner_collator (Callable): The inner collator to use by the packing collator.

    Returns:
        An iterable of tuples of packing ratio, padding, and waste, sorted by smallest to largest packing ratio.
    
llmfoundry/data/packing.py