Samuka007/asterinas
1
0
Fork 0
mirror of https://github.com/asterinas/asterinas.git synced 2025-06-29 04:13:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enable scheduler to fetch scheduling information directly from Thread

Browse Source
This commit is contained in:
jellllly420
2024-09-23 15:35:21 +08:00
committed by Tate, Hongliang Tian
parent 9cc63149f1
commit 5c8e369057
3 changed files with 96 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
kernel/src/sched/priority_scheduler.rs
View File

@ -4,8 +4,11 @@ use ostd::{
cpu::{num_cpus, CpuSet, PinCurrentCpu}, cpu::{num_cpus, CpuSet, PinCurrentCpu},
sync::PreemptDisabled, sync::PreemptDisabled,
task::{ task::{
scheduler::{inject_scheduler, EnqueueFlags, LocalRunQueue, Scheduler, UpdateFlags}, scheduler::{
AtomicCpuId, Task, info::CommonSchedInfo, inject_scheduler, EnqueueFlags, LocalRunQueue, Scheduler,
UpdateFlags,
},
Task,
}, },
trap::disable_local, trap::disable_local,
}; };
@ -15,7 +18,7 @@ use crate::{prelude::*, thread::Thread};
pub fn init() { pub fn init() {
let preempt_scheduler = Box::new(PreemptScheduler::default()); let preempt_scheduler = Box::new(PreemptScheduler::default());
let scheduler = Box::<PreemptScheduler<Task>>::leak(preempt_scheduler); let scheduler = Box::<PreemptScheduler<Thread, Task>>::leak(preempt_scheduler);
inject_scheduler(scheduler); inject_scheduler(scheduler);
} }
@ -25,11 +28,11 @@ pub fn init() {
/// are always prioritized during scheduling. /// are always prioritized during scheduling.
/// Normal tasks are placed in the `normal_entities` queue and are only /// Normal tasks are placed in the `normal_entities` queue and are only
/// scheduled for execution when there are no real-time tasks. /// scheduled for execution when there are no real-time tasks.
struct PreemptScheduler<T: PreemptSchedInfo> { struct PreemptScheduler<T: PreemptSchedInfo + FromTask<U>, U: CommonSchedInfo> {
rq: Vec<SpinLock<PreemptRunQueue<T>>>, rq: Vec<SpinLock<PreemptRunQueue<T, U>>>,
} }
impl<T: PreemptSchedInfo> PreemptScheduler<T> { impl<T: PreemptSchedInfo + FromTask<U>, U: CommonSchedInfo> PreemptScheduler<T, U> {
fn new(nr_cpus: u32) -> Self { fn new(nr_cpus: u32) -> Self {
let mut rq = Vec::with_capacity(nr_cpus as usize); let mut rq = Vec::with_capacity(nr_cpus as usize);
for _ in 0..nr_cpus { for _ in 0..nr_cpus {
@ -39,11 +42,11 @@ impl<T: PreemptSchedInfo> PreemptScheduler<T> {
} }
/// Selects a CPU for task to run on for the first time. /// Selects a CPU for task to run on for the first time.
fn select_cpu(&self, runnable: &Arc<T>) -> u32 { fn select_cpu(&self, entity: &PreemptSchedEntity<T, U>) -> u32 {
// If the CPU of a runnable task has been set before, keep scheduling // If the CPU of a runnable task has been set before, keep scheduling
// the task to that one. // the task to that one.
// TODO: Consider migrating tasks between CPUs for load balancing. // TODO: Consider migrating tasks between CPUs for load balancing.
if let Some(cpu_id) = runnable.cpu().get() { if let Some(cpu_id) = entity.task.cpu().get() {
return cpu_id; return cpu_id;
} }
@ -51,7 +54,7 @@ impl<T: PreemptSchedInfo> PreemptScheduler<T> {
let mut selected = irq_guard.current_cpu(); let mut selected = irq_guard.current_cpu();
let mut minimum_load = usize::MAX; let mut minimum_load = usize::MAX;
for candidate in runnable.cpu_affinity().iter() { for candidate in entity.thread.cpu_affinity().iter() {
let rq = self.rq[candidate as usize].lock(); let rq = self.rq[candidate as usize].lock();
// A wild guess measuring the load of a runqueue. We assume that // A wild guess measuring the load of a runqueue. We assume that
// real-time tasks are 4-times as important as normal tasks. // real-time tasks are 4-times as important as normal tasks.
@ -68,12 +71,15 @@ impl<T: PreemptSchedInfo> PreemptScheduler<T> {
} }
} }
impl<T: Sync + Send + PreemptSchedInfo> Scheduler<T> for PreemptScheduler<T> { impl<T: Sync + Send + PreemptSchedInfo + FromTask<U>, U: Sync + Send + CommonSchedInfo> Scheduler<U>
fn enqueue(&self, runnable: Arc<T>, flags: EnqueueFlags) -> Option<u32> { for PreemptScheduler<T, U>
{
fn enqueue(&self, task: Arc<U>, flags: EnqueueFlags) -> Option<u32> {
let entity = PreemptSchedEntity::new(task);
let mut still_in_rq = false; let mut still_in_rq = false;
let target_cpu = { let target_cpu = {
let mut cpu_id = self.select_cpu(&runnable); let mut cpu_id = self.select_cpu(&entity);
if let Err(task_cpu_id) = runnable.cpu().set_if_is_none(cpu_id) { if let Err(task_cpu_id) = entity.task.cpu().set_if_is_none(cpu_id) {
debug_assert!(flags != EnqueueFlags::Spawn); debug_assert!(flags != EnqueueFlags::Spawn);
still_in_rq = true; still_in_rq = true;
cpu_id = task_cpu_id; cpu_id = task_cpu_id;
@ -83,13 +89,12 @@ impl<T: Sync + Send + PreemptSchedInfo> Scheduler<T> for PreemptScheduler<T> {
}; };
let mut rq = self.rq[target_cpu as usize].disable_irq().lock(); let mut rq = self.rq[target_cpu as usize].disable_irq().lock();
if still_in_rq && let Err(_) = runnable.cpu().set_if_is_none(target_cpu) { if still_in_rq && let Err(_) = entity.task.cpu().set_if_is_none(target_cpu) {
return None; return None;
} }
let entity = PreemptSchedEntity::new(runnable); if entity.thread.is_real_time() {
if entity.is_real_time() {
rq.real_time_entities.push_back(entity); rq.real_time_entities.push_back(entity);
} else if entity.is_lowest() { } else if entity.thread.is_lowest() {
rq.lowest_entities.push_back(entity); rq.lowest_entities.push_back(entity);
} else { } else {
rq.normal_entities.push_back(entity); rq.normal_entities.push_back(entity);
@ -98,34 +103,34 @@ impl<T: Sync + Send + PreemptSchedInfo> Scheduler<T> for PreemptScheduler<T> {
Some(target_cpu) Some(target_cpu)
} }
fn local_rq_with(&self, f: &mut dyn FnMut(&dyn LocalRunQueue<T>)) { fn local_rq_with(&self, f: &mut dyn FnMut(&dyn LocalRunQueue<U>)) {
let irq_guard = disable_local(); let irq_guard = disable_local();
let local_rq: &PreemptRunQueue<T> = &self.rq[irq_guard.current_cpu() as usize].lock(); let local_rq: &PreemptRunQueue<T, U> = &self.rq[irq_guard.current_cpu() as usize].lock();
f(local_rq); f(local_rq);
} }
fn local_mut_rq_with(&self, f: &mut dyn FnMut(&mut dyn LocalRunQueue<T>)) { fn local_mut_rq_with(&self, f: &mut dyn FnMut(&mut dyn LocalRunQueue<U>)) {
let irq_guard = disable_local(); let irq_guard = disable_local();
let local_rq: &mut PreemptRunQueue<T> = let local_rq: &mut PreemptRunQueue<T, U> =
&mut self.rq[irq_guard.current_cpu() as usize].lock(); &mut self.rq[irq_guard.current_cpu() as usize].lock();
f(local_rq); f(local_rq);
} }
} }
impl Default for PreemptScheduler<Task> { impl Default for PreemptScheduler<Thread, Task> {
fn default() -> Self { fn default() -> Self {
Self::new(num_cpus()) Self::new(num_cpus())
} }
} }
struct PreemptRunQueue<T: PreemptSchedInfo> { struct PreemptRunQueue<T: PreemptSchedInfo + FromTask<U>, U: CommonSchedInfo> {
current: Option<PreemptSchedEntity<T>>, current: Option<PreemptSchedEntity<T, U>>,
real_time_entities: VecDeque<PreemptSchedEntity<T>>, real_time_entities: VecDeque<PreemptSchedEntity<T, U>>,
normal_entities: VecDeque<PreemptSchedEntity<T>>, normal_entities: VecDeque<PreemptSchedEntity<T, U>>,
lowest_entities: VecDeque<PreemptSchedEntity<T>>, lowest_entities: VecDeque<PreemptSchedEntity<T, U>>,
} }
impl<T: PreemptSchedInfo> PreemptRunQueue<T> { impl<T: PreemptSchedInfo + FromTask<U>, U: CommonSchedInfo> PreemptRunQueue<T, U> {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
current: None, current: None,
@ -136,9 +141,11 @@ impl<T: PreemptSchedInfo> PreemptRunQueue<T> {
} }
} }
impl<T: Sync + Send + PreemptSchedInfo> LocalRunQueue<T> for PreemptRunQueue<T> { impl<T: PreemptSchedInfo + FromTask<U>, U: CommonSchedInfo> LocalRunQueue<U>
fn current(&self) -> Option<&Arc<T>> { for PreemptRunQueue<T, U>
self.current.as_ref().map(|entity| &entity.runnable) {
fn current(&self) -> Option<&Arc<U>> {
self.current.as_ref().map(|entity| &entity.task)
} }
fn update_current(&mut self, flags: UpdateFlags) -> bool { fn update_current(&mut self, flags: UpdateFlags) -> bool {
@ -148,13 +155,14 @@ impl<T: Sync + Send + PreemptSchedInfo> LocalRunQueue<T> for PreemptRunQueue<T>
return false; return false;
}; };
current_entity.tick() current_entity.tick()
|| (!current_entity.is_real_time() && !self.real_time_entities.is_empty()) || (!current_entity.thread.is_real_time()
&& !self.real_time_entities.is_empty())
} }
_ => true, _ => true,
} }
} }
fn pick_next_current(&mut self) -> Option<&Arc<T>> { fn pick_next_current(&mut self) -> Option<&Arc<U>> {
let next_entity = if !self.real_time_entities.is_empty() { let next_entity = if !self.real_time_entities.is_empty() {
self.real_time_entities.pop_front() self.real_time_entities.pop_front()
} else if !self.normal_entities.is_empty() { } else if !self.normal_entities.is_empty() {
@ -163,58 +171,54 @@ impl<T: Sync + Send + PreemptSchedInfo> LocalRunQueue<T> for PreemptRunQueue<T>
self.lowest_entities.pop_front() self.lowest_entities.pop_front()
}?; }?;
if let Some(prev_entity) = self.current.replace(next_entity) { if let Some(prev_entity) = self.current.replace(next_entity) {
if prev_entity.is_real_time() { if prev_entity.thread.is_real_time() {
self.real_time_entities.push_back(prev_entity); self.real_time_entities.push_back(prev_entity);
} else if prev_entity.is_lowest() { } else if prev_entity.thread.is_lowest() {
self.lowest_entities.push_back(prev_entity); self.lowest_entities.push_back(prev_entity);
} else { } else {
self.normal_entities.push_back(prev_entity); self.normal_entities.push_back(prev_entity);
} }
} }
Some(&self.current.as_ref().unwrap().runnable) Some(&self.current.as_ref().unwrap().task)
} }
fn dequeue_current(&mut self) -> Option<Arc<T>> { fn dequeue_current(&mut self) -> Option<Arc<U>> {
self.current.take().map(|entity| { self.current.take().map(|entity| {
let runnable = entity.runnable; let runnable = entity.task;
runnable.cpu().set_to_none(); runnable.cpu().set_to_none();
runnable runnable
}) })
} }
} }
struct PreemptSchedEntity<T: PreemptSchedInfo + FromTask<U>, U: CommonSchedInfo> {
struct PreemptSchedEntity<T: PreemptSchedInfo> { task: Arc<U>,
runnable: Arc<T>, thread: Arc<T>,
time_slice: TimeSlice, time_slice: TimeSlice,
} }
impl<T: PreemptSchedInfo> PreemptSchedEntity<T> { impl<T: PreemptSchedInfo + FromTask<U>, U: CommonSchedInfo> PreemptSchedEntity<T, U> {
fn new(runnable: Arc<T>) -> Self { fn new(task: Arc<U>) -> Self {
let thread = T::from_task(&task);
let time_slice = TimeSlice::default();
Self { Self {
runnable, task,
time_slice: TimeSlice::default(), thread,
time_slice,
} }
} }
fn is_real_time(&self) -> bool {
self.runnable.is_real_time()
}
fn is_lowest(&self) -> bool {
self.runnable.is_lowest()
}
fn tick(&mut self) -> bool { fn tick(&mut self) -> bool {
self.time_slice.elapse() self.time_slice.elapse()
} }
} }
impl<T: PreemptSchedInfo> Clone for PreemptSchedEntity<T> { impl<T: PreemptSchedInfo + FromTask<U>, U: CommonSchedInfo> Clone for PreemptSchedEntity<T, U> {
fn clone(&self) -> Self { fn clone(&self) -> Self {
Self { Self {
runnable: self.runnable.clone(), task: self.task.clone(),
thread: self.thread.clone(),
time_slice: self.time_slice, time_slice: self.time_slice,
} }
} }
@ -245,26 +249,16 @@ impl Default for TimeSlice {
} }
} }
impl PreemptSchedInfo for Task { impl PreemptSchedInfo for Thread {
const REAL_TIME_TASK_PRIORITY: Priority = Priority::new(PriorityRange::new(100)); const REAL_TIME_TASK_PRIORITY: Priority = Priority::new(PriorityRange::new(100));
const LOWEST_TASK_PRIORITY: Priority = Priority::new(PriorityRange::new(PriorityRange::MAX)); const LOWEST_TASK_PRIORITY: Priority = Priority::new(PriorityRange::new(PriorityRange::MAX));
fn priority(&self) -> Priority { fn priority(&self) -> Priority {
self.data() self.priority()
.downcast_ref::<Arc<Thread>>()
.unwrap()
.priority()
}
fn cpu(&self) -> &AtomicCpuId {
&self.schedule_info().cpu
} }
fn cpu_affinity(&self) -> SpinLockGuard<CpuSet, PreemptDisabled> { fn cpu_affinity(&self) -> SpinLockGuard<CpuSet, PreemptDisabled> {
self.data() self.lock_cpu_affinity()
.downcast_ref::<Arc<Thread>>()
.unwrap()
.lock_cpu_affinity()
} }
} }
@ -274,8 +268,6 @@ trait PreemptSchedInfo {
fn priority(&self) -> Priority; fn priority(&self) -> Priority;
fn cpu(&self) -> &AtomicCpuId;
fn cpu_affinity(&self) -> SpinLockGuard<CpuSet, PreemptDisabled>; fn cpu_affinity(&self) -> SpinLockGuard<CpuSet, PreemptDisabled>;
fn is_real_time(&self) -> bool { fn is_real_time(&self) -> bool {
@ -286,3 +278,13 @@ trait PreemptSchedInfo {
self.priority() == Self::LOWEST_TASK_PRIORITY self.priority() == Self::LOWEST_TASK_PRIORITY
} }
} }
impl FromTask<Task> for Thread {
fn from_task(task: &Arc<Task>) -> Arc<Self> {
task.data().downcast_ref::<Arc<Self>>().unwrap().clone()
}
}
trait FromTask<U> {
fn from_task(task: &Arc<U>) -> Arc<Self>;
}

28
ostd/src/task/scheduler/fifo_scheduler.rs
View File

@ -2,11 +2,13 @@
use alloc::{boxed::Box, collections::VecDeque, sync::Arc, vec::Vec}; use alloc::{boxed::Box, collections::VecDeque, sync::Arc, vec::Vec};
use super::{inject_scheduler, EnqueueFlags, LocalRunQueue, Scheduler, UpdateFlags}; use super::{
info::CommonSchedInfo, inject_scheduler, EnqueueFlags, LocalRunQueue, Scheduler, UpdateFlags,
};
use crate::{ use crate::{
cpu::{num_cpus, PinCurrentCpu}, cpu::{num_cpus, PinCurrentCpu},
sync::SpinLock, sync::SpinLock,
task::{disable_preempt, AtomicCpuId, Task}, task::{disable_preempt, Task},
}; };
pub fn init() { pub fn init() {
@ -16,12 +18,12 @@ pub fn init() {
} }
/// A simple FIFO (First-In-First-Out) task scheduler. /// A simple FIFO (First-In-First-Out) task scheduler.
struct FifoScheduler<T: FifoSchedInfo> { struct FifoScheduler<T: CommonSchedInfo> {
/// A thread-safe queue to hold tasks waiting to be executed. /// A thread-safe queue to hold tasks waiting to be executed.
rq: Vec<SpinLock<FifoRunQueue<T>>>, rq: Vec<SpinLock<FifoRunQueue<T>>>,
} }
impl<T: FifoSchedInfo> FifoScheduler<T> { impl<T: CommonSchedInfo> FifoScheduler<T> {
/// Creates a new instance of `FifoScheduler`. /// Creates a new instance of `FifoScheduler`.
fn new(nr_cpus: u32) -> Self { fn new(nr_cpus: u32) -> Self {
let mut rq = Vec::new(); let mut rq = Vec::new();
@ -37,7 +39,7 @@ impl<T: FifoSchedInfo> FifoScheduler<T> {
} }
} }
impl<T: FifoSchedInfo + Send + Sync> Scheduler<T> for FifoScheduler<T> { impl<T: CommonSchedInfo + Send + Sync> Scheduler<T> for FifoScheduler<T> {
fn enqueue(&self, runnable: Arc<T>, flags: EnqueueFlags) -> Option<u32> { fn enqueue(&self, runnable: Arc<T>, flags: EnqueueFlags) -> Option<u32> {
let mut still_in_rq = false; let mut still_in_rq = false;
let target_cpu = { let target_cpu = {
@ -77,12 +79,12 @@ impl<T: FifoSchedInfo + Send + Sync> Scheduler<T> for FifoScheduler<T> {
} }
} }
struct FifoRunQueue<T: FifoSchedInfo> { struct FifoRunQueue<T: CommonSchedInfo> {
current: Option<Arc<T>>, current: Option<Arc<T>>,
queue: VecDeque<Arc<T>>, queue: VecDeque<Arc<T>>,
} }
impl<T: FifoSchedInfo> FifoRunQueue<T> { impl<T: CommonSchedInfo> FifoRunQueue<T> {
pub const fn new() -> Self { pub const fn new() -> Self {
Self { Self {
current: None, current: None,
@ -91,7 +93,7 @@ impl<T: FifoSchedInfo> FifoRunQueue<T> {
} }
} }
impl<T: FifoSchedInfo> LocalRunQueue<T> for FifoRunQueue<T> { impl<T: CommonSchedInfo> LocalRunQueue<T> for FifoRunQueue<T> {
fn current(&self) -> Option<&Arc<T>> { fn current(&self) -> Option<&Arc<T>> {
self.current.as_ref() self.current.as_ref()
} }
@ -119,13 +121,3 @@ impl Default for FifoScheduler<Task> {
Self::new(num_cpus()) Self::new(num_cpus())
} }
} }
impl FifoSchedInfo for Task {
fn cpu(&self) -> &AtomicCpuId {
&self.schedule_info().cpu
}
}
trait FifoSchedInfo {
fn cpu(&self) -> &AtomicCpuId;
}

14
ostd/src/task/scheduler/info.rs
View File

@ -4,6 +4,8 @@
use core::sync::atomic::{AtomicU32, Ordering}; use core::sync::atomic::{AtomicU32, Ordering};
use crate::task::Task;
/// Fields of a task that OSTD will never touch. /// Fields of a task that OSTD will never touch.
/// ///
/// The type ought to be defined by the OSTD user and injected into the task. /// The type ought to be defined by the OSTD user and injected into the task.
@ -61,3 +63,15 @@ impl Default for AtomicCpuId {
Self::new(Self::NONE) Self::new(Self::NONE)
} }
} }
impl CommonSchedInfo for Task {
fn cpu(&self) -> &AtomicCpuId {
&self.schedule_info().cpu
}
}
/// Trait for fetching common scheduling information.
pub trait CommonSchedInfo {
/// Gets the CPU that the task is running on or lately ran on.
fn cpu(&self) -> &AtomicCpuId;
}