# mypy: 允许未类型化定义
导入
复制
导入 json
导入
正则表达式
导入
弱引用
来自
集合
导入 defaultdict
来自
打字
导入
任何
导入
火炬
导入 torch.nn
来自 torch._guards
导入
检测伪造模式
来自 torch.autograd.graph
导入
注册多梯度钩子
来自 torch.distributed._tools.mod_tracker
导入 ModTracker
来自 torch.distributed.tensor._api
导入 DTensor
来自
torch.nn.modules 下的 module 模块
导入 (
注册模块前向钩子,
注册模块前向预处理钩子,
register_module_full_backward_pre_hook,
)
来自 torch.utils._python_dispatch
导入
Torch 分发模式
来自 torch.utils._pytree
导入
树扁平化
__all__ = ["CommDebugMode"]
funcol_本地 =
火炬.
操作.
_c10d_功能
funcol_Python =
火炬.
操作.
c10d_功能
funcol_autograd = 火炬.
操作._c10d_functional_autograd
c10d_ops = 火炬.
操作.c10d
原生到 Python 映射 = {
funcol 原生.all_gather_into_tensor:
funcolPython.all_gather_into_tensor,
funcol 原生.
将所有数据聚集到合并的 tensor 中: funcol_py.
将所有数据聚集到合并的 tensor 中,
funcol_native.all_reduce: funcol_py.all_reduce,
funcol_native.全局归约合并: funcol_py.
全局归约合并,
funcol_native.全向单播: funcol_py.
全向单播,
funcol_native.广播: funcol_py.
广播,
funcol_native.reduce_scatter_tensor: funcol_py.reduce_scatter_tensor,
funcol_native.简化散列张量: funcol_py.
简化散列张量,
功能操作
funcol_自动微分.
全向单播:
funcol_Python.
全向单播,
}
c10d_集体操作 = {
c10d 操作.
_allgather 基础_,
c10d 操作.
_reduce_scatter 基础_,
c10d 操作.allgather_,
c10d 操作.allgather_coalesced_,
c10d_ops.allgather_into_tensor_coalesced_,
c10d_ops.allreduce_,
c10d_ops.allreduce_coalesced_,
c10d_ops.alltoall_,
c10d_ops.alltoall_base_,
c10d_ops.broadcast_,
c10d 操作.gather_,
c10d 操作.
散射_,
c10d 操作.
减少_,
c10d 操作.
reduce_scatter_减少散射_,
c10d 操作.reduce_scatter_tensor_coalesced_,
}
trivial_ops = {
aten.detach.default,
aten.t.default,
aten.view 默认,
aten._to_copy 默认,
aten.as_strided 默认,
aten.transpose.int,
}
类 _CommModeModuleTracker(
模追踪器):
""
继承 ModuleTracker 并扩展其功能以跟踪
模型在模块级别的参数和分片信息
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
def 初始化(
自身):
超级().
初始化()
自身.
模块助手字典 = {}
自身.
模块参数字典 = {}
自身.
模块父字典 = {}
自身.
注册前向钩子处理程序 = {}
自身.
父字典 = {}
自身.
父列表 = []
自身.
分片字典 = {}
自身.
激活检查点 =
假
自身.
名称 =
请提供需要翻译的文本
def _fw_set_module_hook(自身,
模块,
输入,
输出):
""
更新模块在模块运行完成后以及所有其他钩子解决后
激活检查点
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
如果
自身.
是黑白的:
自身.activation_checkpointing =
真实
否则:
自身.
激活检查点 =
假
如果 not
自身.
激活检查点:
# 模块不再是下一个模块的父模块
自身.
父列表.
弹出()
将当前模块设置为上一个父模块
自身.
名称 =
自身.
父列表[-1]
def _fw_pre_hook(自身,
模块,
输入):
""
这个函数在模块的前向传播之前被调用。
收集模块的参数和分片信息
存储到字典中。
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
如果
自身.
是黑白的:
自身.
激活检查点 =
真实
否则:
自身.
激活检查点 =
假
自身.
名称 =
超级()._get_mod_name(
模块)
w 模 =
弱引用.ref(
模块)
将当前子模块添加到模块跟踪父类
超级()._get_append_fn(w_mod,
自身.
名称,
错误)()
参数, _ =
树扁平化(
输入)
张量 = [a
为 a
在 args
如果 isinstance(a,
火炬.
张量)
和 a.
需要梯度]
如果 not
自身.is_bw
和
张量:
注册多梯度钩子(
张量,
超级()._get_pop_fn(w_mod,
自身.
名称, True)
)
如果 not
自身.
激活检查点:
包含有关模块树中模块排序和深度的信息
如果
自身.
名称 not
在
自身.
模块辅助字典:
自身.
模块辅助字典[
自身.
名称] = {}
自身.
模块辅助字典[
自身.
名称
]
[模块类型] = (
字符串(
类型(
模块)).
替换("<", "").
替换(">", "")
)
自身.
模块助手字典[
自身.
名称
]
[深度] =
长度(
自身.parents) - 1
为
` 的类型为 List[torch.Tensor],
参数
在
模块.
命名参数。(
递归=
错误):
如果
自身.
名称 not
在
自身.
模块参数字典:
自身.
模块参数字典[
自身.
名称] = {}
自身.
模块参数字典[
自身.
名称
]
[` 的类型为 List[torch.Tensor]] =
参数.
数据
如果 isinstance(
参数.
数据, DTensor):
关键名称 =
自身.
名称 + "." +
参数名称
自身.
分片字典[
键名] =
参数.
数据.
布局
如果
参数 not
在
自身.
模块辅助字典[
自身.
名称
]：
自身.
模块辅助字典[
自身.
名称
]
[参数] = {}
自身.
模块辅助字典[
自身.
名称
]
[参数
]
[` 的类型为 List[torch.Tensor]] =
字符串(
参数.
数据.
布局
)
用于存储模块的父级以确保反向传递/检查点中的正确性
如果
自身.
名称 not
在
自身.
模块父字典:
自身.
模块父字典[
自身.
名称] =
复制.
深拷贝(
自身.parents)
# 用于创建用于 json 输出的父-子模块关联
父级 =
自身.
父列表[-1]
如果
父级 not
在
自身.
父字典:
自身.
父字典[
父节点] = []
自身.
父字典[
父节点].
追加(
自身.
名称)
自身.
父列表.
追加(
自身.
名称)
自身.
注册前向钩子处理[
自身.
名称] =
模块.
注册前向钩子(
自身._fw_set_module_hook
)
def _fw_post_hook(自身,
模块,
输入,
输出):
""
当模块的前向传播被调用时，此函数会被调用。
它更新模块跟踪器并从父数据中移除模块。
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
超级()._fw_post_hook(
模块,
输入,
输出)
def _bw_hook(自身,
模块,
输出):
""
这个函数在模块的逆向传播被调用时执行。它
更新当前模块以进行逆向传播
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
自身.
激活检查点 =
假
自身.
名称 =
超级()._get_mod_name(
模块)
def __进入__(
自身):
自身.
激活检查点 =
假
自身.
模块参数字典.
清晰()
自身.
分片字典.
清晰()
自身.
父字典.
清晰()
自身.
父列表 = [
全球]
自身.
模块助手字典.
清晰()
自身.
模块助手字典[
全球] = {
深度: 0}
自身.
模块父字典.
清晰()
自身.
模块父字典[
全球] =
集合()
自身._fw_pre_handle =
注册模块前向预处理钩子(
自身._fw_pre_hook)
自身.
_fw 后处理 =
注册模块前向钩子(
自身._fw_post_hook)
自身.
注册前向钩子处理程序.
清晰()
自身.
_bw 处理程序 =
注册模块完整反向钩子(
自身.
_bw 钩子)
自身.
名称 =
"全局"
def __退出__(
自身, *
参数):
超级().
__退出__(*
参数)
自身.
_bw 处理.
删除()
# 移除在预钩子中添加的所有 forward_hook 处理程序
为 handle
在
自身.
注册 forward_hook 处理程序.
值():
处理.
删除()
def 打印参数信息(
自身):
打印(
自身.
模块参数字典)
def 打印分片信息(
自身):
为
键, value
在
自身.
分片字典.
项目():
打印(key + ": " +
字符串(
值))
类
通信调试模式(
火炬调度模式):
""
`CommDebugMode` 是一个上下文管理器，用于统计其上下文中的功能集体数量。
它使用 `TorchDispatchMode` 来完成这项工作。
```
注意：并非所有集体都受到支持。
示例用法
.. 代码块 :: python
mod = ...
comm_mode = CommDebugMode()
with comm_mode:
mod.sum().backward()
打印(comm_mode.get_comm_counts())
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
def 初始化(
自身):
自身.comm_counts:
字典[
任意, int] = defaultdict(int)
自身.comm_module_counts = {}
自身.comm_module_operation_counts = {}
自身.
通信注册表 =
集合()
为
原生操作,
Python 操作
在
原生到 Python 映射.
项目():
自身.
通信注册表.
添加(
原生操作)
自身.
通信注册表.
添加(
Python 操作)
自身.
通信注册表.
添加(
火炬.
操作.
_张量.
分片维度全对全)
自身.
高级模块跟踪器 =
_通信模式模块跟踪器()
[文档] def
生成 JSON 转储(
自身,
文件名=
"通信模式日志.json",
噪音水平=3):
""
创建用于构建浏览器可视化的 JSON 文件
0. 打印模块级别的集体计数
1. 打印不包括在简单操作中的 dTensor 操作
2. 打印不包括在简单操作中的操作
3. 打印所有操作
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
(
include_DTensor_ops,
包含模块数据,
包含操作,
包含简单操作,
) = 自身.
设置噪声参数(
噪音水平)
递归构建 JSON 数据
def 添加 JSON 信息(
JSON 字典,
完全限定名):
JSON 字典[
"全限定名"] =
完全限定名
JSON 字典[
模块类型] =
请提供需要翻译的文本
JSON 字典[
参数] = []
JSON 字典["children"] = []
JSON 字典[
"集体前进"] = []
json 字典[
"集体后退"] = []
json 字典[
"前向操作"] = []
JSON 字典[
"后向操作"] = []
添加模块层类型及其参数，以及它们的分片
如果 (
模块类型
在
自身.
高级模块跟踪器.
模块辅助字典[
完全限定名]
和
包含模块数据
):
JSON 字典[
模块类型] = (
自身.
高级模块跟踪器.
模块辅助字典[
完全限定名
]
[模块类型]
)
如果
参数
在
自身.
高级模块跟踪器.
模块辅助字典[
完全限定名
]：
为 (
` 的类型为 List[torch.Tensor],
布置,
) 在
自身.
高级模块跟踪器.
模块辅助字典[
完全限定名
]
参数
].项目():
JSON 字典[
参数].
追加((
` 的类型为 List[torch.Tensor],
布置))
# 添加模块集体信息
如果
完全限定名
在
自身.
通信模块数量:
为
群体,
计算
在
自身.
通信模块数量[
完全限定名
]
向前
].项目():
JSON 字典[
群体前进].
追加((
字符串(
群体),
数量))
为
群体,
计算
在
自身.
通信模块数量[
完全限定名
]
“向后”
].项目():
JSON 字典[
群体反向].
追加((
字符串(
群体),
数量))
添加模块操作信息
前向操作 = []
向后操作 = []
检查点操作 = []
仅当最小操作噪声级别设置为 true 时获取操作
如果
包含 DTensor 操作:
如果
完全限定名
在
自身.
通信模块操作次数:
(
前向操作,
后向操作,
检查点操作,
) = 自身.
获取操作列表(
自身.
通信模块操作计数[
完全限定名]
)
# 移除所有没有 DTensor 输入的操作
如果 not
包含操作:
前向操作 = [
操作符
为
操作符
在
前向操作
如果
长度(
操作[
输入分片])
]
向后操作 = [
操作符
为
操作符
在
向后操作
如果
长度(
操作[
输入分片])
]
检查点操作 = [
操作符
为
操作符
在
检查点操作
如果
长度(
操作[
输入分片])
]
# 删除所有平凡操作集中的操作
如果 not
包含平凡操作:
前向操作 = [
操作符
为
操作符
在
前向操作
如果
字符串(
操作[
名称]) not
在
简单操作
]
向后操作 = [
操作符
为
操作符
在
向后操作
如果
字符串(
操作[
名称]) not
在
简单操作
]
检查点操作 = [
操作符
为
操作符
在
检查点操作
如果
字符串(
操作[
名称]) not
在
简单操作
]
将操作信息转换为字符串格式以供 json.dumps()使用
前向操作 =
复制.
深拷贝(
前向操作)
为
操作符
在
前置操作:
操作[
名称] =
字符串(
操作[
名称])
为 i
在
范围(
长度(
操作[
输入分片])):
操作[
输入分片
]
[i] = 字符串(
操作[
输入分片
]
[i])
操作[
输入形状
]
[i] = 字符串(
操作[
输入形状
]
[i])
反向操作 =
复制.
深拷贝(
反向操作)
为
操作符
在
向后操作:
操作[
名称] =
字符串(
操作[
名称])
为 i
在
范围(
长度(
操作[
输入分片])):
操作[
输入分片
]
[i] = 字符串(
操作[
输入分片
]
[i])
操作[
输入形状
]
[i] = 字符串(
操作[
输入形状
]
[i])
检查点操作 =
复制.
深拷贝(
检查点操作)
为
操作符
在
检查点操作:
操作[
名称] =
字符串(
操作[
名称])
为 i
在
范围(
长度(
操作[
输入分片])):
操作[
输入分片
]
[i] = 字符串(
操作[
输入分片
]
[i])
操作[
输入形状
]
[i] = 字符串(
操作[
输入形状
]
[i])
json 字典[
前向操作] =
前向操作
JSON 字典[
"反向操作"] =
反向操作
json 字典[
操作检查点] =
检查点操作
如果
完全限定名 not
在
自身.
高级模块跟踪器.
父字典:
返回
JSON 字典
递归添加模块的子项
为
元素
在
自身.
高级模块跟踪器.
父字典[
完全限定名
]：
JSON 字典["children"].
追加(
添加 JSON 信息({}, ele))
返回 json_dict
json_dict: 字典[
字符串,
任意] = {}
添加 JSON 信息(
json 字典,
全球)
# 将字典转换为 json 文件
与
打开(
文件名,
w)
作为
json 文件:
json.导出(
json 字典,
json 文件,
缩进=4)
[文档] def
生成通信调试跟踪表(
自身,
噪声水平=3):
""
生成显示操作和集体跟踪信息的详细表格
在模块级别。信息量取决于噪声级别
0. 打印模块级别的集体计数
1. 打印不包括在平凡操作中的 dTensor 操作，模块信息
2. 打印不包括在平凡操作中的操作
打印所有操作
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
(
包含 DTensor 操作,
包含模块数据,
包含操作,
包含琐碎操作,
) = 自身.
设置噪声参数(
噪声水平)
表格 =
请提供需要翻译的文本
为
完全限定名
在
自身.
高级模块跟踪器.
模块辅助字典:
# 设置表格格式化缩进
indent = " " * (
2 * 自身.
高级模块跟踪器.
模块助手字典[
完全限定名
]
[深度]
)
表格 += f"{
缩进}{
完全限定名}
输入文本翻译为简体中文为：\n"
如果
包含模块数据:
如果 (
模块类型
在
自身.
高级模块跟踪器.
模块辅助字典[
完全限定名]
):
模块类型 =
自身.
高级模块跟踪器.
模块辅助字典[
完全限定名
]
"模块类型"
]
表格 += f"{
缩进}
*模块类型：{
模块类型}
输入文本翻译为简体中文为：\n"
如果
参数
在
自身.
高级模块跟踪器.
模块辅助字典[
完全限定名
]：
表格 += f"{
缩进}
*参数列表
输入文本翻译为简体中文为：\n"
为 (
` 的类型为 List[torch.Tensor],
布置,
) 在
自身.
高级模块跟踪器.
模块辅助字典[
完全限定名
]
参数
].项目():
表格 += f"{
缩进} *{
` 的类型为 List[torch.Tensor]}: {
布置}
输入文本翻译为简体中文为：\n"
indent += " "
集体缩进 = " " * (
2 * 自身.
高级模块跟踪器.
模块辅助字典[
完全限定名
]
[深度] + 2
)
操作缩进 = " " * (
2 * 自身.
高级模块跟踪器.
模块辅助字典[
完全限定名
]
[深度] + 3
)
用前后符号区分模块的集合和操作
前向集体 = {}
后向集体 = {}
如果
完全限定名
在
自身.
通信模块数量:
前向集体 =
自身.
通信模块数量[
完全限定名
]
[前进]
后向集体 =
自身.
通信模块数量[
完全限定名
]
["向后"]
前向操作 = []
向后操作 = []
检查点操作 = []
如果
包含 DTensor 操作:
如果
完全限定名
在
自身.
通信模块操作计数:
(
前向操作,
向后操作,
检查点操作,
) = 自身.
_获取操作列表(
自身.
通信模块操作计数[
完全限定名]
)
def 添加追踪信息(
表格,
集合字典,
操作列表):
""
为模块的前向或反向添加追踪信息
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
为
群体,
计算
在
群体字典.
项目():
表格 += (
f"退格键
\033[1;33m{
集体缩进}*{
集体}: {
数量}
退格键
\033[0m
输入文本翻译为简体中文为：\n"
)
def 添加操作(
表格,
操作,
集体缩进,
操作缩进
):
""
向表中添加操作信息
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
表格 += f"
退格键[1;33m{
集体缩进}**{
操作名称}
退格键
空行
输入文本翻译为简体中文为：\n"
如果
长度(
操作[
输入形状
)]
操作形状 =
操作[
输入形状]
操作分片 =
操作[
输入分片]
操作设备网格 =
操作[
设备网格]
表格 += f"
退格键
颜色代码{
操作缩进}
形状：{
操作形状}
退格键[0m
输入文本翻译为简体中文为：\n"
表格 += f"
退格键
\033[1;31m{
操作缩进}sharding: {
操作分片}
退格键[0m
输入文本翻译为简体中文为：\n"
表格 += f"
退格键[1;31m{
操作缩进}
设备网格:{
运营设备网格}
退格键[0m
输入文本翻译为简体中文为：\n"
返回
表格
为
运营
在
运营列表:
操作名称 =
字符串(
操作[
名称])
# 包含所有操作
如果
包含简单操作:
表格 =
添加操作(
表格,
操作,
集体缩进,
操作缩进
)
# 不包含琐碎操作的所有操作
elif 包含操作
和
操作名称 not
在
琐碎操作:
表格 =
添加操作(
表格,
操作,
集体缩进,
操作缩进
)
仅包含非平凡集的 dTensor 操作
elif (
包含_DTensor_ops
和 (
操作名称 not
在
平凡操作)
和
长度(
操作[
输入形状])
):
表格 =
添加操作(
表格,
操作,
集体缩进,
操作缩进
)
返回
表格
如果
长度(
前进集体)
或
长度(
前进操作):
表格 += f"{
缩进}
前向传递
输入文本翻译为简体中文为：\n"
表格 =
添加跟踪信息(
表格,
前向集体操作,
前向操作
)
如果
长度(
向后集合)
或
长度(
向后操作):
表格 += f"{
缩进}
反向传播
输入文本翻译为简体中文为：\n"
表格 =
添加跟踪信息(
表格,
向后集合,
向后操作
)
如果
长度(
检查点操作):
表格 += f"{
缩进}
激活检查点
输入文本翻译为简体中文为：\n"
表格 =
添加跟踪信息(
表格, {},
检查点操作)
返回
表格
def _获取操作列表(
自身,
模块操作次数):
前向操作 = [
操作符
为
操作符
在
模块操作次数[
操作列表]
如果 not
操作[
是否为向后]
]
向后操作 = [
操作符
为
操作符
在
模块操作次数[
操作列表]
如果
操作[
是否为向后]
和 not
操作[
"激活检查点"]
]
检查点操作 = [
操作符
为
操作符
在
模块操作计数[
"操作列表"]
如果
操作[
激活检查点]
]
返回
前向操作,
反向操作,
检查点操作
[文档] def get_total_counts(self) -> int:
return sum(self.comm_counts.values())
[文档] def get_comm_counts(self) -> dict[Any, int]:
"""返回通信计数作为字典。
返回值：
Dict[Any, int]: 通信计数被视为字典。
"""
返回 self.comm_counts
[文档] def get_parameter_info(self) -> dict[str, dict[str, Any]]:
return self.advanced_module_tracker.module_parameters_dict
[文档] def get_sharding_info(self) -> dict[str, dict[str, Any]]:
return self.advanced_module_tracker.sharding_dict
def __进入__(
自身):
自身.
通信计数.
清晰()
自身.
通信模块计数.
清晰()
自身.
通信模块计数[
全球] = {}
自身.
通信模块计数[
全球
]
[前进] = defaultdict(int)
自身.
通信模块计数[
全球
]
["向后"] = defaultdict(int)
自身.
通信模块操作计数.
清晰()
超级().
__进入__()
自身.
高级模块跟踪器.
__进入__()
返回 self
def __退出__(
自身, *
参数):
自身.
高级模块跟踪器.
__退出__()
超级().
__退出__(*
参数)
[文档] def log_comm_debug_tracing_table_to_file(
self, 文件名="comm_mode_log.txt", 噪声级别=3
):
"""
用户指定的文件中写入，作为控制台 CommDebugMode 输出的替代方案
"""
ansi_escape = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")
table = ansi_escape.sub("", self.generate_comm_debug_tracing_table(noise_level))
使用 open(file_name, "w") as log_file:
log_file.write(table)
def 设置噪声参数(
自身,
噪音水平):
""
设置根据噪音水平显示信息的变量
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
包含 DTensor 操作 =
假
包含模块数据 =
假
包含操作 =
假
包含琐碎操作 =
假
如果
噪声水平 > 0:
包含 DTensor 操作 =
真实
包含模块数据 =
真实
如果
噪声水平 > 1:
包含操作 =
真实
如果
噪声水平 > 2:
包含简单操作 =
真实
返回 (
包含 DTensor 操作,
包含模块数据,
包含操作,
include_trivial_ops,
)
def __torch_dispatch__(自身,
函数,
类型,
参数=(), kwargs=
无):
当以 DTensor 模式运行此模式时，通常所有模式都会
在子类有机会运行之前
在这里返回 NotImplemented，给我们机会让 DTensor
在 CommDebugMode 看到它们之前，运行并解糖为通信操作。
设置操作级别的集体计数。
如果
自身.
高级模块跟踪器.
名称 not
在
自身.
通信模块操作计数:
# 模块输入输出形状、操作列表和集体计数器
自身.comm_module_operation_counts[
自身.advanced_module_tracker.
名称] = {
"操作列表": []
}
操作字典 = {}
操作字典[
名称] =
函数
操作字典[
输入形状] = []
操作字典[
输入分片] = []
操作字典[
设备网格] =
请提供需要翻译的文本
# 跟踪操作是否为反向传播的一部分
操作字典[
is_bw] =
自身.
高级模块跟踪器.is_bw
# 跟踪操作是否为激活检查点的一部分
操作字典[
"是否激活检查点"] = (
自身.
高级模块跟踪器.
激活检查点
)
如果
任何(t == DTensor
为 t
在
类型):
为
元素
在
参数:
如果 isinstance(
元素, DTensor):
保存所有 DTensor 参数的形状和位置
操作字典[
输入形状].
追加(
元素.shape)
操作字典[
输入分片].
追加(
元素.
安排)
操作字典[
设备网格] =
字符串(
元素.
装置网状结构)
自身.
通信模块操作次数[
自身.
高级模块跟踪器.
名称
]
操作列表
].追加(
操作字典)
返回 NotImplemented
kwargs = kwargs 如果 kwargs
否则 {}
out = 函数(*
参数, **kwargs)
功能包 =
函数._overloadpacket
我们有很多测试使用 CommDebugMode 来验证发生的
集合。这些测试通过查询 comm_counts 与传统的
funcol 操作作为键。为了原生 funcol 迁移，我们
需要这些测试同时适用于遗留和原生 funcol 功能。为了避免
需要修改所有测试以适应两种实现，我们
将 CommDebugMode 翻译为在迁移完成之前将原生 funcol 操作转换为遗留 funcol 操作。
操作。
如果
功能包
在
自身.
通信注册表
或
功能包
在
c10d 集体操作:
如果 func_packet
在 NATIVE_TO_PY_MAPPING:
func_packet = NATIVE_TO_PY_MAPPING[功能包]
自身.
通信计数[
功能包] += 1
key = 向前
如果
自身.
高级模块跟踪器.
是黑白的:
key = “向后”
为当前模块添加集体计数
如果
自身.
高级模块跟踪器.
名称 not
在
自身.
通信模块计数:
自身.
通信模块计数[
自身.
高级模块跟踪器.
名称] = {}
自身.
通信模块数量[
自身.
高级模块跟踪器.
名称
]
向前
] = defaultdict(int)
自身.
通信模块数量[
自身.
高级模块跟踪器.
名称
]
“向后”
] = defaultdict(int)
自身.comm_module_counts[
自身.
高级模块跟踪器.
名称
]
[键
]
func_packet
] += 1
# 为父模块添加集体计数
为 par
在
自身.
高级模块跟踪器.module_parents_dict[
自身.
高级模块跟踪器.
名称
]：
确保在当前子模块尚未从父模块中删除时，我们不重复计数
如果 par !=
自身.
高级模块跟踪器.
名称:
如果
句子 not
在
自身.
通信模块计数:
自身.
通信模块计数[
句子] = {}
自身.comm_module_counts[par
]]翻译
前进] = defaultdict(int)
自身.
通信模块计数[
段
]
["向后"] = defaultdict(int)
自身.
通信模块计数[
段
]
[键
]
[func_packet] += 1
如果张量操作使用伪造的张量，则返回
如果
检测伪造模式(
参数):
返回 out
将模块操作列表添加张量操作
自身.
通信模块操作次数[
自身.
高级模块跟踪器.
名称
]
"操作列表"
].追加(
操作字典)
返回 out
