Replace the wait_queue data structure VecQueue with RawList

--------
1 use Rawlist for wait queue
2 fix mutex size after wait_queue modify
3 add unit test for axsync/mutex size
Note: must use an unique &T as rawlist node

Signed-off-by: guoweikang <[email protected]>

api/arceos_posix_api/build.rs

@@ -5,9 +5,9 @@ fn main() {
         // TODO: generate size and initial content automatically.
         let (mutex_size, mutex_init) = if cfg!(feature = "multitask") {
             if cfg!(feature = "smp") {
-                (6, "{0, 0, 8, 0, 0, 0}") // core::mem::transmute::<_, [usize; 6]>(axsync::Mutex::new(()))
+                (3, "{0, 0, 0}") // core::mem::transmute::<_, [usize; 3]>(axsync::Mutex::new(()))
             } else {
-                (5, "{0, 8, 0, 0, 0}") // core::mem::transmute::<_, [usize; 5]>(axsync::Mutex::new(()))
+                (2, "{0, 0}") // core::mem::transmute::<_, [usize; 2]>(axsync::Mutex::new(()))
             }
         } else {
             (1, "{0}")

modules/axsync/src/mutex.rs

@@ -249,4 +249,22 @@ mod tests {
         assert_eq!(*M.lock(), NUM_ITERS * NUM_TASKS * 3);
         println!("Mutex test OK");
     }
+
+    fn assert_mem_content<T>(val: &T, content: &[usize]) {
+        let val_ptr = val as *const T as *const usize;
+        let size = core::mem::size_of::<T>() / core::mem::size_of::<usize>();
+        assert_eq!(size, content.len());
+        let usize_slice = unsafe { core::slice::from_raw_parts(val_ptr, size) };
+        for (i, chunk) in usize_slice.iter().enumerate() {
+            assert_eq!(*chunk, content[i]);
+        }
+    }
+
+    #[test]
+    fn mutex_test_for_posix() {
+        // Test mutex size is equal api/arceos_posix_api/build.rs
+        let mutex_tuple = axsync::Mutex::new(());
+        let content: [usize; 2] = [0, 0];
+        assert_mem_content(&mutex_tuple, &content);
+    }
 }

modules/axtask/Cargo.toml

@@ -23,6 +23,7 @@ multitask = [
     "kernel_guard",
     "dep:crate_interface",
     "dep:cpumask",
+    "dep:linked_list_r4l"
 ]
 irq = []
 tls = ["axhal/tls"]
@@ -48,7 +49,8 @@ timer_list = { version = "0.1", optional = true }
 kernel_guard = { version = "0.1", optional = true }
 crate_interface = { version = "0.1", optional = true }
 cpumask = { version = "0.1", optional = true }
-scheduler = { git = "https://github.com/arceos-org/scheduler.git", tag = "v0.1.0", optional = true }
+scheduler = { git = "https://github.com/arceos-org/scheduler.git", tag = "v0.2.0", optional = true }
+linked_list_r4l = { version = "0.2.1" , optional = true }
 
 [dev-dependencies]
 rand = "0.8"

modules/axtask/src/run_queue.rs

@@ -10,7 +10,7 @@ use scheduler::BaseScheduler;
 use axhal::cpu::this_cpu_id;
 
 use crate::task::{CurrentTask, TaskState};
-use crate::wait_queue::WaitQueueGuard;
+use crate::wait_queue::{WaitQueueGuard, WaitTaskNode};
 use crate::{AxCpuMask, AxTaskRef, Scheduler, TaskInner, WaitQueue};
 
 macro_rules! percpu_static {
@@ -367,7 +367,7 @@ impl<'a, G: BaseGuard> CurrentRunQueueRef<'a, G> {
     ///     2. The caller must ensure that the current task is in the running state.
     ///     3. The caller must ensure that the current task is not the idle task.
     ///     4. The lock of the wait queue will be released explicitly after current task is pushed into it.
-    pub fn blocked_resched(&mut self, mut wq_guard: WaitQueueGuard) {
+    pub fn blocked_resched(&mut self, mut wq_guard: WaitQueueGuard, curr_waiter: &WaitTaskNode) {
         let curr = &self.current_task;
         assert!(curr.is_running());
         assert!(!curr.is_idle());
@@ -380,9 +380,11 @@ impl<'a, G: BaseGuard> CurrentRunQueueRef<'a, G> {
         // Mark the task as blocked, this has to be done before adding it to the wait queue
         // while holding the lock of the wait queue.
         curr.set_state(TaskState::Blocked);
-        curr.set_in_wait_queue(true);
-
-        wq_guard.push_back(curr.clone());
+        // SAFETY: The waiter was on caller stack, the lifetime ends
+        // only when the task resumes running.
+        unsafe {
+            assert!(wq_guard.push_back(curr_waiter));
+        }
         // Drop the lock of wait queue explictly.
         drop(wq_guard);
 

modules/axtask/src/task.rs

@@ -1,8 +1,11 @@
 use alloc::{boxed::Box, string::String, sync::Arc};
 use core::ops::Deref;
-use core::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, AtomicU8, Ordering};
+use core::sync::atomic::{AtomicI32, AtomicU64, AtomicU8, Ordering};
 use core::{alloc::Layout, cell::UnsafeCell, fmt, ptr::NonNull};
 
+#[cfg(any(feature = "smp", feature = "preempt"))]
+use core::sync::atomic::AtomicBool;
+
 #[cfg(feature = "smp")]
 use alloc::sync::Weak;
 #[cfg(feature = "preempt")]
@@ -50,9 +53,6 @@ pub struct TaskInner {
     /// CPU affinity mask.
     cpumask: SpinNoIrq<AxCpuMask>,
 
-    /// Mark whether the task is in the wait queue.
-    in_wait_queue: AtomicBool,
-
     /// Used to indicate whether the task is running on a CPU.
     #[cfg(feature = "smp")]
     on_cpu: AtomicBool,
@@ -228,7 +228,6 @@ impl TaskInner {
             state: AtomicU8::new(TaskState::Ready as u8),
             // By default, the task is allowed to run on all CPUs.
             cpumask: SpinNoIrq::new(AxCpuMask::full()),
-            in_wait_queue: AtomicBool::new(false),
             #[cfg(feature = "irq")]
             timer_ticket_id: AtomicU64::new(0),
             #[cfg(feature = "smp")]
@@ -317,16 +316,6 @@ impl TaskInner {
         self.is_idle
     }
 
-    #[inline]
-    pub(crate) fn in_wait_queue(&self) -> bool {
-        self.in_wait_queue.load(Ordering::Acquire)
-    }
-
-    #[inline]
-    pub(crate) fn set_in_wait_queue(&self, in_wait_queue: bool) {
-        self.in_wait_queue.store(in_wait_queue, Ordering::Release);
-    }
-
     /// Returns task's current timer ticket ID.
     #[inline]
     #[cfg(feature = "irq")]

modules/axtask/src/tests.rs

@@ -81,8 +81,6 @@ fn test_wait_queue() {
             WQ1.notify_one(true); // WQ1.wait_until()
             WQ2.wait();
 
-            assert!(!current().in_wait_queue());
-
             COUNTER.fetch_sub(1, Ordering::Relaxed);
             println!("wait_queue: task {:?} finished", current().id());
             WQ1.notify_one(true); // WQ1.wait_until()
@@ -92,7 +90,6 @@ fn test_wait_queue() {
     println!("task {:?} is waiting for tasks to start...", current().id());
     WQ1.wait_until(|| COUNTER.load(Ordering::Relaxed) == NUM_TASKS);
     assert_eq!(COUNTER.load(Ordering::Relaxed), NUM_TASKS);
-    assert!(!current().in_wait_queue());
     WQ2.notify_all(true); // WQ2.wait()
 
     println!(
@@ -101,7 +98,6 @@ fn test_wait_queue() {
     );
     WQ1.wait_until(|| COUNTER.load(Ordering::Relaxed) == 0);
     assert_eq!(COUNTER.load(Ordering::Relaxed), 0);
-    assert!(!current().in_wait_queue());
 }
 
 #[test]