greenie 0.4.0

Green threads and coroutines in stable Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

use crate::common::mutex::*;
use crate::ctx::*;
use crate::ptr::*;
use crate::scheduler::*;
/// Synchronization primitive that can be used to block a thread, or multiple threads at the same time,
/// until another thread both modifies a shared variable (the condition), and notifies the condition_variable.
#[derive(Clone)]
pub struct Condvar {
    pub(crate) wait_queue: Ptr<std::collections::LinkedList<crate::ptr::Ptr<Context>>>,
}

#[derive(Clone, Copy, PartialEq, Eq)]
pub enum CvStatus {
    NoTimeout = 1,
    Timeout,
}

impl Condvar {
    pub fn new() -> Self {
        Self {
            wait_queue: Ptr::new(std::collections::LinkedList::new()),
        }
    }
    /// wait causes the current thread to block until the condition variable is notified or a spurious wakeup occurs, optionally
    /// looping until some predicate is satisfied.
    ///
    /// Atomically unlocks lock, blocks the current executing thread, and adds it to the list of threads waiting on `self`.
    /// The thread will be unblocked when notify_all() or notify_one() is executed. It may also be unblocked spuriously.
    /// When unblocked, regardless of the reason, lock is reacquired and wait exits.
    pub fn wait_for_mutex(&self, m: &Mutex) {
        let active_ctx = RUNTIME.with(|rt| rt.active_ctx);
        self.wait_queue.get().push_back(active_ctx);
        active_ctx.get().twstatus.store(
            self as *const Condvar as *mut i8,
            std::sync::atomic::Ordering::Release,
        );

        m.unlock();

        yield_thread();

        m.lock();
    }
    /// Equivalent to
    /// ```c
    /// while !pred() {
    ///       self.wait_for_mutex(m);
    /// }
    /// ```
    pub fn wait_pred(&self, m: &Mutex, mut pred: impl FnMut() -> bool) {
        while !pred() {
            self.wait_for_mutex(m);
        }
    }
    /// If any threads are waiting on this condvar, calling notify_one unblocks one of the waiting threads.
    pub fn notify_one(&self) {
        //let active_ctx = RUNTIME.with(|rt| rt.get().threads[rt.get().current]);

        while let Some(ctx) = self.wait_queue.get().pop_front() {
            let expected = self as *const Condvar as *mut i8;
            let result = ctx.get().twstatus.compare_exchange(
                expected,
                -1i8 as *mut i8,
                std::sync::atomic::Ordering::Acquire,
                std::sync::atomic::Ordering::Relaxed,
            );
            match result {
                Ok(_) => {
                    yield_thread();
                    break;
                }
                Err(x) => {
                    if x.is_null() {
                        yield_thread();
                    }
                    break;
                }
            }
        }
    }
    /// Unblocks all threads currently waiting for this condvar.
    pub fn notify_all(&self) {
        //let active_ctx = RUNTIME.with(|rt| rt.get().threads[rt.get().current]);

        while let Some(ctx) = self.wait_queue.get().pop_front() {
            let expected = self as *const Condvar as *mut i8;
            let result = ctx.get().twstatus.compare_exchange(
                expected,
                -1i8 as *mut i8,
                std::sync::atomic::Ordering::Acquire,
                std::sync::atomic::Ordering::Relaxed,
            );
            match result {
                Ok(_) => {
                    crate::yield_thread();
                    break;
                }
                Err(x) => {
                    if x.is_null() {
                        crate::yield_thread();
                        break;
                    }
                }
            }
        }
    }
}

impl Drop for Condvar {
    fn drop(&mut self) {
        unsafe { std::intrinsics::drop_in_place(self.wait_queue.0) }
    }
}