framework: code clean

framework/benchmark/benchmark_gaussian_elimination.cpp

@@ -66,7 +66,7 @@ int MSG_SEND_GAP_US = 0;
 int PROC_TIME_US = 0;
 // 0: usleep; 1: cpu computer; 2: gpu computer
 int PROC_TYPE = 0;
-int MSG_NUM = 0;
+uint64_t MSG_NUM = 0;
 
 void LocalProc(const std::string& node_name,
                const std::shared_ptr<apollo::cybertron::proto::Chatter>& msg) {
@@ -143,7 +143,7 @@ class GaussianComponent
   }
 };
 
-int GenGaussianDag() {
+void GenGaussianDag() {
   std::shared_ptr<apollo::cybertron::Node> start_node;
   std::shared_ptr<apollo::cybertron::Writer<apollo::cybertron::proto::Chatter>>
       start_node_writer;
@@ -232,7 +232,7 @@ int GenGaussianDag() {
       ++idx;
     }
   }
-  for (int i = 0; i < MSG_NUM; ++i) {
+  for (uint64_t i = 0; i < MSG_NUM; ++i) {
     if (apollo::cybertron::IsShutdown()) {
       break;
     }

framework/cybertron/scheduler/policy/cfs.cpp

@@ -62,13 +62,13 @@ std::shared_ptr<CRoutine> CFSContext::NextLocalRoutine() {
         cur_croutine_->priority());
     cur_croutine_->set_exec_time(cur_croutine_->vruntime() *
                                  cur_croutine_->priority());
-    local_rb_map_.insert(std::pair<double, std::shared_ptr<CRoutine>>(
+    local_rt_queue_.insert(std::pair<double, std::shared_ptr<CRoutine>>(
         cur_croutine_->vruntime(), cur_croutine_));
     cur_croutine_ = nullptr;
   }
 
   std::shared_ptr<CRoutine> croutine = nullptr;
-  for (auto it = local_rb_map_.begin(); it != local_rb_map_.end();) {
+  for (auto it = local_rt_queue_.begin(); it != local_rt_queue_.end();) {
     auto cr = it->second;
     auto lock = cr->TryLock();
     if (!lock) {
@@ -77,13 +77,9 @@ std::shared_ptr<CRoutine> CFSContext::NextLocalRoutine() {
     }
 
     cr->UpdateState();
-    if (cr->state() == RoutineState::RUNNING) {
-      ++it;
-      continue;
-    }
 
     if (cr->state() == RoutineState::FINISHED) {
-      it = local_rb_map_.erase(it);
+      it = local_rt_queue_.erase(it);
       continue;
     }
 
@@ -92,11 +88,12 @@ std::shared_ptr<CRoutine> CFSContext::NextLocalRoutine() {
       croutine = cr;
       croutine->set_state(RoutineState::RUNNING);
       cur_croutine_ = croutine;
-      local_rb_map_.erase(it);
+      local_rt_queue_.erase(it);
       break;
     }
     ++it;
   }
+
   if (croutine == nullptr) {
     notified_.store(false);
   } else {
@@ -115,7 +112,7 @@ std::shared_ptr<CRoutine> CFSContext::NextAffinityRoutine() {
   auto start_perf_time = apollo::cybertron::Time::Now().ToNanosecond();
   std::shared_ptr<CRoutine> croutine = nullptr;
 
-  for (auto it = affinity_rb_map_.begin(); it != affinity_rb_map_.end();) {
+  for (auto it = affinity_rt_queue_.begin(); it != affinity_rt_queue_.end();) {
     auto cr = it->second;
     auto lock = cr->TryLock();
     if (!lock) {
@@ -124,13 +121,9 @@ std::shared_ptr<CRoutine> CFSContext::NextAffinityRoutine() {
     }
 
     cr->UpdateState();
-    if (cr->state() == RoutineState::FINISHED) {
-      it = affinity_rb_map_.erase(it);
-      continue;
-    }
 
-    if (cr->state() == RoutineState::RUNNING) {
-      ++it;
+    if (cr->state() == RoutineState::FINISHED) {
+      it = affinity_rt_queue_.erase(it);
       continue;
     }
 
@@ -141,6 +134,7 @@ std::shared_ptr<CRoutine> CFSContext::NextAffinityRoutine() {
     }
     ++it;
   }
+
   if (croutine) {
     PerfEventCache::Instance()->AddSchedEvent(SchedPerf::NEXT_AFFINITY_R,
                                               croutine->id(), proc_index_, 0,
@@ -165,7 +159,7 @@ bool CFSContext::Enqueue(const std::shared_ptr<CRoutine>& cr) {
   std::lock_guard<std::mutex> lg(mtx_run_queue_);
   cr->set_vruntime(min_vruntime_ + cr->normalized_vruntime());
   cr->set_exec_time(cr->vruntime() * cr->priority());
-  local_rb_map_.insert(
+  local_rt_queue_.insert(
       std::pair<double, std::shared_ptr<CRoutine>>(cr->vruntime(), cr));
   return true;
 }
@@ -173,7 +167,7 @@ bool CFSContext::Enqueue(const std::shared_ptr<CRoutine>& cr) {
 bool CFSContext::EnqueueAffinityRoutine(const std::shared_ptr<CRoutine>& cr) {
   std::lock_guard<std::mutex> lg(rw_affinity_lock_);
   if (cr->IsAffinity(id())) {
-    affinity_rb_map_.insert(
+    affinity_rt_queue_.insert(
         std::pair<uint32_t, std::shared_ptr<CRoutine>>(cr->priority(), cr));
     return true;
   }

framework/cybertron/scheduler/policy/cfs.h

@@ -18,13 +18,11 @@
 #define CYBERTRON_SCHEDULER_POLICY_CFS_CONTEXT_H_
 
 #include <cstdint>
-#include <future>
-#include <list>
+#include <functional>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <string>
-#include <vector>
 
 #include "cybertron/scheduler/processor_context.h"
 
@@ -53,8 +51,9 @@ class CFSContext : public ProcessorContext {
 
   std::mutex mtx_run_queue_;
   std::mutex rw_affinity_lock_;
-  std::multimap<double, std::shared_ptr<CRoutine>> local_rb_map_;
-  std::multimap<double, std::shared_ptr<CRoutine>> affinity_rb_map_;
+  std::multimap<double, std::shared_ptr<CRoutine>> local_rt_queue_;
+  std::multimap<double, std::shared_ptr<CRoutine>, std::greater<double>>
+      affinity_rt_queue_;
   std::shared_ptr<CRoutine> cur_croutine_ = nullptr;
   double min_vruntime_ = 0;
 };

framework/cybertron/scheduler/policy/classic.cpp

@@ -48,7 +48,7 @@ std::shared_ptr<CRoutine> ClassicContext::NextRoutine() {
   auto start_perf_time = apollo::cybertron::Time::Now().ToNanosecond();
 
   std::shared_ptr<CRoutine> croutine = nullptr;
-  for (auto it = rb_map_.begin(); it != rb_map_.end();) {
+  for (auto it = rt_queue_.begin(); it != rt_queue_.end();) {
     auto cr = it->second;
     auto lock = cr->TryLock();
     if (!lock) {
@@ -56,8 +56,9 @@ std::shared_ptr<CRoutine> ClassicContext::NextRoutine() {
       continue;
     }
 
+    cr->UpdateState();
     if (cr->state() == RoutineState::FINISHED) {
-      it = rb_map_.erase(it);
+      it = rt_queue_.erase(it);
       continue;
     }
 
@@ -90,7 +91,7 @@ bool ClassicContext::Enqueue(const std::shared_ptr<CRoutine>& cr) {
 bool ClassicContext::EnqueueAffinityRoutine(
     const std::shared_ptr<CRoutine>& cr) {
   std::lock_guard<std::mutex> lg(mtx_run_queue_);
-  rb_map_.insert(
+  rt_queue_.insert(
       std::pair<double, std::shared_ptr<CRoutine>>(cr->priority(), cr));
   return true;
 }

framework/cybertron/scheduler/policy/classic.h

@@ -18,13 +18,11 @@
 #define CYBERTRON_SCHEDULER_POLICY_CLASSIC_CONTEXT_H_
 
 #include <cstdint>
-#include <future>
-#include <list>
+#include <functional>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <string>
-#include <vector>
 
 #include "cybertron/scheduler/processor_context.h"
 
@@ -49,7 +47,8 @@ class ClassicContext : public ProcessorContext {
 
  private:
   std::mutex mtx_run_queue_;
-  std::multimap<double, std::shared_ptr<CRoutine>> rb_map_;
+  std::multimap<double, std::shared_ptr<CRoutine>, std::greater<double>>
+      rt_queue_;
 };
 
 }  // namespace scheduler

scripts/apollo_base.sh

@@ -84,7 +84,7 @@ function set_lib_path() {
     PY_LIB_PATH=/apollo/lib
     PY_TOOLS_PATH=/apollo/modules/tools
   else
-    local CYBERTRON_SETUP="/apollo/framework/cybertron/setup.bash"
+    local CYBERTRON_SETUP="/apollo/framework/install/setup.bash"
     if [ -e "${CYBERTRON_SETUP}" ]; then
       source "${CYBERTRON_SETUP}"
     fi

scripts/record_bag.py

@@ -203,7 +203,7 @@ class Recorder(object):
         cmd = '''
             cd "{}"
             source /apollo/scripts/apollo_base.sh
-            source /apollo/framework/cybertron/setup.bash
+            source /apollo/framework/install/setup.bash
             nohup cyber_recorder record -c {} >{} 2>&1 &
         '''.format(task_dir, topics_str, log_file)
         shell_cmd(cmd)