feat(sched): enforce realtime runtime bandwidth (#2248)

Add per-CPU realtime bandwidth accounting with Linux-compatible default period and runtime limits. Track elapsed RT execution under the runqueue lock, throttle exhausted RT queues, and restore their eligibility after period replenishment.

Integrate bandwidth updates into task selection, scheduler ticks, RT enqueue transitions, and class switch lifecycle. Keep throttled tasks queued locally while excluding them from the top-level runnable count and preemption decisions.

Extend the FIFO scheduler demo with a deterministic Fair observer scenario that verifies both throttling and tick-driven RT recovery on each exercised CPU.

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

kernel/src/sched/fifo_demo.rs

@@ -211,6 +211,110 @@ fn run_fifo_pair(cpu: ProcessorId) {
     log::info!("fifo_demo status=ok cpu={}", cpu.data());
 }
 
+struct BandwidthState {
+    runner_started: AtomicBool,
+    observer_ran: AtomicBool,
+    runner_resumed: AtomicBool,
+    abort: AtomicBool,
+}
+
+impl BandwidthState {
+    fn new() -> Arc<Self> {
+        Arc::new(Self {
+            runner_started: AtomicBool::new(false),
+            observer_ran: AtomicBool::new(false),
+            runner_resumed: AtomicBool::new(false),
+            abort: AtomicBool::new(false),
+        })
+    }
+}
+
+fn bandwidth_runner_closure(state: Arc<BandwidthState>) -> KernelThreadClosure {
+    KernelThreadClosure::EmptyClosure((
+        Box::new(move || {
+            state.runner_started.store(true, Ordering::Release);
+            let started = clock();
+            while !state.observer_ran.load(Ordering::Acquire)
+                && !state.abort.load(Ordering::Acquire)
+                && clock().wrapping_sub(started) < TEST_TIMEOUT_TICKS
+            {
+                core::hint::spin_loop();
+            }
+
+            if state.observer_ran.load(Ordering::Acquire) {
+                state.runner_resumed.store(true, Ordering::Release);
+                0
+            } else {
+                1
+            }
+        }),
+        (),
+    ))
+}
+
+fn bandwidth_observer_closure(state: Arc<BandwidthState>) -> KernelThreadClosure {
+    KernelThreadClosure::EmptyClosure((
+        Box::new(move || {
+            state.observer_ran.store(true, Ordering::Release);
+            let started = clock();
+            while !state.runner_resumed.load(Ordering::Acquire)
+                && !state.abort.load(Ordering::Acquire)
+                && clock().wrapping_sub(started) < TEST_TIMEOUT_TICKS
+            {
+                // Remain runnable until the FIFO runner becomes eligible
+                // again. This prevents observer exit from being mistaken for
+                // period replenishment, and avoids advancing the period from
+                // a voluntary schedule path so the tick hook is exercised.
+                core::hint::spin_loop();
+            }
+            i32::from(
+                !state.runner_resumed.load(Ordering::Acquire)
+                    && !state.abort.load(Ordering::Acquire),
+            )
+        }),
+        (),
+    ))
+}
+
+fn run_runtime_throttling(cpu: ProcessorId) {
+    let state = BandwidthState::new();
+    let runner = KernelThreadMechanism::create_on_cpu(
+        bandwidth_runner_closure(state.clone()),
+        "fifo_bandwidth_runner".into(),
+        cpu,
+    )
+    .expect("failed to create bandwidth runner");
+    let observer = KernelThreadMechanism::create_on_cpu(
+        bandwidth_observer_closure(state.clone()),
+        "fifo_bandwidth_observer".into(),
+        cpu,
+    )
+    .expect("failed to create bandwidth observer");
+    let workers = [runner.clone(), observer.clone()];
+
+    let mut ok = workers.iter().all(wait_off_rq);
+    ok &= ProcessManager::set_fifo_policy(&runner, FIFO_PRIO).is_ok();
+    ok &= ProcessManager::wakeup(&runner).is_ok();
+    ok &= wait_until(|| state.runner_started.load(Ordering::Acquire));
+
+    // The runner never yields, sleeps, or exits before this observer runs.
+    // Therefore the Fair observer can run only after the local RT class is
+    // throttled and skipped by task selection.
+    ok &= ProcessManager::wakeup(&observer).is_ok();
+    ok &= wait_until(|| state.observer_ran.load(Ordering::Acquire));
+
+    // The FIFO runner remains queued while throttled. Seeing it resume proves
+    // that period replenishment made the RT class eligible again.
+    ok &= wait_until(|| state.runner_resumed.load(Ordering::Acquire));
+    if !ok {
+        state.abort.store(true, Ordering::Release);
+    }
+
+    reap_workers(&workers);
+    assert!(ok, "realtime runtime throttling scenario failed");
+    log::info!("fifo_demo runtime_throttling=ok cpu={}", cpu.data());
+}
+
 struct RemoteTransitionState {
     runner_started: AtomicBool,
     runner_release: AtomicBool,
@@ -847,6 +951,7 @@ pub fn fifo_demo_init() {
 
     for &cpu in cpus.iter().take(2) {
         run_fifo_pair(cpu);
+        run_runtime_throttling(cpu);
     }
 
     let control_cpu = smp_get_processor_id();

kernel/src/sched/mod.rs

@@ -706,6 +706,10 @@ impl CpuRunQueue {
         let current_class = self.current().sched_info().sched_class();
         let waking_class = pcb.sched_info().sched_class();
 
+        if waking_class == SchedClass::Realtime && self.rt.is_throttled() {
+            return;
+        }
+
         if waking_class == current_class {
             match current_class {
                 SchedClass::Fair => {
@@ -744,6 +748,10 @@ impl CpuRunQueue {
         let current_class = current.sched_info().sched_class();
         let next_class = pcb.sched_info().sched_class();
 
+        if next_class == SchedClass::Realtime && self.rt.is_throttled() {
+            return;
+        }
+
         if current.flags().contains(ProcessFlags::NEED_SCHEDULE) {
             if current_class == SchedClass::Idle {
                 self.resched_current();
@@ -931,6 +939,12 @@ impl CpuRunQueue {
     pub fn pick_next_task(&mut self, prev: Arc<ProcessControlBlock>) -> Arc<ProcessControlBlock> {
         debug_assert_eq!(prev.sched_info().on_cpu(), Some(self.cpu));
 
+        // RT runtime must be current before class selection. DragonOS picks
+        // `next` before calling put_prev_task(), so accounting only there can
+        // select another RT task from a runqueue which has already exhausted
+        // its bandwidth.
+        RealtimeScheduler::update_bandwidth(self, prev.sched_info().sched_class());
+
         let mut next: Option<Arc<ProcessControlBlock>> = None;
 
         if self.rt.nr_running() > 0 {
@@ -945,6 +959,7 @@ impl CpuRunQueue {
             && !task_is_idle(&prev)
             && prev.sched_info().state().is_runnable()
             && *prev.sched_info().on_rq.lock_irqsave() == OnRq::Queued
+            && !(prev.sched_info().sched_class() == SchedClass::Realtime && self.rt.is_throttled())
         {
             next = Some(prev.clone());
         }
@@ -958,8 +973,10 @@ impl CpuRunQueue {
                 SchedClass::Idle => IdleScheduler::put_prev_task(self, prev),
             }
 
-            if next.sched_info().sched_class() == SchedClass::Fair {
-                CompletelyFairScheduler::set_next_task(self, next.clone());
+            match next.sched_info().sched_class() {
+                SchedClass::Realtime => RealtimeScheduler::set_next_task(self, next.clone()),
+                SchedClass::Fair => CompletelyFairScheduler::set_next_task(self, next.clone()),
+                SchedClass::Idle => {}
             }
         }
 
@@ -1093,7 +1110,9 @@ pub fn scheduler_tick() {
 
     // 更新请求队列时钟
     rq.update_rq_clock();
-    match current.sched_info().sched_class() {
+    let current_class = current.sched_info().sched_class();
+    RealtimeScheduler::update_bandwidth(rq, current_class);
+    match current_class {
         SchedClass::Realtime => RealtimeScheduler::tick(rq, current, false),
         SchedClass::Fair => CompletelyFairScheduler::tick(rq, current, false),
         SchedClass::Idle => IdleScheduler::tick(rq, current, false),

kernel/src/sched/realtime.rs

@@ -4,6 +4,11 @@ use crate::{process::ProcessControlBlock, sched::prio::MAX_RT_PRIO};
 
 use super::{CpuRunQueue, DequeueFlag, EnqueueFlag, PrioUtil, SchedClass, Scheduler, WakeupFlags};
 
+/// Match Linux's default root realtime bandwidth: reserve 5% of each CPU for
+/// non-RT work in every one-second period.
+const RT_PERIOD_NS: u64 = 1_000_000_000;
+const RT_RUNTIME_NS: u64 = 950_000_000;
+
 const _: () = {
     assert!(MAX_RT_PRIO > 1);
     assert!(MAX_RT_PRIO <= u128::BITS as i32);
@@ -18,6 +23,10 @@ pub struct RealtimeRunQueue {
     queues: Vec<VecDeque<Arc<ProcessControlBlock>>>,
     active: u128,
     nr_running: usize,
+    runtime_used: u64,
+    period_start: Option<u64>,
+    exec_start: Option<u64>,
+    throttled: bool,
 }
 
 impl RealtimeRunQueue {
@@ -28,6 +37,10 @@ impl RealtimeRunQueue {
             queues,
             active: 0,
             nr_running: 0,
+            runtime_used: 0,
+            period_start: None,
+            exec_start: None,
+            throttled: false,
         }
     }
 
@@ -36,6 +49,67 @@ impl RealtimeRunQueue {
         self.nr_running
     }
 
+    #[inline]
+    pub fn is_throttled(&self) -> bool {
+        self.throttled
+    }
+
+    /// Account the current RT task before advancing the bandwidth period.
+    ///
+    /// This order preserves runtime debt when a non-preemptible kernel path
+    /// crosses one or more period boundaries before reaching the scheduler.
+    fn update_current(&mut self, clock_task: u64, clock: u64) {
+        if let Some(exec_start) = self.exec_start {
+            self.runtime_used = self
+                .runtime_used
+                .saturating_add(clock_task.saturating_sub(exec_start));
+            self.exec_start = Some(clock_task);
+        }
+        self.update_period(clock);
+    }
+
+    fn update_period(&mut self, clock: u64) {
+        let Some(period_start) = self.period_start.as_mut() else {
+            self.period_start = Some(clock - clock % RT_PERIOD_NS);
+            self.update_throttled();
+            return;
+        };
+        let elapsed = clock.saturating_sub(*period_start);
+
+        if elapsed >= RT_PERIOD_NS {
+            let elapsed_periods = elapsed / RT_PERIOD_NS;
+            *period_start =
+                period_start.saturating_add(elapsed_periods.saturating_mul(RT_PERIOD_NS));
+            self.runtime_used = self
+                .runtime_used
+                .saturating_sub(elapsed_periods.saturating_mul(RT_RUNTIME_NS));
+        }
+
+        self.update_throttled();
+    }
+
+    fn update_throttled(&mut self) {
+        // Linux uses hysteresis at the exact runtime boundary: `>` enters
+        // throttling, while an already-throttled rq needs `<` to leave it.
+        if self.throttled {
+            if self.runtime_used < RT_RUNTIME_NS {
+                self.throttled = false;
+            }
+        } else if self.runtime_used > RT_RUNTIME_NS {
+            self.throttled = true;
+        }
+    }
+
+    #[inline]
+    fn set_current(&mut self, clock_task: u64) {
+        self.exec_start = Some(clock_task);
+    }
+
+    #[inline]
+    fn clear_current(&mut self) {
+        self.exec_start = None;
+    }
+
     #[inline]
     fn prio_index(pcb: &ProcessControlBlock) -> usize {
         let prio = pcb.sched_info().prio();
@@ -185,29 +259,61 @@ impl RealtimeScheduler {
         pcb.sched_info().prio()
     }
 
-    /// Set a realtime task as the current task for its class.
-    ///
-    /// DragonOS does not yet maintain a separate current RT entity, so there
-    /// is no class-local state to update here.
+    /// Bring the local RT bandwidth state up to date under the rq lock.
+    pub fn update_bandwidth(rq: &mut CpuRunQueue, current_class: SchedClass) {
+        let was_throttled = rq.rt.is_throttled();
+        if current_class == SchedClass::Realtime {
+            rq.rt.update_current(rq.clock_task, rq.clock);
+        } else {
+            rq.rt.update_period(rq.clock);
+        }
+
+        if !was_throttled && rq.rt.is_throttled() {
+            // Match Linux's dequeue_top_rt_rq(): throttled RT tasks stay on
+            // their class queue but no longer contribute to the top-level rq.
+            let nr_running = rq.rt.nr_running();
+            rq.sub_nr_running(nr_running);
+            rq.resched_current();
+        } else if was_throttled && !rq.rt.is_throttled() {
+            let nr_running = rq.rt.nr_running();
+            rq.add_nr_running(nr_running);
+            if nr_running > 0 {
+                rq.resched_current();
+            }
+        }
+    }
+
+    /// Start charging the selected realtime task on this runqueue.
     pub fn set_next_task(
-        _rq: &mut super::CpuRunQueue,
+        rq: &mut super::CpuRunQueue,
         _pcb: alloc::sync::Arc<crate::process::ProcessControlBlock>,
     ) {
+        rq.rt.set_current(rq.clock_task);
+        if rq.rt.is_throttled() {
+            // This is reachable when a running Fair task is changed to RT on
+            // an already-throttled rq. Charge its bounded tail, then switch it
+            // out instead of letting the policy transaction bypass bandwidth.
+            rq.resched_current();
+        }
     }
 }
 
 impl Scheduler for RealtimeScheduler {
     fn enqueue(rq: &mut CpuRunQueue, pcb: Arc<ProcessControlBlock>, flags: EnqueueFlag) {
+        let current_class = rq.current().sched_info().sched_class();
+        Self::update_bandwidth(rq, current_class);
         if flags.contains(EnqueueFlag::ENQUEUE_HEAD) {
             rq.rt.enqueue_head(pcb);
         } else {
             rq.rt.enqueue_tail(pcb);
         }
+        if !rq.rt.is_throttled() {
             rq.add_nr_running(1);
         }
+    }
 
     fn dequeue(rq: &mut CpuRunQueue, pcb: Arc<ProcessControlBlock>, _flags: DequeueFlag) {
-        if rq.rt.dequeue(&pcb) {
+        if rq.rt.dequeue(&pcb) && !rq.rt.is_throttled() {
             rq.sub_nr_running(1);
         }
     }
@@ -243,6 +349,9 @@ impl Scheduler for RealtimeScheduler {
         rq: &mut CpuRunQueue,
         _pcb: Option<Arc<ProcessControlBlock>>,
     ) -> Option<Arc<ProcessControlBlock>> {
+        if rq.rt.is_throttled() {
+            return None;
+        }
         rq.rt.pick_next()
     }
 
@@ -261,5 +370,8 @@ impl Scheduler for RealtimeScheduler {
 
     fn task_fork(_pcb: Arc<ProcessControlBlock>) {}
 
-    fn put_prev_task(_rq: &mut CpuRunQueue, _prev: Arc<ProcessControlBlock>) {}
+    fn put_prev_task(rq: &mut CpuRunQueue, _prev: Arc<ProcessControlBlock>) {
+        Self::update_bandwidth(rq, SchedClass::Realtime);
+        rq.rt.clear_current();
+    }
 }