framekwork: Specify/set scheduling policy/nice for choreography policy. (#1932)

cyber/conf/example_choreagraphy_sched.conf

@@ -12,6 +12,8 @@ scheduler_conf {
     pool_processor_num: 8
     pool_affinity: "range"
     pool_cpuset: "16-23"
+    pool_processor_policy: "SCHED_OTHER"
+    pool_processor_prio: -10
 
     tasks: [
       {

cyber/proto/choreography_conf.proto

@@ -19,11 +19,13 @@ message ChoreographyConf {
   optional uint32 choreography_processor_num = 1;
   optional string choreography_affinity = 2;
   optional string choreography_processor_policy = 3;
-  optional uint32 choreography_processor_prio = 4;
+  optional int32 choreography_processor_prio = 4;
   optional string choreography_cpuset = 5;
   optional uint32 pool_processor_num = 6;
   optional string pool_affinity = 7;
-  optional string pool_cpuset = 8; 
-  repeated ChoreographyTask tasks = 9;
-  repeated InnerThread threads = 10;
+  optional string pool_processor_policy = 8;
+  optional int32 pool_processor_prio = 9;
+  optional string pool_cpuset = 10; 
+  repeated ChoreographyTask tasks = 11;
+  repeated InnerThread threads = 12;
 }

cyber/scheduler/policy/scheduler_choreography.cc

@@ -61,6 +61,10 @@ SchedulerChoreography::SchedulerChoreography() {
         .pool_processor_num();
     pool_affinity_ = cfg.scheduler_conf().choreography_conf()
         .pool_affinity();
+    pool_processor_policy_ = cfg.scheduler_conf().choreography_conf()
+        .pool_processor_policy();
+    pool_processor_prio_ = cfg.scheduler_conf().choreography_conf()
+        .pool_processor_prio();
     ParseCpuset(cfg.scheduler_conf().choreography_conf().pool_cpuset(),
         &pool_cpuset_);
 
@@ -103,6 +107,8 @@ void SchedulerChoreography::CreateProcessor() {
 
     proc->BindContext(ctx);
     proc->SetAffinity(pool_cpuset_, pool_affinity_, i);
+    proc->SetSchedPolicy(pool_processor_policy_,
+                        pool_processor_prio_);
     ctx->BindProc(proc);
     pctxs_.emplace_back(ctx);
   }

cyber/scheduler/policy/scheduler_choreography.h

@@ -48,10 +48,12 @@ class SchedulerChoreography : public Scheduler {
 
   std::string choreography_affinity_;
   std::string choreography_processor_policy_;
-  uint32_t choreography_processor_prio_;
+  int32_t choreography_processor_prio_;
   std::vector<int> choreography_cpuset_;
   std::string pool_affinity_;
   std::vector<int> pool_cpuset_;
+  std::string pool_processor_policy_;
+  int32_t pool_processor_prio_;
 };
 
 }  // namespace scheduler

cyber/scheduler/processor.cc

@@ -16,6 +16,8 @@
 
 #include "cyber/scheduler/processor.h"
 
+#include <sys/syscall.h>
+#include <sys/resource.h>
 #include <sched.h>
 #include <chrono>
 
@@ -67,21 +69,26 @@ void Processor::SetSchedPolicy(std::string spolicy, int sched_priority) {
   struct sched_param sp;
   int policy;
 
+  memset(reinterpret_cast<void *>(&sp), 0, sizeof(sp));
+  sp.sched_priority = sched_priority;
+
   if (!spolicy.compare("SCHED_FIFO")) {
     policy = SCHED_FIFO;
+    pthread_setschedparam(thread_.native_handle(), policy, &sp);
   } else if (!spolicy.compare("SCHED_RR")) {
     policy = SCHED_RR;
-  } else {
-    return;
+    pthread_setschedparam(thread_.native_handle(), policy, &sp);
+  } else if (!spolicy.compare("SCHED_OTHER")) {
+    // Set normal thread nice value.
+    while (tid_.load() == -1) {
+      cpu_relax();
+    }
+    setpriority(PRIO_PROCESS, tid_.load(), sched_priority);
   }
-
-  memset(reinterpret_cast<void *>(&sp), 0, sizeof(sp));
-  sp.sched_priority = sched_priority;
-
-  pthread_setschedparam(thread_.native_handle(), policy, &sp);
 }
 
 void Processor::Run() {
+  tid_.store(static_cast<int>(syscall(SYS_gettid)));
   CRoutine::SetMainContext(routine_context_);
 
   while (likely(running_)) {

cyber/scheduler/processor.h

@@ -58,6 +58,7 @@ class Processor {
   std::atomic<bool> running_{false};
   std::mutex mtx_ctx_;
   std::condition_variable cv_ctx_;
+  std::atomic<pid_t> tid_{-1};
 };
 
 }  // namespace scheduler