wasmer_wasix/fs/
notification.rs

1use std::{
2    collections::VecDeque,
3    sync::Mutex,
4    task::{Poll, Waker},
5};
6
7use virtual_mio::{InterestHandler, InterestType};
8
9#[derive(Debug)]
10struct NotificationState {
11    /// Used for event notifications by the user application or operating system
12    /// (positive number means there are events waiting to be processed)
13    counter: u64,
14    /// Flag that indicates if this is operating
15    is_semaphore: bool,
16    /// All the registered wakers
17    wakers: VecDeque<Waker>,
18    /// InterestHandler for use with epoll
19    interest_handler: Option<Box<dyn InterestHandler>>,
20}
21
22impl NotificationState {
23    fn add_waker(&mut self, waker: &Waker) {
24        if !self.wakers.iter().any(|a| a.will_wake(waker)) {
25            self.wakers.push_front(waker.clone());
26        }
27    }
28
29    fn wake_all(&mut self) {
30        while let Some(waker) = self.wakers.pop_front() {
31            waker.wake();
32        }
33        if let Some(handler) = self.interest_handler.as_mut() {
34            handler.push_interest(InterestType::Readable);
35        }
36    }
37
38    fn inc(&mut self, val: u64) {
39        self.counter += val;
40        self.wake_all();
41    }
42
43    fn dec(&mut self) -> u64 {
44        let val = self.counter;
45        if self.is_semaphore {
46            if self.counter > 0 {
47                self.counter -= 1;
48                if self.counter > 0 {
49                    self.wake_all();
50                }
51            }
52        } else {
53            self.counter = 0;
54        }
55        val
56    }
57}
58
59#[derive(Debug)]
60pub struct NotificationInner {
61    /// Receiver that wakes sleeping threads
62    state: Mutex<NotificationState>,
63}
64
65impl NotificationInner {
66    pub fn new(initial_val: u64, is_semaphore: bool) -> Self {
67        Self {
68            state: Mutex::new(NotificationState {
69                counter: initial_val,
70                is_semaphore,
71                wakers: Default::default(),
72                interest_handler: None,
73            }),
74        }
75    }
76    pub fn poll(&self, waker: &Waker) -> Poll<usize> {
77        let mut state = self.state.lock().unwrap();
78        state.add_waker(waker);
79
80        if state.counter > 0 {
81            Poll::Ready(state.counter as usize)
82        } else {
83            Poll::Pending
84        }
85    }
86
87    pub fn write(&self, val: u64) {
88        let mut state = self.state.lock().unwrap();
89        state.inc(val);
90    }
91
92    pub fn read(&self, waker: &Waker) -> Poll<u64> {
93        let mut state = self.state.lock().unwrap();
94        state.add_waker(waker);
95        match state.dec() {
96            0 => Poll::Pending,
97            res => Poll::Ready(res),
98        }
99    }
100
101    pub fn try_read(&self) -> Option<u64> {
102        let mut state = self.state.lock().unwrap();
103        match state.dec() {
104            0 => None,
105            res => Some(res),
106        }
107    }
108
109    pub fn reset(&self) {
110        let mut state = self.state.lock().unwrap();
111        state.counter = 0;
112    }
113
114    pub fn set_interest_handler(&self, handler: Box<dyn InterestHandler>) {
115        let mut state = self.state.lock().unwrap();
116        state.interest_handler.replace(handler);
117    }
118
119    pub fn remove_interest_handler(&self) -> Option<Box<dyn InterestHandler>> {
120        let mut state = self.state.lock().unwrap();
121        state.interest_handler.take()
122    }
123}
wasmer_wasix/fs/notification.rs

wasmer_wasix/fs/
notification.rs
