大模型算法实战教程

Appearance

48. Communication Hotspots and Mitigation | 通信热点与缓解策略 ​

难度： Medium | 环境： CPU-first | 标签： 并行通信, 热点分析, 缓解策略 | 目标人群： 并行通信学习者

🚀 云端运行环境

本章节的实战代码可以点击以下链接在免费 GPU 算力平台上直接运行：

Open In ColabOpen In Studio (国内推荐：魔搭社区免费实例)

本节导读 ​

48 聚焦多卡训练和推理里最容易停留在“感觉通信很慢”这一层的问题：到底是哪类 collective 最慢、等待时间主要卡在哪里、应该优先改哪种缓解动作。它把通信 profiling 进一步收敛成可执行的热点判断和缓解策略。

关键词： collective, wait time, hotspot, communication plan

前置阅读 ​

相关阅读 ​

导语： 学完通信热点判断后，下一步重点不是继续背 collective 名词，而是看这些热点与缓解动作怎样进入 benchmark 和 MoE 场景，确认“看到了瓶颈”是否真的能转成有效收益。

Step 1: 先识别通信热点 ​

  • 区分带宽受限、延迟受限和等待链路。
  • 不同 collective 的热点位置不一样，不能用一个结论覆盖全部场景。
  • 先知道通信慢在哪里，后面才有替换空间。

Step 2: 写清替换和缓解策略 ​

  • 可能的动作包括改 collective、改 bucket、改 overlap 或改分组方式。
  • 这些策略要和热点类型一一对应，而不是泛化成“多做 overlap”。

Step 3: 判断是否值得继续单页展开 ​

  • 如果热点识别和缓解策略已经形成独立判断框架，就值得升级为正式通信补页。

Step 4: 动手实战 ​

  1. 补全 summarize_comm_hotspots，识别通信热点。
  2. 补全 choose_comm_mitigation，选择缓解策略。
  3. 补全 recommend_comm_followup，输出是否继续展开。

提示 ​

  • 这页不是让你实现完整通信优化器，而是先固定三步判断：最慢的 collective 是谁、对应该采取什么缓解动作、这条链路是否已经值得单独扩页。
  • TODO 1 只需要找出 wait_ms 最大的事件，并返回对应 collective 和等待时间。
  • TODO 2 只要按最慢 collective 的类型选择一个最小缓解动作，不需要在这里展开复杂调参。
  • TODO 3 先判断等待时间是否已经明显超标，再结合 mitigation 是否存在，给出是否值得继续展开的结论。
python
from typing import Dict, List
python
def summarize_comm_hotspots(events: List[Dict[str, float]]) -> Dict[str, object]:
    """
    TODO 1: 找出等待时间最长的 collective。
    """
    # 提示：可以先用 max 找到 wait_ms 最大的事件，再返回 collective 和 wait_ms。
    # hotspot = ???
    raise NotImplementedError


def choose_comm_mitigation(summary: Dict[str, object]) -> Dict[str, object]:
    """
    TODO 2: 根据最慢 collective 选择缓解策略。
    """
    # 提示：先取出 largest_collective，再分别判断 all_to_all、all_reduce 和其他情况。
    # collective = ???
    raise NotImplementedError


def recommend_comm_followup(summary: Dict[str, object], mitigation: Dict[str, object]) -> Dict[str, object]:
    """
    TODO 3: 输出是否值得继续扩成独立通信页。
    """
    # 提示：先判断 largest_wait_ms 是否大于阈值，再结合 mitigation 是否存在给出 needs_dedicated_page 和 reason。
    # needs_dedicated_page = ???
    # reason = ???
    raise NotImplementedError
python
def test_comm_hotspots_and_mitigation():
    try:
        events = [
            {'collective': 'all_reduce', 'wait_ms': 18.0},
            {'collective': 'all_to_all', 'wait_ms': 32.0},
            {'collective': 'broadcast', 'wait_ms': 5.0},
        ]
        summary = summarize_comm_hotspots(events)
        assert summary['largest_collective'] == 'all_to_all'
        assert summary['largest_wait_ms'] == 32.0
        mitigation = choose_comm_mitigation(summary)
        assert mitigation['strategy'] == 'reduce_routing_traffic'
        assert mitigation['target'] == 'all_to_all'
        decision = recommend_comm_followup(summary, mitigation)
        assert decision['needs_dedicated_page'] is True
        print('测试通过：通信热点与缓解策略页面模板可以工作。')
    except NotImplementedError:
        raise
    except (AttributeError, NameError, TypeError, ValueError, AssertionError) as e:
        raise NotImplementedError('请先完成 TODO 代码！') from e


test_comm_hotspots_and_mitigation()

🛑 STOP HERE 🛑









请先尝试自己完成代码并跑通测试。
如果你正在 Colab 中运行，并且遇到困难没有思路，可以向下滚动查看参考答案。









参考代码与解析 ​

代码 ​

python
def summarize_comm_hotspots(events: List[Dict[str, float]]) -> Dict[str, object]:
    """
    TODO 1: 找出等待时间最长的 collective。
    """
    # 提示：可以先用 max 找到 wait_ms 最大的事件，再返回 collective 和 wait_ms。
    # hotspot = ???
    hotspot = max(events, key=lambda item: item.get('wait_ms', 0.0))
    return {'largest_collective': hotspot.get('collective', ''), 'largest_wait_ms': hotspot.get('wait_ms', 0.0)}


def choose_comm_mitigation(summary: Dict[str, object]) -> Dict[str, object]:
    """
    TODO 2: 根据最慢 collective 选择缓解策略。
    """
    # 提示：先取出 largest_collective，再分别判断 all_to_all、all_reduce 和其他情况。
    # collective = ???
    collective = summary.get('largest_collective', '')
    if collective == 'all_to_all':
        return {'strategy': 'reduce_routing_traffic', 'target': collective}
    if collective == 'all_reduce':
        return {'strategy': 'increase_overlap', 'target': collective}
    return {'strategy': 'rebalance_message_groups', 'target': collective}


def recommend_comm_followup(summary: Dict[str, object], mitigation: Dict[str, object]) -> Dict[str, object]:
    """
    TODO 3: 输出是否值得继续扩成独立通信页。
    """
    # 提示：先判断 largest_wait_ms 是否大于阈值，再结合 mitigation 是否存在给出 needs_dedicated_page 和 reason。
    # needs_dedicated_page = ???
    # reason = ???
    needs_page = summary.get('largest_wait_ms', 0.0) > 10 and bool(mitigation.get('strategy'))
    reason = '通信热点和缓解策略已经形成独立分析链路' if needs_page else '当前通信问题仍可由现有页面覆盖'
    return {'needs_dedicated_page': needs_page, 'reason': reason}

解析 ​

TODO 1：summarize_comm_hotspots 先回答“最慢的是哪段 collective”。只有把最大的等待热点定位出来，后面的优化动作才不会停留在泛泛而谈的通信调优。

TODO 2：choose_comm_mitigation 负责把热点映射成最小可执行动作。这里故意不展开复杂参数搜索，而是先把 all_to_all、all_reduce 和其他 collective 的首选缓解方向固定下来。

TODO 3：recommend_comm_followup 用来判断这条通信链路是否已经复杂到值得单独扩页。如果热点定位和缓解动作已经形成稳定判断框架，就说明它不再只是 profiling 注释，而是一页完整的通信分析主题。

Released under the MIT License.

Copyright © 2024-present