[feat] 支持host网卡的多QP配置的获取-hcomm-AtomGit

Msrc/coll_communicator_mgr/api_c_adpt/coll_comm_res_c_adpt.cc+10-1

	@@ -71,7 +71,16 @@ static u32 ResolveQueueNum(const Hccl::EnvRdmaConfig& rdmaConfig, const HcclChan
71	if (channelDesc.roceAttr.queueNum != INVALID_UINT) { // 用户有配置qp数量，使用用户配置的	71	if (channelDesc.roceAttr.queueNum != INVALID_UINT) { // 用户有配置qp数量，使用用户配置的
72	return channelDesc.roceAttr.queueNum;	72	return channelDesc.roceAttr.queueNum;
73	}	73	}
74	- // 查询channelDesc，localEndpoint与remoteEndpoint的CommAddr字段，得到ip对	74	+
		75	+ if (channelDesc.channelProtocol == COMM_PROTOCOL_ROCE
		76	+ && channelDesc.localEndpoint.loc.locType == ENDPOINT_LOC_TYPE_HOST) {
		77	+ u32 hostQpCount = 0;
		78	+ MyRankUtils::ReadHostNicMultiQpCount(hostQpCount);
		79	+ if (hostQpCount > 0) {
		80	+ return hostQpCount;
		81	+ }
		82	+ }
		83	+
75	const auto& qpSrcPortConfig = rdmaConfig.GetMultiQpSrcPortConfig();	84	const auto& qpSrcPortConfig = rdmaConfig.GetMultiQpSrcPortConfig();
76	const CommAddr& localCommAddr = channelDesc.localEndpoint.commAddr;	85	const CommAddr& localCommAddr = channelDesc.localEndpoint.commAddr;
77	const CommAddr& remoteCommAddr = channelDesc.remoteEndpoint.commAddr;	86	const CommAddr& remoteCommAddr = channelDesc.remoteEndpoint.commAddr;

Msrc/coll_communicator_mgr/resource_mgr/local/my_rank/my_rank.cc+185-21

	@@ -108,6 +108,165 @@ inline const char* GetTlsStatusStr(Hccl::TlsStatus tlsStatus)
108	return "UNKNOWN";	108	return "UNKNOWN";
109	}	109	}
110		110
		111	+bool ParseStrictDecimal(const std::string& value, uint32_t minValue, uint32_t maxValue, uint32_t& parsed)
		112	+{
		113	+ const auto isDecimalDigit = [](unsigned char character) {
		114	+ return std::isdigit(character) != 0;
		115	+ };
		116	+ const bool hasValidDecimalFormat = !value.empty() && std::all_of(value.begin(), value.end(), isDecimalDigit);
		117	+ if (!hasValidDecimalFormat) {
		118	+ HCCL_WARNING("[%s] value[%s] is not a valid decimal number", __func__, value.c_str());
		119	+ return false;
		120	+ }
		121	+
		122	+ errno = 0;
		123	+ char* end = nullptr;
		124	+ const unsigned long number = std::strtoul(value.c_str(), &end, 10);
		125	+ if (errno == ERANGE \|\| end != value.c_str() + value.size()) {
		126	+ HCCL_WARNING("[%s] parse value[%s] failed", __func__, value.c_str());
		127	+ return false;
		128	+ }
		129	+ if (number < minValue \|\| number > maxValue) {
		130	+ HCCL_WARNING("[%s] value[%s] is out of range[%u, %u]", __func__, value.c_str(), minValue, maxValue);
		131	+ return false;
		132	+ }
		133	+
		134	+ parsed = static_cast<uint32_t>(number);
		135	+ return true;
		136	+}
		137	+
		138	+bool ParseMultiQpUdpPorts(const std::string& value, std::vector<std::uint16_t>& ports)
		139	+{
		140	+ if (value.empty()) {
		141	+ HCCL_WARNING("[%s] ports config is empty", __func__);
		142	+ return false;
		143	+ }
		144	+
		145	+ std::vector<std::uint16_t> parsedPorts;
		146	+ std::size_t start = 0;
		147	+ while (start <= value.size()) {
		148	+ const std::size_t comma = value.find(',', start);
		149	+ const std::string token = value.substr(start, comma == std::string::npos ? std::string::npos : comma - start);
		150	+ uint32_t port = 0;
		151	+ if (!ParseStrictDecimal(token, 1, Hccl::MultiQpSrcPortConfig::CONFIG_SRC_PORT_ID_MAX, port)) {
		152	+ HCCL_WARNING("[%s] invalid port[%s]", __func__, token.c_str());
		153	+ return false;
		154	+ }
		155	+ parsedPorts.emplace_back(static_cast<std::uint16_t>(port));
		156	+ if (parsedPorts.size() > Hccl::MultiQpSrcPortConfig::CONFIG_SRC_PORT_NUM_MAX) {
		157	+ HCCL_WARNING(
		158	+ "[%s] ports count[%zu] exceeds the maximum[%u]", __func__, parsedPorts.size(),
		159	+ Hccl::MultiQpSrcPortConfig::CONFIG_SRC_PORT_NUM_MAX);
		160	+ return false;
		161	+ }
		162	+ if (comma == std::string::npos) {
		163	+ break;
		164	+ }
		165	+ start = comma + 1;
		166	+ }
		167	+
		168	+ ports = std::move(parsedPorts);
		169	+ return true;
		170	+}
		171	+
		172	+bool ReadHostConfigValue(RaInfo& info, HccnCfgKey key, std::string& value)
		173	+{
		174	+ std::vector<char> buffer(HOST_NIC_CONFIG_BUFFER_SIZE, '\0');
		175	+ uint32_t valueLen = static_cast<uint32_t>(buffer.size());
		176	+ const int ret = RaGetHccnCfg(&info, key, buffer.data(), &valueLen);
		177	+ // 配置文件不存在或配置项为空时，RaGetHccnCfg返回成功和空值，不会进入该分支
		178	+ if (ret != 0) {
		179	+ HCCL_WARNING("[%s] RaGetHccnCfg failed, phyId[%u], key[%d], ret[%d]", __func__, info.phyId, key, ret);
		180	+ return false;
		181	+ }
		182	+
		183	+ valueLen = std::min<uint32_t>(valueLen, static_cast<uint32_t>(buffer.size()));
		184	+ value.assign(buffer.data(), valueLen);
		185	+ if (!value.empty() && value.back() == '\0') {
		186	+ value.pop_back();
		187	+ }
		188	+ return true;
		189	+}
		190	+
		191	+static bool
		192	+ReadHostNicMultiQpConfigItem(uint32_t devicePhyId, HccnCfgKey key, const char* itemName, std::string& itemValue)
		193	+{
		194	+ RaInfo info{};
		195	+ info.mode = NETWORK_PEER_ONLINE;
		196	+ info.phyId = devicePhyId;
		197	+ std::string mode;
		198	+ if (!ReadHostConfigValue(info, HCCN_CFG_UDP_PORT_MODE, mode)) {
		199	+ return false;
		200	+ }
		201	+ if (mode.empty()) {
		202	+ HCCL_WARNING("[%s] no host multi-qp config, phyId[%u]", __func__, devicePhyId);
		203	+ return false;
		204	+ }
		205	+ if (mode != "multi_qp") {
		206	+ HCCL_WARNING("[%s] invalid udp_port_mode[%s], phyId[%u]", __func__, mode.c_str(), devicePhyId);
		207	+ return false;
		208	+ }
		209	+
		210	+ if (!ReadHostConfigValue(info, key, itemValue)) {
		211	+ return false;
		212	+ }
		213	+ if (itemValue.empty()) {
		214	+ HCCL_WARNING("[%s] incomplete host multi-qp config, phyId[%u] %s[0]", __func__, devicePhyId, itemName);
		215	+ return false;
		216	+ }
		217	+ return true;
		218	+}
		219	+
		220	+void ReadHostNicMultiQpCount(uint32_t& qpCount)
		221	+{
		222	+ qpCount = 0;
		223	+ s32 deviceLogicId = INVALID_INT;
		224	+ uint32_t devicePhyId = INVALID_UINT;
		225	+ if ((hrtGetDevice(&deviceLogicId) != HCCL_SUCCESS)
		226	+ \|\| (hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devicePhyId, false) != HCCL_SUCCESS)) {
		227	+ return;
		228	+ }
		229	+
		230	+ std::string count;
		231	+ if (!ReadHostNicMultiQpConfigItem(devicePhyId, HCCN_CFG_MULTI_QP_COUNT, "multi_qp_count", count)) {
		232	+ return;
		233	+ }
		234	+
		235	+ uint32_t parsedQpCount = 0;
		236	+ if (!ParseStrictDecimal(count, 1, Hccl::MultiQpSrcPortConfig::CONFIG_SRC_PORT_NUM_MAX, parsedQpCount)) {
		237	+ HCCL_WARNING("[%s] invalid multi_qp_count[%s], phyId[%u]", __func__, count.c_str(), devicePhyId);
		238	+ return;
		239	+ }
		240	+
		241	+ qpCount = parsedQpCount;
		242	+ HCCL_INFO("[%s] host config valid, phyId[%u] count[%u]", __func__, devicePhyId, qpCount);
		243	+}
		244	+
		245	+void ReadHostNicMultiQpUdpPorts(std::vector<std::uint16_t>& qpUdpPorts)
		246	+{
		247	+ qpUdpPorts.clear();
		248	+ s32 deviceLogicId = INVALID_INT;
		249	+ uint32_t devicePhyId = INVALID_UINT;
		250	+ if ((hrtGetDevice(&deviceLogicId) != HCCL_SUCCESS)
		251	+ \|\| (hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devicePhyId, false) != HCCL_SUCCESS)) {
		252	+ return;
		253	+ }
		254	+
		255	+ std::string ports;
		256	+ if (!ReadHostNicMultiQpConfigItem(devicePhyId, HCCN_CFG_MULTI_QP_UDP_PORTS, "multi_qp_udp_ports", ports)) {
		257	+ return;
		258	+ }
		259	+
		260	+ std::vector<std::uint16_t> parsedQpUdpPorts;
		261	+ if (!ParseMultiQpUdpPorts(ports, parsedQpUdpPorts)) {
		262	+ HCCL_WARNING("[%s] invalid multi_qp_udp_ports[%s], phyId[%u]", __func__, ports.c_str(), devicePhyId);
		263	+ return;
		264	+ }
		265	+
		266	+ qpUdpPorts = std::move(parsedQpUdpPorts);
		267	+ HCCL_INFO("[%s] host config valid, phyId[%u] ports[%zu]", __func__, devicePhyId, qpUdpPorts.size());
		268	+}
		269	+
111	} // namespace MyRankUtils	270	} // namespace MyRankUtils
112		271
113	HcclResult MyRankUtils::FillRoceSrcPortList(	272	HcclResult MyRankUtils::FillRoceSrcPortList(
	@@ -120,28 +279,33 @@ HcclResult MyRankUtils::FillRoceSrcPortList(
120	hcommDesc.exchangeAllMems);	279	hcommDesc.exchangeAllMems);
121	return HCCL_SUCCESS;	280	return HCCL_SUCCESS;
122	}	281	}
123	- const auto& qpSrcPortConfig = Hccl::EnvConfig::GetInstance().GetRdmaConfig().GetMultiQpSrcPortConfig();	282	+
124	- if (!qpSrcPortConfig.IsAvailable()) {	283	+ std::vector<std::uint16_t> ports;
125	- HCCL_INFO(	284	+ ReadHostNicMultiQpUdpPorts(ports);
126	- "[%s] skip: multiQpSrcPortConfig not available (env HCCL_RDMA_QP_PORT_CONFIG_PATH unset or "
127	- "MultiQpSrcPort.cfg empty)",
128	- __func__);
129	- return HCCL_SUCCESS;
130	- }
131	- Hccl::IpAddress localIp;
132	- Hccl::IpAddress remoteIp;
133	- HcclResult localRet = CommAddrToIpAddress(hcclDesc.localEndpoint.commAddr, localIp);
134	- HcclResult remoteRet = CommAddrToIpAddress(hcclDesc.remoteEndpoint.commAddr, remoteIp);
135	- CHK_PRT_RET(
136	- localRet != HCCL_SUCCESS \|\| remoteRet != HCCL_SUCCESS,
137	- HCCL_ERROR("[%s] CommAddrToIpAddress failed: localRet[%d] remoteRet[%d]", __func__, localRet, remoteRet),
138	- HCCL_E_INTERNAL);
139	- auto ports = Hccl::GetMultiQpSrcPortsByIpPair(qpSrcPortConfig, localIp, remoteIp);
140	if (ports.empty()) {	285	if (ports.empty()) {
141	- HCCL_INFO(	286	+ const auto& qpSrcPortConfig = Hccl::EnvConfig::GetInstance().GetRdmaConfig().GetMultiQpSrcPortConfig();
142	- "[%s] skip: no matching ports for localIp[%s] remoteIp[%s]", __func__, localIp.GetIpStr().c_str(),	287	+ if (!qpSrcPortConfig.IsAvailable()) {
143	- remoteIp.GetIpStr().c_str());	288	+ HCCL_INFO(
144	- return HCCL_SUCCESS;	289	+ "[%s] skip: multiQpSrcPortConfig not available (env HCCL_RDMA_QP_PORT_CONFIG_PATH unset or "
		290	+ "MultiQpSrcPort.cfg empty)",
		291	+ __func__);
		292	+ return HCCL_SUCCESS;
		293	+ }
		294	+ Hccl::IpAddress localIp;
		295	+ Hccl::IpAddress remoteIp;
		296	+ HcclResult localRet = CommAddrToIpAddress(hcclDesc.localEndpoint.commAddr, localIp);
		297	+ HcclResult remoteRet = CommAddrToIpAddress(hcclDesc.remoteEndpoint.commAddr, remoteIp);
		298	+ CHK_PRT_RET(
		299	+ localRet != HCCL_SUCCESS \|\| remoteRet != HCCL_SUCCESS,
		300	+ HCCL_ERROR("[%s] CommAddrToIpAddress failed: localRet[%d] remoteRet[%d]", __func__, localRet, remoteRet),
		301	+ HCCL_E_INTERNAL);
		302	+ ports = Hccl::GetMultiQpSrcPortsByIpPair(qpSrcPortConfig, localIp, remoteIp);
		303	+ if (ports.empty()) {
		304	+ HCCL_INFO(
		305	+ "[%s] skip: no matching ports for localIp[%s] remoteIp[%s]", __func__, localIp.GetIpStr().c_str(),
		306	+ remoteIp.GetIpStr().c_str());
		307	+ return HCCL_SUCCESS;
		308	+ }
145	}	309	}
146	u32 queueNum = hcommDesc.roceAttr.queueNum;	310	u32 queueNum = hcommDesc.roceAttr.queueNum;
147	srcPortBuf.resize(queueNum);	311	srcPortBuf.resize(queueNum);

Msrc/coll_communicator_mgr/resource_mgr/local/my_rank/my_rank.h+18-0

 namespace MyRankUtils {
++constexpr std::size_t HOST_NIC_CONFIG_BUFFER_SIZE = 2048;
++
++// 将 HCCL 通道描述转换为 HCOMM 通道描述。
 HcommChannelDesc ChannelDescHccl2Hcomm(const HcclChannelDesc& hcclDesc, const hccl::CommConfig& commConfig);
++// 将字符串解析为指定范围内的十进制整数。
++bool ParseStrictDecimal(const std::string& value, uint32_t minValue, uint32_t maxValue, uint32_t& parsed);
++
++// 将逗号分隔的字符串解析为多 QP UDP 端口列表。
++bool ParseMultiQpUdpPorts(const std::string& value, std::vector<std::uint16_t>& ports);
++
++// 读取指定的 Host 网卡配置项。
++bool ReadHostConfigValue(RaInfo& info, HccnCfgKey key, std::string& value);
++
++// 读取并校验 Host 网卡的多 QP 数量。
++void ReadHostNicMultiQpCount(uint32_t& qpCount);
++
++// 读取并校验 Host 网卡的多 QP UDP 端口列表。
++void ReadHostNicMultiQpUdpPorts(std::vector<std::uint16_t>& qpUdpPorts);
++
 HcclResult
 FillRoceSrcPortList(const HcclChannelDesc& hcclDesc, HcommChannelDesc& hcommDesc, std::vector<uint16_t>& srcPortBuf);

Mtest/legacy/st/algorithm/utils/adapter_v2/adapter_hccp/orion_adapter_hccp_stub.cc+14-0

	@@ -176,3 +176,17 @@ HcclResult HrtRaGetTpAttrAsync(
176	}	176	}
177		177
178	} // namespace Hccl	178	} // namespace Hccl
		179	+
		180	+extern "C" HCCP_ATTRI_VISI_DEF int
		181	+RaGetHccnCfg(struct RaInfo* info, enum HccnCfgKey key, char* value, unsigned int* valueLen)
		182	+{
		183	+ (void)info;
		184	+ (void)key;
		185	+ if (value != nullptr && valueLen != nullptr && *valueLen > 0U) {
		186	+ value[0] = '\0';
		187	+ }
		188	+ if (valueLen != nullptr) {
		189	+ *valueLen = 0U;
		190	+ }
		191	+ return 0;
		192	+}

Mtest/ut/framework/next/coll_comms/api_c_adpt/ut_coll_comm_res_c_adpt.cc+151-0

	@@ -62,6 +62,86 @@ static int StubRaGetHccnCfgRoceQosDscp(struct RaInfo* info, enum HccnCfgKey key,
62	return 0;	62	return 0;
63	}	63	}
64		64
		65	+namespace {
		66	+// UT stub state is only used by the single-threaded ResolveQueueNum test.
		67	+s32 g_hostConfigDeviceLogicId = 0;
		68	+u32 g_hostConfigDevicePhyId = 0;
		69	+HcclResult g_getDeviceResult = HCCL_SUCCESS;
		70	+HcclResult g_getDevicePhyIdResult = HCCL_SUCCESS;
		71	+std::string g_hostMultiQpMode;
		72	+std::string g_hostMultiQpCount;
		73	+std::vector<HccnCfgKey> g_hostConfigReadKeys;
		74	+u32 g_getDeviceCallCount = 0;
		75	+u32 g_getDevicePhyIdCallCount = 0;
		76	+
		77	+void ResetHostMultiQpCountStub()
		78	+{
		79	+ g_hostConfigDeviceLogicId = 6;
		80	+ g_hostConfigDevicePhyId = 3;
		81	+ g_getDeviceResult = HCCL_SUCCESS;
		82	+ g_getDevicePhyIdResult = HCCL_SUCCESS;
		83	+ g_hostMultiQpMode = "multi_qp";
		84	+ g_hostMultiQpCount = "4";
		85	+ g_hostConfigReadKeys.clear();
		86	+ g_getDeviceCallCount = 0;
		87	+ g_getDevicePhyIdCallCount = 0;
		88	+}
		89	+
		90	+HcclResult StubGetDeviceForHostMultiQp(s32* deviceLogicId)
		91	+{
		92	+ ++g_getDeviceCallCount;
		93	+ if (g_getDeviceResult == HCCL_SUCCESS && deviceLogicId != nullptr) {
		94	+ *deviceLogicId = g_hostConfigDeviceLogicId;
		95	+ }
		96	+ return g_getDeviceResult;
		97	+}
		98	+
		99	+HcclResult StubGetDevicePhyIdForHostMultiQp(u32 deviceLogicId, u32& devicePhyId, bool isRefresh)
		100	+{
		101	+ ++g_getDevicePhyIdCallCount;
		102	+ EXPECT_EQ(deviceLogicId, static_cast<u32>(g_hostConfigDeviceLogicId));
		103	+ EXPECT_FALSE(isRefresh);
		104	+ if (g_getDevicePhyIdResult == HCCL_SUCCESS) {
		105	+ devicePhyId = g_hostConfigDevicePhyId;
		106	+ }
		107	+ return g_getDevicePhyIdResult;
		108	+}
		109	+
		110	+int StubRaGetHostMultiQpCount(RaInfo* info, HccnCfgKey key, char* value, unsigned int* valueLen)
		111	+{
		112	+ if (info == nullptr \|\| value == nullptr \|\| valueLen == nullptr) {
		113	+ return -1;
		114	+ }
		115	+ EXPECT_EQ(info->mode, NETWORK_PEER_ONLINE);
		116	+ EXPECT_EQ(info->phyId, g_hostConfigDevicePhyId);
		117	+ EXPECT_EQ(*valueLen, MyRankUtils::HOST_NIC_CONFIG_BUFFER_SIZE);
		118	+ g_hostConfigReadKeys.emplace_back(key);
		119	+
		120	+ const std::string* configValue = nullptr;
		121	+ if (key == HCCN_CFG_UDP_PORT_MODE) {
		122	+ configValue = &g_hostMultiQpMode;
		123	+ } else if (key == HCCN_CFG_MULTI_QP_COUNT) {
		124	+ configValue = &g_hostMultiQpCount;
		125	+ }
		126	+ if (configValue == nullptr) {
		127	+ *valueLen = 0;
		128	+ return 0;
		129	+ }
		130	+ if (configValue->empty()) {
		131	+ *valueLen = 0;
		132	+ return 0;
		133	+ }
		134	+ const std::size_t requiredValueLen = configValue->size() + 1U;
		135	+ if (*valueLen < requiredValueLen) {
		136	+ return -1;
		137	+ }
		138	+ std::copy(configValue->begin(), configValue->end(), value);
		139	+ value[configValue->size()] = '\0';
		140	+ *valueLen = static_cast<unsigned int>(requiredValueLen);
		141	+ return 0;
		142	+}
		143	+} // namespace
		144	+
65	static HcclMemHandle g_userMemHandle = reinterpret_cast<HcclMemHandle>(0x1111);	145	static HcclMemHandle g_userMemHandle = reinterpret_cast<HcclMemHandle>(0x1111);
66	static HcclMemHandle g_symMemHandle = reinterpret_cast<HcclMemHandle>(0x2222);	146	static HcclMemHandle g_symMemHandle = reinterpret_cast<HcclMemHandle>(0x2222);
67	static ChannelHandle g_testChannel = static_cast<ChannelHandle>(0x3333);	147	static ChannelHandle g_testChannel = static_cast<ChannelHandle>(0x3333);
	@@ -279,6 +359,77 @@ TEST_F(HcclChannelDescTest, Ut_ProcessRoceChannelDesc_When_IsCommunicatorV2_Is_T
279	EXPECT_EQ(ret, HCCL_SUCCESS);	359	EXPECT_EQ(ret, HCCL_SUCCESS);
280	}	360	}
281		361
		362	+TEST_F(HcclChannelDescTest, Ut_ProcessRoceChannelDesc_HostCountPriority)
		363	+{
		364	+ auto& portConfig = const_cast<Hccl::MultiQpSrcPortConfig&>(
		365	+ Hccl::EnvConfig::GetInstance().GetRdmaConfig().GetMultiQpSrcPortConfig());
		366	+ portConfig.ipPairToPorts.clear();
		367	+ portConfig.ipPairToPorts["1.0.0.0,2.0.0.0"] = {10001, 10002, 10003};
		368	+ ResetHostMultiQpCountStub();
		369	+ MOCKER(hrtGetDevice).stubs().will(invoke(StubGetDeviceForHostMultiQp));
		370	+ MOCKER(hrtGetDevicePhyIdByIndex).stubs().will(invoke(StubGetDevicePhyIdForHostMultiQp));
		371	+ MOCKER(RaGetHccnCfg).stubs().will(invoke(StubRaGetHostMultiQpCount));
		372	+
		373	+ EndpointLocType localLocType = ENDPOINT_LOC_TYPE_HOST;
		374	+ auto resolveQueueNum = [this, &localLocType](uint32_t requestedQueueNum) {
		375	+ HcclChannelDesc input{};
		376	+ HcclChannelDesc output{};
		377	+ InitRoceChannelDesc(input);
		378	+ EXPECT_EQ(HcclChannelDescInit(&output, 1), HCCL_SUCCESS);
		379	+ input.roceAttr.queueNum = requestedQueueNum;
		380	+ input.localEndpoint.loc.locType = localLocType;
		381	+ input.localEndpoint.commAddr.type = COMM_ADDR_TYPE_IP_V4;
		382	+ input.localEndpoint.commAddr.addr = Hccl::IpAddress("1.0.0.0").GetBinaryAddress().addr;
		383	+ input.remoteEndpoint.commAddr.type = COMM_ADDR_TYPE_IP_V4;
		384	+ input.remoteEndpoint.commAddr.addr = Hccl::IpAddress("2.0.0.0").GetBinaryAddress().addr;
		385	+ EXPECT_EQ(ProcessHcclResPackReq(input, output, hcclCommPtr.get()), HCCL_SUCCESS);
		386	+ return output.roceAttr.queueNum;
		387	+ };
		388	+
		389	+ EXPECT_EQ(resolveQueueNum(7), 7U);
		390	+ EXPECT_EQ(g_getDeviceCallCount, 0U);
		391	+ EXPECT_TRUE(g_hostConfigReadKeys.empty());
		392	+
		393	+ EXPECT_EQ(resolveQueueNum(INVALID_UINT), 4U);
		394	+ EXPECT_EQ(g_getDeviceCallCount, 1U);
		395	+ EXPECT_EQ(g_getDevicePhyIdCallCount, 1U);
		396	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE, HCCN_CFG_MULTI_QP_COUNT}));
		397	+
		398	+ localLocType = ENDPOINT_LOC_TYPE_DEVICE;
		399	+ g_hostConfigReadKeys.clear();
		400	+ const u32 getDeviceCallCount = g_getDeviceCallCount;
		401	+ EXPECT_EQ(resolveQueueNum(INVALID_UINT), 3U);
		402	+ EXPECT_EQ(g_getDeviceCallCount, getDeviceCallCount);
		403	+ EXPECT_TRUE(g_hostConfigReadKeys.empty());
		404	+
		405	+ localLocType = ENDPOINT_LOC_TYPE_HOST;
		406	+ g_hostConfigReadKeys.clear();
		407	+ g_hostMultiQpMode.clear();
		408	+ EXPECT_EQ(resolveQueueNum(INVALID_UINT), 3U);
		409	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE}));
		410	+
		411	+ g_hostConfigReadKeys.clear();
		412	+ g_getDevicePhyIdResult = HCCL_E_RUNTIME;
		413	+ EXPECT_EQ(resolveQueueNum(INVALID_UINT), 3U);
		414	+ EXPECT_TRUE(g_hostConfigReadKeys.empty());
		415	+
		416	+ g_getDevicePhyIdResult = HCCL_SUCCESS;
		417	+ g_getDeviceResult = HCCL_E_RUNTIME;
		418	+ const u32 phyIdCallCount = g_getDevicePhyIdCallCount;
		419	+ EXPECT_EQ(resolveQueueNum(INVALID_UINT), 3U);
		420	+ EXPECT_EQ(g_getDevicePhyIdCallCount, phyIdCallCount);
		421	+ EXPECT_TRUE(g_hostConfigReadKeys.empty());
		422	+
		423	+ g_getDeviceResult = HCCL_SUCCESS;
		424	+ portConfig.ipPairToPorts.clear();
		425	+ EXPECT_EQ(resolveQueueNum(INVALID_UINT), Hccl::EnvConfig::GetInstance().GetRdmaConfig().GetRdmaQueueNum());
		426	+
		427	+ MOCKER_CPP(&hccl::hcclComm::IsCommunicatorV2).stubs().will(returnValue(false));
		428	+ EXPECT_EQ(resolveQueueNum(6), 6U);
		429	+
		430	+ portConfig.ipPairToPorts.clear();
		431	+}
		432	+
282	TEST_F(HcclChannelDescTest, Ut_ProcessRoceChannelDesc_When_TcIsInvaild_ReturnHCCLEPARA)	433	TEST_F(HcclChannelDescTest, Ut_ProcessRoceChannelDesc_When_TcIsInvaild_ReturnHCCLEPARA)
283	{	434	{
284	hcclCommPtr->collComm_->config_.trafficClass_ = 10; // 单独赋非法值	435	hcclCommPtr->collComm_->config_.trafficClass_ = 10; // 单独赋非法值

Mtest/ut/framework/next/coll_comms/rank/ut_my_rank.cc+238-0

	@@ -16,6 +16,7 @@
16	#include "rank_graph_v2.h"	16	#include "rank_graph_v2.h"
17	#include "hccl/hccl_res.h"	17	#include "hccl/hccl_res.h"
18	#include <hccl/hccl_comm.h>	18	#include <hccl/hccl_comm.h>
		19	+#include "hcomm_adapter_hccp.h"
19	#include "hcomm_c_adpt.h"	20	#include "hcomm_c_adpt.h"
20	#include "channel_process.h"	21	#include "channel_process.h"
21	#include "base_config_legacy.h"	22	#include "base_config_legacy.h"
	@@ -32,6 +33,91 @@
32		33
33	using namespace hccl;	34	using namespace hccl;
34		35
		36	+namespace {
		37	+// UT stub state is only used by the single-threaded host multi-QP tests.
		38	+u32 g_hostConfigDeviceLogicId = 0;
		39	+u32 g_hostConfigDevicePhyId = 0;
		40	+HcclResult g_getDeviceResult = HCCL_SUCCESS;
		41	+HcclResult g_getDevicePhyIdResult = HCCL_SUCCESS;
		42	+std::string g_hostMultiQpMode;
		43	+std::string g_hostMultiQpCount;
		44	+std::string g_hostMultiQpPorts;
		45	+int g_hostConfigFailureKey = -1;
		46	+std::vector<HccnCfgKey> g_hostConfigReadKeys;
		47	+u32 g_getDeviceCallCount = 0;
		48	+u32 g_getDevicePhyIdCallCount = 0;
		49	+
		50	+void ResetHostMultiQpStub()
		51	+{
		52	+ g_hostConfigDeviceLogicId = 4;
		53	+ g_hostConfigDevicePhyId = 3;
		54	+ g_getDeviceResult = HCCL_SUCCESS;
		55	+ g_getDevicePhyIdResult = HCCL_SUCCESS;
		56	+ g_hostMultiQpMode = "multi_qp";
		57	+ g_hostMultiQpCount = "4";
		58	+ g_hostMultiQpPorts = "20001,20002";
		59	+ g_hostConfigFailureKey = -1;
		60	+ g_hostConfigReadKeys.clear();
		61	+ g_getDeviceCallCount = 0;
		62	+ g_getDevicePhyIdCallCount = 0;
		63	+}
		64	+
		65	+HcclResult StubGetDeviceForHostMultiQp(s32* deviceLogicId)
		66	+{
		67	+ ++g_getDeviceCallCount;
		68	+ if (g_getDeviceResult == HCCL_SUCCESS && deviceLogicId != nullptr) {
		69	+ *deviceLogicId = static_cast<s32>(g_hostConfigDeviceLogicId);
		70	+ }
		71	+ return g_getDeviceResult;
		72	+}
		73	+
		74	+HcclResult StubGetDevicePhyIdForHostMultiQp(u32 deviceLogicId, u32& devicePhyId, bool isRefresh)
		75	+{
		76	+ ++g_getDevicePhyIdCallCount;
		77	+ EXPECT_EQ(deviceLogicId, g_hostConfigDeviceLogicId);
		78	+ EXPECT_FALSE(isRefresh);
		79	+ if (g_getDevicePhyIdResult == HCCL_SUCCESS) {
		80	+ devicePhyId = g_hostConfigDevicePhyId;
		81	+ }
		82	+ return g_getDevicePhyIdResult;
		83	+}
		84	+
		85	+int StubRaGetHostMultiQpConfig(RaInfo* info, HccnCfgKey key, char* value, unsigned int* valueLen)
		86	+{
		87	+ if (info == nullptr \|\| value == nullptr \|\| valueLen == nullptr) {
		88	+ return -1;
		89	+ }
		90	+ EXPECT_EQ(info->mode, NETWORK_PEER_ONLINE);
		91	+ EXPECT_EQ(info->phyId, g_hostConfigDevicePhyId);
		92	+ EXPECT_EQ(*valueLen, MyRankUtils::HOST_NIC_CONFIG_BUFFER_SIZE);
		93	+ g_hostConfigReadKeys.emplace_back(key);
		94	+ if (g_hostConfigFailureKey == static_cast<int>(key)) {
		95	+ return -1;
		96	+ }
		97	+
		98	+ const std::string* configValue = nullptr;
		99	+ if (key == HCCN_CFG_UDP_PORT_MODE) {
		100	+ configValue = &g_hostMultiQpMode;
		101	+ } else if (key == HCCN_CFG_MULTI_QP_COUNT) {
		102	+ configValue = &g_hostMultiQpCount;
		103	+ } else if (key == HCCN_CFG_MULTI_QP_UDP_PORTS) {
		104	+ configValue = &g_hostMultiQpPorts;
		105	+ }
		106	+ if (configValue == nullptr \|\| configValue->empty()) {
		107	+ *valueLen = 0;
		108	+ return 0;
		109	+ }
		110	+ const std::size_t contentLength = configValue->size();
		111	+ if (contentLength >= *valueLen) {
		112	+ return -1;
		113	+ }
		114	+ configValue->copy(value, contentLength);
		115	+ value[contentLength] = '\0';
		116	+ *valueLen = static_cast<unsigned int>(contentLength + 1U);
		117	+ return 0;
		118	+}
		119	+} // namespace
		120	+
35	HcclResult GetAllMemoryStub(CommMems*, std::vector<HcclMem>& mems, std::vector<std::string>& tags, uint64_t& version)	121	HcclResult GetAllMemoryStub(CommMems*, std::vector<HcclMem>& mems, std::vector<std::string>& tags, uint64_t& version)
36	{	122	{
37	HcclMem m;	123	HcclMem m;
	@@ -182,6 +268,15 @@ protected:
182	}	268	}
183	}	269	}
184		270
		271	+ void ExpectFillRoceSrcPortListSuccess(
		272	+ HcclChannelDesc& hcclDesc, HcommChannelDesc& hcommDesc, std::vector<uint16_t>& srcPortBuf,
		273	+ const std::vector<uint16_t>& expectedPorts)
		274	+ {
		275	+ EXPECT_EQ(MyRankUtils::FillRoceSrcPortList(hcclDesc, hcommDesc, srcPortBuf), HCCL_SUCCESS);
		276	+ ASSERT_NE(hcommDesc.roceAttr.srcPortList, nullptr);
		277	+ EXPECT_EQ(srcPortBuf, expectedPorts);
		278	+ }
		279	+
185	uint32_t DEFAULT_MODE = 0;	280	uint32_t DEFAULT_MODE = 0;
186	uint32_t AICPU_TS_MODE = 2;	281	uint32_t AICPU_TS_MODE = 2;
187	uint32_t CCU_MS_MODE = 5;	282	uint32_t CCU_MS_MODE = 5;
	@@ -2028,6 +2123,7 @@ TEST_F(MyRankTest, Ut_FillRoceSrcPortList_When_CommAddrToIpFail_Expect_SrcPortLi
2028	portConfig.ipPairToPorts.clear();	2123	portConfig.ipPairToPorts.clear();
2029	portConfig.ipPairToPorts["1.0.0.0,2.0.0.0"] = {10001};	2124	portConfig.ipPairToPorts["1.0.0.0,2.0.0.0"] = {10001};
2030		2125
		2126	+ MOCKER(hrtGetDevice).stubs().will(returnValue(HCCL_E_RUNTIME));
2031	MOCKER(hcomm::CommAddrToIpAddress).stubs().will(returnValue(HCCL_E_NOT_SUPPORT));	2127	MOCKER(hcomm::CommAddrToIpAddress).stubs().will(returnValue(HCCL_E_NOT_SUPPORT));
2032		2128
2033	HcclChannelDesc hcclDesc{};	2129	HcclChannelDesc hcclDesc{};
	@@ -2053,6 +2149,7 @@ TEST_F(MyRankTest, Ut_FillRoceSrcPortList_When_ConfigAvailable_Expect_SrcPortLis
2053	Hccl::EnvConfig::GetInstance().GetRdmaConfig().GetMultiQpSrcPortConfig());	2149	Hccl::EnvConfig::GetInstance().GetRdmaConfig().GetMultiQpSrcPortConfig());
2054	portConfig.ipPairToPorts.clear();	2150	portConfig.ipPairToPorts.clear();
2055	portConfig.ipPairToPorts["1.0.0.0,2.0.0.0"] = {10001, 10002};	2151	portConfig.ipPairToPorts["1.0.0.0,2.0.0.0"] = {10001, 10002};
		2152	+ MOCKER(hrtGetDevice).stubs().will(returnValue(HCCL_E_RUNTIME));
2056		2153
2057	HcclChannelDesc hcclDesc{};	2154	HcclChannelDesc hcclDesc{};
2058	ASSERT_EQ(HcclChannelDescInit(&hcclDesc, 1), HCCL_SUCCESS);	2155	ASSERT_EQ(HcclChannelDescInit(&hcclDesc, 1), HCCL_SUCCESS);
	@@ -2074,3 +2171,144 @@ TEST_F(MyRankTest, Ut_FillRoceSrcPortList_When_ConfigAvailable_Expect_SrcPortLis
2074		2171
2075	portConfig.ipPairToPorts.clear();	2172	portConfig.ipPairToPorts.clear();
2076	}	2173	}
		2174	+
		2175	+TEST_F(MyRankTest, Ut_ParseHostMultiQpConfig_CoversValidAndInvalidValues)
		2176	+{
		2177	+ uint32_t parsed = 0;
		2178	+ EXPECT_TRUE(MyRankUtils::ParseStrictDecimal("1", 1, 32, parsed));
		2179	+ EXPECT_EQ(parsed, 1U);
		2180	+ EXPECT_TRUE(MyRankUtils::ParseStrictDecimal("32", 1, 32, parsed));
		2181	+ EXPECT_EQ(parsed, 32U);
		2182	+ EXPECT_FALSE(MyRankUtils::ParseStrictDecimal("", 1, 32, parsed));
		2183	+ EXPECT_FALSE(MyRankUtils::ParseStrictDecimal("0", 1, 32, parsed));
		2184	+ EXPECT_FALSE(MyRankUtils::ParseStrictDecimal("+1", 1, 32, parsed));
		2185	+ EXPECT_FALSE(MyRankUtils::ParseStrictDecimal("33", 1, 32, parsed));
		2186	+
		2187	+ std::vector<uint16_t> ports;
		2188	+ EXPECT_TRUE(MyRankUtils::ParseMultiQpUdpPorts("10001,10001,65535", ports));
		2189	+ EXPECT_EQ(ports, (std::vector<uint16_t>{10001, 10001, 65535}));
		2190	+ EXPECT_FALSE(MyRankUtils::ParseMultiQpUdpPorts("10001,", ports));
		2191	+ EXPECT_FALSE(MyRankUtils::ParseMultiQpUdpPorts("0", ports));
		2192	+
		2193	+ std::string tooManyPorts = "1";
		2194	+ for (u32 i = 0; i < Hccl::MultiQpSrcPortConfig::CONFIG_SRC_PORT_NUM_MAX; ++i) {
		2195	+ tooManyPorts += ",1";
		2196	+ }
		2197	+ EXPECT_FALSE(MyRankUtils::ParseMultiQpUdpPorts(tooManyPorts, ports));
		2198	+}
		2199	+
		2200	+TEST_F(MyRankTest, Ut_ReadHostNicMultiQpCount_ReadsLatestCountOnly)
		2201	+{
		2202	+ ResetHostMultiQpStub();
		2203	+ MOCKER(hrtGetDevice).stubs().will(invoke(StubGetDeviceForHostMultiQp));
		2204	+ MOCKER(hrtGetDevicePhyIdByIndex).stubs().will(invoke(StubGetDevicePhyIdForHostMultiQp));
		2205	+ MOCKER(RaGetHccnCfg).stubs().will(invoke(StubRaGetHostMultiQpConfig));
		2206	+
		2207	+ uint32_t qpCount = 0;
		2208	+ MyRankUtils::ReadHostNicMultiQpCount(qpCount);
		2209	+ EXPECT_EQ(qpCount, 4U);
		2210	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE, HCCN_CFG_MULTI_QP_COUNT}));
		2211	+
		2212	+ g_hostConfigReadKeys.clear();
		2213	+ g_hostMultiQpCount = "5";
		2214	+ MyRankUtils::ReadHostNicMultiQpCount(qpCount);
		2215	+ EXPECT_EQ(qpCount, 5U);
		2216	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE, HCCN_CFG_MULTI_QP_COUNT}));
		2217	+
		2218	+ g_hostConfigReadKeys.clear();
		2219	+ g_hostMultiQpMode.clear();
		2220	+ MyRankUtils::ReadHostNicMultiQpCount(qpCount);
		2221	+ EXPECT_EQ(qpCount, 0U);
		2222	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE}));
		2223	+}
		2224	+
		2225	+TEST_F(MyRankTest, Ut_ReadHostNicMultiQpUdpPorts_ReadsLatestPortsOnly)
		2226	+{
		2227	+ ResetHostMultiQpStub();
		2228	+ MOCKER(hrtGetDevice).stubs().will(invoke(StubGetDeviceForHostMultiQp));
		2229	+ MOCKER(hrtGetDevicePhyIdByIndex).stubs().will(invoke(StubGetDevicePhyIdForHostMultiQp));
		2230	+ MOCKER(RaGetHccnCfg).stubs().will(invoke(StubRaGetHostMultiQpConfig));
		2231	+
		2232	+ std::vector<uint16_t> ports;
		2233	+ MyRankUtils::ReadHostNicMultiQpUdpPorts(ports);
		2234	+ EXPECT_EQ(ports, (std::vector<uint16_t>{20001, 20002}));
		2235	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE, HCCN_CFG_MULTI_QP_UDP_PORTS}));
		2236	+
		2237	+ g_hostConfigReadKeys.clear();
		2238	+ g_hostMultiQpPorts = "30001";
		2239	+ MyRankUtils::ReadHostNicMultiQpUdpPorts(ports);
		2240	+ EXPECT_EQ(ports, (std::vector<uint16_t>{30001}));
		2241	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE, HCCN_CFG_MULTI_QP_UDP_PORTS}));
		2242	+
		2243	+ g_hostConfigReadKeys.clear();
		2244	+ g_hostConfigFailureKey = HCCN_CFG_MULTI_QP_UDP_PORTS;
		2245	+ MyRankUtils::ReadHostNicMultiQpUdpPorts(ports);
		2246	+ EXPECT_TRUE(ports.empty());
		2247	+}
		2248	+
		2249	+TEST_F(MyRankTest, Ut_FillRoceSrcPortList_HostConfigPriority)
		2250	+{
		2251	+ auto& portConfig = const_cast<Hccl::MultiQpSrcPortConfig&>(
		2252	+ Hccl::EnvConfig::GetInstance().GetRdmaConfig().GetMultiQpSrcPortConfig());
		2253	+ portConfig.ipPairToPorts.clear();
		2254	+ portConfig.ipPairToPorts["1.0.0.0,2.0.0.0"] = {10001, 10002};
		2255	+ ResetHostMultiQpStub();
		2256	+ MOCKER(hrtGetDevice).stubs().will(invoke(StubGetDeviceForHostMultiQp));
		2257	+ MOCKER(hrtGetDevicePhyIdByIndex).stubs().will(invoke(StubGetDevicePhyIdForHostMultiQp));
		2258	+ MOCKER(RaGetHccnCfg).stubs().will(invoke(StubRaGetHostMultiQpConfig));
		2259	+
		2260	+ HcclChannelDesc hcclDesc{};
		2261	+ ASSERT_EQ(HcclChannelDescInit(&hcclDesc, 1), HCCL_SUCCESS);
		2262	+ hcclDesc.channelProtocol = COMM_PROTOCOL_ROCE;
		2263	+ CreateEndpointDesc(hcclDesc.localEndpoint, COMM_PROTOCOL_ROCE, "1.0.0.0");
		2264	+ CreateEndpointDesc(hcclDesc.remoteEndpoint, COMM_PROTOCOL_ROCE, "2.0.0.0");
		2265	+
		2266	+ HcommChannelDesc hcommDesc = MyRankUtils::ChannelDescHccl2Hcomm(hcclDesc, config);
		2267	+ hcommDesc.exchangeAllMems = false;
		2268	+ hcommDesc.roceAttr.queueNum = 5;
		2269	+ std::vector<uint16_t> srcPortBuf;
		2270	+
		2271	+ ASSERT_NO_FATAL_FAILURE(
		2272	+ ExpectFillRoceSrcPortListSuccess(hcclDesc, hcommDesc, srcPortBuf, {20001, 20002, 20001, 20002, 20001}));
		2273	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE, HCCN_CFG_MULTI_QP_UDP_PORTS}));
		2274	+
		2275	+ g_hostConfigReadKeys.clear();
		2276	+ g_hostMultiQpMode.clear();
		2277	+ ASSERT_NO_FATAL_FAILURE(
		2278	+ ExpectFillRoceSrcPortListSuccess(hcclDesc, hcommDesc, srcPortBuf, {10001, 10002, 10001, 10002, 10001}));
		2279	+ EXPECT_EQ(g_hostConfigReadKeys, (std::vector<HccnCfgKey>{HCCN_CFG_UDP_PORT_MODE}));
		2280	+
		2281	+ g_hostConfigReadKeys.clear();
		2282	+ g_getDevicePhyIdResult = HCCL_E_RUNTIME;
		2283	+ ASSERT_NO_FATAL_FAILURE(
		2284	+ ExpectFillRoceSrcPortListSuccess(hcclDesc, hcommDesc, srcPortBuf, {10001, 10002, 10001, 10002, 10001}));
		2285	+ EXPECT_TRUE(g_hostConfigReadKeys.empty());
		2286	+
		2287	+ g_getDevicePhyIdResult = HCCL_SUCCESS;
		2288	+ g_getDeviceResult = HCCL_E_RUNTIME;
		2289	+ const u32 phyIdCallCount = g_getDevicePhyIdCallCount;
		2290	+ ASSERT_NO_FATAL_FAILURE(
		2291	+ ExpectFillRoceSrcPortListSuccess(hcclDesc, hcommDesc, srcPortBuf, {10001, 10002, 10001, 10002, 10001}));
		2292	+ EXPECT_EQ(g_getDevicePhyIdCallCount, phyIdCallCount);
		2293	+ EXPECT_TRUE(g_hostConfigReadKeys.empty());
		2294	+
		2295	+ g_getDeviceResult = HCCL_SUCCESS;
		2296	+ portConfig.ipPairToPorts.clear();
		2297	+ EXPECT_EQ(MyRankUtils::FillRoceSrcPortList(hcclDesc, hcommDesc, srcPortBuf), HCCL_SUCCESS);
		2298	+ EXPECT_EQ(hcommDesc.roceAttr.srcPortList, nullptr);
		2299	+
		2300	+ g_hostMultiQpMode = "multi_qp";
		2301	+ g_hostMultiQpPorts = "30001";
		2302	+ hcommDesc.roceAttr.queueNum = 3;
		2303	+ ASSERT_NO_FATAL_FAILURE(ExpectFillRoceSrcPortListSuccess(hcclDesc, hcommDesc, srcPortBuf, {30001, 30001, 30001}));
		2304	+
		2305	+ hcommDesc.exchangeAllMems = true;
		2306	+ const u32 getDeviceCallCount = g_getDeviceCallCount;
		2307	+ const u32 skipPhyIdCallCount = g_getDevicePhyIdCallCount;
		2308	+ EXPECT_EQ(MyRankUtils::FillRoceSrcPortList(hcclDesc, hcommDesc, srcPortBuf), HCCL_SUCCESS);
		2309	+ EXPECT_EQ(hcommDesc.roceAttr.srcPortList, nullptr);
		2310	+ EXPECT_EQ(g_getDeviceCallCount, getDeviceCallCount);
		2311	+ EXPECT_EQ(g_getDevicePhyIdCallCount, skipPhyIdCallCount);
		2312	+
		2313	+ portConfig.ipPairToPorts.clear();
		2314	+}

Mtest/ut/impl/ut_coll_batch_send_recv_group_executor.cc+4-2

	@@ -1256,6 +1256,7 @@ TEST_F(CollBatchSendRecvGroupExecutorTest, ProcessNewRankSendSlice_SingleSlice)
1256	{	1256	{
1257	SetStreamNum(2, 2);	1257	SetStreamNum(2, 2);
1258	SetupAlgResResp(6, 4 * 1024 * 1024);	1258	SetupAlgResResp(6, 4 * 1024 * 1024);
		1259	+ MOCKER(hrtMemAsyncCopy).stubs().will(returnValue(HCCL_SUCCESS));
1259	executor_->bufferSliceSize_ = 16384;	1260	executor_->bufferSliceSize_ = 16384;
1260	LINK fakeLink = MakeFakeLink();	1261	LINK fakeLink = MakeFakeLink();
1261	SetupOpTransportForRank(1, fakeLink); // remoteRank=1 < userRank=4 -> send	1262	SetupOpTransportForRank(1, fakeLink); // remoteRank=1 < userRank=4 -> send
	@@ -1264,7 +1265,6 @@ TEST_F(CollBatchSendRecvGroupExecutorTest, ProcessNewRankSendSlice_SingleSlice)
1264	executor_->sendDataSlicesBySendStream_.resize(2);	1265	executor_->sendDataSlicesBySendStream_.resize(2);
1265	executor_->sendDataSlicesBySendStream_[0].emplace_back(buf, 16, 1);	1266	executor_->sendDataSlicesBySendStream_[0].emplace_back(buf, 16, 1);
1266		1267
1267	- // hrtMemAsyncCopy is already stubbed in runtime_stub.cc
1268	// Initialize per-stream state vectors	1268	// Initialize per-stream state vectors
1269	executor_->sendCurPhase_.assign(2, 0);	1269	executor_->sendCurPhase_.assign(2, 0);
1270	executor_->sendLoadedSize_.assign(2, 0);	1270	executor_->sendLoadedSize_.assign(2, 0);
	@@ -1282,6 +1282,7 @@ TEST_F(CollBatchSendRecvGroupExecutorTest, ProcessNewRankSendSlice_TwoSlicesSame
1282	{	1282	{
1283	SetStreamNum(2, 2);	1283	SetStreamNum(2, 2);
1284	SetupAlgResResp(6, 4 * 1024 * 1024);	1284	SetupAlgResResp(6, 4 * 1024 * 1024);
		1285	+ MOCKER(hrtMemAsyncCopy).stubs().will(returnValue(HCCL_SUCCESS));
1285	executor_->bufferSliceSize_ = 16384;	1286	executor_->bufferSliceSize_ = 16384;
1286	LINK fakeLink = MakeFakeLink();	1287	LINK fakeLink = MakeFakeLink();
1287	SetupOpTransportForRank(1, fakeLink);	1288	SetupOpTransportForRank(1, fakeLink);
	@@ -1301,7 +1302,6 @@ TEST_F(CollBatchSendRecvGroupExecutorTest, ProcessNewRankSendSlice_TwoSlicesSame
1301	EXPECT_EQ(ret, HCCL_SUCCESS);	1302	EXPECT_EQ(ret, HCCL_SUCCESS);
1302	EXPECT_EQ(executor_->sendLoadedSize_[0], 32u);	1303	EXPECT_EQ(executor_->sendLoadedSize_[0], 32u);
1303	EXPECT_EQ(pending, 1u);	1304	EXPECT_EQ(pending, 1u);
1304	- algRes_.slaveStreams.clear();
1305	}	1305	}
1306		1306
1307	// ============================================================================	1307	// ============================================================================
	@@ -1312,6 +1312,7 @@ TEST_F(CollBatchSendRecvGroupExecutorTest, ProcessPreloadedSendSlice_Basic)
1312	{	1312	{
1313	SetStreamNum(2, 2);	1313	SetStreamNum(2, 2);
1314	SetupAlgResResp(6, 4 * 1024 * 1024);	1314	SetupAlgResResp(6, 4 * 1024 * 1024);
		1315	+ MOCKER(hrtMemAsyncCopy).stubs().will(returnValue(HCCL_SUCCESS));
1315	executor_->bufferSliceSize_ = 16384;	1316	executor_->bufferSliceSize_ = 16384;
1316	LINK fakeLink = MakeFakeLink();	1317	LINK fakeLink = MakeFakeLink();
1317	SetupOpTransportForRank(1, fakeLink);	1318	SetupOpTransportForRank(1, fakeLink);
	@@ -1415,6 +1416,7 @@ TEST_F(CollBatchSendRecvGroupExecutorTest, ProcessRdmaSendSlice_Basic)
1415	{	1416	{
1416	SetStreamNum(2, 2);	1417	SetStreamNum(2, 2);
1417	SetupAlgResResp(6, 4 * 1024 * 1024);	1418	SetupAlgResResp(6, 4 * 1024 * 1024);
		1419	+ MOCKER(hrtMemAsyncCopy).stubs().will(returnValue(HCCL_SUCCESS));
1418	executor_->rdmaDataBlockSize_ = 32768;	1420	executor_->rdmaDataBlockSize_ = 32768;
1419	LINK fakeLink = MakeFakeLink();	1421	LINK fakeLink = MakeFakeLink();
1420	SetupOpTransportForRank(5, fakeLink);	1422	SetupOpTransportForRank(5, fakeLink);