已合并
[fa_v3]skip cpu check, register_sharding for npu_fusion_attention_v3 #30662
[fa_v3]skip cpu check, register_sharding for npu_fusion_attention_v3 #30662
已合并
王超创建于 2月9日
6 个文件变更+511-21
Mtest/distributed/tensor/test_attention_ops.py+187-0
@@ -262,6 +262,193 @@ class TestAttentionOps(NPUDTensorTestBase):
262 for placement in placements:262 for placement in placements:
263 test_placement_comb([placement], [placement], [placement])263 test_placement_comb([placement], [placement], [placement])
264 264 
265+ @SupportedDevices(['Ascend910B'])
266+ @skipIfUnsupportMultiNPU(2)
267+ @with_comms
268+ @parametrize(
269+ "sparse_mode,pre_tokens,next_tokens",
270+ [
271+ (0, 128, 128),
272+ (1, 65536, 65536),
273+ (2, 65536, 0),
274+ (3, 65536, 0),
275+ (4, 128, 128)
276+ ]
277+ )
278+ def test_npu_fusion_attention_v3_forward_bnsd(self, sparse_mode, pre_tokens, next_tokens):
279+ device_mesh = self.build_device_mesh()
280+ 
281+ B, N, S, D = 4, 8, 32, 32
282+ shape = (B, N, S, D)
283+ query = torch.randn(shape, dtype=torch.float32, device="npu")
284+ key = torch.randn(shape, dtype=torch.float32, device="npu")
285+ value = torch.randn(shape, dtype=torch.float32, device="npu")
286+ 
287+ scale = 0.08838
288+ 
289+ atten_mask = get_atten_mask(shape, sparse_mode, pre_tokens, next_tokens)
290+ result = torch_npu.npu_fusion_attention_v3(
291+ query, key, value, head_num=N, input_layout="BNSD", scale=scale, sparse_mode=sparse_mode,
292+ atten_mask=atten_mask, pre_tockens=pre_tokens, next_tockens=next_tokens
293+ )
294+ 
295+ def test_placement_comb(query_placements, key_placements, value_placements, atten_mask_placements):
296+ dist_query = distribute_tensor(query, device_mesh, query_placements)
297+ dist_key = distribute_tensor(key, device_mesh, key_placements)
298+ dist_value = distribute_tensor(value, device_mesh, value_placements)
299+ dist_atten_mask = distribute_tensor(
300+ atten_mask, device_mesh, atten_mask_placements
301+ ) if atten_mask is not None else None
302+ dist_result = torch_npu.npu_fusion_attention_v3(
303+ dist_query, dist_key, dist_value, head_num=N, input_layout="BNSD", scale=scale,
304+ sparse_mode=sparse_mode, atten_mask=dist_atten_mask,
305+ pre_tockens=pre_tokens, next_tockens=next_tokens
306+ )
307+ self.assertEqual(dist_result[0].full_tensor(), result[0])
308+ self.assertEqual(dist_result[1].full_tensor(), result[1])
309+ self.assertEqual(dist_result[2].full_tensor(), result[2])
310+ 
311+ placements = [Shard(0), Shard(1), Shard(2), Shard(3), Replicate()]
312+ for placement in placements:
313+ if atten_mask is None or (isinstance(placement, Shard) and atten_mask.ndim <= placement.dim):
314+ test_placement_comb([placement], [placement], [placement], [Replicate()])
315+ else:
316+ test_placement_comb([placement], [placement], [placement], [placement])
317+ 
318+ @SupportedDevices(['Ascend910B'])
319+ @skipIfUnsupportMultiNPU(2)
320+ @with_comms
321+ @parametrize(
322+ "sparse_mode,pre_tokens,next_tokens",
323+ [
324+ (0, 128, 128),
325+ (1, 65536, 65536),
326+ (2, 65536, 0),
327+ (3, 65536, 0),
328+ (4, 128, 128)
329+ ]
330+ )
331+ def test_npu_fusion_attention_v3_backward_bnsd(self, sparse_mode, pre_tokens, next_tokens):
332+ device_mesh = self.build_device_mesh()
333+ 
334+ B, N, S, D = 4, 8, 32, 32
335+ shape = (B, N, S, D)
336+ query = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
337+ key = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
338+ value = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
339+ 
340+ scale = 0.08838
341+ 
342+ atten_mask = get_atten_mask(shape, sparse_mode, pre_tokens, next_tokens)
343+ result = torch_npu.npu_fusion_attention_v3(
344+ query, key, value, head_num=N, input_layout="BNSD", scale=scale, sparse_mode=sparse_mode,
345+ atten_mask=atten_mask, pre_tockens=pre_tokens, next_tockens=next_tokens
346+ )
347+ gard_y = torch.ones_like(result[0])
348+ result[0].backward(gard_y)
349+ 
350+ def test_placement_comb(query_placements, key_placements, value_placements, atten_mask_placements):
351+ dist_query = distribute_tensor(query, device_mesh, query_placements)
352+ dist_key = distribute_tensor(key, device_mesh, key_placements)
353+ dist_value = distribute_tensor(value, device_mesh, value_placements)
354+ dist_atten_mask = distribute_tensor(
355+ atten_mask, device_mesh, atten_mask_placements
356+ ) if atten_mask is not None else None
357+ dist_result = torch_npu.npu_fusion_attention_v3(
358+ dist_query, dist_key, dist_value, head_num=N, input_layout="BNSD", scale=scale,
359+ sparse_mode=sparse_mode, atten_mask=dist_atten_mask,
360+ pre_tockens=pre_tokens, next_tockens=next_tokens
361+ )
362+ dist_grad_y = distribute_tensor(gard_y, device_mesh, dist_result[0].placements)
363+ dist_result[0].backward(dist_grad_y)
364+ self.assertEqual(dist_result[0].full_tensor(), result[0])
365+ self.assertEqual(dist_result[1].full_tensor(), result[1])
366+ self.assertEqual(dist_result[2].full_tensor(), result[2])
367+ self.assertEqual(dist_query.grad.full_tensor(), query.grad)
368+ self.assertEqual(dist_key.grad.full_tensor(), key.grad)
369+ self.assertEqual(dist_value.grad.full_tensor(), value.grad)
370+ 
371+ placements = [Shard(0), Shard(1), Shard(2), Shard(3), Replicate()]
372+ for placement in placements:
373+ if atten_mask is None or (isinstance(placement, Shard) and atten_mask.ndim <= placement.dim):
374+ test_placement_comb([placement], [placement], [placement], [Replicate()])
375+ else:
376+ test_placement_comb([placement], [placement], [placement], [placement])
377+ 
378+ @SupportedDevices(['Ascend910B'])
379+ @skipIfUnsupportMultiNPU(2)
380+ @with_comms
381+ def test_npu_fusion_attention_v3_bsnd(self):
382+ device_mesh = self.build_device_mesh()
383+ 
384+ B, N, S, D = 4, 8, 32, 32
385+ shape = (B, S, N, D)
386+ query = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
387+ key = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
388+ value = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
389+ scale = 0.08838
390+ 
391+ result = torch_npu.npu_fusion_attention_v3(query, key, value, head_num=N, input_layout="BSND", scale=scale)
392+ gard_y = torch.ones_like(result[0])
393+ result[0].backward(gard_y)
394+ 
395+ def test_placement_comb(query_placements, key_placements, value_placements):
396+ dist_query = distribute_tensor(query, device_mesh, query_placements)
397+ dist_key = distribute_tensor(key, device_mesh, key_placements)
398+ dist_value = distribute_tensor(value, device_mesh, value_placements)
399+ dist_result = torch_npu.npu_fusion_attention_v3(
400+ dist_query, dist_key, dist_value, head_num=N, input_layout="BSND", scale=scale
401+ )
402+ dist_grad_y = distribute_tensor(gard_y, device_mesh, dist_result[0].placements)
403+ dist_result[0].backward(dist_grad_y)
404+ self.assertEqual(dist_result[0].full_tensor(), result[0])
405+ self.assertEqual(dist_result[1].full_tensor(), result[1])
406+ self.assertEqual(dist_result[2].full_tensor(), result[2])
407+ self.assertEqual(dist_query.grad.full_tensor(), query.grad)
408+ self.assertEqual(dist_key.grad.full_tensor(), key.grad)
409+ self.assertEqual(dist_value.grad.full_tensor(), value.grad)
410+ 
411+ placements = [Shard(0), Shard(1), Shard(2), Shard(3), Replicate()]
412+ for placement in placements:
413+ test_placement_comb([placement], [placement], [placement])
414+ 
415+ @SupportedDevices(['Ascend910B'])
416+ @skipIfUnsupportMultiNPU(2)
417+ @with_comms
418+ def test_npu_fusion_attention_v3_bsh(self):
419+ device_mesh = self.build_device_mesh()
420+ 
421+ B, N, S, D = 4, 8, 32, 32
422+ shape = (B, S, N * D)
423+ query = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
424+ key = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
425+ value = torch.randn(shape, dtype=torch.float32, device="npu", requires_grad=True)
426+ scale = 0.08838
427+ 
428+ result = torch_npu.npu_fusion_attention_v3(query, key, value, head_num=N, input_layout="BSH", scale=scale)
429+ gard_y = torch.ones_like(result[0])
430+ result[0].backward(gard_y)
431+ 
432+ def test_placement_comb(query_placements, key_placements, value_placements):
433+ dist_query = distribute_tensor(query, device_mesh, query_placements)
434+ dist_key = distribute_tensor(key, device_mesh, key_placements)
435+ dist_value = distribute_tensor(value, device_mesh, value_placements)
436+ dist_result = torch_npu.npu_fusion_attention_v3(
437+ dist_query, dist_key, dist_value, head_num=N, input_layout="BSH", scale=scale
438+ )
439+ dist_grad_y = distribute_tensor(gard_y, device_mesh, dist_result[0].placements)
440+ dist_result[0].backward(dist_grad_y)
441+ self.assertEqual(dist_result[0].full_tensor(), result[0])
442+ self.assertEqual(dist_result[1].full_tensor(), result[1])
443+ self.assertEqual(dist_result[2].full_tensor(), result[2])
444+ self.assertEqual(dist_query.grad.full_tensor(), query.grad)
445+ self.assertEqual(dist_key.grad.full_tensor(), key.grad)
446+ self.assertEqual(dist_value.grad.full_tensor(), value.grad)
447+ 
448+ placements = [Shard(0), Shard(1), Shard(2), Replicate()]
449+ for placement in placements:
450+ test_placement_comb([placement], [placement], [placement])
451+ 
265 452 
266instantiate_parametrized_tests(TestAttentionOps)453instantiate_parametrized_tests(TestAttentionOps)
267 454 
Mtorch_npu/_inductor/config.py+20-0
@@ -51,6 +51,26 @@ class aot_inductor:
51 dump_path_py = os.environ.get("AOTI_DUMP_PATH_PY", "aoti_dump_py")51 dump_path_py = os.environ.get("AOTI_DUMP_PATH_PY", "aoti_dump_py")
52 52 
53 53 
54+class _npugraph_trees:
55+ def __init__(self):
56+ # skip cpu node check, eg: npu_fusion_attention_v3
57+ self._disable_cpu_input_check = False
58+ 
59+ @property
60+ def disable_cpu_input_check(self):
61+ return self._disable_cpu_input_check
62+ 
63+ @disable_cpu_input_check.setter
64+ def disable_cpu_input_check(self, value):
65+ self._disable_cpu_input_check = bool(value)
66+ # When disable_cpu_input_check is True, set slow_path_cudagraph_asserts to True to skip the CPU check.
67+ if value:
68+ torch._inductor.config.triton.slow_path_cudagraph_asserts = False
69+ 
70+ 
71+npugraph_trees = _npugraph_trees()
72+ 
73+ 
54traced_fx_graph_cache = os.environ.get("INDUCTOR_ASCEND_FX_GRAPH_CACHE", None)74traced_fx_graph_cache = os.environ.get("INDUCTOR_ASCEND_FX_GRAPH_CACHE", None)
55check_accuracy = os.environ.get("INDUCTOR_ASCEND_CHECK_ACCURACY", False)75check_accuracy = os.environ.get("INDUCTOR_ASCEND_CHECK_ACCURACY", False)
56auto_fallback = os.environ.get("INDUCTOR_ASCEND_AUTO_FALLBACK", True)76auto_fallback = os.environ.get("INDUCTOR_ASCEND_AUTO_FALLBACK", True)
Mtorch_npu/_logging/_internal.py+1-0
@@ -42,3 +42,4 @@ def _add_logging_module():
42 torch._logging._internal.register_log("shmem", "torch_npu.symmetric_memory")42 torch._logging._internal.register_log("shmem", "torch_npu.symmetric_memory")
43 torch._logging._internal.register_log("env", "torch_npu.env")43 torch._logging._internal.register_log("env", "torch_npu.env")
44 torch._logging._internal.register_log("acl", "torch_npu.acl")44 torch._logging._internal.register_log("acl", "torch_npu.acl")
45+ torch._logging._internal.register_log("aclgraph", "torch_npu.aclgraph")
Mtorch_npu/distributed/tensor/_attention.py+281-8
@@ -271,6 +271,267 @@ def npu_fusion_attention_grad_strategy(query, key, value, dy, head_num, input_la
271 return strategies271 return strategies
272 272 
273 273 
274+@register_sharding(npu.npu_fusion_attention_v3.default)
AtlasAccount
AtlasAccountAtlasAccount2月9日

代码重复: npu_fusion_attention_v3_strategy函数与npu_fusion_attention_strategy函数(第27-139行)的代码结构几乎完全相同,存在大量重复代码。两个函数都实现了相同的分片策略逻辑,只是注册的算子不同。这种重复违反了DRY(Don't Repeat Yourself)原则,增加了维护成本和出错风险。

问题类型: 代码重复 文件路径: torch_npu/distributed/tensor/_attention.py 行号: 274 问题代码:

@register_sharding(npu.npu_fusion_attention_v3.default)
# pylint:disable=huawei-too-many-arguments
def npu_fusion_attention_v3_strategy(query, key, value, head_num, input_layout, pse=None, padding_mask=None,
                                  atten_mask=None, scale=1.0, keep_prob=1.0, pre_tockens=2147483647,
                                  next_tockens=2147483647, inner_precise=0, prefix=None, actual_seq_qlen=None,
                                  actual_seq_kvlen=None, sparse_mode=0, gen_mask_parallel=True, sync=False,
                                  softmax_layout="", sink=None):

修改建议:

建议将公共的分片策略逻辑提取为辅助函数,让两个策略函数共享相同的实现。例如:
1. 创建一个通用的_attention_strategy_impl函数,接受算子参数并返回策略列表
2. 让npu_fusion_attention_strategy和npu_fusion_attention_v3_strategy都调用这个通用实现
3. 这样可以减少代码重复,便于未来维护和扩展

此评论由代码审查工具自动生成

likedislike
275+# pylint:disable=huawei-too-many-arguments
276+def npu_fusion_attention_v3_strategy(query, key, value, head_num, input_layout, pse=None, padding_mask=None,
277+ atten_mask=None, scale=1.0, keep_prob=1.0, pre_tockens=2147483647,
AtlasAccount
AtlasAccountAtlasAccount2月9日

拼写错误: 在npu_fusion_attention_v3_strategy函数的参数列表中,参数名'pre_tockens'和'next_tockens'中的'tockens'疑似拼写错误,应该是'tokens'。这个拼写错误在原始函数npu_fusion_attention_strategy中也存在(第30行),但在新添加的v3函数中重复了这个错误。

问题类型: 拼写错误 文件路径: torch_npu/distributed/tensor/_attention.py 行号: 277 问题代码:

pre_tockens=2147483647,
next_tockens=2147483647,

修改建议:

修正参数名拼写错误,将'tockens'改为'tokens',以保持与常见术语的一致性。如果这是API的一部分且不能更改,至少应在注释中说明。

此评论由代码审查工具自动生成

likedislike
278+ next_tockens=2147483647, inner_precise=0, prefix=None, actual_seq_qlen=None,
279+ actual_seq_kvlen=None, sparse_mode=0, gen_mask_parallel=True, sync=False,
280+ softmax_layout="", sink=None):
281+ strategies = []
282+ 
283+ # all replicate strategy
284+ replicate_strategy = (
285+ [
286+ Replicate(), # attention_out
287+ Replicate(), # softmax_max
288+ Replicate(), # softmax_sum
289+ Replicate(), # softmax_out(reserve, unused now)
290+ Replicate(), # seed
291+ Replicate() # offset
292+ ],
293+ [
294+ Replicate(), # query
295+ Replicate(), # key
296+ Replicate(), # value
297+ None, # head_num
298+ None, # input_layout
299+ None if pse is None else Replicate(), # pse
300+ None if padding_mask is None else Replicate(), # padding_mask
301+ None if atten_mask is None else Replicate(), # atten_mask
302+ None, None, None, None, None, None, # others
303+ None if actual_seq_qlen is None else Replicate(), # actual_seq_qlen
304+ None if actual_seq_kvlen is None else Replicate(), # actual_seq_kvlen
305+ None, None, None, None, # others
306+ None if sink is None else Replicate() # sink
307+ ]
308+ )
309+ strategies.append(replicate_strategy)
310+ 
311+ # only support sharding for sdpa currently, in which pse and padding_mask are not used
312+ # keep_prob < 1.0 may effect different results under sharding
313+ unused_args_in_sdpa = [pse, padding_mask, prefix, actual_seq_qlen, actual_seq_kvlen, sink]
314+ if not all(arg is None for arg in unused_args_in_sdpa) or keep_prob < 1.0:
315+ return strategies
316+ 
317+ # input layout: BSH, SBH, BSND, BNSD, TND
318+ # atten_mask layout: BNSS, B1SS, 11SS, SS
319+ # dp sharding strategy
320+ if 'B' in input_layout:
321+ batch_dim = input_layout.index('B')
322+ atten_mask_sharding = None
323+ if atten_mask is not None:
324+ if atten_mask.ndim == 4 and atten_mask.shape[0] != 1: # BNSS, B1SS
325+ atten_mask_sharding = Shard(0)
326+ else: # 11SS, SS
327+ atten_mask_sharding = Replicate()
328+ dp_sharding_strategy = (
329+ [
330+ Shard(batch_dim), # attention_out
331+ Shard(0), # softmax_max layout: BNS8
332+ Shard(0), # softmax_sum layout: BNS8
333+ Replicate(), # softmax_out(reserve, unused now)
334+ Replicate(), # seed
335+ Replicate() # offset
336+ ],
337+ [
338+ Shard(batch_dim), # query
339+ Shard(batch_dim), # key
340+ Shard(batch_dim), # value
341+ None, # head_num
342+ None, # input_layout
343+ None, # pse
344+ None, # padding_mask
345+ atten_mask_sharding, # atten_mask
346+ None, None, None, None, None, None, # others
347+ None if actual_seq_qlen is None else Replicate(), # actual_seq_qlen
348+ None if actual_seq_kvlen is None else Replicate(), # actual_seq_kvlen
349+ None, None, None, None, # others
350+ None # sink
351+ ]
352+ )
353+ strategies.append(dp_sharding_strategy)
354+ 
355+ # add tp sharding strategy
356+ if 'N' in input_layout:
357+ head_dim = input_layout.index('N')
358+ atten_mask_sharding = None
359+ if atten_mask is not None:
360+ if atten_mask.ndim == 4 and atten_mask.shape[1] != 1: # BNSS
361+ atten_mask_sharding = Shard(1)
362+ else:
363+ atten_mask_sharding = Replicate() # B1SS, 11SS, SS
364+ tp_sharding_strategy = (
365+ [
366+ Shard(head_dim), # attention_out
367+ Shard(1), # softmax_max layout: BNS8
368+ Shard(1), # softmax_sum layout: BNS8
369+ Replicate(), # softmax_out(reserve, unused now)
370+ Replicate(), # seed
371+ Replicate() # offset
372+ ],
373+ [
374+ Shard(head_dim), # query
375+ Shard(head_dim), # key
376+ Shard(head_dim), # value
377+ None, # head_num
378+ None, # input_layout
379+ None, # pse
380+ None, # padding_mask
381+ atten_mask_sharding, # atten_mask
382+ None, None, None, None, None, None, # others
383+ None if actual_seq_qlen is None else Replicate(), # actual_seq_qlen
384+ None if actual_seq_kvlen is None else Replicate(), # actual_seq_kvlen
385+ None, None, None, None, # others
386+ None # sink
387+ ]
388+ )
389+ strategies.append(tp_sharding_strategy)
390+ 
391+ return strategies
392+ 
393+ 
394+@register_sharding(npu.npu_fusion_attention_grad_v3.default)
AtlasAccount
AtlasAccountAtlasAccount2月9日

代码重复: npu_fusion_attention_grad_v3_strategy函数与npu_fusion_attention_grad_strategy函数(第142-271行)的代码结构几乎完全相同,存在大量重复代码。两个函数都实现了相同的梯度分片策略逻辑,只是注册的算子不同。这种重复违反了DRY原则,增加了维护成本和出错风险。

问题类型: 代码重复 文件路径: torch_npu/distributed/tensor/_attention.py 行号: 382 问题代码:

@register_sharding(npu.npu_fusion_attention_grad_v3.default)
def npu_fusion_attention_grad_v3_strategy(query, key, value, dy, head_num, input_layout, pse=None, padding_mask=None,
                                       atten_mask=None, softmax_max=None, softmax_sum=None, softmax_in=None,
                                       attention_in=None, scale_value=1., keep_prob=1., pre_tockens=2147483647,
                                       next_tockens=2147483647, inner_precise=0, seed=None, offset=None,
                                       prefix=None, actual_seq_qlen=None, actual_seq_kvlen=None, sparse_mode=0,
                                       gen_mask_parallel=True, sync=False, softmax_layout="", sink=None):

修改建议:

建议将公共的梯度分片策略逻辑提取为辅助函数,让两个梯度策略函数共享相同的实现。例如:
1. 创建一个通用的_attention_grad_strategy_impl函数,接受算子参数并返回策略列表
2. 让npu_fusion_attention_grad_strategy和npu_fusion_attention_grad_v3_strategy都调用这个通用实现
3. 这样可以减少代码重复,便于未来维护和扩展

此评论由代码审查工具自动生成

likedislike
395+def npu_fusion_attention_grad_v3_strategy(query, key, value, dy, head_num, input_layout, pse=None, padding_mask=None,
396+ atten_mask=None, softmax_max=None, softmax_sum=None, softmax_in=None,
397+ attention_in=None, scale_value=1., keep_prob=1., pre_tockens=2147483647,
AtlasAccount
AtlasAccountAtlasAccount2月9日

拼写错误: 在npu_fusion_attention_grad_v3_strategy函数的参数列表中,参数名'pre_tockens'和'next_tockens'中的'tockens'疑似拼写错误,应该是'tokens'。这个拼写错误在原始函数npu_fusion_attention_grad_strategy中也存在(第145行),但在新添加的v3函数中重复了这个错误。

问题类型: 拼写错误 文件路径: torch_npu/distributed/tensor/_attention.py 行号: 385 问题代码:

pre_tockens=2147483647,
next_tockens=2147483647,

修改建议:

修正参数名拼写错误,将'tockens'改为'tokens',以保持与常见术语的一致性。如果这是API的一部分且不能更改,至少应在注释中说明。

此评论由代码审查工具自动生成

likedislike
398+ next_tockens=2147483647, inner_precise=0, seed=None, offset=None,
AtlasAccount
AtlasAccountAtlasAccount2月9日

参数默认值不一致: 在npu_fusion_attention_grad_v3_strategy函数中,参数'seed'和'offset'的默认值为None(第386-387行),但在原始的npu_fusion_attention_grad_strategy函数中(第146行),这些参数的默认值是0。这种不一致可能导致调用方困惑,特别是当两个算子应该具有相似的接口时。

问题类型: 参数默认值不一致 文件路径: torch_npu/distributed/tensor/_attention.py 行号: 386 问题代码:

seed=None, offset=None,

修改建议:

检查算子接口定义,确保v3版本和原始版本的参数默认值保持一致。如果v3算子确实接受None值,应在注释中说明与原始版本的区别。

此评论由代码审查工具自动生成

likedislike
399+ prefix=None, actual_seq_qlen=None, actual_seq_kvlen=None, sparse_mode=0,
400+ gen_mask_parallel=True, sync=False, softmax_layout="", sink=None):
401+ strategies = []
402+ 
403+ # all replicate strategy
404+ replicate_strategy = (
405+ [
406+ Replicate(), # grad_query
407+ Replicate(), # grad_key
408+ Replicate(), # grad_value
409+ Replicate(), # grad_pse(reserve, unused now)
410+ Replicate() # grad_sink
411+ ],
412+ [
413+ Replicate(), # query
414+ Replicate(), # key
415+ Replicate(), # value
416+ Replicate(), # dy
417+ None, # head_num
418+ None, # input_layout
419+ None if pse is None else Replicate(), # pse
420+ None if padding_mask is None else Replicate(), # padding_mask
421+ None if atten_mask is None else Replicate(), # atten_mask
422+ None if softmax_max is None else Replicate(), # softmax_max
423+ None if softmax_sum is None else Replicate(), # softmax_sum
424+ None if softmax_in is None else Replicate(), # softmax_in(reserve, unused now)
425+ None if attention_in is None else Replicate(), # attention_in
426+ None, None, None, None, None, # others
427+ None if seed is None else Replicate(), # seed
428+ None if offset is None else Replicate(), # offset
429+ None,
430+ None if actual_seq_qlen is None else Replicate(), # actual_seq_qlen
431+ None if actual_seq_kvlen is None else Replicate(), # actual_seq_kvlen
432+ None, None, None, None, # others
433+ None if sink is None else Replicate() # sink
434+ ]
435+ )
436+ strategies.append(replicate_strategy)
437+ 
438+ # only support sharding for sdpa currently, in which pse and padding_mask are not used
439+ # keep_prob < 1.0 may effect different results under sharding
440+ unused_args_in_sdpa = [pse, padding_mask, prefix, actual_seq_qlen, actual_seq_kvlen, sink]
441+ if not all(arg is None for arg in unused_args_in_sdpa) or keep_prob < 1.0:
442+ return strategies
443+ 
444+ # input layout: BSH, SBH, BSND, BNSD, TND
445+ # atten_mask layout: BNSS, B1SS, 11SS, SS
446+ # dp sharding strategy
447+ if 'B' in input_layout:
448+ batch_dim = input_layout.index('B')
449+ atten_mask_sharding = None
450+ if atten_mask is not None:
451+ if atten_mask.ndim == 4 and atten_mask.shape[0] != 1: # BNSS, B1SS
452+ atten_mask_sharding = Shard(0)
453+ else: # 11SS, SS
454+ atten_mask_sharding = Replicate()
455+ dp_sharding_strategy = (
456+ [
457+ Shard(batch_dim), # grad_query
458+ Shard(batch_dim), # grad_key
459+ Shard(batch_dim), # grad_value
460+ Replicate(), # grad_pse(reserve, unused now)
461+ Replicate() # grad_sink(unsupported now)
462+ ],
463+ [
464+ Shard(batch_dim), # query
465+ Shard(batch_dim), # key
466+ Shard(batch_dim), # value
467+ Shard(batch_dim), # dy
468+ None, # head_num
469+ None, # input_layout
470+ None, # pse
471+ None, # padding_mask
472+ atten_mask_sharding, # atten_mask
473+ Shard(0) if softmax_max is not None else None, # softmax_max layout: BNS8
474+ Shard(0) if softmax_sum is not None else None, # softmax_sum layout: BNS8
475+ None if softmax_in is None else Replicate(), # softmax_in(reserve, unused now)
476+ Shard(batch_dim) if attention_in is not None else None, # attention_in
477+ None, None, None, None, None, # others
478+ None if seed is None else Replicate(), # seed
479+ None if offset is None else Replicate(), # offset
480+ None,
481+ None if actual_seq_qlen is None else Replicate(), # actual_seq_qlen
482+ None if actual_seq_kvlen is None else Replicate(), # actual_seq_kvlen
483+ None, None, None, None, # others
484+ None # sink
485+ ]
486+ )
487+ strategies.append(dp_sharding_strategy)
488+ 
489+ # add tp sharding strategy
490+ if 'N' in input_layout:
491+ head_dim = input_layout.index('N')
492+ atten_mask_sharding = None
493+ if atten_mask is not None:
494+ if atten_mask.ndim == 4 and atten_mask.shape[1] != 1: # BNSS
495+ atten_mask_sharding = Shard(1)
496+ else:
497+ atten_mask_sharding = Replicate() # B1SS, 11SS, SS
498+ tp_sharding_strategy = (
499+ [
500+ Shard(head_dim), # grad_query
501+ Shard(head_dim), # grad_key
502+ Shard(head_dim), # grad_value
503+ Replicate(), # grad_pse(reserve, unused now)
504+ Replicate() # grad_sink(unsupported now)
505+ ],
506+ [
507+ Shard(head_dim), # query
508+ Shard(head_dim), # key
509+ Shard(head_dim), # value
510+ Shard(head_dim), # dy
511+ None, # head_num
512+ None, # input_layout
513+ None, # pse
514+ None, # padding_mask
515+ atten_mask_sharding, # atten_mask
516+ Shard(1) if softmax_max is not None else None, # softmax_max layout: BNS8
517+ Shard(1) if softmax_sum is not None else None, # softmax_sum layout: BNS8
518+ None if softmax_in is None else Replicate(), # softmax_in(reserve, unused now)
519+ Shard(head_dim) if attention_in is not None else None, # attention_in
520+ None, None, None, None, None, # others
521+ None if seed is None else Replicate(), # seed
522+ None if offset is None else Replicate(), # offset
523+ None,
524+ None if actual_seq_qlen is None else Replicate(), # actual_seq_qlen
525+ None if actual_seq_kvlen is None else Replicate(), # actual_seq_kvlen
526+ None, None, None, None, # others
527+ None # sink
528+ ]
529+ )
530+ strategies.append(tp_sharding_strategy)
531+ 
532+ return strategies
533+ 
534+ 
274def _infer_npu_fusion_attention_grad_kwargs_spec(535def _infer_npu_fusion_attention_grad_kwargs_spec(
275 op_schema: OpSchema,536 op_schema: OpSchema,
276 output_sharding: OutputSharding537 output_sharding: OutputSharding
@@ -383,7 +644,7 @@ def _npu_fusion_attention_handler(
383 # computation that happens in the current rank of the mesh, normal case644 # computation that happens in the current rank of the mesh, normal case
384 local_args = get_redistributed_local_args(op_info, output_sharding)645 local_args = get_redistributed_local_args(op_info, output_sharding)
385 local_kwargs = op_info.local_kwargs646 local_kwargs = op_info.local_kwargs
386- if op_call == npu.npu_fusion_attention.default:647+ if op_call == npu.npu_fusion_attention.default or op_call == npu.npu_fusion_attention_v3.default:
387 # if sharding head_dim in qkv, need recalculate head_num in local args648 # if sharding head_dim in qkv, need recalculate head_num in local args
388 input_layout = op_info.local_args[4]649 input_layout = op_info.local_args[4]
389 if 'N' in input_layout:650 if 'N' in input_layout:
@@ -394,10 +655,15 @@ def _npu_fusion_attention_handler(
394 local_args = tuple(local_args)655 local_args = tuple(local_args)
395 656 
396 # run local op computation with potentially modified args/kwargs657 # run local op computation with potentially modified args/kwargs
397- local_results = torch_npu.npu_fusion_attention(658+ if op_call == npu.npu_fusion_attention.default:
398- *local_args, **local_kwargs659+ local_results = torch_npu.npu_fusion_attention(
399- )660+ *local_args, **local_kwargs
400- elif op_call == npu.npu_fusion_attention_grad.default:661+ )
662+ else:
663+ local_results = torch_npu.npu_fusion_attention_v3(
664+ *local_args, **local_kwargs
665+ )
666+ elif op_call == npu.npu_fusion_attention_grad.default or op_call == npu.npu_fusion_attention_grad_v3.default:
401 local_kwargs = get_redistributed_local_kwargs(667 local_kwargs = get_redistributed_local_kwargs(
402 _infer_npu_fusion_attention_grad_kwargs_spec, op_info, output_sharding668 _infer_npu_fusion_attention_grad_kwargs_spec, op_info, output_sharding
403 )669 )
@@ -409,9 +675,14 @@ def _npu_fusion_attention_handler(
409 local_query = local_args[0]675 local_query = local_args[0]
410 local_args[4] = local_query.size(head_dim)676 local_args[4] = local_query.size(head_dim)
411 local_args = tuple(local_args)677 local_args = tuple(local_args)
412- local_results = torch_npu.npu_fusion_attention_grad(678+ if op_call == npu.npu_fusion_attention_grad.default:
AtlasAccount
AtlasAccountAtlasAccount2月9日

代码重复: 在_npu_fusion_attention_handler函数中,处理npu_fusion_attention_grad和npu_fusion_attention_grad_v3的逻辑几乎完全相同(第636-655行),只是调用的算子不同。这种重复的if-else分支结构使得代码冗长,难以维护。

问题类型: 代码重复 文件路径: torch_npu/distributed/tensor/_attention.py 行号: 648 问题代码:

if op_call == npu.npu_fusion_attention_grad.default:
    local_results = torch_npu.npu_fusion_attention_grad(
        *local_args, **local_kwargs
    )
else:
    local_results = torch_npu.npu_fusion_attention_grad_v3(
        *local_args, **local_kwargs
    )

修改建议:

建议将算子调用逻辑抽象化,例如:
1. 创建一个算子名称到实际算子的映射字典
2. 根据op_call从映射中获取对应的算子函数
3. 统一调用,减少重复的if-else分支

此评论由代码审查工具自动生成

likedislike
413- *local_args, **local_kwargs679+ local_results = torch_npu.npu_fusion_attention_grad(
414- )680+ *local_args, **local_kwargs
681+ )
682+ else:
683+ local_results = torch_npu.npu_fusion_attention_grad_v3(
684+ *local_args, **local_kwargs
685+ )
415 else:686 else:
416 raise NotImplementedError(687 raise NotImplementedError(
417 "_npu_fusion_attention_handler only supports npu_fusion_attention and npu_fusion_attention_grad now."688 "_npu_fusion_attention_handler only supports npu_fusion_attention and npu_fusion_attention_grad now."
@@ -443,6 +714,8 @@ def _npu_fusion_attention_handler(
443customized_ops = {714customized_ops = {
444 npu.npu_fusion_attention.default: _npu_fusion_attention_handler,715 npu.npu_fusion_attention.default: _npu_fusion_attention_handler,
445 npu.npu_fusion_attention_grad.default: _npu_fusion_attention_handler,716 npu.npu_fusion_attention_grad.default: _npu_fusion_attention_handler,
717+ npu.npu_fusion_attention_v3.default: _npu_fusion_attention_handler,
718+ npu.npu_fusion_attention_grad_v3.default: _npu_fusion_attention_handler,
446}719}
447 720 
448old_handlers = DTensor._op_dispatcher._custom_op_handlers721old_handlers = DTensor._op_dispatcher._custom_op_handlers
Mtorch_npu/npu/_graph_tree.py+8-12
@@ -46,6 +46,7 @@ import threading
46import traceback46import traceback
47import warnings47import warnings
48import weakref48import weakref
49+import logging
49from collections import defaultdict50from collections import defaultdict
50from enum import auto, Enum51from enum import auto, Enum
51from typing import (52from typing import (
@@ -114,7 +115,7 @@ StorageWeakRefPointer = int
114StorageDataPtr = int115StorageDataPtr = int
115NBytes = int116NBytes = int
116S = TypeVar("S", bound="StorageWeakRefWrapper")117S = TypeVar("S", bound="StorageWeakRefWrapper")
117-log = torch._logging.getArtifactLogger(__name__, "cudagraphs")118+log = logging.getLogger("torch_npu.aclgraph")
118 119 
119 120 
120@dataclasses.dataclass(frozen=True)121@dataclasses.dataclass(frozen=True)
@@ -1270,7 +1271,8 @@ class NPUGraphNode:
1270 self.static_output_tensors = [None for _ in range(len(outputs))]1271 self.static_output_tensors = [None for _ in range(len(outputs))]
1271 1272 
1272 for index_, out_ in enumerate(outputs):1273 for index_, out_ in enumerate(outputs):
1273- if out_ is None or not isinstance(out_, torch.Tensor):1274+ from torch_npu._inductor import config as npu_config
1275+ if out_ is None or not isinstance(out_, torch.Tensor) or (npu_config.npugraph_trees.disable_cpu_input_check and out_.is_cpu):
1274 self.output_storage_alias.append(UnaliasedStorage)1276 self.output_storage_alias.append(UnaliasedStorage)
1275 continue1277 continue
1276 1278 
@@ -2184,11 +2186,7 @@ class NPUGraphTreeManager:
2184 if isinstance(self.current_node, NPUWarmupNode):2186 if isinstance(self.current_node, NPUWarmupNode):
2185 raise RuntimeError("self.current_node is NPUWarmupNode object")2187 raise RuntimeError("self.current_node is NPUWarmupNode object")
2186 graph_id = self.new_graph_id()2188 graph_id = self.new_graph_id()
2187- log.debug(2189+ log.debug(f"Recording function {function_id.id} of graph recording id {graph_id.id}")
2188- "Recording function %d of graph recording id %d",
2189- function_id.id,
2190- graph_id.id,
2191- )
2192 torch.npu.synchronize()2190 torch.npu.synchronize()
2193 node = NPUGraphNode(2191 node = NPUGraphNode(
2194 self.ids_to_funcs[function_id],2192 self.ids_to_funcs[function_id],
@@ -2216,6 +2214,7 @@ class NPUGraphTreeManager:
2216 self.current_node = node2214 self.current_node = node
2217 self.path_state = ExecutionState.EXECUTION2215 self.path_state = ExecutionState.EXECUTION
2218 self.update_generation()2216 self.update_generation()
2217+ log.debug(f"execute graph, id is {self.current_node.id}")
2219 return node.run(new_inputs)2218 return node.run(new_inputs)
2220 2219 
2221 def run_eager(2220 def run_eager(
@@ -2225,12 +2224,9 @@ class NPUGraphTreeManager:
2225 # we will deallocate it2224 # we will deallocate it
2226 already_warm = function_id in self.warmed_up_functions2225 already_warm = function_id in self.warmed_up_functions
2227 if not already_warm:2226 if not already_warm:
2228- log.debug("Running warmup of function %d", function_id.id)2227+ log.debug(f"Running warmup of function {function_id}")
2229 else:2228 else:
2230- log.debug(2229+ log.debug(f"Running eager of function {function_id} because ancestor needed to warm up")
2231- "Running eager of function %d because ancestor needed to warm up",
2232- function_id.id,
2233- )
2234 self.warmed_up_functions.add(function_id)2230 self.warmed_up_functions.add(function_id)
2235 node = NPUWarmupNode(2231 node = NPUWarmupNode(
2236 self.ids_to_funcs[function_id],2232 self.ids_to_funcs[function_id],
Mtorch_npu/utils/_graph_tree.py+14-1
@@ -1,4 +1,5 @@
1import functools1import functools
2+import logging
2from collections import defaultdict3from collections import defaultdict
3from typing import (4from typing import (
4 Any,5 Any,
@@ -53,6 +54,9 @@ from torch.multiprocessing.reductions import StorageWeakRef
53import torch_npu.npu.aclnn54import torch_npu.npu.aclnn
54 55 
55 56 
57+log = logging.getLogger("torch_npu.aclgraph")
58+ 
59+ 
56def npugraph_mark_step_begin():60def npugraph_mark_step_begin():
57 from torch_npu.npu._graph_tree import mark_step_begin61 from torch_npu.npu._graph_tree import mark_step_begin
58 mark_step_begin()62 mark_step_begin()
@@ -61,10 +65,15 @@ def npugraph_mark_step_begin():
61def check_multiple_devices_or_any_cpu_nodes(65def check_multiple_devices_or_any_cpu_nodes(
62 device_node_mapping: Dict[torch.device, torch.fx.Node]66 device_node_mapping: Dict[torch.device, torch.fx.Node]
63) -> Optional[str]:67) -> Optional[str]:
68+ from torch_npu._inductor import config as npu_config
69+ if npu_config.npugraph_trees.disable_cpu_input_check:
70+ device_node_mapping.pop(torch.device("cpu"), None)
71+ 
64 cpu_node = device_node_mapping.get(torch.device("cpu"))72 cpu_node = device_node_mapping.get(torch.device("cpu"))
65 if cpu_node:73 if cpu_node:
66 msg = f"cpu device ({cpu_node.name})"74 msg = f"cpu device ({cpu_node.name})"
67 stack_trace = _get_use_stack_trace(cpu_node)75 stack_trace = _get_use_stack_trace(cpu_node)
76+ log.info(f"skip with cpu node, msg is {msg}, stack_trace is {stack_trace}")
68 if stack_trace:77 if stack_trace:
69 return format_default_skip_message(f"{msg}. Found from : \n {stack_trace}")78 return format_default_skip_message(f"{msg}. Found from : \n {stack_trace}")
70 return format_default_skip_message(msg)79 return format_default_skip_message(msg)
@@ -251,7 +260,11 @@ def check_for_skip(aot_model: torch.fx.GraphModule, num_fixed) -> Optional[str]:
251 260 
252 261 
253def get_device_index(gm) -> int:262def get_device_index(gm) -> int:
254- device = next(iter(get_device_node_mapping(gm)))263+ device_node_mapping = get_device_node_mapping(gm)
264+ from torch_npu._inductor import config as npu_config
265+ if npu_config.npugraph_trees.disable_cpu_input_check:
266+ device_node_mapping.pop(torch.device("cpu"), None)
267+ device = next(iter(device_node_mapping))
255 if not (device.type == "npu"):268 if not (device.type == "npu"):
256 raise RuntimeError("check device.type == npu fail", )269 raise RuntimeError("check device.type == npu fail", )
257 return device.index270 return device.index