mod bpf_skel; pub use bpf_skel::*; pub mod bpf_intf; mod mock; mod task_state; mod e_core_selector; #[rustfmt::skip] mod bpf; use bpf::*; use clap::{Arg, ArgAction, Command}; use e_core_selector::{ECoreSelector, RoundRobinSelector}; use scx_utils::UserExitInfo; use libbpf_rs::OpenObject; use task_state::TaskState; use mock::{MockModule, KernelModule, KernelDriver}; use std::collections::{HashMap, VecDeque}; use std::mem::MaybeUninit; use std::process; use std::ops::Range; use anyhow::Result; const SLICE_US: u64 = 5000; struct Scheduler<'a> { bpf: BpfScheduler<'a>, module: Box, task_queue: VecDeque, no_budget_task_queue: VecDeque, managed_tasks: HashMap, maximum_budget: u64, power_cap: u64, own_pid: u32, p_cores: Range, e_cores: Range, e_core_selector: Box } impl<'a> Scheduler<'a> { fn init(open_object: &'a mut MaybeUninit, use_mocking: bool, power_cap: u64) -> Result { let bpf = BpfScheduler::init( open_object, 0, // exit_dump_len (buffer size of exit info, 0 = default) false, // partial (false = include all tasks) false, // debug (false = debug mode off) )?; dbg!("registering rust user space scheduler"); let module: Box = if use_mocking { Box::new(MockModule) } else { Box::new(KernelDriver::default()) }; let selector = Box::new(RoundRobinSelector::new(8..16)); //TODO: get real value Ok(Self { bpf, module, task_queue: VecDeque::new(), no_budget_task_queue: VecDeque::new(), managed_tasks: HashMap::new(), maximum_budget: u64::MAX, power_cap, own_pid: process::id(), p_cores: 0..8, //TODO: real value e_cores: 8..16, //TODO: real value e_core_selector: selector, }) } fn consume_all_tasks(&mut self) { // Consume all tasks that are ready to run. // // Each task contains the following details: // // pub struct QueuedTask { // pub pid: i32, // pid that uniquely identifies a task // pub cpu: i32, // CPU where the task is running // pub sum_exec_runtime: u64, // Total cpu time // pub weight: u64, // Task static priority // pub nvcsw: u64, // Total amount of voluntary context switches // pub slice: u64, // Remaining time slice budget // pub vtime: u64, // Current task vruntime / deadline (set by the scheduler) // } // // Although the FIFO scheduler doesn't use these fields, they can provide valuable data for // implementing more sophisticated scheduling policies. while let Ok(Some(task)) = self.bpf.dequeue_task() { // The scheduler itself has to be scheduled regardless of its energy usage if task.pid as u32 == self.own_pid { self.task_queue.push_back(task); continue; } if let Some(task_state) = self.managed_tasks.get_mut(&(task.pid as u32)) { let energy = self.module.read_consumption(task.pid as u64); let used_energy = energy - task_state.previous_energy_usage; if task_state.budget < used_energy { task_state.budget = 0; self.no_budget_task_queue.push_back(task); } else { task_state.budget -= used_energy; self.task_queue.push_back(task); } } else { self.module.start_trace(task.pid as u64); self.managed_tasks.insert(task.pid as u32, TaskState { previous_energy_usage: 0, budget: self.maximum_budget, children: vec![], }); self.task_queue.push_back(task); } } } fn dispatch_next_task(&mut self) { if let Some(task) = self.task_queue.pop_front() { // Create a new task to be dispatched, derived from the received enqueued task. // // pub struct DispatchedTask { // pub pid: i32, // pid that uniquely identifies a task // pub cpu: i32, // target CPU selected by the scheduler // pub flags: u64, // special dispatch flags // pub slice_ns: u64, // time slice assigned to the task (0 = default) // } // // The dispatched task's information are pre-populated from the QueuedTask and they can // be modified before dispatching it via self.bpf.dispatch_task(). let mut dispatched_task = DispatchedTask::new(&task); // Decide where the task needs to run (target CPU). // // A call to select_cpu() will return the most suitable idle CPU for the task, // considering its previously used CPU. let cpu = self.bpf.select_cpu(task.pid, task.cpu, 0); if cpu >= 0 { dispatched_task.cpu = cpu; } else { dispatched_task.flags |= RL_CPU_ANY as u64; } // Decide for how long the task needs to run (time slice); if not specified // SCX_SLICE_DFL will be used by default. dispatched_task.slice_ns = SLICE_US; // Dispatch the task on the target CPU. self.bpf.dispatch_task(&dispatched_task).unwrap(); // Notify the BPF component of the number of pending tasks and immediately give a // chance to run to the dispatched task. //TODO: should the pending tasks include the ones that don't have any budget left? self.bpf.notify_complete(self.task_queue.len() as u64 + self.no_budget_task_queue.len() as u64); } else if let Some(task) = self.no_budget_task_queue.pop_front() { //TODO: do something with tasks that have no budget left let mut dispatched_task = DispatchedTask::new(&task); let cpu = self.e_core_selector.next_core(); if cpu >= 0 { dispatched_task.cpu = cpu; } else { dispatched_task.flags |= RL_CPU_ANY as u64; } dispatched_task.slice_ns = SLICE_US; self.bpf.dispatch_task(&dispatched_task).unwrap(); self.bpf.notify_complete(self.task_queue.len() as u64 + self.no_budget_task_queue.len() as u64); } } fn dispatch_tasks(&mut self) { loop { //TODO: we should probably not do this every time but after a predefined amount of time self.reset_budgets_and_garbage_collect(); // Consume all tasks before dispatching any. self.consume_all_tasks(); // Dispatch one task from the queue. self.dispatch_next_task(); // If no task is ready to run (or in case of error), stop dispatching tasks and notify // the BPF component that all tasks have been scheduled / dispatched, with no remaining // pending tasks. if self.task_queue.is_empty() { self.bpf.notify_complete(0); break; } } } fn run(&mut self) -> Result { while !self.bpf.exited() { self.dispatch_tasks(); } self.bpf.shutdown_and_report() } fn get_new_budget(&self) -> u64 { //TODO u64::MAX } fn reset_budgets_and_garbage_collect(&mut self) { let old_budget = self.maximum_budget; self.maximum_budget = self.get_new_budget(); self.managed_tasks.retain(|key, value| { let was_scheduled = value.budget == old_budget; if was_scheduled { value.budget = self.maximum_budget; } else { self.module.stop_trace(*key as u64); } was_scheduled }); } } //TODO: get children fn main() -> Result<()> { let matches = Command::new("Energy User Space Scheduler") .arg(Arg::new("mock") .short('m') .long("mock") .help("Use this flag if to activate kernel module mocking") .action(ArgAction::SetTrue) .required(false)) .arg(Arg::new("power_cap") .long("energy_cap") .help("Set a power cap for the processor") .required(false) .value_name("power in watts")) .get_matches(); let power_cap = *matches.get_one::("power_cap").unwrap_or(&u64::MAX); let use_mocking = matches.get_flag("mock"); // Initialize and load the FIFO scheduler. let mut open_object = MaybeUninit::uninit(); loop { let mut sched = Scheduler::init(&mut open_object, use_mocking, power_cap)?; if !sched.run()?.should_restart() { break; } } Ok(()) }