fix sendrequest core

3rd/CMakeLists.txt

@@ -23,6 +23,7 @@ ExternalProject_Add(
         -DBUILD_SHARED_LIBS=ON
         -DCMAKE_BUILD_TYPE=Release
         -DCMAKE_INSTALL_PREFIX=${CMAKE_INSTALL_PREFIX}
+        -DCMAKE_TOOLCHAIN_FILE=${CMAKE_TOOLCHAIN_FILE}
 )
 
 ExternalProject_Add(
@@ -42,6 +43,7 @@ ExternalProject_Add(
         -DWITH_GTEST=OFF
         -DCMAKE_INSTALL_PREFIX=${CMAKE_INSTALL_PREFIX}
         -DCMAKE_PREFIX_PATH=${CMAKE_INSTALL_PREFIX}
+        -DCMAKE_TOOLCHAIN_FILE=${CMAKE_TOOLCHAIN_FILE}
 )
 ExternalProject_Add_StepDependencies(glog install gflags)
 
@@ -64,4 +66,5 @@ ExternalProject_Add(
         -DJSONCPP_WITH_POST_BUILD_UNITTEST=OFF
         -DJSONCPP_WITH_PKGCONFIG_SUPPORT=OFF
         -DCMAKE_INSTALL_PREFIX=${CMAKE_INSTALL_PREFIX}
+        -DCMAKE_TOOLCHAIN_FILE=${CMAKE_TOOLCHAIN_FILE}
 )

myframe/app.cpp

@@ -347,8 +347,7 @@ const std::shared_ptr<const Msg> App::SendRequest(
   poller_->Add(conn);
   auto resp = conn->SendRequest(std::move(msg));
   poller_->Del(conn);
-  // 不需要调用ev_conn_mgr_->Release()
-  // 系统会主动释放
+  ev_conn_mgr_->Release(std::move(conn));
   return resp;
 }
 

myframe/event_conn_manager.cpp

@@ -81,8 +81,6 @@ void EventConnManager::Notify(
     LOG(WARNING) << "event " << ev->GetName() << " need't resp msg";
     return;
   }
-  // need release immediately
-  Release(ev);
   // push msg to event_conn
   ev->GetMailbox()->Recv(std::move(msg));
   // send cmd to event_conn

test/CMakeLists.txt

@@ -14,6 +14,11 @@ target_link_libraries(app_send_test
     myframe
 )
 
+add_executable(app_send_req_test app_send_req_test.cpp)
+target_link_libraries(app_send_req_test
+    myframe
+)
+
 add_executable(performance_trans1_cost_test performance_trans1_cost_test.cpp)
 target_link_libraries(performance_trans1_cost_test
     myframe
@@ -43,6 +48,7 @@ target_link_libraries(performance_trans100_fullspeed_test
 INSTALL(TARGETS
     common_test
     app_send_test
+    app_send_req_test
     performance_trans1_cost_test
     performance_trans10_cost_test
     performance_trans1_fullspeed_test

test/app_send_req_test.cpp

@@ -0,0 +1,98 @@
+/****************************************************************************
+Copyright (c) 2019, 李柯鹏
+All rights reserved.
+
+Author: 李柯鹏 <[email protected]>
+****************************************************************************/
+#include <algorithm>
+#include <numeric>
+#include <vector>
+#include <thread>
+#include <chrono>
+#include <glog/logging.h>
+#include "myframe/common.h"
+#include "myframe/log.h"
+#include "myframe/msg.h"
+#include "myframe/actor.h"
+#include "myframe/mod_manager.h"
+#include "myframe/app.h"
+
+#include "performance_test_config.h"
+
+class EchoActorTest : public myframe::Actor {
+ public:
+  int Init(const char* param) override {
+    (void)param;
+    LOG(INFO) << "init EchoActorTest";
+    return 0;
+  }
+
+  void Proc(const std::shared_ptr<const myframe::Msg>& msg) override {
+    if (msg->GetData() == "hello") {
+      auto re = std::make_shared<myframe::Msg>(
+        "resp:" + std::to_string(seq_++));
+      auto mailbox = GetMailbox();
+      mailbox->Send(msg->GetSrc(), std::move(re));
+    }
+  }
+
+ private:
+  int seq_{0};
+};
+
+int main() {
+  auto lib_dir =
+      myframe::Common::GetAbsolutePath(MYFRAME_LIB_DIR).string();
+  auto log_dir =
+      myframe::Common::GetAbsolutePath(MYFRAME_LOG_DIR).string();
+
+  myframe::InitLog(log_dir, "app_send_req_test");
+
+  auto app = std::make_shared<myframe::App>();
+  if (false == app->Init(lib_dir, 4)) {
+    LOG(ERROR) << "Init failed";
+    return -1;
+  }
+
+  // mod manager
+  auto& mod = app->GetModManager();
+
+  // 注册echo Actor
+  mod->RegActor("EchoActorTest", [](const std::string&) {
+      return std::make_shared<EchoActorTest>();
+  });
+  auto actor = mod->CreateActorInst("class", "EchoActorTest");
+  app->AddActor("1", "", actor);
+
+  // 压力测试SendRequest函数
+  std::mutex mtx;
+  int th_cnt = 5;
+  int exit_th_cnt = 0;
+  int send_cnt = 10000;
+  std::vector<std::thread> th_vec;
+  for (int i = 0; i < th_cnt; ++i) {
+    th_vec.push_back(std::thread([&, i](){
+      int cnt = send_cnt;
+      while (cnt--) {
+        auto msg = std::make_shared<myframe::Msg>("hello");
+        msg->SetDst("actor.EchoActorTest.1");
+        auto resp = app->SendRequest(std::move(msg));
+        LOG(INFO) << "thread " << i << " resp: " << resp->GetData();
+      }
+      std::lock_guard<std::mutex> g(mtx);
+      LOG(INFO) << "user thread " << i << " exit";
+      ++exit_th_cnt;
+      if (exit_th_cnt == th_cnt) {
+        app->Quit();
+      }
+    }));
+  }
+
+  app->Exec();
+  for (int i = 0; i < th_cnt; ++i) {
+    if (th_vec[i].joinable()) {
+      th_vec[i].join();
+    }
+  }
+  return 0;
+}

test/app_send_test.cpp

@@ -9,7 +9,6 @@ Author: 李柯鹏 <[email protected]>
 #include <vector>
 #include <thread>
 #include <chrono>
-
 #include "myframe/common.h"
 #include "myframe/log.h"
 #include "myframe/msg.h"
@@ -29,16 +28,7 @@ class EchoActorTest : public myframe::Actor {
 
   void Proc(const std::shared_ptr<const myframe::Msg>& msg) override {
     LOG(INFO) << "recv " << msg->GetSrc() << ":" << msg->GetData();
-    if (msg->GetData() == "hello") {
-      auto re = std::make_shared<myframe::Msg>(
-        "resp:" + std::to_string(seq_++));
-      auto mailbox = GetMailbox();
-      mailbox->Send(msg->GetSrc(), re);
-    }
   }
-
- private:
-  int seq_{0};
 };
 
 int main() {
@@ -65,23 +55,36 @@ int main() {
   auto actor = mod->CreateActorInst("class", "EchoActorTest");
   app->AddActor("1", "", actor);
 
-  // 创建一个线程请求服务
-  std::thread th([&]() {
-    int cnt = 100;
-    while (cnt--) {
-      auto msg = std::make_shared<myframe::Msg>("hello");
-      msg->SetDst("actor.EchoActorTest.1");
-      auto resp = app->SendRequest(msg);
-      LOG(INFO) << "get resp: " << resp->GetData();
-      auto msg2 = std::make_shared<myframe::Msg>("world");
-      msg2->SetDst("actor.EchoActorTest.1");
-      app->Send(msg2);
-      std::this_thread::sleep_for(std::chrono::milliseconds(10));
-    }
-    app->Quit();
-  });
+  // 测试Send函数
+  std::mutex mtx;
+  int th_cnt = 5;
+  int exit_th_cnt = 0;
+  int send_cnt = 10000;
+  std::vector<std::thread> th_vec;
+  for (int i = 0; i < th_cnt; ++i) {
+    th_vec.push_back(std::thread([&, i](){
+      int cnt = send_cnt;
+      while (cnt--) {
+        auto msg = std::make_shared<myframe::Msg>(
+          "world " + std::to_string(i));
+        msg->SetDst("actor.EchoActorTest.1");
+        app->Send(std::move(msg));
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+      }
+      std::lock_guard<std::mutex> g(mtx);
+      LOG(INFO) << "user thread " << i << " exit";
+      ++exit_th_cnt;
+      if (exit_th_cnt == th_cnt) {
+        app->Quit();
+      }
+    }));
+  }
 
   app->Exec();
-  th.join();
+  for (int i = 0; i < th_cnt; ++i) {
+    if (th_vec[i].joinable()) {
+      th_vec[i].join();
+    }
+  }
   return 0;
 }