fix gatewayNodeInfo encode concurrency bug

WeBankBlockchain · Sep 6, 2024 · 89d3ec5 · 89d3ec5
1 parent 145eb51
commit 89d3ec5
Show file tree

Hide file tree

Showing 13 changed files with 172 additions and 120 deletions.
diff --git a/cpp/wedpr-helper/ppc-utilities/Utilities.h b/cpp/wedpr-helper/ppc-utilities/Utilities.h
@@ -42,9 +42,6 @@ inline uint64_t decodeNetworkBuffer(
         return curOffset;
     }
     CHECK_OFFSET_WITH_THROW_EXCEPTION(curOffset, bufferLen);
-    //_result.insert(
-    //    _result.end(), (bcos::byte*)buffer + curOffset, (bcos::byte*)buffer + curOffset +
-    //    dataLen);
     _result.assign((bcos::byte*)buffer + curOffset, (bcos::byte*)buffer + curOffset + dataLen);
     curOffset += dataLen;
     return curOffset;

diff --git a/cpp/wedpr-transport/ppc-gateway/ppc-gateway/Common.h b/cpp/wedpr-transport/ppc-gateway/ppc-gateway/Common.h
@@ -31,6 +31,8 @@
 namespace ppc::gateway
 {
 #define GATEWAY_LOG(LEVEL) BCOS_LOG(LEVEL) << "[GATEWAY]"
+#define SERVICE_LOG(LEVEL) BCOS_LOG(LEVEL) << "[GATEWAY][SERVICE]"
+#define SERVICE_ROUTER_LOG(LEVEL) BCOS_LOG(LEVEL) << "[GATEWAY][SERVICE][ROUTER]"
 
 // HTTP HEADER DEFINE
 #define HEAD_TASK_ID "x-ptp-session-id"

diff --git a/cpp/wedpr-transport/ppc-gateway/ppc-gateway/GatewayFactory.cpp b/cpp/wedpr-transport/ppc-gateway/ppc-gateway/GatewayFactory.cpp
@@ -60,7 +60,7 @@ Service::Ptr GatewayFactory::buildService() const
     wsInitializer->setConfig(wsConfig);
     auto p2pService = std::make_shared<Service>(m_contextConfig->nodeID(),
         std::make_shared<RouterTableFactoryImpl>(), m_config->gatewayConfig().unreachableDistance,
-        "Service");
+        "Gateway-Service");
     p2pService->setTimerFactory(std::make_shared<bcos::timer::TimerFactory>());
     p2pService->setNodeEndpoints(m_gatewayConfig->nodeIPEndpointSet());
 

diff --git a/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/GatewayNodeInfoImpl.cpp b/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/GatewayNodeInfoImpl.cpp
@@ -58,18 +58,35 @@ INodeInfo::Ptr GatewayNodeInfoImpl::nodeInfo(bcos::bytes const& nodeID) const
     return nullptr;
 }
 
+void GatewayNodeInfoImpl::updateNodeList()
+{
+    // Note: can't use clear_nodelist here, for clear_nodelist will destroy the allocated nodelist,
+    // and cause double release coredump
+    releaseWithoutDestory();
+    // re-encode nodeList
+    for (auto const& it : m_nodeList)
+    {
+        auto nodeInfo = std::dynamic_pointer_cast<NodeInfoImpl>(it.second);
+        m_rawGatewayInfo->mutable_nodelist()->UnsafeArenaAddAllocated(
+            nodeInfo->rawNodeInfo().get());
+    }
+}
+
 bool GatewayNodeInfoImpl::tryAddNodeInfo(INodeInfo::Ptr const& info)
 {
     auto nodeID = info->nodeID().toBytes();
     auto existedNodeInfo = nodeInfo(nodeID);
-    // update the info
-    if (existedNodeInfo == nullptr || !existedNodeInfo->equal(info))
+    // the node info has not been updated
+    if (existedNodeInfo != nullptr && existedNodeInfo->equal(info))
+    {
+        return false;
+    }
     {
         bcos::WriteGuard l(x_nodeList);
         m_nodeList[nodeID] = info;
-        return true;
+        updateNodeList();
     }
-    return false;
+    return true;
 }
 
 void GatewayNodeInfoImpl::removeNodeInfo(bcos::bytes const& nodeID)
@@ -84,6 +101,7 @@ void GatewayNodeInfoImpl::removeNodeInfo(bcos::bytes const& nodeID)
         }
         bcos::UpgradeGuard ul(l);
         m_nodeList.erase(it);
+        updateNodeList();
     }
     // remove the topic info
     {
@@ -184,17 +202,6 @@ void GatewayNodeInfoImpl::unRegisterTopic(bcos::bytes const& nodeID, std::string
 
 void GatewayNodeInfoImpl::encode(bcos::bytes& data) const
 {
-    m_rawGatewayInfo->clear_nodelist();
-    {
-        bcos::ReadGuard l(x_nodeList);
-        // encode nodeList
-        for (auto const& it : m_nodeList)
-        {
-            auto nodeInfo = std::dynamic_pointer_cast<NodeInfoImpl>(it.second);
-            m_rawGatewayInfo->mutable_nodelist()->UnsafeArenaAddAllocated(
-                nodeInfo->rawNodeInfo().get());
-        }
-    }
     encodePBObject(data, m_rawGatewayInfo);
 }
 

diff --git a/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/GatewayNodeInfoImpl.h b/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/GatewayNodeInfoImpl.h
@@ -39,15 +39,7 @@ class GatewayNodeInfoImpl : public GatewayNodeInfo
 
     GatewayNodeInfoImpl(bcos::bytesConstRef data) : GatewayNodeInfoImpl() { decode(data); }
 
-    ~GatewayNodeInfoImpl() override
-    {
-        // return back the ownership to nodeList
-        auto allocatedNodeListSize = m_rawGatewayInfo->nodelist_size();
-        for (int i = 0; i < allocatedNodeListSize; i++)
-        {
-            m_rawGatewayInfo->mutable_nodelist()->UnsafeArenaReleaseLast();
-        }
-    }
+    ~GatewayNodeInfoImpl() override { releaseWithoutDestory(); }
 
     // the gateway nodeID
     std::string const& p2pNodeID() const override;
@@ -88,6 +80,19 @@ class GatewayNodeInfoImpl : public GatewayNodeInfo
         return m_nodeList.size();
     }
 
+private:
+    void updateNodeList();
+
+    void releaseWithoutDestory()
+    {
+        // return back the ownership to nodeList to shared_ptr
+        auto allocatedNodeListSize = m_rawGatewayInfo->nodelist_size();
+        for (int i = 0; i < allocatedNodeListSize; i++)
+        {
+            m_rawGatewayInfo->mutable_nodelist()->UnsafeArenaReleaseLast();
+        }
+    }
+
 private:
     std::shared_ptr<ppc::proto::GatewayNodeInfo> m_rawGatewayInfo;
     // NodeID => nodeInfo

diff --git a/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/GatewayRouterManager.cpp b/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/GatewayRouterManager.cpp
@@ -93,7 +93,8 @@ void GatewayRouterManager::onReceiveNodeSeqMessage(MessageFace::Ptr msg, WsSessi
         return;
     }
     // status changed, request for the nodeStatus
-    GATEWAY_LOG(TRACE) << LOG_DESC("onReceiveNodeSeqMessage") << LOG_KV("from", from)
+    GATEWAY_LOG(TRACE) << LOG_DESC("onReceiveNodeSeqMessage")
+                       << LOG_KV("from", printP2PIDElegantly(from))
                        << LOG_KV("statusSeq", statusSeq);
     m_service->asyncSendMessageByP2PNodeID(
         (uint16_t)GatewayPacketType::RequestNodeStatus, from, std::make_shared<bcos::bytes>());
@@ -138,11 +139,11 @@ void GatewayRouterManager::onReceiveRequestNodeStatusMsg(
     if (!nodeStatusData)
     {
         GATEWAY_LOG(WARNING) << LOG_DESC("onReceiveRequestNodeStatusMsg: generate nodeInfo error")
-                             << LOG_KV("from", from);
+                             << LOG_KV("from", printP2PIDElegantly(from));
         return;
     }
     GATEWAY_LOG(TRACE) << LOG_DESC("onReceiveRequestNodeStatusMsg: response the latest nodeStatus")
-                       << LOG_KV("from", from);
+                       << LOG_KV("from", printP2PIDElegantly(from));
     m_service->asyncSendMessageByP2PNodeID(
         (uint16_t)GatewayPacketType::ResponseNodeStatus, from, nodeStatusData);
 }
@@ -157,7 +158,8 @@ void GatewayRouterManager::onRecvResponseNodeStatusMsg(MessageFace::Ptr msg, WsS
                            p2pMessage->header()->srcGwNode() :
                            session->nodeId();
 
-    GATEWAY_LOG(INFO) << LOG_DESC("onRecvResponseNodeStatusMsg") << LOG_KV("from", from)
+    GATEWAY_LOG(INFO) << LOG_DESC("onRecvResponseNodeStatusMsg")
+                      << LOG_KV("from", printP2PIDElegantly(from))
                       << LOG_KV("statusSeq", nodeStatus->statusSeq())
                       << LOG_KV("agency", nodeStatus->agency());
     updatePeerNodeStatus(from, nodeStatus);
@@ -176,7 +178,8 @@ void GatewayRouterManager::updatePeerNodeStatus(
         UpgradeGuard ul(l);
         m_p2pID2Seq[p2pID] = statusSeq;
     }
-    GATEWAY_LOG(INFO) << LOG_DESC("updatePeerNodeStatus") << LOG_KV("from", p2pID)
+    GATEWAY_LOG(INFO) << LOG_DESC("updatePeerNodeStatus")
+                      << LOG_KV("from", printP2PIDElegantly(p2pID))
                       << LOG_KV("statusSeq", status->statusSeq())
                       << LOG_KV("agency", status->agency());
     m_peerRouter->updateGatewayInfo(status);

diff --git a/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/LocalRouter.h b/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/LocalRouter.h
@@ -46,7 +46,7 @@ class LocalRouter
         increaseSeq();
     }
 
-    virtual void registerTopic(std::string_view nodeID, std::string const& topic);
+    virtual void registerTopic(bcos::bytesConstRef nodeID, std::string const& topic);
     virtual void unRegisterTopic(bcos::bytesConstRef nodeID, std::string const& topic);
 
     virtual std::vector<ppc::front::IFrontClient::Ptr> chooseReceiver(