fix select nullptr front when dispatcherMessage caused coredump (#95)

cyjseagull · web-flow · commit 75b56b730903 · 2024-11-19T20:52:53.000+08:00
* fix waitress version

* fix select nullptr front when dispatcherMessage caused coredump

* fix model can't send message when some gateway down
diff --git a/cpp/ppc-framework/protocol/INodeInfo.h b/cpp/ppc-framework/protocol/INodeInfo.h
@@ -57,7 +57,7 @@ class INodeInfo
     virtual void decode(bcos::bytesConstRef data) = 0;
 
     virtual void setFront(std::shared_ptr<ppc::front::IFrontClient>&& front) = 0;
-    virtual std::shared_ptr<ppc::front::IFrontClient> const& getFront() const = 0;
+    virtual std::shared_ptr<ppc::front::IFrontClient> getFront() const = 0;
 
     virtual bool equal(INodeInfo::Ptr const& info)
     {
diff --git a/cpp/wedpr-protocol/grpc/client/GrpcClient.cpp b/cpp/wedpr-protocol/grpc/client/GrpcClient.cpp
@@ -75,6 +75,7 @@ bcos::Error::Ptr GrpcClient::broadCast(
     std::function<bcos::Error::Ptr(ChannelInfo const& channel)> callback)
 {
     auto result = std::make_shared<bcos::Error>(0, "");
+    int successCount = 0;
     for (auto const& channel : m_broadcastChannels)
     {
         try
@@ -91,6 +92,10 @@ bcos::Error::Ptr GrpcClient::broadCast(
                 result->setErrorCode(error->errorCode());
                 result->setErrorMessage(result->errorMessage() + error->errorMessage() + "; ");
             }
+            else
+            {
+                successCount++;
+            }
         }
         catch (std::exception const& e)
         {
@@ -99,5 +104,10 @@ bcos::Error::Ptr GrpcClient::broadCast(
                 << LOG_KV("error", boost::diagnostic_information(e));
         }
     }
+    // at least one success
+    if (successCount > 0)
+    {
+        return std::make_shared<bcos::Error>(0, "success");
+    }
     return result;
 }
diff --git a/cpp/wedpr-protocol/protobuf/src/NodeInfoImpl.h b/cpp/wedpr-protocol/protobuf/src/NodeInfoImpl.h
@@ -117,9 +117,14 @@ class NodeInfoImpl : public INodeInfo
 
     void setFront(std::shared_ptr<ppc::front::IFrontClient>&& front) override
     {
+        bcos::WriteGuard l(x_front);
         m_front = std::move(front);
     }
-    std::shared_ptr<ppc::front::IFrontClient> const& getFront() const override { return m_front; }
+    std::shared_ptr<ppc::front::IFrontClient> getFront() const override
+    {
+        bcos::ReadGuard l(x_front);
+        return m_front;
+    }
 
     void toJson(Json::Value& jsonObject) const override;
 
@@ -142,6 +147,8 @@ class NodeInfoImpl : public INodeInfo
 
 private:
     std::shared_ptr<ppc::front::IFrontClient> m_front;
+    mutable bcos::SharedMutex x_front;
+
     std::set<std::string> m_components;
     mutable bcos::SharedMutex x_components;
 
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
@@ -155,9 +155,10 @@ std::vector<std::shared_ptr<ppc::front::IFrontClient>> GatewayNodeInfoImpl::choo
     bcos::ReadGuard l(x_nodeList);
     for (auto const& it : m_nodeList)
     {
-        if (it.second->components().count(component))
+        auto front = it.second->getFront();
+        if (front && it.second->components().count(component))
         {
-            result.emplace_back(it.second->getFront());
+            result.emplace_back(front);
         }
         if (!result.empty() && !selectAll)
         {
@@ -175,7 +176,11 @@ std::vector<std::shared_ptr<ppc::front::IFrontClient>> GatewayNodeInfoImpl::choo
     bcos::ReadGuard l(x_nodeList);
     for (auto const& it : m_nodeList)
     {
-        result.emplace_back(it.second->getFront());
+        auto front = it.second->getFront();
+        if (front)
+        {
+            result.emplace_back(front);
+        }
         if (!result.empty() && !selectAll)
         {
             break;
@@ -194,7 +199,11 @@ std::vector<std::shared_ptr<ppc::front::IFrontClient>> GatewayNodeInfoImpl::choo
         bcos::ReadGuard l(x_nodeList);
         for (auto const& it : m_nodeList)
         {
-            result.emplace_back(it.second->getFront());
+            auto front = it.second->getFront();
+            if (front)
+            {
+                result.emplace_back(front);
+            }
         }
         return result;
     }
@@ -210,7 +219,11 @@ std::vector<std::shared_ptr<ppc::front::IFrontClient>> GatewayNodeInfoImpl::choo
         // ignore the fromNode
         if (selectedNode != nullptr && selectedNode->nodeID().toBytes() != fromNode)
         {
-            result.emplace_back(selectedNode->getFront());
+            auto front = selectedNode->getFront();
+            if (front)
+            {
+                result.emplace_back(front);
+            }
         }
         if (!result.empty() && !selectAll)
         {
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
@@ -103,9 +103,8 @@ void GatewayRouterManager::onReceiveNodeSeqMessage(MessageFace::Ptr msg, WsSessi
     auto statusSeq =
         boost::asio::detail::socket_ops::network_to_host_long(*((uint32_t*)msg->payload()->data()));
     auto p2pMessage = std::dynamic_pointer_cast<P2PMessage>(msg);
-    auto const& from = (p2pMessage->header()->srcGwNode().size() > 0) ?
-                           p2pMessage->header()->srcGwNode() :
-                           session->nodeId();
+    auto from = (p2pMessage->header()->srcGwNode().size() > 0) ? p2pMessage->header()->srcGwNode() :
+                                                                 session->nodeId();
     auto statusSeqChanged = statusChanged(from, statusSeq);
     if (!statusSeqChanged)
     {
@@ -151,9 +150,8 @@ void GatewayRouterManager::onReceiveRequestNodeStatusMsg(
     MessageFace::Ptr msg, WsSession::Ptr session)
 {
     auto p2pMessage = std::dynamic_pointer_cast<P2PMessage>(msg);
-    auto const& from = (!p2pMessage->header()->srcGwNode().empty()) ?
-                           p2pMessage->header()->srcGwNode() :
-                           session->nodeId();
+    auto from = (!p2pMessage->header()->srcGwNode().empty()) ? p2pMessage->header()->srcGwNode() :
+                                                               session->nodeId();
 
     auto nodeStatusData = m_localRouter->generateNodeStatus();
     if (!nodeStatusData)
diff --git a/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/LocalRouter.cpp b/cpp/wedpr-transport/ppc-gateway/ppc-gateway/gateway/router/LocalRouter.cpp
@@ -98,6 +98,10 @@ bool LocalRouter::dispatcherMessage(
         int i = 0;
         for (auto const& front : frontList)
         {
+            if (!front)
+            {
+                continue;
+            }
             if (i == 0)
             {
                 front->onReceiveMessage(msg->msg(), callback);
@@ -153,9 +157,10 @@ std::vector<ppc::front::IFrontClient::Ptr> LocalRouter::chooseReceiver(
     case (uint16_t)RouteType::ROUTE_THROUGH_NODEID:
     {
         auto gatewayInfo = m_routerInfo->nodeInfo(msg->header()->optionalField()->dstNode());
-        if (gatewayInfo != nullptr)
+        auto front = gatewayInfo ? gatewayInfo->getFront() : nullptr;
+        if (gatewayInfo != nullptr && front)
         {
-            receivers.emplace_back(gatewayInfo->getFront());
+            receivers.emplace_back(front);
         }
         return receivers;
     }
diff --git a/python/requirements.txt b/python/requirements.txt
@@ -27,7 +27,7 @@ pycryptodome==3.9.9
 pyjwt
 pyyaml
 mysqlclient==2.1.0
-waitress==3.0.1
+waitress==3.0.0
 sqlparse~=0.4.1
 toolz~=0.11.1
 tenacity==7.0.0

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,7 @@ class INodeInfo`
`57`	`57`	`virtual void decode(bcos::bytesConstRef data) = 0;`
`58`	`58`
`59`	`59`	`virtual void setFront(std::shared_ptr<ppc::front::IFrontClient>&& front) = 0;`
`60`		`- virtual std::shared_ptr<ppc::front::IFrontClient> const& getFront() const = 0;`
	`60`	`+ virtual std::shared_ptr<ppc::front::IFrontClient> getFront() const = 0;`
`61`	`61`
`62`	`62`	`virtual bool equal(INodeInfo::Ptr const& info)`
`63`	`63`	`{`
Original file line number	Diff line number	Diff line change
`@@ -75,6 +75,7 @@ bcos::Error::Ptr GrpcClient::broadCast(`
`75`	`75`	`std::function<bcos::Error::Ptr(ChannelInfo const& channel)> callback)`
`76`	`76`	`{`
`77`	`77`	`auto result = std::make_shared<bcos::Error>(0, "");`
	`78`	`+ int successCount = 0;`
`78`	`79`	`for (auto const& channel : m_broadcastChannels)`
`79`	`80`	`{`
`80`	`81`	`try`
`@@ -91,6 +92,10 @@ bcos::Error::Ptr GrpcClient::broadCast(`
`91`	`92`	`result->setErrorCode(error->errorCode());`
`92`	`93`	`result->setErrorMessage(result->errorMessage() + error->errorMessage() + "; ");`
`93`	`94`	`}`
	`95`	`+ else`
	`96`	`+ {`
	`97`	`+ successCount++;`
	`98`	`+ }`
`94`	`99`	`}`
`95`	`100`	`catch (std::exception const& e)`
`96`	`101`	`{`
`@@ -99,5 +104,10 @@ bcos::Error::Ptr GrpcClient::broadCast(`
`99`	`104`	`<< LOG_KV("error", boost::diagnostic_information(e));`
`100`	`105`	`}`
`101`	`106`	`}`
	`107`	`+ // at least one success`
	`108`	`+ if (successCount > 0)`
	`109`	`+ {`
	`110`	`+ return std::make_shared<bcos::Error>(0, "success");`
	`111`	`+ }`
`102`	`112`	`return result;`
`103`	`113`	`}`
Original file line number	Diff line number	Diff line change
`@@ -155,9 +155,10 @@ std::vector<std::shared_ptr<ppc::front::IFrontClient>> GatewayNodeInfoImpl::choo`
`155`	`155`	`bcos::ReadGuard l(x_nodeList);`
`156`	`156`	`for (auto const& it : m_nodeList)`
`157`	`157`	`{`
`158`		`- if (it.second->components().count(component))`
	`158`	`+ auto front = it.second->getFront();`
	`159`	`+ if (front && it.second->components().count(component))`
`159`	`160`	`{`
`160`		`- result.emplace_back(it.second->getFront());`
	`161`	`+ result.emplace_back(front);`
`161`	`162`	`}`
`162`	`163`	`if (!result.empty() && !selectAll)`
`163`	`164`	`{`
`@@ -175,7 +176,11 @@ std::vector<std::shared_ptr<ppc::front::IFrontClient>> GatewayNodeInfoImpl::choo`
`175`	`176`	`bcos::ReadGuard l(x_nodeList);`
`176`	`177`	`for (auto const& it : m_nodeList)`
`177`	`178`	`{`
`178`		`- result.emplace_back(it.second->getFront());`
	`179`	`+ auto front = it.second->getFront();`
	`180`	`+ if (front)`
	`181`	`+ {`
	`182`	`+ result.emplace_back(front);`
	`183`	`+ }`
`179`	`184`	`if (!result.empty() && !selectAll)`
`180`	`185`	`{`
`181`	`186`	`break;`
`@@ -194,7 +199,11 @@ std::vector<std::shared_ptr<ppc::front::IFrontClient>> GatewayNodeInfoImpl::choo`
`194`	`199`	`bcos::ReadGuard l(x_nodeList);`
`195`	`200`	`for (auto const& it : m_nodeList)`
`196`	`201`	`{`
`197`		`- result.emplace_back(it.second->getFront());`
	`202`	`+ auto front = it.second->getFront();`
	`203`	`+ if (front)`
	`204`	`+ {`
	`205`	`+ result.emplace_back(front);`
	`206`	`+ }`
`198`	`207`	`}`
`199`	`208`	`return result;`
`200`	`209`	`}`
`@@ -210,7 +219,11 @@ std::vector<std::shared_ptr<ppc::front::IFrontClient>> GatewayNodeInfoImpl::choo`
`210`	`219`	`// ignore the fromNode`
`211`	`220`	`if (selectedNode != nullptr && selectedNode->nodeID().toBytes() != fromNode)`
`212`	`221`	`{`
`213`		`- result.emplace_back(selectedNode->getFront());`
	`222`	`+ auto front = selectedNode->getFront();`
	`223`	`+ if (front)`
	`224`	`+ {`
	`225`	`+ result.emplace_back(front);`
	`226`	`+ }`
`214`	`227`	`}`
`215`	`228`	`if (!result.empty() && !selectAll)`
`216`	`229`	`{`
Original file line number	Diff line number	Diff line change
`@@ -98,6 +98,10 @@ bool LocalRouter::dispatcherMessage(`
`98`	`98`	`int i = 0;`
`99`	`99`	`for (auto const& front : frontList)`
`100`	`100`	`{`
	`101`	`+ if (!front)`
	`102`	`+ {`
	`103`	`+ continue;`
	`104`	`+ }`
`101`	`105`	`if (i == 0)`
`102`	`106`	`{`
`103`	`107`	`front->onReceiveMessage(msg->msg(), callback);`
`@@ -153,9 +157,10 @@ std::vector<ppc::front::IFrontClient::Ptr> LocalRouter::chooseReceiver(`
`153`	`157`	`case (uint16_t)RouteType::ROUTE_THROUGH_NODEID:`
`154`	`158`	`{`
`155`	`159`	`auto gatewayInfo = m_routerInfo->nodeInfo(msg->header()->optionalField()->dstNode());`
`156`		`- if (gatewayInfo != nullptr)`
	`160`	`+ auto front = gatewayInfo ? gatewayInfo->getFront() : nullptr;`
	`161`	`+ if (gatewayInfo != nullptr && front)`
`157`	`162`	`{`
`158`		`- receivers.emplace_back(gatewayInfo->getFront());`
	`163`	`+ receivers.emplace_back(front);`
`159`	`164`	`}`
`160`	`165`	`return receivers;`
`161`	`166`	`}`