qiurui/DragonOS
1
0
Fork 0
mirror of https://github.com/DragonOS-Community/DragonOS.git synced 2026-09-08 23:57:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(sched): make scheduler changes atomic (#2246)

Browse Source 
* refactor(sched): make scheduler changes atomic

Introduce a single scheduler-change transaction that holds the task PI lock and a stable runqueue lock while updating policy, class, priority, queue placement, and reset-on-fork state.

Preserve Linux 6.6 scheduler semantics by aging and transferring Fair PELT accounting across class changes, removing dead-task contributions, honoring realtime head placement on priority drops, and requesting the required local or remote reschedule.

Centralize affinity publication under the same PI-lock boundary and make IRQ-time cross-CPU reads sound with atomic totals plus owner-local accounting.

Extend the bounded fifo_demo coverage for remote queued and running tasks, realtime ordering, policy-affinity publication and migration, policy-exit races, cleanup results, and 1/2/3-CPU behavior.

Validation:

- make fmt ARCH=x86_64 (including all-features clippy)

- x86_64 default and fifo_demo kernel builds

- x86_64 fifo_demo boots with 1, 2, and 3 vCPUs

- scheduler, affinity, tracepoint, process, and RCU dunitests

- loongarch64 kernel build

- riscv64 kernel compile, link, and ELF generation

Signed-off-by: longjin <longjin@dragonos.org>

* fix(sched): preserve migration state through switch tail

Keep running tasks marked as migrating from source runqueue dequeue until the context-switch tail owns the task PI lock. This prevents concurrent scheduler class changes from accounting against a stale source runqueue.

Make scheduler changes wait for the migration owner without contending on the PI lock, mirroring Linux task_rq_lock semantics. The switch tail now clears the transient state only while committing the destination enqueue or a completed stop, so no migrating state leaks past the transaction.

Signed-off-by: longjin <longjin@dragonos.org>

* fix(sched): migrate fair PELT accounting between runqueues

Use the task-level migrating state to detach Fair PELT contributions from the source runqueue and reset the entity timestamp before rebinding it. Preserve ENQUEUE_MIGRATED through destination activation so the contribution is attached before the task becomes queued.

Handle queued, current, and sleeping Fair migrations under the appropriate runqueue locks, and restore the source attachment when an asynchronous stop cancels a current-task migration. Keep dormant realtime entities and new-task placement outside the Fair migration transaction.

Extend the scheduler feature test with deterministic Fair and FIFO migrations and enforce the migrated Fair timestamp invariant before destination enqueue.

Signed-off-by: longjin <longjin@dragonos.org>

---------

Signed-off-by: longjin <longjin@dragonos.org>
This commit is contained in:
LoGin
2026-09-01 16:42:39 +08:00
committed by GitHub
parent 564b3cb412
commit 43f4d72808
9 changed files with 1211 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kernel/src/process/manager/mod.rs
+29 -8
View File
@@ -23,13 +23,14 @@ use crate::{
cred::SUID_DUMP_USER, namespace::user_namespace::INIT_USER_NAMESPACE, ProcessControlBlock,
RawPid,
},
sched::{cpu_rq, enqueue_task_on_cpu, select_task_rq, OnRq, WakeupFlags},
sched::{cpu_rq, enqueue_task_on_cpu, select_task_rq, OnRq, SchedClass, WakeupFlags},
smp::{core::smp_get_processor_id, cpu::ProcessorId, kick_cpu},
syscall::user_access::write_one_to_user_protected,
};
mod exit;
mod sched;
pub(crate) use sched::SchedChangeRequest;
#[derive(Debug)]
pub struct ProcessManager;
@@ -431,20 +432,40 @@ impl ProcessManager {
// that lock through enqueue. A later affinity update will then
// either observe this legal placement or migrate it again.
let pi_guard = prev_pcb.sched_info().pi_lock_irqsave();
// stop_task() can win after schedule dequeues prev but before this
// tail obtains pi_lock. Likewise, wakeup_stop() can enqueue it
// after finish_running() and before this check. Only a still
// runnable, still off-rq task belongs to this migration owner;
// otherwise leave the stopped or already-enqueued state intact.
// DEQUEUE_MOVE keeps on_rq=Migrating until this owner holds
// pi_lock. This excludes task_rq_lock-style mutations against the
// old rq throughout the switch tail. stop_task() may still make
// the task non-runnable first; wakeup_stop() may make it runnable
// again, but leaves the migration enqueue to this owner.
let still_owns_migration = prev_pcb.sched_info().state().is_runnable()
&& *prev_pcb.sched_info().on_rq.lock_irqsave() == OnRq::None;
&& *prev_pcb.sched_info().on_rq.lock_irqsave() == OnRq::Migrating;
if still_owns_migration {
let allowed = pi_guard.cpus_allowed.clone();
let dest_cpu =
select_task_rq(&prev_pcb, dest_cpu, WakeupFlags::WF_MIGRATED, &allowed);
prev_pcb.sched_info().set_on_cpu(None);
// Keep Migrating through activate_task() so the destination
// class receives ENQUEUE_MIGRATED and publishes Queued only
// after its accounting is committed.
enqueue_task_on_cpu(&prev_pcb, dest_cpu, WakeupFlags::WF_MIGRATED, false);
} else if *prev_pcb.sched_info().on_rq.lock_irqsave() == OnRq::Migrating {
// An asynchronous stop can win after a Fair task detached
// from the source rq. Restore Linux's sleeping-task PELT
// attachment before publishing a genuine off-rq state.
if prev_pcb.sched_info().sched_class() == SchedClass::Fair {
let src_cpu = prev_pcb
.sched_info()
.on_cpu()
.expect("a cancelled current migration must retain its source CPU");
let src_rq = cpu_rq(src_cpu.data() as usize);
let (src_rq, _src_rq_guard) = src_rq.self_lock();
src_rq.update_rq_clock();
crate::sched::fair::CompletelyFairScheduler::cancel_task_rq_migration(
src_rq, &prev_pcb,
);
}
*prev_pcb.sched_info().on_rq.lock_irqsave() = OnRq::None;
}
debug_assert_ne!(*prev_pcb.sched_info().on_rq.lock_irqsave(), OnRq::Migrating);
drop(pi_guard);
}
kernel/src/process/manager/sched.rs
+163 -87
View File
@@ -6,14 +6,26 @@ use system_error::SystemError;
use crate::{
arch::{cpu::current_cpu_id, CurrentIrqArch},
exception::InterruptArch,
libs::cpumask::CpuMask,
process::{ProcessControlBlock, ProcessFlags, ProcessManager, ProcessState},
sched::{
cpu_rq, enqueue_task_on_cpu, select_task_rq, DequeueFlag, EnqueueFlag, LinuxSchedPolicy,
OnRq, SchedClass, Scheduler, WakeupFlags,
OnRq, SchedClass, WakeupFlags,
},
smp::{core::smp_get_processor_id, kick_cpu},
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SchedChangeRequest {
#[allow(dead_code)]
Normal {
reset_on_fork: bool,
},
Fifo {
priority: i32,
},
}
impl ProcessManager {
/// Wake up a process.
pub fn wakeup(pcb: &Arc<ProcessControlBlock>) -> Result<(), SystemError> {
@@ -131,107 +143,171 @@ impl ProcessManager {
Ok(())
}
/// Set the specified kernel thread to the SCHED_FIFO scheduling policy.
///
/// task_rq_lock → update_rq_clock → read queued/running →
/// dequeue/put_prev → modify parameters → enqueue → check_class_changed →
/// unlock
/// Atomically change the base scheduler parameters of an already placed task.
pub(crate) fn set_scheduler(
pcb: &Arc<ProcessControlBlock>,
request: SchedChangeRequest,
) -> Result<(), SystemError> {
if let SchedChangeRequest::Fifo { priority } = request {
if !(0..crate::sched::prio::MAX_RT_PRIO - 1).contains(&priority) {
return Err(SystemError::EINVAL);
}
}
let _irq_guard = unsafe { CurrentIrqArch::save_and_disable_irq() };
loop {
// Lock ordering matches Linux task_rq_lock(): pi_lock -> rq_lock.
let mut pi_guard = pcb.sched_info().pi_lock_irqsave();
let old_class = pcb.sched_info().sched_class();
if old_class == SchedClass::Idle {
return Err(SystemError::EINVAL);
}
// Published tasks retain their last rq while sleeping or exiting.
// A task without a CPU can have a Fair entity bound to another rq,
// so this PR deliberately rejects that unplaced state.
let Some(target_cpu) = pcb.sched_info().on_cpu() else {
return Err(SystemError::EINVAL);
};
let rq = cpu_rq(target_cpu.data() as usize);
let (rq, rq_guard) = rq.self_lock();
let migrating = *pcb.sched_info().on_rq.lock_irqsave() == OnRq::Migrating;
let stable = pcb.sched_info().on_cpu() == Some(target_cpu) && !migrating;
if !stable {
drop(rq_guard);
drop(pi_guard);
// Match Linux task_rq_lock(): do not repeatedly contend on
// pi_lock while the migration owner is moving the task
// between runqueues. The owner publishes a stable rq before
// clearing Migrating, after which the outer loop samples the
// task CPU again.
if migrating {
while *pcb.sched_info().on_rq.lock_irqsave() == OnRq::Migrating {
core::hint::spin_loop();
}
}
continue;
}
rq.update_rq_clock();
let old_policy = pcb.sched_info().policy();
let old_prio = pcb.sched_info().prio();
let old_reset = pi_guard.sched_reset_on_fork();
let (new_policy, new_prio, new_reset) = match request {
SchedChangeRequest::Normal { reset_on_fork } => (
LinuxSchedPolicy::Normal,
pcb.sched_info().static_prio(),
reset_on_fork,
),
SchedChangeRequest::Fifo { priority } => (LinuxSchedPolicy::Fifo, priority, false),
};
let new_class = new_policy.base_sched_class();
if old_policy == new_policy && old_prio == new_prio && old_reset == new_reset {
drop(rq_guard);
drop(pi_guard);
return Ok(());
}
let queued = *pcb.sched_info().on_rq.lock_irqsave() == OnRq::Queued;
let running = Arc::ptr_eq(&rq.current(), pcb);
// The final switch-out removes a dead Fair entity's PELT
// contribution. Linux still permits scheduler parameter updates
// for dead tasks, but such an off-rq task must not be attached
// again after its task-dead accounting has completed.
let account_class_change = !pcb.sched_info().state().is_exited() || queued || running;
if queued {
rq.dequeue_task(
pcb.clone(),
DequeueFlag::DEQUEUE_SAVE
| DequeueFlag::DEQUEUE_MOVE
| DequeueFlag::DEQUEUE_NOCLOCK,
);
}
if running {
rq.put_prev_task_for_class(old_class, pcb.clone());
}
if account_class_change
&& old_class == SchedClass::Fair
&& new_class != SchedClass::Fair
{
crate::sched::fair::CompletelyFairScheduler::switched_from_fair(rq, pcb);
}
pcb.sched_info().set_policy(new_policy);
pcb.sched_info().set_prio(new_prio);
pcb.sched_info().set_normal_prio(new_prio);
pi_guard.set_sched_reset_on_fork(new_reset);
if account_class_change
&& old_class != SchedClass::Fair
&& new_class == SchedClass::Fair
{
crate::sched::fair::CompletelyFairScheduler::switched_to_fair(rq, pcb);
}
if queued {
let mut flags = EnqueueFlag::ENQUEUE_RESTORE
| EnqueueFlag::ENQUEUE_MOVE
| EnqueueFlag::ENQUEUE_NOCLOCK;
if old_prio < new_prio {
flags |= EnqueueFlag::ENQUEUE_HEAD;
}
rq.enqueue_task(pcb.clone(), flags);
}
if running {
rq.set_next_task_for_class(new_class, pcb.clone());
}
rq.check_scheduler_changed(pcb, old_class, old_prio);
drop(rq_guard);
drop(pi_guard);
return Ok(());
}
}
/// Set a trusted kernel thread to the SCHED_FIFO policy.
pub fn set_fifo_policy(pcb: &Arc<ProcessControlBlock>, prio: i32) -> Result<(), SystemError> {
if !pcb.flags().contains(ProcessFlags::KTHREAD) {
return Err(SystemError::EPERM);
}
if !(0..crate::sched::prio::MAX_RT_PRIO - 1).contains(&prio) {
return Err(SystemError::EINVAL);
}
Self::set_scheduler(pcb, SchedChangeRequest::Fifo { priority: prio })
}
let _irq_guard = unsafe { CurrentIrqArch::save_and_disable_irq() };
// Lock ordering: pi_lock → rq_lock, matching Linux task_rq_lock().
/// Publish a validated affinity mask and apply any required migration.
pub(crate) fn set_cpus_allowed(
pcb: &Arc<ProcessControlBlock>,
mask: CpuMask,
) -> Result<(), SystemError> {
let mut pi_guard = pcb.sched_info().pi_lock_irqsave();
pi_guard.set_cpus_allowed(mask.clone());
let target_cpu = pcb.sched_info().on_cpu().unwrap_or(current_cpu_id());
let update_clock = target_cpu == smp_get_processor_id();
let rq = cpu_rq(target_cpu.data() as usize);
let (rq, rq_guard) = rq.self_lock();
if update_clock {
rq.update_rq_clock();
if pcb.sched_info().is_new_task() {
return Ok(());
}
// Read task state under the lock.
let old_class = pcb.sched_info().sched_class();
let queued = *pcb.sched_info().on_rq.lock_irqsave() == OnRq::Queued;
// Determine whether the target is the currently running task on this rq.
let running = Arc::ptr_eq(&rq.current(), pcb);
// First dequeue the task from the scheduler, then modify parameters,
// and finally re-enqueue.
if queued {
rq.dequeue_task(
pcb.clone(),
DequeueFlag::DEQUEUE_NOCLOCK | DequeueFlag::DEQUEUE_SAVE,
);
if pcb
.sched_info()
.migrate_to()
.is_some_and(|cpu| !mask.get(cpu).unwrap_or(false))
{
pcb.sched_info().set_migrate_to(None);
pcb.flags().remove(ProcessFlags::NEED_MIGRATE);
}
// A running task must first be put_prev_task to yield its current
// execution position.
if running {
match old_class {
SchedClass::Realtime => {
crate::sched::realtime::RealtimeScheduler::put_prev_task(rq, pcb.clone())
}
SchedClass::Fair => {
crate::sched::fair::CompletelyFairScheduler::put_prev_task(rq, pcb.clone())
}
SchedClass::Idle => {
crate::sched::idle::IdleScheduler::put_prev_task(rq, pcb.clone())
}
if let Some(cpu) = pcb.sched_info().on_cpu() {
if !mask.get(cpu).unwrap_or(false) {
let dest_cpu = select_task_rq(pcb, cpu, WakeupFlags::WF_TTWU, &mask);
crate::sched::request_task_migration(pcb, dest_cpu)?;
}
}
// Modify scheduling parameters (under rq_lock protection).
// Matches Linux __setscheduler_params + __setscheduler_prio:
// - policy set to FIFO
// - prio = normal_prio = MAX_RT_PRIO - 1 - rt_priority (the caller
// already passes the kernel prio)
// - static_prio is left unchanged (Linux only modifies static_prio for
// fair_policy, core.c:7528-7529)
pcb.sched_info().set_policy(LinuxSchedPolicy::Fifo);
pcb.sched_info().set_prio(prio);
pcb.sched_info().set_normal_prio(prio);
// This internal API has no RESET_ON_FORK argument. Do not carry a
// stale CFS flag into FIFO, where this PR deliberately does not
// implement the otherwise unreachable RT fork-reset path.
pi_guard.set_sched_reset_on_fork(false);
debug_assert!(!pi_guard.sched_reset_on_fork());
// Re-enqueue.
if queued {
rq.enqueue_task(
pcb.clone(),
EnqueueFlag::ENQUEUE_NOCLOCK | EnqueueFlag::ENQUEUE_RESTORE,
);
}
// Matches Linux __sched_setscheduler: after a running task changes its
// policy, set_next_task is required.
if running {
crate::sched::realtime::RealtimeScheduler::set_next_task(rq, pcb.clone());
}
// check_class_changed → preemption check.
if update_clock {
rq.check_preempt_current(pcb, WakeupFlags::empty());
} else {
rq.check_preempt_remote(pcb, WakeupFlags::empty());
}
// Release order: rq_lock first, then pi_lock, matching Linux
// task_rq_unlock().
drop(rq_guard);
drop(pi_guard);
Ok(())
}
kernel/src/process/mod.rs
+2
View File
@@ -49,6 +49,8 @@ pub use info::{
};
#[allow(unused_imports)]
pub use kstack::{KernelStack, KernelStackType};
#[allow(unused_imports)]
pub(crate) use manager::SchedChangeRequest;
pub(crate) use manager::{
account_context_switch, account_successful_fork, all_process, dec_visible_thread_count,
inc_visible_thread_count, lock_fs_refs_copy, lock_fs_refs_pivot, FsRefsReadGuard,
kernel/src/sched/cputime.rs
+70 -55
View File
@@ -1,4 +1,7 @@
use core::sync::atomic::{compiler_fence, AtomicU64, AtomicUsize, Ordering};
use core::{
cell::UnsafeCell,
sync::atomic::{AtomicU64, Ordering},
};
use crate::{
arch::{ipc::signal::Signal, CurrentIrqArch},
@@ -108,78 +111,90 @@ pub fn init_kernel_cpu_stat() {
}
pub fn irq_time_read(cpu: ProcessorId) -> u64 {
compiler_fence(Ordering::SeqCst);
let irqtime = cpu_irq_time(cpu);
let mut total;
loop {
let seq = irqtime.sync.load(Ordering::SeqCst);
total = irqtime.total;
if seq == irqtime.sync.load(Ordering::SeqCst) {
break;
}
}
compiler_fence(Ordering::SeqCst);
total
cpu_irq_time(cpu).total.load(Ordering::Relaxed)
}
#[derive(Debug, Default)]
pub struct IrqTime {
pub total: u64,
pub tick_delta: u64,
pub hardirq_delta: u64,
pub softirq_delta: u64,
pub irq_start_time: u64,
pub sync: AtomicUsize,
struct IrqTimeLocal {
tick_delta: u64,
hardirq_delta: u64,
softirq_delta: u64,
irq_start_time: u64,
}
impl IrqTime {
pub fn account_delta(&mut self, delta: u64, is_hardirq: bool) {
// 开始更改时增加序列号
self.sync.fetch_add(1, Ordering::SeqCst);
self.total += delta;
self.tick_delta += delta;
#[derive(Debug)]
pub(crate) struct IrqTime {
owner: ProcessorId,
total: AtomicU64,
local: UnsafeCell<IrqTimeLocal>,
}
// 根据中断类型分别记录
if is_hardirq {
self.hardirq_delta += delta;
} else {
self.softirq_delta += delta;
// `total` is the only cross-CPU field and is atomic. `local` is accessed only
// by `owner` with local IRQs disabled, as enforced by `with_local()`.
unsafe impl Sync for IrqTime {}
impl IrqTime {
pub(crate) fn new(owner: ProcessorId) -> Self {
Self {
owner,
total: AtomicU64::new(0),
local: UnsafeCell::new(IrqTimeLocal::default()),
}
}
pub fn irqtime_tick_accounted(&mut self, max: u64) -> (u64, u64, u64) {
let total_delta = self.tick_delta.min(max);
let hardirq_delta = self.hardirq_delta.min(total_delta);
let softirq_delta = self.softirq_delta.min(total_delta - hardirq_delta);
#[inline]
fn with_local<R>(&self, f: impl FnOnce(&mut IrqTimeLocal) -> R) -> R {
debug_assert_eq!(self.owner, smp_get_processor_id());
debug_assert!(!CurrentIrqArch::is_irq_enabled());
self.tick_delta -= total_delta;
self.hardirq_delta -= hardirq_delta;
self.softirq_delta -= softirq_delta;
(total_delta, hardirq_delta, softirq_delta)
// SAFETY: each instance has one owner CPU. All callers run on that CPU
// with IRQs disabled, so local fields cannot be accessed concurrently.
f(unsafe { &mut *self.local.get() })
}
pub fn irqtime_start() {
let cpu = smp_get_processor_id();
let irq_time = cpu_irq_time(cpu);
compiler_fence(Ordering::SeqCst);
irq_time.irq_start_time = SchedClock::sched_clock_cpu(cpu) as u64;
compiler_fence(Ordering::SeqCst);
fn account_delta(&self, delta: u64, is_hardirq: bool) {
self.total.fetch_add(delta, Ordering::Relaxed);
self.with_local(|local| {
local.tick_delta += delta;
// 根据中断类型分别记录
if is_hardirq {
local.hardirq_delta += delta;
} else {
local.softirq_delta += delta;
}
});
}
pub fn irqtime_account_irq(_pcb: Arc<ProcessControlBlock>, is_hardirq: bool) {
compiler_fence(Ordering::SeqCst);
fn irqtime_tick_accounted(&self, max: u64) -> (u64, u64, u64) {
self.with_local(|local| {
let total_delta = local.tick_delta.min(max);
let hardirq_delta = local.hardirq_delta.min(total_delta);
let softirq_delta = local.softirq_delta.min(total_delta - hardirq_delta);
local.tick_delta -= total_delta;
local.hardirq_delta -= hardirq_delta;
local.softirq_delta -= softirq_delta;
(total_delta, hardirq_delta, softirq_delta)
})
}
pub(crate) fn irqtime_start() {
let cpu = smp_get_processor_id();
let irq_time = cpu_irq_time(cpu);
compiler_fence(Ordering::SeqCst);
let delta = SchedClock::sched_clock_cpu(cpu) as u64 - irq_time.irq_start_time;
compiler_fence(Ordering::SeqCst);
irq_time.with_local(|local| {
local.irq_start_time = SchedClock::sched_clock_cpu(cpu) as u64;
});
}
pub(crate) fn irqtime_account_irq(_pcb: Arc<ProcessControlBlock>, is_hardirq: bool) {
let cpu = smp_get_processor_id();
let irq_time = cpu_irq_time(cpu);
let irq_start_time = irq_time.with_local(|local| local.irq_start_time);
let delta = SchedClock::sched_clock_cpu(cpu) as u64 - irq_start_time;
irq_time.account_delta(delta, is_hardirq);
compiler_fence(Ordering::SeqCst);
}
}
kernel/src/sched/fair.rs
+114 -16
View File
@@ -858,43 +858,64 @@ impl CfsRunQueue {
se.force_mut().update_load_avg(self, now);
}
let mut decayed = self.update_self_load_avg(now);
decayed |= se.force_mut().propagate_entity_load_avg() as u32;
let mut _decayed = self.update_self_load_avg(now);
_decayed |= se.force_mut().propagate_entity_load_avg() as u32;
if se.avg.last_update_time > 0 && flags.contains(UpdateAvgFlags::DO_ATTACH) {
todo!()
} else if flags.contains(UpdateAvgFlags::DO_ATTACH) {
if se.avg.last_update_time == 0 && flags.contains(UpdateAvgFlags::DO_ATTACH) {
self.attach_entity_load_avg(se);
} else if flags.contains(UpdateAvgFlags::DO_DETACH) {
self.detach_entity_load_avg(se);
} else if decayed > 0 {
// cfs_rq_util_change
todo!()
}
}
/// Attach an entity's PELT contribution to this CFS runqueue.
///
/// The CFS rq average must be current before this method is called.
fn attach_entity_load_avg(&mut self, se: &Arc<FairSchedEntity>) {
let divider = self.avg.get_pelt_divider();
let scaled_weight = LoadWeight::scale_load_down(se.load.weight);
let se_mut = se.force_mut();
se_mut.avg.last_update_time = self.avg.last_update_time;
se_mut.avg.period_contrib = self.avg.period_contrib;
se_mut.avg.util_sum = (se_mut.avg.util_avg * divider) as u64;
se_mut.avg.runnable_sum = (se_mut.avg.runnable_avg * divider) as u64;
se_mut.avg.load_sum = (se_mut.avg.load_avg * divider) as u64;
se_mut.avg.load_sum = if scaled_weight < se_mut.avg.load_sum {
se_mut.avg.load_sum / scaled_weight
} else {
1
};
self.enqueue_load_avg(se.clone());
self.avg.util_avg += se.avg.util_avg;
self.avg.util_sum += se.avg.util_sum;
self.avg.runnable_avg += se.avg.runnable_avg;
self.avg.runnable_sum += se.avg.runnable_sum;
self.propagate = 1;
self.prop_runnable_sum += se.avg.load_sum as isize;
}
/// 将实体的负载均值与对应cfs分离
fn detach_entity_load_avg(&mut self, se: &Arc<FairSchedEntity>) {
self.dequeue_load_avg(se);
sub_positive(&mut self.avg.util_avg, se.avg.util_avg);
sub_positive(&mut (self.avg.util_sum as usize), se.avg.util_sum as usize);
self.avg.util_sum = self.avg.util_sum.saturating_sub(se.avg.util_sum);
self.avg.util_sum = self
.avg
.util_sum
.max((self.avg.util_avg * PELT_MIN_DIVIDER) as u64);
sub_positive(&mut self.avg.runnable_avg, se.avg.runnable_avg);
sub_positive(
&mut (self.avg.runnable_sum as usize),
se.avg.runnable_sum as usize,
);
self.avg.runnable_sum = self.avg.runnable_sum.saturating_sub(se.avg.runnable_sum);
self.avg.runnable_sum = self
.avg
.runnable_sum
.max((self.avg.runnable_avg * PELT_MIN_DIVIDER) as u64);
self.propagate = 1;
self.prop_runnable_sum += se.avg.load_sum as isize;
self.prop_runnable_sum -= se.avg.load_sum as isize;
}
fn update_self_load_avg(&mut self, now: u64) -> u32 {
@@ -976,6 +997,14 @@ impl CfsRunQueue {
/// 将实体加入队列
pub fn enqueue_entity(&mut self, se: &Arc<FairSchedEntity>, flags: EnqueueFlag) {
#[cfg(any(debug_assertions, feature = "fifo_demo"))]
if flags.contains(EnqueueFlag::ENQUEUE_MIGRATED) {
assert_eq!(
se.avg.last_update_time, 0,
"a migrated Fair entity must be detached before destination enqueue"
);
}
let is_curr = self.is_curr(se);
if is_curr {
@@ -1015,7 +1044,7 @@ impl CfsRunQueue {
pub fn dequeue_entity(&mut self, se: &Arc<FairSchedEntity>, flags: DequeueFlag) {
let mut action = UpdateAvgFlags::UPDATE_TG;
if se.is_task() && se.on_rq == OnRq::Migrating {
if se.is_task() && *se.pcb().sched_info().on_rq.lock_irqsave() == OnRq::Migrating {
action |= UpdateAvgFlags::DO_DETACH;
}
@@ -1356,6 +1385,75 @@ impl Default for CfsRunQueue {
pub struct CompletelyFairScheduler;
impl CompletelyFairScheduler {
fn detach_task_load_avg(rq: &mut CpuRunQueue, pcb: &Arc<ProcessControlBlock>) {
let se = pcb.sched_info().sched_entity();
debug_assert!(Arc::ptr_eq(&se.cfs_rq(), &rq.cfs_rq()));
if se.avg.last_update_time == 0 {
return;
}
let cfs = se.cfs_rq();
let cfs = cfs.force_mut();
cfs.update_load_avg(&se, UpdateAvgFlags::empty());
cfs.detach_entity_load_avg(&se);
}
/// Remove a task's PELT contribution when it leaves the fair class.
pub fn switched_from_fair(rq: &mut CpuRunQueue, pcb: &Arc<ProcessControlBlock>) {
Self::detach_task_load_avg(rq, pcb);
}
fn attach_task_load_avg(rq: &mut CpuRunQueue, pcb: &Arc<ProcessControlBlock>) {
let se = pcb.sched_info().sched_entity();
debug_assert!(Arc::ptr_eq(&se.cfs_rq(), &rq.cfs_rq()));
let cfs = se.cfs_rq();
let cfs = cfs.force_mut();
// Linux enables ATTACH_AGE_LOAD by default: age a detached entity to
// the destination rq clock before restoring its contribution.
cfs.update_load_avg(&se, UpdateAvgFlags::empty());
cfs.attach_entity_load_avg(&se);
}
/// Attach a task's PELT contribution before it enters the fair class.
pub fn switched_to_fair(rq: &mut CpuRunQueue, pcb: &Arc<ProcessControlBlock>) {
Self::attach_task_load_avg(rq, pcb);
}
/// Prepare a Fair entity before changing its CPU/CFS runqueue binding.
///
/// A runnable migration has already detached under the source rq lock.
/// Sleeping tasks remain attached on Linux, so a wakeup migration must
/// synchronize and detach that contribution from the retained source rq.
pub(crate) fn prepare_task_rq_migration(pcb: &Arc<ProcessControlBlock>) {
let se = pcb.sched_info().sched_entity();
if se.avg.last_update_time == 0 {
return;
}
if *pcb.sched_info().on_rq.lock_irqsave() != OnRq::Migrating {
let old_rq = se.cfs_rq().rq();
let (old_rq, _old_rq_guard) = old_rq.self_lock();
old_rq.update_rq_clock();
Self::detach_task_load_avg(old_rq, pcb);
}
// A zero timestamp tells destination enqueue_entity() to attach the
// migrated entity to its new CFS runqueue.
se.force_mut().avg.last_update_time = 0;
}
/// Restore the source PELT attachment when an asynchronous stop cancels
/// a current-task migration after source dequeue.
pub(crate) fn cancel_task_rq_migration(rq: &mut CpuRunQueue, pcb: &Arc<ProcessControlBlock>) {
Self::attach_task_load_avg(rq, pcb);
}
/// Remove the final PELT contribution after a Fair task exits and leaves
/// its runqueue. Sleeping tasks deliberately remain attached.
pub(crate) fn task_dead_fair(rq: &mut CpuRunQueue, pcb: &Arc<ProcessControlBlock>) {
Self::detach_task_load_avg(rq, pcb);
}
pub fn set_next_task(_rq: &mut CpuRunQueue, next: Arc<ProcessControlBlock>) {
let mut se = next.sched_info().sched_entity();
FairSchedEntity::for_each_in_group(&mut se, |se| {
kernel/src/sched/fifo_demo.rs
+721 -26
View File
@@ -1,23 +1,81 @@
#![allow(dead_code)]
use core::sync::atomic::{AtomicUsize, Ordering};
use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use alloc::{boxed::Box, string::String, sync::Arc, vec::Vec};
use crate::{
libs::cpumask::CpuMask,
process::{
kthread::{KernelThreadClosure, KernelThreadMechanism},
ProcessControlBlock, ProcessManager,
ProcessControlBlock, ProcessFlags, ProcessManager, SchedChangeRequest,
},
sched::{completion::Completion, prio::MAX_RT_PRIO, OnRq},
smp::cpu::{smp_cpu_manager, ProcessorId},
time::{sleep::nanosleep, PosixTimeSpec},
sched::{completion::Completion, prio::MAX_RT_PRIO, LinuxSchedPolicy, OnRq, SchedClass},
smp::{
core::smp_get_processor_id,
cpu::{smp_cpu_manager, ProcessorId},
},
time::{clocksource::HZ, sleep::nanosleep, timer::clock, PosixTimeSpec},
};
const YIELD_ROUNDS: usize = 16;
const EVENT_COUNT: usize = YIELD_ROUNDS * 2;
const EVENT_UNSET: usize = usize::MAX;
const FIFO_PRIO: i32 = MAX_RT_PRIO - 50;
const TEST_TIMEOUT_TICKS: u64 = 5 * HZ;
fn wait_until(mut predicate: impl FnMut() -> bool) -> bool {
let start = clock();
while clock().wrapping_sub(start) < TEST_TIMEOUT_TICKS {
if predicate() {
return true;
}
crate::sched::sched_yield();
}
predicate()
}
fn wait_completion(completion: &Completion) -> bool {
completion
.wait_for_completion_timeout(TEST_TIMEOUT_TICKS as i64)
.is_ok_and(|remaining| remaining > 0)
}
fn wait_off_rq(pcb: &Arc<ProcessControlBlock>) -> bool {
wait_until(|| {
!pcb.sched_info().is_running() && *pcb.sched_info().on_rq.lock_irqsave() == OnRq::None
})
}
fn reap_workers(workers: &[Arc<ProcessControlBlock>]) {
for worker in workers {
if !worker.sched_info().state().is_exited() {
let _ = ProcessManager::set_scheduler(
worker,
SchedChangeRequest::Normal {
reset_on_fork: false,
},
);
let _ = KernelThreadMechanism::request_stop(worker);
}
}
assert!(
wait_until(|| workers
.iter()
.all(|worker| worker.sched_info().state().is_exited())),
"fifo demo workers did not exit before the cleanup deadline"
);
// Reap every worker before asserting so one failure cannot leave later
// workers unreaped. A non-zero closure result represents a scenario
// failure (usually its own deadline), not successful cleanup.
let mut all_results_ok = true;
for worker in workers {
all_results_ok &= matches!(KernelThreadMechanism::stop(worker), Ok(0));
}
assert!(all_results_ok, "fifo demo worker reported failure");
}
struct FifoPairState {
first_start: Completion,
@@ -26,6 +84,7 @@ struct FifoPairState {
events: Vec<AtomicUsize>,
resumed: AtomicUsize,
results: [AtomicUsize; 2],
abort: AtomicBool,
}
impl FifoPairState {
@@ -39,6 +98,7 @@ impl FifoPairState {
.collect(),
resumed: AtomicUsize::new(0),
results: [AtomicUsize::new(EVENT_UNSET), AtomicUsize::new(EVENT_UNSET)],
abort: AtomicBool::new(false),
})
}
}
@@ -57,6 +117,10 @@ fn worker_closure(worker: usize, state: Arc<FifoPairState>) -> KernelThreadClosu
if result == 0 {
for _ in 0..YIELD_ROUNDS {
if state.abort.load(Ordering::Acquire) {
result = 4;
break;
}
let slot = state.next_event.fetch_add(1, Ordering::AcqRel);
if slot >= EVENT_COUNT {
result = 2;
@@ -95,15 +159,7 @@ fn create_fifo_worker(
let pcb = KernelThreadMechanism::create_on_cpu(worker_closure(worker, state), name, cpu)
.expect("fifo demo failed to create worker");
// create_on_cpu publishes the PCB before the child has necessarily
// finished its initial blocked switch-out. Wait for stable off-rq state
// before changing its scheduling policy.
pcb.sched_info().wait_until_not_running();
assert_eq!(
*pcb.sched_info().on_rq.lock_irqsave(),
OnRq::None,
"fifo demo worker did not become off-rq"
);
assert!(wait_off_rq(&pcb), "fifo demo worker did not become off-rq");
ProcessManager::set_fifo_policy(&pcb, FIFO_PRIO).expect("fifo demo failed to set FIFO policy");
pcb
}
@@ -116,12 +172,15 @@ fn run_fifo_pair(cpu: ProcessorId) {
ProcessManager::wakeup(&first).expect("fifo demo failed to wake first worker");
ProcessManager::wakeup(&second).expect("fifo demo failed to wake second worker");
for completion in &state.finished {
completion
.wait_for_completion()
.expect("fifo demo worker completion failed");
let completed = state.finished.iter().all(wait_completion);
if !completed {
state.abort.store(true, Ordering::Release);
state.first_start.complete_all();
}
reap_workers(&[first.clone(), second.clone()]);
assert!(completed, "fifo demo worker completion timed out");
assert_eq!(
state.next_event.load(Ordering::Acquire),
EVENT_COUNT,
@@ -141,32 +200,668 @@ fn run_fifo_pair(cpu: ProcessorId) {
"fifo demo worker did not resume after sleeping"
);
for (worker, pcb) in [first, second].iter().enumerate() {
for worker in 0..2 {
assert_eq!(
state.results[worker].load(Ordering::Acquire),
0,
"fifo demo worker reported failure"
);
assert_eq!(
KernelThreadMechanism::stop(pcb),
Ok(0),
"fifo demo worker exited with failure"
);
}
log::info!("fifo_demo status=ok cpu={}", cpu.data());
}
struct RemoteTransitionState {
runner_started: AtomicBool,
runner_release: AtomicBool,
candidate_started: AtomicBool,
candidate_release: Completion,
observer_started: AtomicBool,
abort: AtomicBool,
}
impl RemoteTransitionState {
fn new() -> Arc<Self> {
Arc::new(Self {
runner_started: AtomicBool::new(false),
runner_release: AtomicBool::new(false),
candidate_started: AtomicBool::new(false),
candidate_release: Completion::new(),
observer_started: AtomicBool::new(false),
abort: AtomicBool::new(false),
})
}
fn release_all(&self) {
self.abort.store(true, Ordering::Release);
self.runner_release.store(true, Ordering::Release);
self.candidate_release.complete_all();
}
}
fn task_is_current(pcb: &Arc<ProcessControlBlock>) -> bool {
let Some(cpu) = pcb.sched_info().on_cpu() else {
return false;
};
let rq = crate::sched::cpu_rq(cpu.data() as usize);
let (rq, _guard) = rq.self_lock();
Arc::ptr_eq(&rq.current(), pcb)
}
fn remote_runner_closure(state: Arc<RemoteTransitionState>) -> KernelThreadClosure {
KernelThreadClosure::EmptyClosure((
Box::new(move || {
state.runner_started.store(true, Ordering::Release);
let started = clock();
while !state.runner_release.load(Ordering::Acquire)
&& !state.abort.load(Ordering::Acquire)
&& clock().wrapping_sub(started) < TEST_TIMEOUT_TICKS
{
core::hint::spin_loop();
}
i32::from(
!state.runner_release.load(Ordering::Acquire)
&& !state.abort.load(Ordering::Acquire),
)
}),
(),
))
}
fn remote_candidate_closure(state: Arc<RemoteTransitionState>) -> KernelThreadClosure {
KernelThreadClosure::EmptyClosure((
Box::new(move || {
state.candidate_started.store(true, Ordering::Release);
i32::from(!wait_completion(&state.candidate_release))
}),
(),
))
}
fn remote_observer_closure(state: Arc<RemoteTransitionState>) -> KernelThreadClosure {
KernelThreadClosure::EmptyClosure((
Box::new(move || {
state.observer_started.store(true, Ordering::Release);
0
}),
(),
))
}
fn run_remote_class_transitions(cpu: ProcessorId) {
const RUNNER_PRIO: i32 = 20;
const OBSERVER_PRIO: i32 = 30;
const CANDIDATE_PRIO: i32 = 10;
let state = RemoteTransitionState::new();
let runner = KernelThreadMechanism::create_on_cpu(
remote_runner_closure(state.clone()),
"fifo_change_runner".into(),
cpu,
)
.expect("failed to create remote scheduler-change runner");
let candidate = KernelThreadMechanism::create_on_cpu(
remote_candidate_closure(state.clone()),
"fifo_change_candidate".into(),
cpu,
)
.expect("failed to create remote scheduler-change candidate");
let observer = KernelThreadMechanism::create_on_cpu(
remote_observer_closure(state.clone()),
"fifo_change_observer".into(),
cpu,
)
.expect("failed to create remote scheduler-change observer");
let workers = [runner.clone(), candidate.clone(), observer.clone()];
let mut ok = workers.iter().all(wait_off_rq);
ok &= ProcessManager::set_fifo_policy(&runner, RUNNER_PRIO).is_ok();
ok &= ProcessManager::set_fifo_policy(&observer, OBSERVER_PRIO).is_ok();
ok &= ProcessManager::wakeup(&runner).is_ok();
ok &= wait_until(|| state.runner_started.load(Ordering::Acquire));
ok &= ProcessManager::wakeup(&candidate).is_ok();
ok &= ProcessManager::wakeup(&observer).is_ok();
ok &= wait_until(|| {
*candidate.sched_info().on_rq.lock_irqsave() == OnRq::Queued
&& *observer.sched_info().on_rq.lock_irqsave() == OnRq::Queued
&& task_is_current(&runner)
});
let invalid_snapshot = (
observer.sched_info().policy(),
observer.sched_info().sched_class(),
observer.sched_info().prio(),
*observer.sched_info().on_rq.lock_irqsave(),
);
ok &= ProcessManager::set_scheduler(&observer, SchedChangeRequest::Fifo { priority: -1 })
== Err(system_error::SystemError::EINVAL);
ok &= invalid_snapshot
== (
observer.sched_info().policy(),
observer.sched_info().sched_class(),
observer.sched_info().prio(),
*observer.sched_info().on_rq.lock_irqsave(),
);
let ipi_before = crate::smp::kick_cpu_received(cpu);
ok &= ProcessManager::set_scheduler(
&candidate,
SchedChangeRequest::Fifo {
priority: CANDIDATE_PRIO,
},
)
.is_ok();
ok &= wait_until(|| state.candidate_started.load(Ordering::Acquire));
if crate::smp::kick_cpu_supported() {
ok &= crate::smp::kick_cpu_received(cpu) > ipi_before;
}
ok &= wait_until(|| candidate.sched_info().state().is_blocked());
ok &= ProcessManager::set_scheduler(
&candidate,
SchedChangeRequest::Normal {
reset_on_fork: false,
},
)
.is_ok();
ok &= candidate.sched_info().policy() == LinuxSchedPolicy::Normal
&& candidate.sched_info().sched_class() == SchedClass::Fair
&& candidate.sched_info().prio() == candidate.sched_info().static_prio();
ok &= wait_until(|| task_is_current(&runner));
ok &= ProcessManager::set_scheduler(
&runner,
SchedChangeRequest::Normal {
reset_on_fork: false,
},
)
.is_ok();
ok &= wait_until(|| state.observer_started.load(Ordering::Acquire));
state.runner_release.store(true, Ordering::Release);
state.candidate_release.complete_all();
let exited = wait_until(|| {
workers
.iter()
.all(|worker| worker.sched_info().state().is_exited())
});
if !exited {
state.release_all();
}
reap_workers(&workers);
assert!(ok && exited, "remote scheduler-change scenario failed");
log::info!("fifo_demo scheduler_change_remote=ok cpu={}", cpu.data());
}
struct PriorityOrderState {
runner_started: AtomicBool,
runner_release: AtomicBool,
target_started: AtomicBool,
target_release: AtomicBool,
peer_started: AtomicBool,
next_event: AtomicUsize,
events: [AtomicUsize; 2],
abort: AtomicBool,
}
impl PriorityOrderState {
fn new() -> Arc<Self> {
Arc::new(Self {
runner_started: AtomicBool::new(false),
runner_release: AtomicBool::new(false),
target_started: AtomicBool::new(false),
target_release: AtomicBool::new(false),
peer_started: AtomicBool::new(false),
next_event: AtomicUsize::new(0),
events: [AtomicUsize::new(EVENT_UNSET), AtomicUsize::new(EVENT_UNSET)],
abort: AtomicBool::new(false),
})
}
fn record(&self, worker: usize) -> i32 {
let slot = self.next_event.fetch_add(1, Ordering::AcqRel);
if slot >= self.events.len() {
return 1;
}
self.events[slot].store(worker, Ordering::Release);
0
}
fn release_all(&self) {
self.abort.store(true, Ordering::Release);
self.runner_release.store(true, Ordering::Release);
self.target_release.store(true, Ordering::Release);
}
}
fn gated_priority_worker(state: Arc<PriorityOrderState>, worker: usize) -> KernelThreadClosure {
KernelThreadClosure::EmptyClosure((
Box::new(move || {
match worker {
0 => state.runner_started.store(true, Ordering::Release),
1 => state.target_started.store(true, Ordering::Release),
_ => state.peer_started.store(true, Ordering::Release),
}
let mut result = if worker == 0 { 0 } else { state.record(worker) };
let release = if worker == 0 {
&state.runner_release
} else {
&state.target_release
};
if worker != 2 {
let started = clock();
while !release.load(Ordering::Acquire)
&& !state.abort.load(Ordering::Acquire)
&& clock().wrapping_sub(started) < TEST_TIMEOUT_TICKS
{
core::hint::spin_loop();
}
if !release.load(Ordering::Acquire) && !state.abort.load(Ordering::Acquire) {
result = 2;
}
}
result
}),
(),
))
}
fn run_priority_order_changes(cpu: ProcessorId) {
const RUNNER_PRIO: i32 = 10;
const TARGET_PRIO: i32 = 20;
const SHARED_PRIO: i32 = 30;
const LOWERED_PRIO: i32 = 40;
let state = PriorityOrderState::new();
let runner = KernelThreadMechanism::create_on_cpu(
gated_priority_worker(state.clone(), 0),
"fifo_head_runner".into(),
cpu,
)
.expect("failed to create head-order runner");
let target = KernelThreadMechanism::create_on_cpu(
gated_priority_worker(state.clone(), 1),
"fifo_head_target".into(),
cpu,
)
.expect("failed to create head-order target");
let peer = KernelThreadMechanism::create_on_cpu(
gated_priority_worker(state.clone(), 2),
"fifo_head_peer".into(),
cpu,
)
.expect("failed to create head-order peer");
let workers = [runner.clone(), target.clone(), peer.clone()];
let mut ok = workers.iter().all(wait_off_rq);
ok &= ProcessManager::set_fifo_policy(&runner, RUNNER_PRIO).is_ok();
ok &= ProcessManager::set_fifo_policy(&target, TARGET_PRIO).is_ok();
ok &= ProcessManager::set_fifo_policy(&peer, SHARED_PRIO).is_ok();
ok &= ProcessManager::wakeup(&runner).is_ok();
ok &= wait_until(|| state.runner_started.load(Ordering::Acquire));
ok &= ProcessManager::wakeup(&target).is_ok();
ok &= ProcessManager::wakeup(&peer).is_ok();
ok &= wait_until(|| {
*target.sched_info().on_rq.lock_irqsave() == OnRq::Queued
&& *peer.sched_info().on_rq.lock_irqsave() == OnRq::Queued
&& task_is_current(&runner)
});
let reset_before = target.sched_info().pi_lock_irqsave().sched_reset_on_fork();
let invalid_snapshot = (
target.sched_info().policy(),
target.sched_info().sched_class(),
target.sched_info().prio(),
*target.sched_info().on_rq.lock_irqsave(),
reset_before,
);
ok &= ProcessManager::set_scheduler(
&target,
SchedChangeRequest::Fifo {
priority: MAX_RT_PRIO,
},
) == Err(system_error::SystemError::EINVAL);
let reset_after = target.sched_info().pi_lock_irqsave().sched_reset_on_fork();
ok &= invalid_snapshot
== (
target.sched_info().policy(),
target.sched_info().sched_class(),
target.sched_info().prio(),
*target.sched_info().on_rq.lock_irqsave(),
reset_after,
);
// Lower target into peer's existing bucket. Linux places it at the head.
ok &= ProcessManager::set_scheduler(
&target,
SchedChangeRequest::Fifo {
priority: SHARED_PRIO,
},
)
.is_ok();
state.runner_release.store(true, Ordering::Release);
ok &= wait_until(|| state.target_started.load(Ordering::Acquire));
ok &= !state.peer_started.load(Ordering::Acquire);
// Target is current. Lower it below the queued peer and require preemption.
ok &= ProcessManager::set_scheduler(
&target,
SchedChangeRequest::Fifo {
priority: LOWERED_PRIO,
},
)
.is_ok();
ok &= wait_until(|| state.peer_started.load(Ordering::Acquire));
ok &= !state.target_release.load(Ordering::Acquire);
state.target_release.store(true, Ordering::Release);
let exited = wait_until(|| {
workers
.iter()
.all(|worker| worker.sched_info().state().is_exited())
});
if !exited {
state.release_all();
}
reap_workers(&workers);
ok &= state.next_event.load(Ordering::Acquire) == 2;
ok &= state.events[0].load(Ordering::Acquire) == 1;
ok &= state.events[1].load(Ordering::Acquire) == 2;
assert!(ok && exited, "FIFO priority-order scenario failed");
log::info!("fifo_demo scheduler_change_order=ok cpu={}", cpu.data());
}
const AFFINITY_RACE_ROUNDS: usize = 32;
struct AffinityRaceState {
start: AtomicBool,
abort: AtomicBool,
target_started: AtomicBool,
visited_cpus: AtomicUsize,
controller_done: Completion,
controller_result: AtomicUsize,
}
impl AffinityRaceState {
fn new() -> Arc<Self> {
Arc::new(Self {
start: AtomicBool::new(false),
abort: AtomicBool::new(false),
target_started: AtomicBool::new(false),
visited_cpus: AtomicUsize::new(0),
controller_done: Completion::new(),
controller_result: AtomicUsize::new(EVENT_UNSET),
})
}
}
fn affinity_target_closure(state: Arc<AffinityRaceState>) -> KernelThreadClosure {
KernelThreadClosure::EmptyClosure((
Box::new(move || {
state.target_started.store(true, Ordering::Release);
let mut race_started = None;
while !state.abort.load(Ordering::Acquire) {
if state.start.load(Ordering::Acquire) {
let started = *race_started.get_or_insert_with(clock);
if clock().wrapping_sub(started) >= TEST_TIMEOUT_TICKS {
break;
}
}
let cpu = smp_get_processor_id().data() as usize;
if cpu < usize::BITS as usize {
state.visited_cpus.fetch_or(1usize << cpu, Ordering::AcqRel);
}
core::hint::spin_loop();
}
i32::from(!state.abort.load(Ordering::Acquire))
}),
(),
))
}
fn affinity_controller_closure(
state: Arc<AffinityRaceState>,
target: Arc<ProcessControlBlock>,
masks: [CpuMask; 2],
) -> KernelThreadClosure {
KernelThreadClosure::EmptyClosure((
Box::new(move || {
let started = clock();
while !state.start.load(Ordering::Acquire)
&& !state.abort.load(Ordering::Acquire)
&& clock().wrapping_sub(started) < TEST_TIMEOUT_TICKS
{
core::hint::spin_loop();
}
let mut result = usize::from(!state.start.load(Ordering::Acquire));
if result == 0 {
for round in 0..AFFINITY_RACE_ROUNDS {
if ProcessManager::set_cpus_allowed(&target, masks[round & 1].clone()).is_err()
{
result = 2;
break;
}
crate::sched::sched_yield();
}
}
state.controller_result.store(result, Ordering::Release);
state.controller_done.complete();
result as i32
}),
(),
))
}
fn run_policy_affinity_race(control_cpu: ProcessorId, remote_cpus: &[ProcessorId]) {
let first_remote = remote_cpus[0];
let state = AffinityRaceState::new();
let target = KernelThreadMechanism::create(
affinity_target_closure(state.clone()),
"fifo_affinity_target".into(),
)
.expect("failed to create affinity race target");
let initial_mask = CpuMask::from_cpu(first_remote);
let mut ok = wait_off_rq(&target);
ok &= ProcessManager::set_cpus_allowed(&target, initial_mask.clone()).is_ok();
let masks = if remote_cpus.len() >= 2 {
[
CpuMask::from_cpu(remote_cpus[0]),
CpuMask::from_cpu(remote_cpus[1]),
]
} else {
let mut broad = CpuMask::from_cpu(first_remote);
broad.set(control_cpu, true);
[initial_mask.clone(), broad]
};
let controller = KernelThreadMechanism::create_on_cpu(
affinity_controller_closure(state.clone(), target.clone(), masks),
"fifo_affinity_controller".into(),
control_cpu,
)
.expect("failed to create affinity race controller");
let workers = [target.clone(), controller.clone()];
ok &= wait_off_rq(&controller);
ok &= ProcessManager::wakeup(&target).is_ok();
ok &= wait_until(|| state.target_started.load(Ordering::Acquire));
// Deterministically cover both Fair and FIFO migrated enqueue before the
// policy/affinity race. The concurrent rounds below then stress the same
// transactions without relying on their relative scheduling order for
// basic migration coverage.
if remote_cpus.len() >= 2 {
let second_remote = remote_cpus[1];
assert!(
ProcessManager::set_cpus_allowed(&target, CpuMask::from_cpu(second_remote)).is_ok(),
"Fair migration preflight affinity update failed"
);
assert!(
wait_until(|| target.sched_info().on_cpu() == Some(second_remote)),
"Fair migration preflight did not reach the destination CPU"
);
assert!(
ProcessManager::set_scheduler(&target, SchedChangeRequest::Fifo { priority: 50 })
.is_ok(),
"FIFO migration preflight policy update failed"
);
assert!(
ProcessManager::set_cpus_allowed(&target, initial_mask.clone()).is_ok(),
"FIFO migration preflight affinity update failed"
);
assert!(
wait_until(|| target.sched_info().on_cpu() == Some(first_remote)),
"FIFO migration preflight did not reach the destination CPU"
);
assert!(
ProcessManager::set_scheduler(
&target,
SchedChangeRequest::Normal {
reset_on_fork: false,
},
)
.is_ok(),
"migration preflight failed to restore the Fair policy"
);
}
ok &= ProcessManager::wakeup(&controller).is_ok();
state.start.store(true, Ordering::Release);
for round in 0..AFFINITY_RACE_ROUNDS {
let request = if round & 1 == 0 {
SchedChangeRequest::Fifo { priority: 50 }
} else {
SchedChangeRequest::Normal {
reset_on_fork: false,
}
};
if ProcessManager::set_scheduler(&target, request).is_err() {
ok = false;
break;
}
crate::sched::sched_yield();
}
ok &= wait_completion(&state.controller_done);
ok &= state.controller_result.load(Ordering::Acquire) == 0;
ok &= ProcessManager::set_cpus_allowed(&target, initial_mask).is_ok();
ok &= wait_until(|| target.sched_info().on_cpu() == Some(first_remote));
ok &= ProcessManager::set_scheduler(
&target,
SchedChangeRequest::Normal {
reset_on_fork: false,
},
)
.is_ok();
state.abort.store(true, Ordering::Release);
let exited = wait_until(|| {
workers
.iter()
.all(|worker| worker.sched_info().state().is_exited())
});
reap_workers(&workers);
let visited = state.visited_cpus.load(Ordering::Acquire);
ok &= visited & (1usize << first_remote.data()) != 0;
if remote_cpus.len() >= 2 {
ok &= visited & (1usize << remote_cpus[1].data()) != 0;
}
assert!(ok && exited, "policy/affinity race scenario failed");
let mode = if remote_cpus.len() >= 2 {
"migration"
} else {
"publication"
};
log::info!(
"fifo_demo scheduler_change_affinity=ok mode={} remote_cpus={}",
mode,
remote_cpus.len()
);
}
fn run_policy_exit_race(cpu: ProcessorId) {
const EXIT_RACE_ROUNDS: usize = 8;
let mut ok = true;
let mut observed_live_exit = false;
for round in 0..EXIT_RACE_ROUNDS {
let worker = KernelThreadMechanism::create_on_cpu(
KernelThreadClosure::EmptyClosure((Box::new(|| 0), ())),
alloc::format!("fifo_exit_race_{round}"),
cpu,
)
.expect("failed to create exit race worker");
ok &= wait_off_rq(&worker);
ok &= ProcessManager::wakeup(&worker).is_ok();
ok &= wait_until(|| worker.flags().contains(ProcessFlags::EXITING));
// EXITING is persistent. Require evidence that at least one
// transaction was initiated before the final Exited state, rather
// than accepting a sequence of post-exit lifecycle updates as a race.
if !worker.sched_info().state().is_exited() {
observed_live_exit = true;
}
// Once EXITING is visible, race several complete transactions against
// the remaining teardown and the final Exited -> schedule transition.
for change in 0..4 {
let request = if change & 1 == 0 {
SchedChangeRequest::Fifo { priority: 45 }
} else {
SchedChangeRequest::Normal {
reset_on_fork: false,
}
};
ok &= ProcessManager::set_scheduler(&worker, request).is_ok();
}
ok &= ProcessManager::set_scheduler(
&worker,
SchedChangeRequest::Normal {
reset_on_fork: false,
},
)
.is_ok();
ok &= wait_until(|| worker.sched_info().state().is_exited());
reap_workers(&[worker]);
}
assert!(
ok && observed_live_exit,
"policy/exit race scenario did not overlap live teardown"
);
log::info!("fifo_demo scheduler_change_exit=ok cpu={}", cpu.data());
}
pub fn fifo_demo_init() {
let cpus: Vec<ProcessorId> = smp_cpu_manager()
.present_cpus()
.iter_cpu()
.filter(|&cpu| smp_cpu_manager().is_online_cpu(cpu))
.take(2)
.collect();
assert!(!cpus.is_empty(), "fifo demo found no online CPU");
for cpu in cpus {
for &cpu in cpus.iter().take(2) {
run_fifo_pair(cpu);
}
let control_cpu = smp_get_processor_id();
let remote_cpus: Vec<ProcessorId> = cpus
.iter()
.copied()
.filter(|&cpu| cpu != control_cpu)
.take(2)
.collect();
if let Some(&remote_cpu) = remote_cpus.first() {
run_remote_class_transitions(remote_cpu);
run_priority_order_changes(remote_cpu);
run_policy_affinity_race(control_cpu, &remote_cpus);
run_policy_exit_race(remote_cpu);
} else {
log::info!("fifo_demo scheduler_change_remote=skip reason=no_remote_cpu");
}
}
kernel/src/sched/mod.rs
+94 -20
View File
@@ -62,7 +62,7 @@ use self::{
pub use policy::{LinuxSchedPolicy, SchedClass};
static mut CPU_IRQ_TIME: Option<Vec<&'static mut IrqTime>> = None;
static mut CPU_IRQ_TIME: Option<Vec<&'static IrqTime>> = None;
pub static IDLE_CPUS: AtomicCpuMask = AtomicCpuMask::new();
// 这里虽然rq是percpu的，但是在负载均衡的时候需要修改对端cpu的rq，所以仍需加锁
@@ -77,8 +77,8 @@ pub const SCHED_CAPACITY_SHIFT: u64 = SCHED_FIXEDPOINT_SHIFT;
pub const SCHED_CAPACITY_SCALE: u64 = 1 << SCHED_CAPACITY_SHIFT;
#[inline]
pub fn cpu_irq_time(cpu: ProcessorId) -> &'static mut IrqTime {
unsafe { CPU_IRQ_TIME.as_mut().unwrap()[cpu.data() as usize] }
pub(crate) fn cpu_irq_time(cpu: ProcessorId) -> &'static IrqTime {
unsafe { CPU_IRQ_TIME.as_ref().unwrap()[cpu.data() as usize] }
}
#[inline]
@@ -623,6 +623,70 @@ impl CpuRunQueue {
}
}
pub(crate) fn put_prev_task_for_class(
&mut self,
class: SchedClass,
pcb: Arc<ProcessControlBlock>,
) {
match class {
SchedClass::Realtime => RealtimeScheduler::put_prev_task(self, pcb),
SchedClass::Fair => CompletelyFairScheduler::put_prev_task(self, pcb),
SchedClass::Idle => IdleScheduler::put_prev_task(self, pcb),
}
}
pub(crate) fn set_next_task_for_class(
&mut self,
class: SchedClass,
pcb: Arc<ProcessControlBlock>,
) {
match class {
SchedClass::Realtime => RealtimeScheduler::set_next_task(self, pcb),
SchedClass::Fair => CompletelyFairScheduler::set_next_task(self, pcb),
SchedClass::Idle => unreachable!("scheduler changes cannot target the idle class"),
}
}
/// Apply the reachable subset of Linux `check_class_changed()` semantics.
pub(crate) fn check_scheduler_changed(
&mut self,
pcb: &Arc<ProcessControlBlock>,
old_class: SchedClass,
old_prio: i32,
) {
if *pcb.sched_info().on_rq.lock_irqsave() != OnRq::Queued {
return;
}
let new_class = pcb.sched_info().sched_class();
let new_prio = pcb.sched_info().prio();
if old_class == new_class && old_prio == new_prio {
return;
}
if !Arc::ptr_eq(&self.current(), pcb) {
self.check_preempt_current(pcb, WakeupFlags::empty());
return;
}
if old_class != new_class {
if old_class == SchedClass::Realtime && new_class == SchedClass::Fair {
self.resched_current();
}
return;
}
if new_class == SchedClass::Realtime
&& new_prio > old_prio
&& self
.rt
.highest_prio()
.is_some_and(|highest| highest < new_prio as usize)
{
self.resched_current();
}
}
/// 将任务加入运行队列，设置 on_rq = Queued。
pub fn activate_task(&mut self, pcb: &Arc<ProcessControlBlock>, mut flags: EnqueueFlag) {
// 1. 迁移标志处理
@@ -630,10 +694,6 @@ impl CpuRunQueue {
flags |= EnqueueFlag::ENQUEUE_MIGRATED;
}
if flags.contains(EnqueueFlag::ENQUEUE_MIGRATED) {
todo!()
}
// 2. enqueue_task
self.enqueue_task(pcb.clone(), flags);
@@ -950,6 +1010,9 @@ bitflags! {
const ENQUEUE_MOVE = 0x04;
const ENQUEUE_NOCLOCK = 0x08;
/// Place a realtime task at the head of its priority bucket.
const ENQUEUE_HEAD = 0x10;
const ENQUEUE_MIGRATED = 0x40;
const ENQUEUE_INITIAL = 0x80;
@@ -1163,6 +1226,10 @@ fn __schedule_inner(sched_mod: SchedMode, current: Option<Arc<ProcessControlBloc
DequeueFlag::DEQUEUE_SLEEP | DequeueFlag::DEQUEUE_NOCLOCK,
);
if prev_state.is_exited() && prev.sched_info().sched_class() == SchedClass::Fair {
CompletelyFairScheduler::task_dead_fair(rq, &prev);
}
// nr_iowait++ happens after deactivate_task.
if prev_state.is_blocked() && prev.flags().contains(ProcessFlags::IN_IOWAIT) {
rq.nr_iowait.fetch_add(1, Ordering::Relaxed);
@@ -1206,7 +1273,11 @@ fn __schedule_inner(sched_mod: SchedMode, current: Option<Arc<ProcessControlBloc
});
}
*prev.sched_info().on_rq.lock_irqsave() = OnRq::None;
// Keep task_cpu unstable until the switch tail owns pi_lock and
// commits the destination enqueue. External task_rq_lock-style
// operations must not mutate scheduler state against the old rq
// during this lockless migration interval.
debug_assert_eq!(*prev.sched_info().on_rq.lock_irqsave(), OnRq::Migrating);
migrate_prev_to = Some(dest_cpu);
}
}
@@ -1340,6 +1411,14 @@ fn __set_task_cpu(pcb: &Arc<ProcessControlBlock>, cpu: ProcessorId) {
cpu
);
let on_rq = *pcb.sched_info().on_rq.lock_irqsave();
let old_cpu = pcb.sched_info().on_cpu();
if pcb.sched_info().sched_class() == SchedClass::Fair
&& (on_rq == OnRq::Migrating || old_cpu.is_some_and(|old_cpu| old_cpu != cpu))
{
CompletelyFairScheduler::prepare_task_rq_migration(pcb);
}
// TODO: Fixme There is not implement group sched;
let se = pcb.sched_info().sched_entity();
let rq = cpu_rq(cpu.data() as usize);
@@ -1399,12 +1478,9 @@ pub fn request_task_migration(
}
let rq = cpu_rq(src_cpu.data() as usize);
let update_clock = src_cpu == smp_get_processor_id();
let (rq, _guard) = rq.self_lock();
if update_clock {
rq.update_rq_clock();
}
rq.update_rq_clock();
if Arc::ptr_eq(&rq.current(), pcb) {
pcb.sched_info().set_migrate_to(Some(dest_cpu));
@@ -1419,13 +1495,11 @@ pub fn request_task_migration(
}
if *pcb.sched_info().on_rq.lock_irqsave() == OnRq::Queued {
rq.dequeue_task(
rq.deactivate_task(
pcb.clone(),
DequeueFlag::DEQUEUE_MOVE | DequeueFlag::DEQUEUE_NOCLOCK,
);
crate::process::rseq::Rseq::on_migrate(pcb);
*pcb.sched_info().on_rq.lock_irqsave() = OnRq::None;
pcb.sched_info().set_on_cpu(None);
drop(_guard);
pcb.sched_info().wait_until_not_running();
@@ -1455,11 +1529,11 @@ pub fn take_current_migration_target(current: &Arc<ProcessControlBlock>) -> Opti
pub fn sched_init() {
// 初始化percpu变量
unsafe {
CPU_IRQ_TIME = Some(Vec::with_capacity(PerCpu::MAX_CPU_NUM as usize));
CPU_IRQ_TIME
.as_mut()
.unwrap()
.resize_with(PerCpu::MAX_CPU_NUM as usize, || Box::leak(Box::default()));
let mut irq_times = Vec::with_capacity(PerCpu::MAX_CPU_NUM as usize);
for cpu in 0..PerCpu::MAX_CPU_NUM {
irq_times.push(Box::leak(Box::new(IrqTime::new(ProcessorId::new(cpu)))) as &'static _);
}
CPU_IRQ_TIME = Some(irq_times);
let mut cpu_runqueue = Vec::with_capacity(PerCpu::MAX_CPU_NUM as usize);
for cpu in 0..PerCpu::MAX_CPU_NUM as usize {
kernel/src/sched/realtime.rs
+15 -2
View File
@@ -109,6 +109,15 @@ impl RealtimeRunQueue {
self.assert_consistent();
}
pub fn enqueue_head(&mut self, pcb: Arc<ProcessControlBlock>) {
let prio = Self::prio_index(&pcb);
self.assert_not_queued(&pcb);
self.queues[prio].push_front(pcb);
self.set_active(prio);
self.nr_running += 1;
self.assert_consistent();
}
pub fn dequeue(&mut self, pcb: &Arc<ProcessControlBlock>) -> bool {
self.assert_consistent();
let prio = Self::prio_index(pcb);
@@ -188,8 +197,12 @@ impl RealtimeScheduler {
}
impl Scheduler for RealtimeScheduler {
fn enqueue(rq: &mut CpuRunQueue, pcb: Arc<ProcessControlBlock>, _flags: EnqueueFlag) {
rq.rt.enqueue_tail(pcb);
fn enqueue(rq: &mut CpuRunQueue, pcb: Arc<ProcessControlBlock>, flags: EnqueueFlag) {
if flags.contains(EnqueueFlag::ENQUEUE_HEAD) {
rq.rt.enqueue_head(pcb);
} else {
rq.rt.enqueue_tail(pcb);
}
rq.add_nr_running(1);
}
kernel/src/smp/syscall/sys_sched_setaffinity.rs
+3 -32
View File
@@ -7,9 +7,7 @@ use crate::arch::interrupt::TrapFrame;
use crate::arch::syscall::nr::SYS_SCHED_SETAFFINITY;
use crate::libs::cpumask::CpuMask;
use crate::process::{kthread::KernelThreadFlags, ProcessFlags, ProcessManager, RawPid};
use crate::sched::{
request_task_migration, select_task_rq, syscall::util::has_sched_setaffinity_permission,
};
use crate::sched::syscall::util::has_sched_setaffinity_permission;
use crate::smp::cpu::smp_cpu_manager;
use crate::syscall::table::{FormattedSyscallParam, Syscall};
use crate::syscall::user_access::UserBufferReader;
@@ -69,35 +67,8 @@ impl Syscall for SysSchedSetaffinity {
}
// Keep affinity publication and the corresponding placement decision
// in one pi_lock critical section. Otherwise two concurrent callers
// can publish masks in one order but execute their migrations in the
// opposite order.
let mut pi_guard = target_pcb.sched_info().pi_lock_irqsave();
pi_guard.set_cpus_allowed(mask.clone());
if target_pcb.sched_info().is_new_task() {
return Ok(0);
}
// A previous affinity request may have left a migration for a running
// task to be consumed in schedule tail. Do not let a newer mask retain
// an now-illegal destination.
if target_pcb
.sched_info()
.migrate_to()
.is_some_and(|cpu| !mask.get(cpu).unwrap_or(false))
{
target_pcb.sched_info().set_migrate_to(None);
target_pcb.flags().remove(ProcessFlags::NEED_MIGRATE);
}
if let Some(cpu) = target_pcb.sched_info().on_cpu() {
if !mask.get(cpu).unwrap_or(false) {
let dest_cpu =
select_task_rq(&target_pcb, cpu, crate::sched::WakeupFlags::WF_TTWU, &mask);
request_task_migration(&target_pcb, dest_cpu)?;
}
}
// in one pi_lock critical section inside the process manager.
ProcessManager::set_cpus_allowed(&target_pcb, mask)?;
Ok(0)
}