BugFix: Fix the problem that the CLIENT_POST_SCHED_RECV_MSG filter point is not executed in case of decoding failure (#151)

Author: chhy2009

trpc/future/future.h

@@ -452,7 +452,13 @@ class FutureImplBase {
     int64_t timeout_us = timeout * 1000;
     auto start = std::chrono::system_clock::now();
 
-    while (!state_base_->has_result) {
+    while (true) {
+      {
+        std::lock_guard<std::mutex> lock(state_base_->mtx);
+        if (state_base_->has_result) {
+          break;
+        }
+      }
       std::this_thread::sleep_for(std::chrono::microseconds(100));
       auto now = std::chrono::system_clock::now();
       auto diff = std::chrono::duration_cast<std::chrono::microseconds>(now - start);

trpc/transport/client/future/common/utils.cc

@@ -88,11 +88,17 @@ void DispatchResponse(CTransportReqMsg* req_msg, CTransportRspMsg* rsp_msg,
 }
 
 void DispatchException(CTransportReqMsg* req_msg, int ret, std::string&& err_msg,
-                       const TransInfo::RspDispatchFunction& rsp_dispatch_function) {
+                       const TransInfo::RspDispatchFunction& rsp_dispatch_function,
+                       const TransInfo::RunClientFiltersFunction& run_client_filters_function) {
   object_pool::LwUniquePtr<MsgTask> task = object_pool::MakeLwUnique<MsgTask>();
   task->task_type = runtime::kResponseMsg;
   task->param = req_msg;
-  task->handler = [req_msg, ret, msg = std::move(err_msg)]() mutable {
+  task->handler = [req_msg, ret, msg = std::move(err_msg), run_client_filters_function]() mutable {
+    // For rpcz
+    if (run_client_filters_function != nullptr) {
+      run_client_filters_function(FilterPoint::CLIENT_POST_SCHED_RECV_MSG, req_msg);
+    }
+
     auto* backup_promise = req_msg->extend_info->backup_promise;
     Exception ex(CommonException(msg.c_str(), ret));
     req_msg->extend_info->promise.SetException(ex);

trpc/transport/client/future/common/utils.h

@@ -43,8 +43,10 @@ void DispatchResponse(CTransportReqMsg* req_msg, CTransportRspMsg* rsp_msg,
 /// @param ret Error code.
 /// @param err_msg Error message.
 /// @param rsp_dispatch_function Function to decide which handle thread to dispatch to.
+/// @param run_client_filters_function Function to run client filters.
 void DispatchException(CTransportReqMsg* req_msg, int ret, std::string&& err_msg,
-                       const TransInfo::RspDispatchFunction& rsp_dispatch_function);
+                       const TransInfo::RspDispatchFunction& rsp_dispatch_function,
+                       const TransInfo::RunClientFiltersFunction& run_client_filters_function = nullptr);
 
 /// @brief To trigger backup request withdraw.
 /// @param promise Related Promise of backup request.

trpc/transport/client/future/conn_pool/future_conn_pool_connection_handler.cc

@@ -58,7 +58,8 @@ bool FutureTcpConnPoolConnectionHandler::HandleMessage(const ConnectionPtr& conn
     std::string error = "decode failed peer addr = ";
     error += options_.group_options->peer_addr.ToString();
     future::DispatchException(msg, TrpcRetCode::TRPC_CLIENT_DECODE_ERR, std::move(error),
-                              options_.group_options->trans_info->rsp_dispatch_function);
+                              options_.group_options->trans_info->rsp_dispatch_function,
+                              options_.group_options->trans_info->run_client_filters_function);
     // Decode error, let caller close connection.
     return false;
   }
@@ -158,7 +159,8 @@ bool FutureUdpIoPoolConnectionHandler::HandleMessage(const ConnectionPtr& conn,
     std::string error = "decode failed peer addr = ";
     error += options_.group_options->peer_addr.ToString();
     future::DispatchException(msg, TrpcRetCode::TRPC_CLIENT_DECODE_ERR, std::move(error),
-                              options_.group_options->trans_info->rsp_dispatch_function);
+                              options_.group_options->trans_info->rsp_dispatch_function,
+                              options_.group_options->trans_info->run_client_filters_function);
   }
 
   // When there is no pending request to send, just recycle connection.

trpc/transport/client/future/pipeline/future_tcp_pipeline_connector.cc

@@ -74,7 +74,8 @@ bool FuturePipelineConnectionHandler::HandleMessage(const ConnectionPtr& conn,
     std::string error = "decode failed peer addr= ";
     error += options_.group_options->peer_addr.ToString();
     future::DispatchException(msg, TrpcRetCode::TRPC_CLIENT_DECODE_ERR, std::move(error),
-                              options_.group_options->trans_info->rsp_dispatch_function);
+                              options_.group_options->trans_info->rsp_dispatch_function,
+                              options_.group_options->trans_info->run_client_filters_function);
     return false;
   }