framework: Specify/set processor scheduling policy for choreography policy.

Users can specify scheduing policy(SCHED_FIFO/SCHED_RR) and sched priority in
choreography sched conf and scheduler subsystem will set the processor(thread)
scheduing policy according to the conf.

cyber/conf/example_choreagraphy_sched.conf

+
+scheduler_conf {
+  policy: "choreography"
+
+  choreography_conf {
+    choreography_processor_num: 8
+    choreography_affinity: "range"
+    choreography_processor_policy: "SCHED_FIFO"
+    choreography_processor_prio: 10
+    choreography_cpuset: "0-7"
+
+    pool_processor_num: 8
+    pool_affinity: "range"
+    pool_cpuset: "16-23"
+
+    tasks: [
+      {
+        name: "velodyne_128_convert"
+        processor: 0
+        prio: 11
+      },
+      {
+        name: "velodyne128_compensator"
+        processor: 1
+        prio: 12
+      }
+    ]
+  }
+}

cyber/proto/choreography_conf.proto

@@ -11,9 +11,11 @@ message ChoreographyTask {
 message ChoreographyConf {
   optional uint32 choreography_processor_num = 1;
   optional string choreography_affinity = 2;
-  optional string choreography_cpuset = 3;
-  optional uint32 pool_processor_num = 4;
-  optional string pool_affinity = 5;
-  optional string pool_cpuset = 6; 
-  repeated ChoreographyTask tasks = 7;
+  optional string choreography_processor_policy = 3;
+  optional uint32 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;
 }

cyber/scheduler/policy/scheduler_choreography.cc

@@ -49,6 +49,10 @@ SchedulerChoreography::SchedulerChoreography() {
         .choreography_processor_num();
     choreography_affinity_ = cfg.scheduler_conf().choreography_conf()
         .choreography_affinity();
+    choreography_processor_policy_ = cfg.scheduler_conf().choreography_conf()
+        .choreography_processor_policy();
+    choreography_processor_prio_ = cfg.scheduler_conf().choreography_conf()
+        .choreography_processor_prio();
     ParseCpuset(cfg.scheduler_conf().choreography_conf().choreography_cpuset(),
         &choreography_cpuset_);
 
@@ -81,6 +85,8 @@ void SchedulerChoreography::CreateProcessor() {
 
     proc->BindContext(ctx);
     proc->SetAffinity(choreography_cpuset_, choreography_affinity_, i);
+    proc->SetSchedPolicy(choreography_processor_policy_,
+                        choreography_processor_prio_);
     ctx->BindProc(proc);
     pctxs_.emplace_back(ctx);
   }

cyber/scheduler/policy/scheduler_choreography.h

@@ -43,6 +43,8 @@ class SchedulerChoreography : public Scheduler {
   std::unordered_map<std::string, ChoreographyTask> cr_confs_;
 
   std::string choreography_affinity_;
+  std::string choreography_processor_policy_;
+  uint32_t choreography_processor_prio_;
   std::vector<int> choreography_cpuset_;
   std::string pool_affinity_;
   std::vector<int> pool_cpuset_;

cyber/scheduler/processor.cc

@@ -63,6 +63,24 @@ void Processor::SetAffinity(const std::vector<int> &cpus,
   }
 }
 
+void Processor::SetSchedPolicy(std::string spolicy, int sched_priority) {
+  struct sched_param sp;
+  int policy;
+
+  if (!spolicy.compare("SCHED_FIFO")) {
+    policy = SCHED_FIFO;
+  } else if (!spolicy.compare("SCHED_RR")) {
+    policy = SCHED_RR;
+  } else {
+    return;
+  }
+
+  memset(reinterpret_cast<void *>(&sp), 0, sizeof(sp));
+  sp.sched_priority = sched_priority;
+
+  pthread_setschedparam(thread_.native_handle(), policy, &sp);
+}
+
 void Processor::Run() {
   CRoutine::SetMainContext(routine_context_);
 

cyber/scheduler/processor.h

@@ -45,6 +45,7 @@ class Processor {
 
   void Run();
   void SetAffinity(const std::vector<int>&, const std::string&, int);
+  void SetSchedPolicy(std::string spolicy, int sched_priority);
   void Stop() { running_.exchange(false); }
   void BindContext(const std::shared_ptr<ProcessorContext>& context) {
     context_ = context;