v0.9.5

Author: lkpworkspace

.gitignore

@@ -5,7 +5,9 @@
 
 bin/
 
-build/
+build*
+
+output*
 
 *.user
 
@@ -17,5 +19,3 @@ myframe/export.h
 myframe/config.h
 3rd/pkg/
 3rd/src/
-3rd/build/
-output/

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}
 )

CMakeLists.txt

@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 3.10)
 if (POLICY CMP0091)
   cmake_policy(SET CMP0091 NEW)
 endif()
-project(myframe VERSION 0.9.4)
+project(myframe VERSION 0.9.5)
 
 ### option
 option(MYFRAME_USE_CV "Using conditional variables for thread communication" ON)

cpplint.bash

@@ -19,6 +19,9 @@ function main() {
     -name "*.cxx" -or \
     -name "*.cuh" \
     ')' -and -not -path "./build/*" \
+    -and -not -path "./output/*" \
+    -and -not -path "./3rd/*" \
+    -and -not -path "./myframe/export.h" \
     | xargs python3 ./cpplint.py
 }
 

examples/CMakeLists.txt

@@ -34,26 +34,29 @@ add_library(example_trans_obj SHARED example_trans_obj.cpp)
 target_link_libraries(example_trans_obj ${PROJECT_NAME})
 
 ### install
-FILE(GLOB conf_files "*.json")
-INSTALL(FILES 
+file(GLOB conf_files "*.json")
+# 在unix like系统下禁用MYFRAME_USE_CV才能用,有需要再打开
+list(REMOVE_ITEM conf_files "${CMAKE_CURRENT_SOURCE_DIR}/example_worker_interactive_with_3rd_frame.json")
+install(FILES
   ${conf_files}
   PERMISSIONS
   OWNER_READ OWNER_WRITE
   GROUP_READ
   WORLD_READ
   DESTINATION ${MYFRAME_SERVICE_DIR}
 )
-INSTALL(TARGETS 
+install(TARGETS
   example_actor_helloworld
   example_actor_timer
   example_actor_serial
   example_actor_concurrent
   example_actor_subscribe
   example_node
-  example_worker_actor_interactive 
+  example_worker_actor_interactive
   example_worker_publish
-  example_worker_talk 
-  example_worker_interactive_with_3rd_frame
+  example_worker_talk
+  # 在unix like系统下禁用MYFRAME_USE_CV才能用,有需要再打开
+  # example_worker_interactive_with_3rd_frame
   example_worker_quit
   example_config
   example_trans_obj

examples/example_config.cpp

@@ -17,6 +17,10 @@ class ExampleActorConfig : public myframe::Actor {
   int Init(const char* param) override {
     (void)param;
     auto conf = GetConfig();
+    LOG(INFO) << GetActorName() << " pending queue size "
+      << GetMailbox()->GetPendingQueueSize();
+    LOG(INFO) << GetActorName() << " run queue size "
+      << GetMailbox()->GetRunQueueSize();
     LOG(INFO) << GetActorName() << " conf " << conf->toStyledString();
     return 0;
   }

examples/example_config.json

@@ -6,6 +6,8 @@
             {
                 "instance_name":"#1",
                 "instance_config":{
+                    "pending_queue_size":-1,
+                    "run_queue_size":-1,
                     "key1":"hello",
                     "key2":"world"
                 }

launcher/conf/sys.json

@@ -1,4 +1,6 @@
 {
+    "default_pending_queue_size":-1,
+    "default_run_queue_size":-1,
     "thread_poll_size":4,
     "conn_event_size":2,
     "warning_msg_size":10

launcher/launcher.cpp

@@ -77,7 +77,9 @@ int main(int argc, char** argv) {
     lib_dir.string(),
     module_args.GetThreadPoolSize(),
     module_args.GetConnEventSize(),
-    module_args.GetWarningMsgSize())) {
+    module_args.GetWarningMsgSize(),
+    module_args.GetDefaultPendingQueueSize(),
+    module_args.GetDefaultRunQueueSize())) {
     LOG(ERROR) << "Init failed";
     return -1;
   }

launcher/module_argument.cpp

@@ -131,6 +131,14 @@ bool ModuleArgument::ParseSysConf(const std::string& sys_conf) {
       && root["service_dir"].isString()) {
     conf_dir_ = root["service_dir"].asString();
   }
+  if (root.isMember("default_pending_queue_size")
+      && root["default_pending_queue_size"].isInt()) {
+    default_pending_queue_size_ = root["default_pending_queue_size"].asInt();
+  }
+  if (root.isMember("default_run_queue_size")
+      && root["default_run_queue_size"].isInt()) {
+    default_run_queue_size_ = root["default_run_queue_size"].asInt();
+  }
   return true;
 }
 

launcher/module_argument.h

@@ -28,13 +28,21 @@ class ModuleArgument final {
   inline int GetThreadPoolSize() const { return thread_poll_size_; }
   inline int GetConnEventSize() const { return conn_event_size_; }
   inline int GetWarningMsgSize() const { return warning_msg_size_; }
+  inline int GetDefaultPendingQueueSize() const {
+    return default_pending_queue_size_;
+  }
+  inline int GetDefaultRunQueueSize() const {
+    return default_run_queue_size_;
+  }
 
  private:
   bool ParseSysConf(const std::string&);
 
   int thread_poll_size_{4};
   int conn_event_size_{2};
   int warning_msg_size_{10};
+  int default_pending_queue_size_{-1};
+  int default_run_queue_size_{-1};
   std::string log_dir_;
   std::string lib_dir_;
   std::string conf_dir_;

myframe/actor_context.cpp

@@ -25,6 +25,17 @@ ActorContext::ActorContext(
     , app_(app) {
   actor_->SetContext(this);
   mailbox_.SetAddr(actor_->GetActorName());
+  int pending_queue_size = app->GetDefaultPendingQueueSize();
+  int run_queue_size = app->GetDefaultRunQueueSize();
+  auto cfg = actor_->GetConfig();
+  if (cfg->isMember("pending_queue_size")) {
+    pending_queue_size = cfg->get("pending_queue_size", -1).asInt();
+  }
+  if (cfg->isMember("run_queue_size")) {
+    run_queue_size = cfg->get("run_queue_size", -1).asInt();
+  }
+  mailbox_.SetPendingQueueSize(pending_queue_size);
+  mailbox_.SetRunQueueSize(run_queue_size);
   LOG(INFO) << mailbox_.Addr() << " context create";
 }
 

myframe/actor_context_manager.h

@@ -39,11 +39,12 @@ class ActorContextManager final {
   std::vector<std::string> GetAllActorAddr();
   bool HasActor(const std::string& name);
 
+  /* 将有消息的actor放入链表 */
+  void PushContext(std::shared_ptr<ActorContext> ctx);
+
  private:
   /* 获得actor名对应的actor */
   std::shared_ptr<ActorContext> GetContext(const std::string& actor_name);
-  /* 将有消息的actor放入链表 */
-  void PushContext(std::shared_ptr<ActorContext> ctx);
   void PrintWaitQueue();
 
   /// 当前注册actor数量

myframe/app.cpp

@@ -34,7 +34,7 @@ std::shared_ptr<WorkerTimer> App::GetTimerWorker() {
     LOG(ERROR) << "worker context manager is nullptr";
     return nullptr;
   }
-  std::string worker_timer_name = "worker.timer.#1";
+  std::string worker_timer_name = "worker.T.1";
   auto w = ev_mgr_->Get<WorkerContext>(worker_timer_name);
   if (w == nullptr) {
     LOG(ERROR)
@@ -63,14 +63,18 @@ bool App::Init(
   const std::string& lib_dir,
   int thread_pool_size,
   int event_conn_size,
-  int warning_msg_size) {
+  int warning_msg_size,
+  int default_pending_queue_size,
+  int default_run_queue_size) {
   if (state_.load() != kUninitialized) {
     return true;
   }
 
   bool ret = true;
   lib_dir_ = lib_dir;
   warning_msg_size_.store(warning_msg_size);
+  default_pending_queue_size_ = default_pending_queue_size;
+  default_run_queue_size_ = default_run_queue_size;
   ret &= poller_->Init();
   ret &= worker_ctx_mgr_->Init(warning_msg_size);
   ret &= ev_conn_mgr_->Init(event_conn_size);
@@ -437,7 +441,7 @@ bool App::StartCommonWorker(int worker_count) {
   for (int i = 0; i < worker_count; ++i) {
     auto worker = std::make_shared<WorkerCommon>();
     worker->SetModName("class");
-    worker->SetTypeName("WorkerCommon");
+    worker->SetTypeName("C");
     if (!AddWorker(std::to_string(i), worker)) {
       LOG(ERROR) << "start common worker " << i << " failed";
       continue;
@@ -451,8 +455,8 @@ bool App::StartCommonWorker(int worker_count) {
 bool App::StartTimerWorker() {
   auto worker = std::make_shared<WorkerTimer>();
   worker->SetModName("class");
-  worker->SetTypeName("timer");
-  if (!AddWorker("#1", worker)) {
+  worker->SetTypeName("T");
+  if (!AddWorker("1", worker)) {
     LOG(ERROR) << "start timer worker failed";
     return false;
   }
@@ -557,6 +561,10 @@ void App::CheckStopWorkers() {
       worker_ctx_mgr_->PopFrontIdleWorker();
       auto common_idle_worker = worker_ctx->GetWorker<WorkerCommon>();
       common_idle_worker->SetActorContext(actor_ctx);
+      // 接收队列不空，重新加入等待执行队列
+      if (!actor_mailbox->RecvEmpty()) {
+        actor_ctx_mgr_->PushContext(std::move(actor_ctx));
+      }
       worker_ctx->GetCmdChannel()->SendToOwner(CmdChannel::Cmd::kRun);
     } else {
       LOG(ERROR) << actor_ctx->GetActor()->GetActorName() << " has no msg";
@@ -816,4 +824,12 @@ std::string App::GetLibName(const std::string& name) {
 #endif
 }
 
+int App::GetDefaultPendingQueueSize() const {
+  return default_pending_queue_size_;
+}
+
+int App::GetDefaultRunQueueSize() const {
+  return default_run_queue_size_;
+}
+
 }  // namespace myframe

myframe/app.h

@@ -59,7 +59,9 @@ class MYFRAME_EXPORT App final : public std::enable_shared_from_this<App> {
     const std::string& lib_dir,
     int thread_pool_size = 4,
     int event_conn_size = 2,
-    int warning_msg_size = 10);
+    int warning_msg_size = 10,
+    int default_pending_queue_size = -1,
+    int default_run_queue_size = -1);
 
   int LoadServiceFromDir(const std::string& path);
 
@@ -91,6 +93,9 @@ class MYFRAME_EXPORT App final : public std::enable_shared_from_this<App> {
 
   void Quit();
 
+  int GetDefaultPendingQueueSize() const;
+  int GetDefaultRunQueueSize() const;
+
  private:
   bool CreateActorContext(
     const std::string& mod_name,
@@ -139,6 +144,8 @@ class MYFRAME_EXPORT App final : public std::enable_shared_from_this<App> {
   /// node地址
   std::string node_addr_;
   ///
+  int default_pending_queue_size_{-1};
+  int default_run_queue_size_{-1};
   std::atomic<std::size_t> warning_msg_size_{10};
   std::atomic<State> state_{kUninitialized};
   std::recursive_mutex local_mtx_;