Star683
459
代码
代码
Issues34
Pull Requests111
Actions
项目讨论
Wiki
分析
Star683
459
已合并
fix: DataDist DataRace修复 #4541
youxiao创建于 12 天前
fix: DataDist DataRace修复 #4541
已合并
youxiao创建于 12 天前
从youyx/gefix/ip-to-conns-unlock-race合入到cann/gedevelop
共 3 个文件变更+67-21
Mdflow/udf/built_in/entity/llm_comm_entity_mgr.cpp+36-20
@@ -50,7 +50,8 @@ EntityPtr LlmCommEntityMgr::GetEntityByConn(HcclConn conn) {
50 return nullptr;50 return nullptr;
51}51}
52 52 
53-HcclConn LlmCommEntityMgr::GetEntityByIp(uint32_t ip) const {53+HcclConn LlmCommEntityMgr::GetEntityByIp(uint32_t ip) {
54+ std::lock_guard<std::mutex> lock(entity_mutex_);
54 auto iter = ip_to_conns_.find(ip);55 auto iter = ip_to_conns_.find(ip);
55 if (iter != ip_to_conns_.end()) {56 if (iter != ip_to_conns_.end()) {
56 return iter->second;57 return iter->second;
@@ -301,25 +302,23 @@ void LlmCommEntityMgr::HandleLinkRequest() {
301 }302 }
302 // accept new link303 // accept new link
303 const uint32_t remote_ip = remote_hccl_addr.info.tcp.ipv4Addr;304 const uint32_t remote_ip = remote_hccl_addr.info.tcp.ipv4Addr;
304- auto iter = ip_to_conns_.find(remote_ip);305+ bool cleared_residual = false;
305- if (iter != ip_to_conns_.end()) {306+ auto entity = FindServerEntityByIp(remote_ip, cleared_residual);
306- auto entity = GetEntityByConn(iter->second);307+ if (entity != nullptr) {
307- if (entity == nullptr) {308+ (void)HcclRawForceClose(entity->GetConn());
308- EraseIpToConnMap(remote_ip, iter->second);309+ entity->SetConn(hccl_conn);
309- UDF_RUN_LOG_INFO("Success to accept new link with residual data in ip_to_conns map, remote hccl addr:%s.",310+ entity->SetLinkEstablished(false);
310- ToDesc(remote_hccl_addr).c_str());311+ entity->SetProbeLinkClusterInfoFlag(false);
311- } else {312+ entity->ClearResource();
312- (void)HcclRawForceClose(entity->GetConn());313+ entity->ChangeState(FsmState::kFsmLinkState);
313- entity->SetConn(hccl_conn);314+ UDF_RUN_LOG_INFO("Success to accept new force link, remote hccl addr:%s.", ToDesc(remote_hccl_addr).c_str());
314- entity->SetLinkEstablished(false);315+ return;
315- entity->SetProbeLinkClusterInfoFlag(false);
316- entity->ClearResource();
317- entity->ChangeState(FsmState::kFsmLinkState);
318- UDF_RUN_LOG_INFO("Success to accept new force link, remote hccl addr:%s.", ToDesc(remote_hccl_addr).c_str());
319- return;
320- }
321 }316 }
322- EntityPtr entity = this->CreateEntity(EntityType::kEntityServer, hccl_conn, listen_hccl_addr_, remote_hccl_addr);317+ if (cleared_residual) {
318+ UDF_RUN_LOG_INFO("Success to accept new link with residual data in ip_to_conns map, remote hccl addr:%s.",
319+ ToDesc(remote_hccl_addr).c_str());
320+ }
321+ entity = this->CreateEntity(EntityType::kEntityServer, hccl_conn, listen_hccl_addr_, remote_hccl_addr);
323 if (entity == nullptr) {322 if (entity == nullptr) {
324 UDF_LOG_ERROR("failed to create server comm entity.");323 UDF_LOG_ERROR("failed to create server comm entity.");
325 return;324 return;
@@ -415,12 +414,29 @@ FsmStatus LlmCommEntityMgr::UnRegisterHcclMr(std::vector<uint64_t> &mem_addrs) {
415}414}
416 415 
417void LlmCommEntityMgr::ClearEntities() {416void LlmCommEntityMgr::ClearEntities() {
418- ip_to_conns_.clear();
419 std::lock_guard<std::mutex> lock(entity_mutex_);417 std::lock_guard<std::mutex> lock(entity_mutex_);
418+ ip_to_conns_.clear();
420 server_entity_map_.clear();419 server_entity_map_.clear();
421 client_entity_map_.clear();420 client_entity_map_.clear();
422}421}
423 422 
423+EntityPtr LlmCommEntityMgr::FindServerEntityByIp(uint32_t ip, bool &cleared_residual) {
424+ cleared_residual = false;
425+ std::lock_guard<std::mutex> lock(entity_mutex_);
426+ auto iter = ip_to_conns_.find(ip);
427+ if (iter == ip_to_conns_.end()) {
428+ return nullptr;
429+ }
430+ const HcclConn conn = iter->second;
431+ auto entity_iter = server_entity_map_.find(conn);
432+ if (entity_iter == server_entity_map_.end()) {
433+ EraseIpToConnMap(ip, conn);
434+ cleared_residual = true;
435+ return nullptr;
436+ }
437+ return entity_iter->second;
438+}
439+ 
424void LlmCommEntityMgr::EraseIpToConnMap(uint32_t ip, const HcclConn conn) {440void LlmCommEntityMgr::EraseIpToConnMap(uint32_t ip, const HcclConn conn) {
425 for (auto multi_iter = ip_to_conns_.find(ip); multi_iter != ip_to_conns_.end(); multi_iter++) {441 for (auto multi_iter = ip_to_conns_.find(ip); multi_iter != ip_to_conns_.end(); multi_iter++) {
426 if (multi_iter->second == conn) {442 if (multi_iter->second == conn) {
Mdflow/udf/built_in/entity/llm_comm_entity_mgr.h+2-1
@@ -25,7 +25,7 @@ class LlmCommEntityMgr {
25 ~LlmCommEntityMgr();25 ~LlmCommEntityMgr();
26 EntityPtr GetEntityByConn(HcclConn conn);26 EntityPtr GetEntityByConn(HcclConn conn);
27 EntityPtr GetEntityByRemoteClusterId(uint64_t remote_cluster_id);27 EntityPtr GetEntityByRemoteClusterId(uint64_t remote_cluster_id);
28- HcclConn GetEntityByIp(uint32_t ip) const;28+ HcclConn GetEntityByIp(uint32_t ip);
29 EntityPtr CreateEntity(EntityType type, HcclConn conn, HcclAddr &local_hccl_addr, HcclAddr &remote_hccl_addr,29 EntityPtr CreateEntity(EntityType type, HcclConn conn, HcclAddr &local_hccl_addr, HcclAddr &remote_hccl_addr,
30 uint64_t remote_cluster_id = 0);30 uint64_t remote_cluster_id = 0);
31 FsmStatus MarkEntityDeletedByConn(HcclConn conn);31 FsmStatus MarkEntityDeletedByConn(HcclConn conn);
@@ -57,6 +57,7 @@ class LlmCommEntityMgr {
57 LlmCommEntityMgr();57 LlmCommEntityMgr();
58 void EraseClientMapByClusterId(uint64_t remote_cluster_id);58 void EraseClientMapByClusterId(uint64_t remote_cluster_id);
59 void EraseIpToConnMap(uint32_t ip, const HcclConn conn);59 void EraseIpToConnMap(uint32_t ip, const HcclConn conn);
60+ EntityPtr FindServerEntityByIp(uint32_t ip, bool &cleared_residual);
60 static FsmStatus QueryCurMemGrp(char **group_name);61 static FsmStatus QueryCurMemGrp(char **group_name);
61 void ReopenServerConn();62 void ReopenServerConn();
62 std::unordered_map<HcclConn, EntityPtr> server_entity_map_;63 std::unordered_map<HcclConn, EntityPtr> server_entity_map_;
Mtests/dflow/udf/ut/testcase/built_in/llm_service_flow_func_utest.cpp+29-0
@@ -4212,4 +4212,33 @@ TEST_F(LlmServiceFlowFuncUTest, test_receive_with_cacheid_push_not_exist_kv) {
4212 EXPECT_EQ(DeallocateCache(run_context, out_msg, llm_service_flow_func, cache_id), FsmStatus::kFsmSuccess);4212 EXPECT_EQ(DeallocateCache(run_context, out_msg, llm_service_flow_func, cache_id), FsmStatus::kFsmSuccess);
4213}4213}
4214 4214 
4215+TEST_F(LlmServiceFlowFuncUTest, get_entity_by_ip_locked_lookup) {
4216+ LlmCommEntityMgr::GetInstance().ClearEntities();
4217+ HcclAddr local_hccl_addr{};
4218+ local_hccl_addr.info.tcp.ipv4Addr = 3232235777; // 192.168.1.1
4219+ local_hccl_addr.info.tcp.port = 8001;
4220+ HcclAddr remote_hccl_addr{};
4221+ remote_hccl_addr.info.tcp.ipv4Addr = 3232235778; // 192.168.1.2
4222+ remote_hccl_addr.info.tcp.port = 8002;
4223+ auto conn = reinterpret_cast<HcclConn>(1);
4224+ auto entity = LlmCommEntityMgr::GetInstance().CreateEntity(EntityType::kEntityServer, conn, local_hccl_addr,
4225+ remote_hccl_addr, 1);
4226+ EXPECT_NE(entity, nullptr);
4227+ EXPECT_EQ(LlmCommEntityMgr::GetInstance().GetEntityByIp(remote_hccl_addr.info.tcp.ipv4Addr), conn);
4228+ EXPECT_EQ(LlmCommEntityMgr::GetInstance().GetEntityByIp(0U), nullptr);
4229+ 
4230+ bool cleared_residual = false;
4231+ auto found =
4232+ LlmCommEntityMgr::GetInstance().FindServerEntityByIp(remote_hccl_addr.info.tcp.ipv4Addr, cleared_residual);
4233+ EXPECT_EQ(found, entity);
4234+ EXPECT_FALSE(cleared_residual);
4235+ 
4236+ LlmCommEntityMgr::GetInstance().server_entity_map_.clear();
4237+ found = LlmCommEntityMgr::GetInstance().FindServerEntityByIp(remote_hccl_addr.info.tcp.ipv4Addr, cleared_residual);
4238+ EXPECT_EQ(found, nullptr);
4239+ EXPECT_TRUE(cleared_residual);
4240+ EXPECT_EQ(LlmCommEntityMgr::GetInstance().GetEntityByIp(remote_hccl_addr.info.tcp.ipv4Addr), nullptr);
4241+ LlmCommEntityMgr::GetInstance().ClearEntities();
4242+}
4243+ 
4215} // namespace FlowFunc4244} // namespace FlowFunc