Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "ccu_context_reduce_tail_block.h"
12 : #include "ccu_instruction_reduce_tail_block.h"
13 :
14 : namespace Hccl {
15 :
16 : constexpr int INPUT_XN_ID = 0;
17 : constexpr int OUTPUT_XN_ID = 1;
18 : constexpr int TOKEN_XN_ID = 2;
19 :
20 : constexpr uint64_t TAIL_BLOCK_LOOP_NUM = 64;
21 : constexpr uint64_t MISSION_NUM_2 = 2;
22 :
23 0 : CcuContextReduceTailBlock::CcuContextReduceTailBlock(const CcuCtxArg &arg,
24 : const std::vector<CcuTransport *> &transports,
25 0 : const CcuTransportGroup &group)
26 0 : : CcuContextAlgBase(arg, transports, group)
27 : {
28 0 : const CcuCtxArgReduceTailBlock *ctxArg = dynamic_cast<const CcuCtxArgReduceTailBlock *>(&arg);
29 0 : if (ctxArg == nullptr) {
30 0 : THROW<NullPtrException>(StringFormat("CcuContextReduceTailBlock::ctxArg ptr is null"));
31 : }
32 0 : rankId_ = ctxArg->rankId_;
33 0 : rankSize_ = ctxArg->dimSize_[0];
34 0 : notifySignal_ = ctxArg->notifySignal_;
35 0 : reduceOp_ = ctxArg->op_.reduceOp;
36 0 : dataType_ = ctxArg->op_.dataType;
37 0 : outputDataType_ = ctxArg->op_.outputDataType;
38 0 : if (outputDataType_ == DataType::INVALID) {
39 0 : outputDataType_ = dataType_;
40 0 : HCCL_INFO("[CcuContextReduceTailBlock] outputDataType is [INVALID], set outputDataType to[%s]",
41 : outputDataType_.Describe().c_str());
42 : }
43 :
44 0 : HCCL_INFO("[CcuContextReduceTailBlock] Init, CtxArgs are rankId[%u], rankSize[%llu], notifySignal[%s], "
45 : "reduceOp[%s], dataType[%s], outputDataType[%s]",
46 : rankId_, rankSize_, notifySignal_.c_str(), reduceOp_.Describe().c_str(), dataType_.Describe().c_str(),
47 : outputDataType_.Describe().c_str());
48 0 : }
49 :
50 0 : void CcuContextReduceTailBlock::Algorithm()
51 : {
52 0 : HCCL_INFO("[CcuContextReduceTailBlock] Algorithm start");
53 0 : InitResource();
54 0 : ExportVariables();
55 0 : SyncMainBlock(0); // 与住任务进行前同步， 主 mission 激活尾块 mission
56 :
57 0 : DoGroupReduce();
58 :
59 0 : SyncMainBlock(1); // 与住任务进行后同步， 通知主 mission 尾块处理完成
60 0 : HCCL_INFO("[CcuContextReduceTailBlock] Algorithm end");
61 0 : return;
62 : }
63 :
64 0 : void CcuContextReduceTailBlock::SyncMainBlock(uint32_t ctxSignalIndex)
65 : {
66 0 : HCCL_INFO("[CcuContextReduceTailBlock] SyncMainBlock start, ctxSignalIndex[%u]", ctxSignalIndex);
67 0 : LocalCtxPost(mainBlockCtxSignal_, 1 << (1 + ctxSignalIndex * MISSION_NUM_2));
68 0 : LocalWait(tailBlockCtxSignal_, 1 << (0 + ctxSignalIndex * MISSION_NUM_2));
69 0 : HCCL_INFO("[CcuContextReduceTailBlock] SyncMainBlock end");
70 0 : }
71 :
72 0 : void CcuContextReduceTailBlock::InitResource()
73 : {
74 0 : HCCL_INFO("[CcuContextReduceTailBlock] InitResource start");
75 : // init resource
76 0 : output_ = CreateVariable();
77 0 : uint16_t transportIdx = 0;
78 0 : if (transports.size() == 0) {
79 0 : THROW<NullPtrException>(StringFormat("CcuContextReduceTailBlock transports is empty"));
80 : }
81 : // 按照rank号从小到大遍历transports，遇到本rank就填充本地资源，否则依次取远端资源，要求给框架返回的Link同样是按顺序排列的
82 0 : for (uint64_t peerId = 0; peerId < rankSize_; peerId++) {
83 0 : if (peerId == rankId_) {
84 0 : input_.push_back(CreateVariable());
85 0 : token_.push_back(CreateVariable());
86 : } else {
87 0 : HCCL_INFO("[CcuContextReduceTailBlock] MyRank[%u], PeerId[%llu], TransportId[%u]",
88 : rankId_, peerId, transportIdx);
89 0 : CHK_PRT_RET(transports[transportIdx] == nullptr,
90 : HCCL_ERROR("[CcuContextReduceTailBlock] Algorithm transport ptr is null"),);
91 0 : input_.push_back(CreateVariable((*transports[transportIdx]), INPUT_XN_ID)); // 获取transport中id=1的Var
92 0 : token_.push_back(CreateVariable((*transports[transportIdx]), TOKEN_XN_ID));
93 0 : transportIdx++;
94 : }
95 : }
96 :
97 0 : inputOffset_ = CreateVariable();
98 0 : outputOffset_ = CreateVariable();
99 0 : groupOpSize_ = CreateGroupOpSize();
100 0 : tailBlockCtxSignal_ = CreateMaskSignal();
101 :
102 : // init loop param
103 0 : AllocGoResource(TAIL_BLOCK_LOOP_NUM);
104 0 : HCCL_INFO("[CcuContextReduceTailBlock] InitResource end");
105 : }
106 :
107 0 : void CcuContextReduceTailBlock::DoGroupReduce()
108 : {
109 0 : HCCL_INFO("[CcuContextReduceTailBlock] DoGroupReduce start");
110 : // 初始化地址寄存器
111 0 : std::vector<CcuRep::Memory> reduceSrc;
112 0 : for (uint32_t rankIdx = 0; rankIdx < rankSize_; rankIdx++) {
113 0 : reduceSrc.push_back(CreateMemory());
114 : }
115 0 : CcuRep::Memory reduceDst = CreateMemory();
116 :
117 : // 填充地址
118 0 : uint32_t dstId = 0;
119 0 : uint32_t curId = 0;
120 : // SRC
121 0 : for (uint32_t rankIdx = 0; rankIdx < rankSize_; rankIdx++) {
122 0 : if (rankIdx != rankId_) {
123 0 : curId = dstId;
124 0 : dstId++;
125 : } else {
126 0 : curId = rankSize_ - 1;
127 : }
128 0 : reduceSrc[curId].addr = input_[rankIdx];
129 0 : reduceSrc[curId].addr += inputOffset_;
130 0 : reduceSrc[curId].token = token_[rankIdx];
131 : }
132 :
133 : // DST
134 0 : reduceDst.addr = output_;
135 0 : reduceDst.addr += outputOffset_;
136 0 : reduceDst.token = token_[rankId_];
137 :
138 : // 执行 reduce 操作
139 0 : GroupReduce(transports, reduceDst, reduceSrc, groupOpSize_, outputDataType_, dataType_, reduceOp_);
140 0 : HCCL_INFO("[CcuContextReduceTailBlock] DoGroupReduce end");
141 0 : return;
142 0 : }
143 :
144 0 : void CcuContextReduceTailBlock::ExportVariables()
145 : {
146 0 : HCCL_INFO("[CcuContextReduceTailBlock] ExportVariables start");
147 0 : ExportVariable(input_[rankId_], notifySignal_ + "_Input_Reduce_Tail_Block");
148 0 : ExportVariable(output_, notifySignal_ + "_Output_Reduce_Tail_Block");
149 0 : ExportVariable(token_[rankId_], notifySignal_ + "_Token_Reduce_Tail_Block");
150 0 : ExportVariable(inputOffset_, notifySignal_ + "_Input_Offset_Reduce_Tail_Block");
151 0 : ExportVariable(outputOffset_, notifySignal_ + "_Output_Offset_Reduce_Tail_Block");
152 0 : ExportVariable(groupOpSize_.addrOffset, notifySignal_ + "_AddrOffset_Reduce_Tail_Block");
153 0 : ExportVariable(groupOpSize_.loopParam, notifySignal_ + "_LoopParam_Reduce_Tail_Block");
154 0 : ExportVariable(groupOpSize_.parallelParam, notifySignal_ + "_ParallelParam_Reduce_Tail_Block");
155 0 : ExportVariable(groupOpSize_.residual, notifySignal_ + "_Residual_Reduce_Tail_Block");
156 :
157 0 : ExportMaskSignal(tailBlockCtxSignal_, notifySignal_ + "_CtxSync_Reduce_Tail_Block");
158 0 : mainBlockCtxSignal_ = ImportMaskSignal(notifySignal_ + "_CtxSync_Main_Block");
159 0 : HCCL_INFO("[CcuContextReduceTailBlock] ExportVariables end");
160 0 : }
161 :
162 0 : std::vector<uint64_t> CcuContextReduceTailBlock::GeneArgs(const CcuTaskArg &arg)
163 : {
164 : (void) arg; // MC2 要求尾块处理不能使用 CTX_ARG，参数需要通过主任务传入
165 0 : HCCL_INFO("[CcuContextReduceTailBlock] GeneArgs, TailBlock, no need to gene args");
166 0 : return {};
167 : }
168 : }
|
