add asyncGetPeers && registerComponent && unRegisterComponent

WeBankBlockchain · Sep 18, 2024 · 6b15919 · 6b15919
1 parent 299f946
commit 6b15919
Show file tree

Hide file tree

Showing 34 changed files with 722 additions and 64 deletions.
diff --git a/cpp/cmake/Options.cmake b/cpp/cmake/Options.cmake
@@ -181,6 +181,7 @@ macro(print_config NAME)
     message("-- BUILD_SDK           BUILD SDK                    ${BUILD_SDK}")
     message("-- BUILD_UDF           BUILD UDF                    ${BUILD_UDF}")
     message("-- BUILD_WEDPR_TOOLKIT BUILD_WEDPR_TOOLKIT          ${BUILD_WEDPR_TOOLKIT}")
+    message("-- AUTO_GENERATE       AUTO_GENERATE                ${AUTO_GENERATE}")
     message("-- DEBUG                                            ${DEBUG}")
     message("------------------------------------------------------------------------")
     message("")

diff --git a/cpp/ppc-framework/front/IFront.h b/cpp/ppc-framework/front/IFront.h
@@ -63,6 +63,7 @@ class MessageDispatcherHandler
     virtual void onMessage(ppc::protocol::Message::Ptr msg) = 0;
 };
 
+
 class SendResponseHandler
 {
 public:
@@ -92,6 +93,16 @@ class IMessageHandler
         SendResponseHandler sendResponseHandler) = 0;
 };
 
+class GetPeersInfoHandler
+{
+public:
+    using Ptr = std::shared_ptr<GetPeersInfoHandler>;
+    GetPeersInfoHandler() = default;
+    virtual ~GetPeersInfoHandler() {}
+
+    virtual void onPeersInfo(bcos::Error::Ptr e, std::string const& peersInfo) = 0;
+};
+
 ///////// the callback definition for sdk wrapper /////////
 
 class IFront : virtual public IFrontClient
@@ -183,14 +194,15 @@ class IFront : virtual public IFrontClient
 
     // !!! Note: the 'payload ' type(char*) should not been changed, since it used to pass-in java
     // byte[] data
-    virtual void async_send_response(char* dstNode, uint64_t dstNodeSize, std::string const& traceID,
-        char* payload, uint64_t payloadSize, int seq, ErrorCallback::Ptr errorCallback)
+    virtual void async_send_response(char* dstNode, uint64_t dstNodeSize,
+        std::string const& traceID, char* payload, uint64_t payloadSize, int seq,
+        ErrorCallback::Ptr errorCallback)
     {
         // TODO: optimize here
         bcos::bytes copiedDstNode(dstNode, dstNode + dstNodeSize);
         bcos::bytes copyedPayload(payload, payload + payloadSize);
-        asyncSendResponse(
-            copiedDstNode, traceID, std::move(copyedPayload), seq, populateErrorCallback(errorCallback));
+        asyncSendResponse(copiedDstNode, traceID, std::move(copyedPayload), seq,
+            populateErrorCallback(errorCallback));
     }
 
     // the sync interface for async_send_message
@@ -209,9 +221,11 @@ class IFront : virtual public IFrontClient
     virtual ppc::protocol::Message::Ptr pop(std::string const& topic, long timeoutMs) = 0;
     virtual ppc::protocol::Message::Ptr peek(std::string const& topic) = 0;
 
-    virtual void asyncGetAgencies(
+    virtual void asyncGetAgencies(std::vector<std::string> const& components,
         std::function<void(bcos::Error::Ptr, std::set<std::string>)> callback) = 0;
 
+    virtual void asyncGetPeers(GetPeersInfoHandler::Ptr getPeersCallback) = 0;
+
     /**
      * @brief register the nodeInfo to the gateway
      * @param nodeInfo the nodeInfo
@@ -223,6 +237,8 @@ class IFront : virtual public IFrontClient
      */
     virtual bcos::Error::Ptr unRegisterNodeInfo() = 0;
 
+    virtual ppc::protocol::INodeInfo::Ptr const& nodeInfo() = 0;
+
     /**
      * @brief register the topic
      *
@@ -237,6 +253,8 @@ class IFront : virtual public IFrontClient
      */
     virtual bcos::Error::Ptr unRegisterTopic(std::string const& topic) = 0;
 
+    virtual void registerComponent(std::string const& component) = 0;
+    virtual void unRegisterComponent(std::string const& component) = 0;
 
 private:
     ppc::protocol::ReceiveMsgFunc populateErrorCallback(ErrorCallback::Ptr errorCallback)

diff --git a/cpp/ppc-framework/gateway/IGateway.h b/cpp/ppc-framework/gateway/IGateway.h
@@ -66,7 +66,7 @@ class IGateway
         bcos::bytes&& payload) = 0;
 
     virtual void asyncGetPeers(std::function<void(bcos::Error::Ptr, std::string)> callback) = 0;
-    virtual void asyncGetAgencies(
+    virtual void asyncGetAgencies(std::vector<std::string> const& components,
         std::function<void(bcos::Error::Ptr, std::set<std::string>)> callback) = 0;
 
     virtual bcos::Error::Ptr registerNodeInfo(ppc::protocol::INodeInfo::Ptr const& nodeInfo) = 0;

diff --git a/cpp/ppc-framework/protocol/INodeInfo.h b/cpp/ppc-framework/protocol/INodeInfo.h
@@ -47,7 +47,10 @@ class INodeInfo
 
     // components
     virtual void setComponents(std::set<std::string> const& components) = 0;
+    virtual bool addComponent(std::string const& component) = 0;
+    virtual bool eraseComponent(std::string const& component) = 0;
     virtual std::set<std::string> const& components() const = 0;
+    virtual std::vector<std::string> copiedComponents() const = 0;
 
     virtual void encode(bcos::bytes& data) const = 0;
     virtual void decode(bcos::bytesConstRef data) = 0;
@@ -58,7 +61,7 @@ class INodeInfo
     virtual bool equal(INodeInfo::Ptr const& info)
     {
         return (nodeID().toBytes() == info->nodeID().toBytes()) &&
-               (components() == info->components());
+               (copiedComponents() == info->copiedComponents());
     }
 
     virtual void toJson(Json::Value& jsonObject) const = 0;

diff --git a/cpp/wedpr-protocol/grpc/client/GatewayClient.cpp b/cpp/wedpr-protocol/grpc/client/GatewayClient.cpp
@@ -65,12 +65,16 @@ void GatewayClient::asyncGetPeers(std::function<void(bcos::Error::Ptr, std::stri
         });
 }
 
-void GatewayClient::asyncGetAgencies(
+void GatewayClient::asyncGetAgencies(std::vector<std::string> const& components,
     std::function<void(bcos::Error::Ptr, std::set<std::string>)> callback)
 {
     auto response = std::make_shared<AgenciesInfo>();
     auto context = std::make_shared<ClientContext>();
-    auto request = std::make_shared<Empty>();
+    auto request = std::make_shared<Condition>();
+    for (auto const& it : components)
+    {
+        request->add_components(it);
+    }
     // lambda keeps the lifecycle for clientContext
     m_stub->async()->asyncGetAgencies(
         context.get(), request.get(), response.get(), [context, callback, response](Status status) {

diff --git a/cpp/wedpr-protocol/grpc/client/GatewayClient.h b/cpp/wedpr-protocol/grpc/client/GatewayClient.h
@@ -54,7 +54,7 @@ class GatewayClient : public ppc::gateway::IGateway, public GrpcClient
         bcos::bytes&& payload, long timeout, ppc::protocol::ReceiveMsgFunc callback) override;
 
     void asyncGetPeers(std::function<void(bcos::Error::Ptr, std::string)> callback) override;
-    void asyncGetAgencies(
+    void asyncGetAgencies(std::vector<std::string> const& components,
         std::function<void(bcos::Error::Ptr, std::set<std::string>)> callback) override;
 
     void asyncSendbroadcastMessage(ppc::protocol::RouteType routeType,

diff --git a/cpp/wedpr-protocol/grpc/server/GatewayServer.cpp b/cpp/wedpr-protocol/grpc/server/GatewayServer.cpp
@@ -75,14 +75,19 @@ grpc::ServerUnaryReactor* GatewayServer::asyncGetPeers(
     return reactor;
 }
 
-grpc::ServerUnaryReactor* GatewayServer::asyncGetAgencies(
-    grpc::CallbackServerContext* context, const ppc::proto::Empty*, ppc::proto::AgenciesInfo* reply)
+grpc::ServerUnaryReactor* GatewayServer::asyncGetAgencies(grpc::CallbackServerContext* context,
+    const ppc::proto::Condition* condition, ppc::proto::AgenciesInfo* reply)
 {
     ServerUnaryReactor* reactor(context->DefaultReactor());
     try
     {
+        std::vector<std::string> components;
+        for (int i = 0; i < condition->components_size(); i++)
+        {
+            components.emplace_back(condition->components(i));
+        }
         m_gateway->asyncGetAgencies(
-            [reactor, reply](bcos::Error::Ptr error, std::set<std::string> agencies) {
+            components, [reactor, reply](bcos::Error::Ptr error, std::set<std::string> agencies) {
                 toSerializedError(reply->mutable_error(), error);
                 for (auto const& it : agencies)
                 {

diff --git a/cpp/wedpr-protocol/grpc/server/GatewayServer.h b/cpp/wedpr-protocol/grpc/server/GatewayServer.h
@@ -42,7 +42,7 @@ class GatewayServer : public ppc::proto::Gateway::CallbackService
     grpc::ServerUnaryReactor* asyncGetPeers(grpc::CallbackServerContext* context,
         const ppc::proto::Empty* request, ppc::proto::PeersInfo* reply) override;
     grpc::ServerUnaryReactor* asyncGetAgencies(grpc::CallbackServerContext* context,
-        const ppc::proto::Empty* request, ppc::proto::AgenciesInfo* reply) override;
+        const ppc::proto::Condition* request, ppc::proto::AgenciesInfo* reply) override;
 
 
     grpc::ServerUnaryReactor* registerNodeInfo(grpc::CallbackServerContext* context,

diff --git a/cpp/wedpr-protocol/proto/pb/Service.proto b/cpp/wedpr-protocol/proto/pb/Service.proto
@@ -41,7 +41,10 @@ message PeersInfo{
     string peersInfo = 2;
 };
 message Empty{
+};
 
+message Condition{
+    repeated string components = 1;
 };
 
 service Front {
@@ -50,7 +53,7 @@ service Front {
 service Gateway{
     rpc asyncSendMessage(SendedMessageRequest) returns(Error){}
     rpc asyncGetPeers(Empty)returns(PeersInfo){}
-    rpc asyncGetAgencies(Empty)returns(AgenciesInfo){}
+    rpc asyncGetAgencies(Condition)returns(AgenciesInfo){}
     rpc registerNodeInfo(NodeInfo) returns(Error){}
     rpc unRegisterNodeInfo(NodeInfo)returns(Error){}
     rpc registerTopic(NodeInfo) returns(Error){}

diff --git a/cpp/wedpr-protocol/protobuf/src/NodeInfoImpl.cpp b/cpp/wedpr-protocol/protobuf/src/NodeInfoImpl.cpp
@@ -25,18 +25,24 @@ using namespace ppc::protocol;
 
 void NodeInfoImpl::encode(bcos::bytes& data) const
 {
-    // set the components
-    for (auto const& component : m_components)
     {
-        m_rawNodeInfo->add_components(component);
+        bcos::ReadGuard l(x_components);
+        // set the components
+        for (auto const& component : m_components)
+        {
+            m_rawNodeInfo->add_components(component);
+        }
     }
     encodePBObject(data, m_rawNodeInfo);
 }
 void NodeInfoImpl::decode(bcos::bytesConstRef data)
 {
     decodePBObject(m_rawNodeInfo, data);
-    m_components = std::set<std::string>(
-        m_rawNodeInfo->components().begin(), m_rawNodeInfo->components().end());
+    {
+        bcos::WriteGuard l(x_components);
+        m_components = std::set<std::string>(
+            m_rawNodeInfo->components().begin(), m_rawNodeInfo->components().end());
+    }
 }
 
 void NodeInfoImpl::toJson(Json::Value& jsonObject) const

diff --git a/cpp/wedpr-protocol/protobuf/src/NodeInfoImpl.h b/cpp/wedpr-protocol/protobuf/src/NodeInfoImpl.h
@@ -54,9 +54,45 @@ class NodeInfoImpl : public INodeInfo
 
     void setComponents(std::set<std::string> const& components) override
     {
+        bcos::WriteGuard l(x_components);
         m_components = components;
     }
-    std::set<std::string> const& components() const override { return m_components; }
+
+    std::set<std::string> const& components() const override
+    {
+        bcos::ReadGuard l(x_components);
+        return m_components;
+    }
+
+    std::vector<std::string> copiedComponents() const override
+    {
+        bcos::ReadGuard l(x_components);
+        return std::vector<std::string>(m_components.begin(), m_components.end());
+    }
+
+    bool addComponent(std::string const& component) override
+    {
+        bcos::UpgradableGuard l(x_components);
+        if (m_components.count(component))
+        {
+            return false;
+        }
+        bcos::UpgradeGuard ul(l);
+        m_components.insert(component);
+        return true;
+    }
+
+    bool eraseComponent(std::string const& component) override
+    {
+        bcos::UpgradableGuard l(x_components);
+        if (!m_components.count(component))
+        {
+            return false;
+        }
+        bcos::UpgradeGuard ul(l);
+        m_components.erase(component);
+        return true;
+    }
 
     std::string const& endPoint() const override { return m_rawNodeInfo->endpoint(); }
 
@@ -81,6 +117,7 @@ class NodeInfoImpl : public INodeInfo
 private:
     std::shared_ptr<ppc::front::IFrontClient> m_front;
     std::set<std::string> m_components;
+    mutable bcos::SharedMutex x_components;
     std::shared_ptr<ppc::proto::NodeInfo> m_rawNodeInfo;
 };
 

diff --git a/cpp/wedpr-transport/ppc-front/ppc-front/Front.cpp b/cpp/wedpr-transport/ppc-front/ppc-front/Front.cpp
@@ -29,7 +29,7 @@ using namespace ppc::front;
 Front::Front(ppc::front::PPCMessageFaceFactory::Ptr ppcMsgFactory, IFront::Ptr front)
   : m_messageFactory(std::move(ppcMsgFactory)), m_front(std::move(front))
 {
-    m_fetcher = std::make_shared<bcos::Timer>(60 * 1000, "metaFetcher");
+    m_fetcher = std::make_shared<bcos::Timer>(10 * 1000, "metaFetcher");
     m_fetcher->registerTimeoutHandler([this]() {
         try
         {
@@ -58,30 +58,31 @@ void Front::stop()
 void Front::fetchGatewayMetaInfo()
 {
     auto self = weak_from_this();
-    m_front->asyncGetAgencies([self](bcos::Error::Ptr error, std::set<std::string> agencies) {
-        auto front = self.lock();
-        if (!front)
-        {
-            return;
-        }
-        if (error && error->errorCode() != 0)
-        {
-            FRONT_LOG(WARNING) << LOG_DESC("asyncGetAgencies failed")
-                               << LOG_KV("code", error->errorCode())
-                               << LOG_KV("msg", error->errorMessage());
-            return;
-        }
-        std::vector agencyList(agencies.begin(), agencies.end());
-        bcos::UpgradableGuard l(front->x_agencyList);
-        if (front->m_agencyList == agencyList)
-        {
-            return;
-        }
-        bcos::UpgradeGuard ul(l);
-        front->m_agencyList = agencyList;
-        FRONT_LOG(INFO) << LOG_DESC("Update agencies information")
-                        << LOG_KV("agencies", printVector(agencyList));
-    });
+    m_front->asyncGetAgencies(m_front->nodeInfo()->copiedComponents(),
+        [self](bcos::Error::Ptr error, std::set<std::string> agencies) {
+            auto front = self.lock();
+            if (!front)
+            {
+                return;
+            }
+            if (error && error->errorCode() != 0)
+            {
+                FRONT_LOG(WARNING)
+                    << LOG_DESC("asyncGetAgencies failed") << LOG_KV("code", error->errorCode())
+                    << LOG_KV("msg", error->errorMessage());
+                return;
+            }
+            std::vector agencyList(agencies.begin(), agencies.end());
+            bcos::UpgradableGuard l(front->x_agencyList);
+            if (front->m_agencyList == agencyList)
+            {
+                return;
+            }
+            bcos::UpgradeGuard ul(l);
+            front->m_agencyList = agencyList;
+            FRONT_LOG(INFO) << LOG_DESC("Update agencies information")
+                            << LOG_KV("agencies", printVector(agencyList));
+        });
     m_fetcher->restart();
 }
 
@@ -192,4 +193,5 @@ void Front::registerMessageHandler(uint8_t _taskType, uint8_t _algorithmType,
                                    << LOG_KV("error", boost::diagnostic_information(e));
             }
         });
+    m_front->registerComponent(std::to_string(type));
 }
diff --git a/cpp/wedpr-transport/ppc-front/ppc-front/FrontImpl.cpp b/cpp/wedpr-transport/ppc-front/ppc-front/FrontImpl.cpp
@@ -275,4 +275,28 @@ bcos::Error::Ptr FrontImpl::push(uint16_t routeType, MessageOptionalHeader::Ptr
         routeType, routeInfo, std::move(payload), seq, timeout,
         [promise](bcos::Error::Ptr error) { promise->set_value(error); }, nullptr);
     return promise->get_future().get();
+}
+
+void FrontImpl::asyncGetPeers(GetPeersInfoHandler::Ptr getPeersCallback)
+{
+    m_gatewayClient->asyncGetPeers(
+        [getPeersCallback](bcos::Error::Ptr error, std::string peersInfo) {
+            getPeersCallback->onPeersInfo(error, peersInfo);
+        });
+}
+
+void FrontImpl::registerComponent(std::string const& component)
+{
+    // Note: the node will report the latest components
+    auto ret = m_nodeInfo->addComponent(component);
+    FRONT_LOG(INFO) << LOG_DESC("registerComponent") << LOG_KV("component", component)
+                    << LOG_KV("insert", ret);
+}
+
+void FrontImpl::unRegisterComponent(std::string const& component)
+{
+    // Note: the node will report the latest components
+    auto ret = m_nodeInfo->eraseComponent(component);
+    FRONT_LOG(INFO) << LOG_DESC("unRegisterComponent") << LOG_KV("component", component)
+                    << LOG_KV("erase", ret);
 }