Changing concord_client_request.proto to concord_client_request.cmf

Author: Tarunkumar Banda

Makefile

@@ -11,6 +11,7 @@ CONCORD_BFT_RECONFIGURATION_CMF_PATHS?=${CONCORD_BFT_TARGET_SOURCE_PATH}/build/r
 CONCORD_BFT_BFTENGINE_CMF_PATHS?=${CONCORD_BFT_TARGET_SOURCE_PATH}/build/bftengine/cmf
 CONCORD_BFT_CCRON_CMF_PATHS?=${CONCORD_BFT_TARGET_SOURCE_PATH}/build/ccron/cmf
 CONCORD_BFT_SKVBC_CMF_PATHS?=${CONCORD_BFT_TARGET_SOURCE_PATH}/build/tests/simpleKVBC/cmf
+CONCORD_BFT_CONCORD_CLIENT_CMF_PATHS?=${CONCORD_BFT_TARGET_SOURCE_PATH}/build/client/concordclient/cmf
 CONCORD_BFT_CLIENT_PROTO_PATH?=${CONCORD_BFT_TARGET_SOURCE_PATH}/build/client/proto
 CONCORD_BFT_THIN_REPLICA_PROTO_PATH?=${CONCORD_BFT_TARGET_SOURCE_PATH}/build/thin-replica-server/proto
 CONCORD_BFT_KVBC_PROTO_PATH?=${CONCORD_BFT_TARGET_SOURCE_PATH}/build/kvbc/proto
@@ -197,6 +198,7 @@ tidy-check: gen_cmake ## Run clang-tidy
 		make -C ${CONCORD_BFT_BFTENGINE_CMF_PATHS} &> /dev/null && \
 		make -C ${CONCORD_BFT_CCRON_CMF_PATHS} &> /dev/null && \
 		make -C ${CONCORD_BFT_SKVBC_CMF_PATHS} &> /dev/null && \
+		make -C ${CONCORD_BFT_CONCORD_CLIENT_CMF_PATHS} &> /dev/null && \
 		make -C ${CONCORD_BFT_CLIENT_PROTO_PATH} &> /dev/null && \
 		make -C ${CONCORD_BFT_THIN_REPLICA_PROTO_PATH} &> /dev/null && \
 		make -C ${CONCORD_BFT_KVBC_PROTO_PATH} &> /dev/null && \

bftengine/include/bftengine/SharedTypes.hpp

@@ -29,4 +29,6 @@ enum class OperationResult : uint32_t {
   INTERNAL_ERROR
 };
 
+enum class RequestType : uint32_t { RAW_MESSAGE, ANY_MESSAGE };
+
 }  // namespace bftEngine

client/bftclient/include/bftclient/config.h

@@ -22,6 +22,7 @@
 #include "bftclient/base_types.h"
 #include "bftclient/quorums.h"
 #include "secret_data.h"
+#include "SharedTypes.hpp"
 
 using namespace std::chrono_literals;
 
@@ -81,6 +82,8 @@ struct RequestConfig {
   std::string span_context = "";
   bool key_exchange = false;
   bool reconfiguration = false;
+  bftEngine::RequestType request_type = bftEngine::RequestType::RAW_MESSAGE;
+  std::string client_service_id = "";
 };
 
 // The configuration for a single write request.

client/client_pool/CMakeLists.txt

@@ -11,6 +11,7 @@ target_link_libraries(concord_client_pool PUBLIC
         bftclient
         bftclient_new
         corebft
+	concord_client_request
       	)
 
 install (TARGETS concord_client_pool DESTINATION lib${LIB_SUFFIX})

client/client_pool/include/client/client_pool/concord_client_pool.hpp

@@ -119,6 +119,8 @@ class ConcordClientPool {
                            uint64_t seq_num,
                            std::string correlation_id = {},
                            const std::string& span_context = std::string(),
+                           const bftEngine::RequestType request_type = bftEngine::RequestType::RAW_MESSAGE,
+                           const std::string& subscriptionId = std::string(),
                            const bftEngine::RequestCallBack& callback = {});
 
   // This method is responsible to get write requests with the new client

client/client_pool/include/client/client_pool/external_client.hpp

@@ -21,6 +21,7 @@
 #include "client_pool_config.hpp"
 #include "communication/StatusInfo.h"
 #include "external_client_exception.hpp"
+#include "concord_client_request.cmf.hpp"
 
 namespace concord {
 
@@ -104,6 +105,12 @@ class ConcordClient {
 
   std::string messageSignature(bft::client::Msg&);
 
+  static void createConcordClientRequest(bft::client::Msg& request,
+                                         bftEngine::RequestType typed_request,
+                                         const std::string& subscriptionId);
+
+  static void createConcordClientResponse(bft::client::Msg& response);
+
  private:
   void CreateClient(std::shared_ptr<concordMetrics::Aggregator> aggregator);
 

client/client_pool/src/concord_client_pool.cpp

@@ -44,6 +44,8 @@ SubmitResult ConcordClientPool::SendRequest(std::vector<uint8_t> &&request,
                                             uint64_t seq_num,
                                             std::string correlation_id,
                                             const std::string &span_context,
+                                            const bftEngine::RequestType request_type,
+                                            const std::string &subscriptionId,
                                             const bftEngine::RequestCallBack &callback) {
   if (callback && timeout_ms.count() == 0) {
     callback(bftEngine::SendResult{static_cast<uint32_t>(OperationResult::INVALID_REQUEST)});
@@ -56,6 +58,7 @@ SubmitResult ConcordClientPool::SendRequest(std::vector<uint8_t> &&request,
 
   while (!clients_.empty() && serving_candidates != 0) {
     auto client = clients_.front();
+    external_client::ConcordClient::createConcordClientRequest(request, request_type, subscriptionId);
     client_id = client->getClientId();
     if (is_overloaded_) {
       is_overloaded_ = false;
@@ -224,6 +227,8 @@ SubmitResult ConcordClientPool::SendRequest(const bft::client::WriteConfig &conf
                      config.request.sequence_number,
                      config.request.correlation_id,
                      config.request.span_context,
+                     config.request.request_type,
+                     config.request.client_service_id,
                      callback);
 }
 
@@ -245,6 +250,8 @@ SubmitResult ConcordClientPool::SendRequest(const bft::client::ReadConfig &confi
                      config.request.sequence_number,
                      config.request.correlation_id,
                      config.request.span_context,
+                     config.request.request_type,
+                     config.request.client_service_id,
                      callback);
 }
 
@@ -487,10 +494,13 @@ void SingleRequestProcessingJob::execute() {
   OperationResult operation_result = processing_client_->getRequestExecutionResult();
   reply_size = res.matched_data.size();
   if (callback_) {
-    if (operation_result == OperationResult::SUCCESS)
+    if (operation_result == OperationResult::SUCCESS) {
+      external_client::ConcordClient::createConcordClientResponse(res.matched_data);
+      reply_size = res.matched_data.size();
       callback_(res);
-    else
+    } else {
       callback_(static_cast<uint32_t>(operation_result));
+    }
   }
   external_client::ConcordClient::PendingReplies replies;
   replies.push_back(ClientReply{static_cast<uint32_t>(request_.size()),

client/client_pool/src/external_client.cpp

@@ -184,7 +184,7 @@ std::pair<int32_t, ConcordClient::PendingReplies> ConcordClient::SendPendingRequ
   try {
     LOG_INFO(logger_, "Batch processing started" << KVLOG(client_id_, batch_cid));
     auto received_replies_map = new_client_->sendBatch(request_queue, batch_cid);
-    for (const auto& received_reply_entry : received_replies_map) {
+    for (auto& received_reply_entry : received_replies_map) {
       const auto received_reply_seq_num = received_reply_entry.first;
       const auto& pending_seq_num_to_cid_entry = seq_num_to_cid.find(received_reply_seq_num);
       if (pending_seq_num_to_cid_entry == seq_num_to_cid.end()) {
@@ -195,6 +195,7 @@ std::pair<int32_t, ConcordClient::PendingReplies> ConcordClient::SendPendingRequ
       }
       auto cid = pending_seq_num_to_cid_entry->second;
       cid_response_map_[cid] = std::chrono::steady_clock::now();
+      createConcordClientResponse(received_reply_entry.second.matched_data);
       auto data_size = received_reply_entry.second.matched_data.size();
       for (auto& pending_reply : pending_replies_) {
         if (pending_reply.cid != cid) continue;
@@ -392,6 +393,24 @@ OperationResult ConcordClient::getRequestExecutionResult() { return clientReques
 
 std::string ConcordClient::messageSignature(bft::client::Msg& message) { return new_client_->signMessage(message); }
 
+void ConcordClient::createConcordClientRequest(bft::client::Msg& request,
+                                               bftEngine::RequestType typed_request,
+                                               const std::string& subscriptionId) {
+  concord::client::request::messages::ConcordClientRequest concord_request;
+  concord_request.type = static_cast<decltype(concord_request.type)>(typed_request);
+  concord_request.client_id = static_cast<decltype(concord_request.client_id)>(subscriptionId);
+  concord_request.application_request = std::vector<uint8_t>(request.begin(), request.end());
+  request.clear();
+  concord::client::request::messages::serialize(request, concord_request);
+}
+
+void ConcordClient::createConcordClientResponse(bft::client::Msg& response) {
+  concord::client::request::messages::ConcordClientResponse concord_response;
+  concord::client::request::messages::deserialize(response, concord_response);
+  response.clear();
+  response.assign(concord_response.application_response.begin(), concord_response.application_response.end());
+}
+
 void ConcordClient::stopClientComm() { new_client_->stop(); }
 
 }  // namespace concord::external_client

client/clientservice/src/request_service.cpp

@@ -14,11 +14,9 @@
 
 #include "client/clientservice/request_service.hpp"
 #include "client/concordclient/concord_client.hpp"
-#include "concord_client_request.pb.h"
 #include "client/thin-replica-client/trace_contexts.hpp"
 
 using namespace client::thin_replica_client;
-using namespace vmware::concord::client::concord_client_request::v1;
 
 namespace concord::client::clientservice {
 
@@ -75,12 +73,9 @@ void RequestServiceCallData::sendToConcordClient() {
   bool is_any_request_type = false;
   bft::client::Msg msg;
   if (request_.has_typed_request()) {
-    ConcordClientRequest concord_request;
-    concord_request.set_client_service_id(client_->getSubscriptionId());
-    concord_request.mutable_application_request()->CopyFrom(request_.typed_request());
-    size_t request_size = concord_request.ByteSizeLong();
+    size_t request_size = request_.typed_request().ByteSizeLong();
     std::string request(request_size, '\0');
-    concord_request.SerializeToArray(request.data(), request_size);
+    request_.typed_request().SerializeToArray(request.data(), request_size);
     msg = bft::client::Msg(request.begin(), request.end());
     is_any_request_type = true;
   } else {
@@ -95,6 +90,12 @@ void RequestServiceCallData::sendToConcordClient() {
   req_config.pre_execute = request_.pre_execute();
   req_config.timeout = timeout;
   req_config.correlation_id = request_.correlation_id();
+  if (request_.has_typed_request()) {
+    req_config.request_type = bftEngine::RequestType::ANY_MESSAGE;
+  } else {
+    req_config.request_type = bftEngine::RequestType::RAW_MESSAGE;
+  }
+  req_config.client_service_id = client_->getSubscriptionId();
 
   auto callback = [this, req_config, is_any_request_type](concord::client::concordclient::SendResult&& send_result) {
     grpc::Status status;
@@ -163,13 +164,12 @@ void RequestServiceCallData::sendToConcordClient() {
 
     // Check if the application response is of Any Type then set it to Any response.
     if (is_any_request_type) {
-      ConcordClientResponse concord_response;
-      if (!concord_response.ParseFromArray(data.c_str(), data.size())) {
+      google::protobuf::Any* app_response = this->response_.mutable_typed_response();
+      if (!app_response->ParseFromArray(data.c_str(), data.size())) {
         status = grpc::Status(grpc::StatusCode::INTERNAL, "Internal error in parsing typed response");
         this->populateResult(status);
         return;
       }
-      this->response_.mutable_typed_response()->CopyFrom(concord_response.application_response());
     } else {
       this->response_.set_raw_response(std::move(data));
     }

client/concordclient/CMakeLists.txt

@@ -1,8 +1,12 @@
 add_library(concordclient "src/concord_client.cpp")
 target_include_directories(concordclient PUBLIC include)
 # TODO: Mark libraries as PRIVATE once the interface is selfcontained
+
+add_subdirectory("cmf")
+
 target_link_libraries(concordclient PUBLIC
   thin_replica_client_lib
+  concord_client_request
   concord_client_pool
   concordclient-event-api
   util