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 <iostream>
12 : #include <memory>
13 :
14 : #include "interpreter.h"
15 : #include "ins_rules.h"
16 : #include "aiv_ins.h"
17 :
18 : namespace Hccl {
19 10 : Interpreter::Interpreter(CommunicatorImpl &communicator)
20 250 : : comm(communicator), insRuleMap({{InstructionType::LOCAL_COPY, GetInsRule<InsLocalCopy>()},
21 10 : {InstructionType::LOCAL_REDUCE, GetInsRule<InsLocalReduce>()},
22 10 : {InstructionType::LOCAL_POST_TO, GetInsRule<InsLocalPostTo>()},
23 10 : {InstructionType::LOCAL_WAIT_FROM, GetInsRule<InsLocalWaitFrom>()},
24 10 : {InstructionType::LOCAL_WAIT_GROUP, GetInsRule<InsLocalWaitGroup>()},
25 10 : {InstructionType::LOCAL_BCAST_POST, GetInsRule<InsLocalBcastPost>()},
26 10 : {InstructionType::POST_READY, GetInsRule<InsPostReady>()},
27 10 : {InstructionType::WAIT_READY, GetInsRule<InsWaitReady>()},
28 10 : {InstructionType::POST_FIN, GetInsRule<InsPostFin>()},
29 10 : {InstructionType::WAIT_FIN, GetInsRule<InsWaitFin>()},
30 10 : {InstructionType::WAIT_GROUP_FIN, GetInsRule<InsWaitGroupFin>()},
31 10 : {InstructionType::POST_FIN_ACK, GetInsRule<InsPostFinAck>()},
32 10 : {InstructionType::WAIT_FIN_ACK, GetInsRule<InsWaitFinAck>()},
33 10 : {InstructionType::WRITE_WITH_FIN, GetInsRule<InsWriteWithFin>()},
34 10 : {InstructionType::WRITE_REDUCE_WITH_FIN, GetInsRule<InsWriteReduceWithFin>()},
35 10 : {InstructionType::WRITE, GetInsRule<InsWrite>()},
36 10 : {InstructionType::WRITE_REDUCE, GetInsRule<InsWriteReduce>()},
37 10 : {InstructionType::READ, GetInsRule<InsRead>()},
38 10 : {InstructionType::READ_REDUCE, GetInsRule<InsReadReduce>()},
39 10 : {InstructionType::CCU_INS, GetInsRule<CcuInstruction>()},
40 10 : {InstructionType::AICPU_INS, GetInsRule<AicpuInstruction>()},
41 10 : {InstructionType::AIV_INS, GetInsRule<AivInstruction>()}
42 10 : })
43 : {
44 10 : if (communicator.GetCurrentCollOperator()->opType == OpType::BARRIER) {
45 0 : taskConfig.SetNotifyWaitTime(communicator.GetNotifyTimeoutCfg().GetBarrierTimeout());
46 : } else {
47 10 : taskConfig.SetNotifyWaitTime(communicator.GetNotifyTimeoutCfg().GetNotifyTimeout());
48 : }
49 20 : }
50 :
51 7 : void Interpreter::Submit(const InsQueue &insQueue)
52 : {
53 7 : list<InsQueue::Iterator> slaveQueueIters;
54 8 : for (auto slaveQueueIter = insQueue.IterSlaves(); slaveQueueIter.HasNext(); ++slaveQueueIter) {
55 1 : slaveQueueIters.emplace_back((*slaveQueueIter).Iter());
56 7 : }
57 :
58 7 : std::set<u32> slaveStreamIndexSet;
59 8 : for (u32 slaveStreamIndex = 0; slaveStreamIndex < slaveQueueIters.size(); ++slaveStreamIndex) {
60 1 : slaveStreamIndexSet.insert(slaveStreamIndex);
61 : }
62 :
63 : // 获取指令规模，填充桶宽及初始化
64 7 : auto& streamMgr = comm.GetStreamManager();
65 7 : auto masterStream = streamMgr.GetMaster();
66 7 : streamMgr.InitBucket(UINT32_MAX);
67 7 : streamMgr.RecordStreamIdToIndex(masterStream->GetId(), UINT32_MAX);
68 7 : u32 index = 0;
69 8 : for (auto slaveIter = insQueue.IterSlaves(); slaveIter.HasNext(); ++slaveIter) {
70 1 : streamMgr.InitBucket(streamMgr.GetSlaveIndex());
71 1 : auto slaveStream = streamMgr.GetSlave();
72 1 : streamMgr.RecordStreamIdToIndex(slaveStream->GetId(), index);
73 1 : streamMgr.CaptureSlaveStream(masterStream, slaveStream);
74 7 : }
75 7 : InsQueue::Iterator masterQueueIter = insQueue.Iter();
76 :
77 : // 交替流下Task，直至全部下完
78 11 : while(!slaveQueueIters.empty() || masterQueueIter.HasNext())
79 : {
80 4 : SubmitSlaveQueueAlternatively(slaveQueueIters, slaveStreamIndexSet);
81 4 : SubmitMasterQueueAlternatively(masterQueueIter);
82 : }
83 :
84 : //销毁桶
85 7 : streamMgr.DestroyRecords();
86 : //销毁流的占用状态
87 7 : streamMgr.ResetSlaveIndex(0);
88 7 : }
89 :
90 4 : void Interpreter::SubmitSlaveQueueAlternatively(list<InsQueue::Iterator> &slaveQueueIters, std::set<u32> &slaveStreamIndexSet)
91 : {
92 4 : auto& streamMgr = comm.GetStreamManager();
93 4 : auto slaveStreamIndexIter = slaveStreamIndexSet.begin();
94 8 : for(auto slaveQueueIter = slaveQueueIters.begin(); slaveQueueIter != slaveQueueIters.end();) {
95 4 : if (!slaveQueueIter->HasNext()) {
96 : //销毁已经完成下发的流迭代器
97 3 : HCCL_INFO("[SubmitSlaveQueueAlternatively] slave stream index(%u) interpret finish", (*slaveStreamIndexIter));
98 1 : slaveQueueIter = slaveQueueIters.erase(slaveQueueIter);
99 1 : slaveStreamIndexIter = slaveStreamIndexSet.erase(slaveStreamIndexIter);
100 1 : continue;
101 1 : }
102 3 : auto& rule = insRuleMap.at((*slaveQueueIter)->GetType());
103 3 : auto stream = streamMgr.GetSlaveByIndex(*slaveStreamIndexIter);
104 3 : rule(**slaveQueueIter, comm, *stream, taskConfig);
105 :
106 9 : HCCL_INFO("[SubmitSlaveQueueAlternatively] slave stream index[%u], stream id[%u]. Instruction start %s",
107 : (*slaveStreamIndexIter), stream->GetId(), (*slaveQueueIter)->Describe().c_str());
108 : // 当前Task下载完毕，跳转这条流上的InsQueue里下一个task
109 3 : ++(*slaveQueueIter);
110 : // 切换至下一条流
111 3 : ++slaveQueueIter;
112 3 : ++slaveStreamIndexIter;
113 : }
114 4 : }
115 :
116 4 : void Interpreter::SubmitMasterQueueAlternatively(InsQueue::Iterator &masterQueueIter)
117 : {
118 4 : if(!masterQueueIter.HasNext()){
119 3 : HCCL_INFO("[SubmitMasterQueueAlternatively] main stream interpret finish");
120 1 : return ;
121 : }
122 3 : auto& streamMgr = comm.GetStreamManager();
123 3 : auto stream = streamMgr.GetMaster();
124 3 : auto &rule = insRuleMap.at(masterQueueIter->GetType());
125 3 : rule(*masterQueueIter, comm, *stream, taskConfig);
126 9 : HCCL_INFO("[SubmitMasterQueueAlternatively] master stream id[%u]. Instruction start %s",
127 : stream->GetId(), masterQueueIter->Describe().c_str());
128 : // 切换至下一个task
129 3 : ++masterQueueIter;
130 : }
131 :
132 : } // namespace Hccl
|
