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 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340
use crate::io::driver::{platform, Direction, Handle};
use crate::util::slab::Address;
use mio::{self, Evented};
use std::io;
use std::task::{Context, Poll};
cfg_io_driver! {
/// Associates an I/O resource with the reactor instance that drives it.
///
/// A registration represents an I/O resource registered with a Reactor such
/// that it will receive task notifications on readiness. This is the lowest
/// level API for integrating with a reactor.
///
/// The association between an I/O resource is made by calling [`new`]. Once
/// the association is established, it remains established until the
/// registration instance is dropped.
///
/// A registration instance represents two separate readiness streams. One
/// for the read readiness and one for write readiness. These streams are
/// independent and can be consumed from separate tasks.
///
/// **Note**: while `Registration` is `Sync`, the caller must ensure that
/// there are at most two tasks that use a registration instance
/// concurrently. One task for [`poll_read_ready`] and one task for
/// [`poll_write_ready`]. While violating this requirement is "safe" from a
/// Rust memory safety point of view, it will result in unexpected behavior
/// in the form of lost notifications and tasks hanging.
///
/// ## Platform-specific events
///
/// `Registration` also allows receiving platform-specific `mio::Ready`
/// events. These events are included as part of the read readiness event
/// stream. The write readiness event stream is only for `Ready::writable()`
/// events.
///
/// [`new`]: method@Self::new
/// [`poll_read_ready`]: method@Self::poll_read_ready`
/// [`poll_write_ready`]: method@Self::poll_write_ready`
#[derive(Debug)]
pub struct Registration {
handle: Handle,
address: Address,
}
}
// ===== impl Registration =====
impl Registration {
/// Registers the I/O resource with the default reactor.
///
/// # Return
///
/// - `Ok` if the registration happened successfully
/// - `Err` if an error was encountered during registration
///
///
/// # Panics
///
/// This function panics if thread-local runtime is not set.
///
/// The runtime is usually set implicitly when this function is called
/// from a future driven by a tokio runtime, otherwise runtime can be set
/// explicitly with [`Handle::enter`](crate::runtime::Handle::enter) function.
pub fn new<T>(io: &T) -> io::Result<Registration>
where
T: Evented,
{
Registration::new_with_ready(io, mio::Ready::all())
}
/// Registers the I/O resource with the default reactor, for a specific `mio::Ready` state.
/// `new_with_ready` should be used over `new` when you need control over the readiness state,
/// such as when a file descriptor only allows reads. This does not add `hup` or `error` so if
/// you are interested in those states, you will need to add them to the readiness state passed
/// to this function.
///
/// An example to listen to read only
///
/// ```rust
/// ##[cfg(unix)]
/// mio::Ready::from_usize(
/// mio::Ready::readable().as_usize()
/// | mio::unix::UnixReady::error().as_usize()
/// | mio::unix::UnixReady::hup().as_usize()
/// );
/// ```
///
/// # Return
///
/// - `Ok` if the registration happened successfully
/// - `Err` if an error was encountered during registration
///
///
/// # Panics
///
/// This function panics if thread-local runtime is not set.
///
/// The runtime is usually set implicitly when this function is called
/// from a future driven by a tokio runtime, otherwise runtime can be set
/// explicitly with [`Handle::enter`](crate::runtime::Handle::enter) function.
pub fn new_with_ready<T>(io: &T, ready: mio::Ready) -> io::Result<Registration>
where
T: Evented,
{
let handle = Handle::current();
let address = if let Some(inner) = handle.inner() {
inner.add_source(io, ready)?
} else {
return Err(io::Error::new(
io::ErrorKind::Other,
"failed to find event loop",
));
};
Ok(Registration { handle, address })
}
/// Deregisters the I/O resource from the reactor it is associated with.
///
/// This function must be called before the I/O resource associated with the
/// registration is dropped.
///
/// Note that deregistering does not guarantee that the I/O resource can be
/// registered with a different reactor. Some I/O resource types can only be
/// associated with a single reactor instance for their lifetime.
///
/// # Return
///
/// If the deregistration was successful, `Ok` is returned. Any calls to
/// `Reactor::turn` that happen after a successful call to `deregister` will
/// no longer result in notifications getting sent for this registration.
///
/// `Err` is returned if an error is encountered.
pub fn deregister<T>(&mut self, io: &T) -> io::Result<()>
where
T: Evented,
{
let inner = match self.handle.inner() {
Some(inner) => inner,
None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
};
inner.deregister_source(io)
}
/// Polls for events on the I/O resource's read readiness stream.
///
/// If the I/O resource receives a new read readiness event since the last
/// call to `poll_read_ready`, it is returned. If it has not, the current
/// task is notified once a new event is received.
///
/// All events except `HUP` are [edge-triggered]. Once `HUP` is returned,
/// the function will always return `Ready(HUP)`. This should be treated as
/// the end of the readiness stream.
///
/// # Return value
///
/// There are several possible return values:
///
/// * `Poll::Ready(Ok(readiness))` means that the I/O resource has received
/// a new readiness event. The readiness value is included.
///
/// * `Poll::Pending` means that no new readiness events have been received
/// since the last call to `poll_read_ready`.
///
/// * `Poll::Ready(Err(err))` means that the registration has encountered an
/// error. This could represent a permanent internal error for example.
///
/// [edge-triggered]: struct@mio::Poll#edge-triggered-and-level-triggered
///
/// # Panics
///
/// This function will panic if called from outside of a task context.
pub fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
// Keep track of task budget
let coop = ready!(crate::coop::poll_proceed(cx));
let v = self.poll_ready(Direction::Read, Some(cx)).map_err(|e| {
coop.made_progress();
e
})?;
match v {
Some(v) => {
coop.made_progress();
Poll::Ready(Ok(v))
}
None => Poll::Pending,
}
}
/// Consume any pending read readiness event.
///
/// This function is identical to [`poll_read_ready`] **except** that it
/// will not notify the current task when a new event is received. As such,
/// it is safe to call this function from outside of a task context.
///
/// [`poll_read_ready`]: method@Self::poll_read_ready
pub fn take_read_ready(&self) -> io::Result<Option<mio::Ready>> {
self.poll_ready(Direction::Read, None)
}
/// Polls for events on the I/O resource's write readiness stream.
///
/// If the I/O resource receives a new write readiness event since the last
/// call to `poll_write_ready`, it is returned. If it has not, the current
/// task is notified once a new event is received.
///
/// All events except `HUP` are [edge-triggered]. Once `HUP` is returned,
/// the function will always return `Ready(HUP)`. This should be treated as
/// the end of the readiness stream.
///
/// # Return value
///
/// There are several possible return values:
///
/// * `Poll::Ready(Ok(readiness))` means that the I/O resource has received
/// a new readiness event. The readiness value is included.
///
/// * `Poll::Pending` means that no new readiness events have been received
/// since the last call to `poll_write_ready`.
///
/// * `Poll::Ready(Err(err))` means that the registration has encountered an
/// error. This could represent a permanent internal error for example.
///
/// [edge-triggered]: struct@mio::Poll#edge-triggered-and-level-triggered
///
/// # Panics
///
/// This function will panic if called from outside of a task context.
pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
// Keep track of task budget
let coop = ready!(crate::coop::poll_proceed(cx));
let v = self.poll_ready(Direction::Write, Some(cx)).map_err(|e| {
coop.made_progress();
e
})?;
match v {
Some(v) => {
coop.made_progress();
Poll::Ready(Ok(v))
}
None => Poll::Pending,
}
}
/// Consumes any pending write readiness event.
///
/// This function is identical to [`poll_write_ready`] **except** that it
/// will not notify the current task when a new event is received. As such,
/// it is safe to call this function from outside of a task context.
///
/// [`poll_write_ready`]: method@Self::poll_write_ready
pub fn take_write_ready(&self) -> io::Result<Option<mio::Ready>> {
self.poll_ready(Direction::Write, None)
}
/// Polls for events on the I/O resource's `direction` readiness stream.
///
/// If called with a task context, notify the task when a new event is
/// received.
fn poll_ready(
&self,
direction: Direction,
cx: Option<&mut Context<'_>>,
) -> io::Result<Option<mio::Ready>> {
let inner = match self.handle.inner() {
Some(inner) => inner,
None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
};
// If the task should be notified about new events, ensure that it has
// been registered
if let Some(ref cx) = cx {
inner.register(self.address, direction, cx.waker().clone())
}
let mask = direction.mask();
let mask_no_hup = (mask - platform::hup() - platform::error()).as_usize();
let sched = inner.io_dispatch.get(self.address).unwrap();
// This consumes the current readiness state **except** for HUP and
// error. HUP and error are excluded because a) they are final states
// and never transitition out and b) both the read AND the write
// directions need to be able to obvserve these states.
//
// # Platform-specific behavior
//
// HUP and error readiness are platform-specific. On epoll platforms,
// HUP has specific conditions that must be met by both peers of a
// connection in order to be triggered.
//
// On epoll platforms, `EPOLLERR` is signaled through
// `UnixReady::error()` and is important to be observable by both read
// AND write. A specific case that `EPOLLERR` occurs is when the read
// end of a pipe is closed. When this occurs, a peer blocked by
// writing to the pipe should be notified.
let curr_ready = sched
.set_readiness(self.address, |curr| curr & (!mask_no_hup))
.unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address));
let mut ready = mask & mio::Ready::from_usize(curr_ready);
if ready.is_empty() {
if let Some(cx) = cx {
// Update the task info
match direction {
Direction::Read => sched.reader.register_by_ref(cx.waker()),
Direction::Write => sched.writer.register_by_ref(cx.waker()),
}
// Try again
let curr_ready = sched
.set_readiness(self.address, |curr| curr & (!mask_no_hup))
.unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address));
ready = mask & mio::Ready::from_usize(curr_ready);
}
}
if ready.is_empty() {
Ok(None)
} else {
Ok(Some(ready))
}
}
}
unsafe impl Send for Registration {}
unsafe impl Sync for Registration {}
impl Drop for Registration {
fn drop(&mut self) {
let inner = match self.handle.inner() {
Some(inner) => inner,
None => return,
};
inner.drop_source(self.address);
}
}