Skip to content

Commit 9999520

Browse files
notgullnotgull
authored
Update for new polling breaking changes (#141)
* Update for new polling breaking changes * Update for polling v3.0.0 Signed-off-by: John Nunley <dev@notgull.net> * Silence clippy Signed-off-by: John Nunley <dev@notgull.net> * Fix book compilation Signed-off-by: John Nunley <dev@notgull.net> * Add more documentation to the NoIoDrop type Signed-off-by: John Nunley <dev@notgull.net> --------- Signed-off-by: John Nunley <dev@notgull.net>
1 parent 6cc56c1 commit 9999520

File tree

6 files changed

+208
-53
lines changed

6 files changed

+208
-53
lines changed

‎Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ io-lifetimes = "1.0.3"
2626
log = "0.4"
2727
nix = { version = "0.26", default-features = false, features = ["signal"], optional = true }
2828
pin-utils = { version = "0.1.0", optional = true }
29-
polling = "2.6.0"
29+
polling = "3.0.0"
3030
rustix = { version = "0.38", default-features = false, features = ["event", "fs", "pipe", "std"] }
3131
slab = "0.4.8"
3232
thiserror = "1.0"

‎doc/src/zmqsource.rs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,7 @@ where
161161
// on the socket that warrants reading the events again.
162162
let events = self
163163
.socket
164-
.file
164+
.get_ref()
165165
.get_events()
166166
.context("Failed to read ZeroMQ events")?;
167167

@@ -170,7 +170,7 @@ where
170170
if events.contains(zmq::POLLOUT) {
171171
if let Some(parts) = self.outbox.pop_front() {
172172
self.socket
173-
.file
173+
.get_ref()
174174
.send_multipart(parts, 0)
175175
.context("Failed to send message")?;
176176
used_socket = true;
@@ -182,7 +182,7 @@ where
182182
// sending, which includes all parts of a multipart message.
183183
let messages = self
184184
.socket
185-
.file
185+
.get_ref()
186186
.recv_multipart(0)
187187
.context("Failed to receive message")?;
188188
used_socket = true;
@@ -247,14 +247,14 @@ where
247247
//
248248
// - https://stackoverflow.com/a/38338578/188535
249249
// - http://api.zeromq.org/4-0:zmq-ctx-term
250-
self.socket.file.set_linger(0).ok();
251-
self.socket.file.set_rcvtimeo(0).ok();
252-
self.socket.file.set_sndtimeo(0).ok();
250+
self.socket.get_ref().set_linger(0).ok();
251+
self.socket.get_ref().set_rcvtimeo(0).ok();
252+
self.socket.get_ref().set_sndtimeo(0).ok();
253253

254254
// Double result because (a) possible failure on call and (b) possible
255255
// failure decoding.
256-
if let Ok(Ok(last_endpoint)) = self.socket.file.get_last_endpoint() {
257-
self.socket.file.disconnect(&last_endpoint).ok();
256+
if let Ok(Ok(last_endpoint)) = self.socket.get_ref().get_last_endpoint() {
257+
self.socket.get_ref().disconnect(&last_endpoint).ok();
258258
}
259259
}
260260
}

‎src/io.rs‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,11 @@ impl<'l, F: AsFd> Async<'l, F> {
6565
}));
6666
let key = inner.sources.borrow_mut().insert(dispatcher.clone());
6767
dispatcher.borrow_mut().token = Some(Token { key });
68-
inner.register(&dispatcher)?;
68+
69+
// SAFETY: We are sure to deregister on drop.
70+
unsafe {
71+
inner.register(&dispatcher)?;
72+
}
6973

7074
// Straightforward casting would require us to add the bound `Data: 'l` but we don't actually need it
7175
// as this module never accesses the dispatch data, so we use transmute to erase it
@@ -168,13 +172,13 @@ impl<'l, F: AsFd> Drop for Async<'l, F> {
168172
impl<'l, F: AsFd> Unpin for Async<'l, F> {}
169173

170174
trait IoLoopInner {
171-
fn register(&self, dispatcher: &RefCell<IoDispatcher>) -> crate::Result<()>;
175+
unsafe fn register(&self, dispatcher: &RefCell<IoDispatcher>) -> crate::Result<()>;
172176
fn reregister(&self, dispatcher: &RefCell<IoDispatcher>) -> crate::Result<()>;
173177
fn kill(&self, dispatcher: &RefCell<IoDispatcher>);
174178
}
175179

176180
impl<'l, Data> IoLoopInner for LoopInner<'l, Data> {
177-
fn register(&self, dispatcher: &RefCell<IoDispatcher>) -> crate::Result<()> {
181+
unsafe fn register(&self, dispatcher: &RefCell<IoDispatcher>) -> crate::Result<()> {
178182
let disp = dispatcher.borrow();
179183
self.poll.borrow_mut().register(
180184
unsafe { BorrowedFd::borrow_raw(disp.fd) },

‎src/sources/generic.rs‎

Lines changed: 129 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,8 @@
3838
//! [`EventSource`](crate::EventSource) implementation to them.
3939
4040
use io_lifetimes::{AsFd, BorrowedFd};
41-
use std::{marker::PhantomData, ops, os::unix::io::AsRawFd};
41+
use polling::Poller;
42+
use std::{borrow, marker::PhantomData, ops, os::unix::io::AsRawFd, sync::Arc};
4243

4344
use crate::{EventSource, Interest, Mode, Poll, PostAction, Readiness, Token, TokenFactory};
4445

@@ -78,16 +79,79 @@ impl<T: AsRawFd> AsFd for FdWrapper<T> {
7879
}
7980
}
8081

82+
/// A wrapper around a type that doesn't expose it mutably safely.
83+
///
84+
/// The [`EventSource`] trait's `Metadata` type demands mutable access to the inner I/O source.
85+
/// However, the inner polling source used by `calloop` keeps the handle-based equivalent of an
86+
/// immutable pointer to the underlying object's I/O handle. Therefore, if the inner source is
87+
/// dropped, this leaves behind a dangling pointer which immediately invokes undefined behavior
88+
/// on the next poll of the event loop.
89+
///
90+
/// In order to prevent this from happening, the [`Generic`] I/O source must not directly expose
91+
/// a mutable reference to the underlying handle. This type wraps around the underlying handle and
92+
/// easily allows users to take immutable (`&`) references to the type, but makes mutable (`&mut`)
93+
/// references unsafe to get. Therefore, it prevents the source from being moved out and dropped
94+
/// while it is still registered in the event loop.
95+
///
96+
/// [`EventSource`]: crate::EventSource
97+
#[derive(Debug)]
98+
pub struct NoIoDrop<T>(T);
99+
100+
impl<T> NoIoDrop<T> {
101+
/// Get a mutable reference.
102+
///
103+
/// # Safety
104+
///
105+
/// The inner type's I/O source must not be dropped.
106+
pub unsafe fn get_mut(&mut self) -> &mut T {
107+
&mut self.0
108+
}
109+
}
110+
111+
impl<T> AsRef<T> for NoIoDrop<T> {
112+
fn as_ref(&self) -> &T {
113+
&self.0
114+
}
115+
}
116+
117+
impl<T> borrow::Borrow<T> for NoIoDrop<T> {
118+
fn borrow(&self) -> &T {
119+
&self.0
120+
}
121+
}
122+
123+
impl<T> ops::Deref for NoIoDrop<T> {
124+
type Target = T;
125+
126+
fn deref(&self) -> &Self::Target {
127+
&self.0
128+
}
129+
}
130+
131+
impl<T: AsFd> AsFd for NoIoDrop<T> {
132+
fn as_fd(&self) -> BorrowedFd<'_> {
133+
// SAFETY: The innter type is not mutated.
134+
self.0.as_fd()
135+
}
136+
}
137+
81138
/// A generic event source wrapping a FD-backed type
82139
#[derive(Debug)]
83140
pub struct Generic<F: AsFd, E = std::io::Error> {
84-
/// The wrapped FD-backed type
85-
pub file: F,
141+
/// The wrapped FD-backed type.
142+
///
143+
/// This must be deregistered before it is dropped.
144+
file: Option<NoIoDrop<F>>,
86145
/// The programmed interest
87146
pub interest: Interest,
88147
/// The programmed mode
89148
pub mode: Mode,
90149

150+
/// Back-reference to the poller.
151+
///
152+
/// This is needed to drop the original file.
153+
poller: Option<Arc<Poller>>,
154+
91155
// This token is used by the event loop logic to look up this source when an
92156
// event occurs.
93157
token: Option<Token>,
@@ -101,30 +165,64 @@ impl<F: AsFd> Generic<F, std::io::Error> {
101165
/// [`std::io::Error`] as its error type.
102166
pub fn new(file: F, interest: Interest, mode: Mode) -> Generic<F, std::io::Error> {
103167
Generic {
104-
file,
168+
file: Some(NoIoDrop(file)),
105169
interest,
106170
mode,
107171
token: None,
172+
poller: None,
108173
_error_type: PhantomData,
109174
}
110175
}
111176

112177
/// Wrap a FD-backed type into a `Generic` event source using an arbitrary error type.
113178
pub fn new_with_error<E>(file: F, interest: Interest, mode: Mode) -> Generic<F, E> {
114179
Generic {
115-
file,
180+
file: Some(NoIoDrop(file)),
116181
interest,
117182
mode,
118183
token: None,
184+
poller: None,
119185
_error_type: PhantomData,
120186
}
121187
}
122188
}
123189

124190
impl<F: AsFd, E> Generic<F, E> {
125191
/// Unwrap the `Generic` source to retrieve the underlying type
126-
pub fn unwrap(self) -> F {
127-
self.file
192+
pub fn unwrap(mut self) -> F {
193+
let NoIoDrop(file) = self.file.take().unwrap();
194+
195+
// Remove it from the poller.
196+
if let Some(poller) = self.poller.take() {
197+
poller.delete(file.as_fd()).ok();
198+
}
199+
200+
file
201+
}
202+
203+
/// Get a reference to the underlying type.
204+
pub fn get_ref(&self) -> &F {
205+
&self.file.as_ref().unwrap().0
206+
}
207+
208+
/// Get a mutable reference to the underlying type.
209+
///
210+
/// # Safety
211+
///
212+
/// This is unsafe because it allows you to modify the underlying type, which
213+
/// allows you to drop the underlying event source. Dropping the underlying source
214+
/// leads to a dangling reference.
215+
pub unsafe fn get_mut(&mut self) -> &mut F {
216+
self.file.as_mut().unwrap().get_mut()
217+
}
218+
}
219+
220+
impl<F: AsFd, E> Drop for Generic<F, E> {
221+
fn drop(&mut self) {
222+
// Remove it from the poller.
223+
if let (Some(file), Some(poller)) = (self.file.take(), self.poller.take()) {
224+
poller.delete(file.as_fd()).ok();
225+
}
128226
}
129227
}
130228

@@ -134,7 +232,7 @@ where
134232
E: Into<Box<dyn std::error::Error + Send + Sync>>,
135233
{
136234
type Event = Readiness;
137-
type Metadata = F;
235+
type Metadata = NoIoDrop<F>;
138236
type Ret = Result<PostAction, E>;
139237
type Error = E;
140238

@@ -152,13 +250,24 @@ where
152250
return Ok(PostAction::Continue);
153251
}
154252

155-
callback(readiness, &mut self.file)
253+
callback(readiness, self.file.as_mut().unwrap())
156254
}
157255

158256
fn register(&mut self, poll: &mut Poll, token_factory: &mut TokenFactory) -> crate::Result<()> {
159257
let token = token_factory.token();
160258

161-
poll.register(&self.file, self.interest, self.mode, token)?;
259+
// Make sure we can use the poller to deregister if need be.
260+
self.poller = Some(poll.poller().clone());
261+
262+
// SAFETY: We've now ensured that we have a poller to deregister with.
263+
unsafe {
264+
poll.register(
265+
&self.file.as_ref().unwrap().0,
266+
self.interest,
267+
self.mode,
268+
token,
269+
)?;
270+
}
162271

163272
self.token = Some(token);
164273
Ok(())
@@ -171,14 +280,20 @@ where
171280
) -> crate::Result<()> {
172281
let token = token_factory.token();
173282

174-
poll.reregister(&self.file, self.interest, self.mode, token)?;
283+
poll.reregister(
284+
&self.file.as_ref().unwrap().0,
285+
self.interest,
286+
self.mode,
287+
token,
288+
)?;
175289

176290
self.token = Some(token);
177291
Ok(())
178292
}
179293

180294
fn unregister(&mut self, poll: &mut Poll) -> crate::Result<()> {
181-
poll.unregister(&self.file)?;
295+
poll.unregister(&self.file.as_ref().unwrap().0)?;
296+
self.poller = None;
182297
self.token = None;
183298
Ok(())
184299
}
@@ -211,7 +326,7 @@ mod tests {
211326
// we have not registered for writability
212327
assert!(!readiness.writable);
213328
let mut buffer = vec![0; 10];
214-
let ret = file.read(&mut buffer).unwrap();
329+
let ret = (&**file).read(&mut buffer).unwrap();
215330
assert_eq!(ret, 6);
216331
assert_eq!(&buffer[..6], &[1, 2, 3, 4, 5, 6]);
217332

@@ -286,7 +401,7 @@ mod tests {
286401
// we have not registered for writability
287402
assert!(!readiness.writable);
288403
let mut buffer = vec![0; 10];
289-
let ret = file.read(&mut buffer).unwrap();
404+
let ret = (&**file).read(&mut buffer).unwrap();
290405
assert_eq!(ret, 6);
291406
assert_eq!(&buffer[..6], &[1, 2, 3, 4, 5, 6]);
292407

‎src/sources/signals.rs‎

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,14 @@ impl Signals {
7676
self.mask.add(s);
7777
}
7878
self.mask.thread_block().map_err(IoError::from)?;
79-
self.sfd.file.set_mask(&self.mask).map_err(IoError::from)?;
79+
80+
// SAFETY: We don't drop the underlying mask.
81+
unsafe {
82+
self.sfd
83+
.get_mut()
84+
.set_mask(&self.mask)
85+
.map_err(IoError::from)?;
86+
}
8087
Ok(())
8188
}
8289

@@ -91,7 +98,14 @@ impl Signals {
9198
removed.add(s);
9299
}
93100
removed.thread_unblock().map_err(IoError::from)?;
94-
self.sfd.file.set_mask(&self.mask).map_err(IoError::from)?;
101+
102+
// SAFETY: We don't drop the underlying mask.
103+
unsafe {
104+
self.sfd
105+
.get_mut()
106+
.set_mask(&self.mask)
107+
.map_err(IoError::from)?;
108+
}
95109
Ok(())
96110
}
97111

@@ -107,7 +121,14 @@ impl Signals {
107121

108122
self.mask.thread_unblock().map_err(IoError::from)?;
109123
new_mask.thread_block().map_err(IoError::from)?;
110-
self.sfd.file.set_mask(&new_mask).map_err(IoError::from)?;
124+
125+
// SAFETY: We don't drop the underlying mask.
126+
unsafe {
127+
self.sfd
128+
.get_mut()
129+
.set_mask(&new_mask)
130+
.map_err(IoError::from)?;
131+
}
111132
self.mask = new_mask;
112133

113134
Ok(())
@@ -141,7 +162,7 @@ impl EventSource for Signals {
141162
self.sfd
142163
.process_events(readiness, token, |_, sfd| {
143164
loop {
144-
match sfd.read_signal() {
165+
match unsafe { sfd.get_mut().read_signal() } {
145166
Ok(Some(info)) => callback(Event { info }, &mut ()),
146167
Ok(None) => break,
147168
Err(e) => {

0 commit comments

Comments
 (0)