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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
//! # KQueue
//! A kqueue per thread, reused by every task that waits on one
use crate::{
RuntimeError,
constants::WAKE_IDENT,
executor,
modules::{
int_check::IntCheck,
kevent::{KEvent, eventlist},
},
};
use std::{cell::Cell, time::Duration};
/// How a wait for a particular event ended
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Waited {
/// The event that was asked for landed
Arrived,
/// Somebody cancelled the task while it waited
Cancelled,
/// The queue itself failed, so nothing is ever arriving
Failed,
}
/// Owns a thread's kqueue, closing it when the thread ends
struct KQueue(Cell<i32>);
impl Drop for KQueue {
fn drop(&mut self) {
let id = self.0.get();
if id >= 0 {
unsafe { libc::close(id) };
}
}
}
thread_local! {
/// This thread's kqueue, or -1 before it has been created
static QUEUE: KQueue = const { KQueue(Cell::new(-1)) };
}
/// This thread's kqueue, creating it on first use
#[inline(always)]
pub(crate) fn id() -> Result<i32, RuntimeError> {
QUEUE.with(|queue| {
let existing = queue.0.get();
if existing >= 0 {
return Ok(existing);
}
let created = unsafe { libc::kqueue() }.check()?;
queue.0.set(created);
Ok(created)
})
}
/// Blocks on a queue until anything lands on it, or `timeout`
/// passes
///
/// Callers re-read what they were waiting on afterwards, so any
/// wake will do
pub(crate) fn wait_any(queue: i32, timeout: Duration) {
let mut events = eventlist();
loop {
match unsafe { KEvent::listen_for(queue, &mut events, timeout) }.check() {
Ok(_) => return,
Err(RuntimeError::CheckError(Some(libc::EINTR))) => continue,
Err(_) => return,
}
}
}
/// Blocks on a queue until a particular event lands on it
///
/// ## Returns
/// `Failed` if the queue itself fails
///
/// #### Note
/// A queue outlives every task that runs on its thread, so a
/// `WAKE_IDENT` can be left over from an earlier task's cancel.
/// It only counts as `Cancelled` if the current task is
pub(crate) fn wait_for(queue: i32, ident: usize, filter: i16) -> Waited {
wait_for_within(queue, ident, filter, None)
}
/// The same wait, with a bound on how long one lap of it blocks
///
/// ## Returns
/// `Arrived` when `within` runs out, so the caller asks whatever it
/// is waiting on again rather than trusting the wake
pub(crate) fn wait_for_upto(queue: i32, ident: usize, filter: i16, within: Duration) -> Waited {
wait_for_within(queue, ident, filter, Some(within))
}
fn wait_for_within(queue: i32, ident: usize, filter: i16, within: Option<Duration>) -> Waited {
let mut events = eventlist();
loop {
let listened = match within {
Some(within) => unsafe { KEvent::listen_for(queue, &mut events, within) },
None => unsafe { KEvent::listen(queue, &mut events) },
};
let count = match listened.check() {
Ok(count) => count as usize,
Err(RuntimeError::CheckError(Some(libc::EINTR))) => continue,
Err(_) => return Waited::Failed,
};
for event in events.iter().take(count) {
if event.flags & libc::EV_ERROR != 0 {
continue;
}
if event.ident == ident && event.filter == filter {
return Waited::Arrived;
}
// Possibly left over from an earlier task, so the task's own
// state decides
if event.ident == WAKE_IDENT && event.filter == libc::EVFILT_USER {
if executor::cancelled() {
return Waited::Cancelled;
}
continue;
}
}
// Nothing of this task's arrived, so a bounded wait hands
// back to whoever asked rather than blocking again
if within.is_some() {
return Waited::Arrived;
}
}
}