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
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
//! Operating System backed readiness event queue.
//!
//! [`OsQueue`] provides an abstraction over platform specific Operating System
//! backed readiness event queues, such as kqueue or epoll.
//!
//! [`OsQueue`]: crate::os::OsQueue
//!
//! # Portability
//!
//! Using [`OsQueue`] provides a portable interface across supported platforms
//! as long as the caller takes the following into consideration:
//!
//! ### Draining readiness
//!
//! When using [edge-triggered] mode, once a readiness event is received, the
//! corresponding operation must be performed repeatedly until it returns
//! [`WouldBlock`]. Unless this is done, there is no guarantee that another
//! readiness event will be delivered, even if further data is received for the
//! [`Evented`] handle. See [`RegisterOption`] for more.
//!
//! [`WouldBlock`]: std::io::ErrorKind::WouldBlock
//! [edge-triggered]: crate::os::RegisterOption::EDGE
//! [`Evented`]: crate::os::Evented
//! [`RegisterOption`]: crate::os::RegisterOption
//!
//! ### Spurious events
//!
//! The [`Source::poll`] implementation may return readiness events even if the
//! associated [`Evented`] handle is not actually ready. Given the same code,
//! this may happen more on some platforms than others. It is important to never
//! assume that, just because a readiness notification was received, that the
//! associated operation will as well.
//!
//! If operation fails with a [`WouldBlock`] error, then the caller should not
//! treat this as an error and wait until another readiness event is received.
//!
//! Furthermore a single call to poll may result in multiple readiness events
//! being returned for a single `Evented` handle. For example, if a TCP socket
//! becomes both readable and writable, it may be possible for a single
//! readiness event to be returned with both [readable] and [writable] readiness
//! **OR** two separate events may be returned, one with readable set and one
//! with writable set.
//!
//! [`Source::poll`]: crate::event::Source::poll
//! [readable]: crate::os::Interests::READABLE
//! [writable]: crate::os::Interests::WRITABLE
//!
//! ### Registering handles
//!
//! Unless otherwise noted, it should be assumed that types implementing
//! [`Evented`] will never become ready unless they are registered with
//! `OsQueue`.
//!
//! For example:
//!
//! ```
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! use std::thread;
//! use std::time::Duration;
//!
//! use gaea::event;
//! use gaea::net::TcpStream;
//! use gaea::os::{OsQueue, RegisterOption};
//!
//! let address = "216.58.193.100:80".parse()?;
//! let mut stream = TcpStream::connect(address)?;
//!
//! // This actually does nothing towards connecting the TCP stream.
//! thread::sleep(Duration::from_secs(1));
//!
//! let mut os_queue = OsQueue::new()?;
//!
//! // The connect is not guaranteed to have started until it is registered at
//! // this point.
//! os_queue.register(&mut stream, event::Id(0), TcpStream::INTERESTS, RegisterOption::EDGE)?;
//! #     Ok(())
//! # }
//! ```
//!
//! ### Timeout granularity
//!
//! The timeout provided to [`event::Source::blocking_poll`] will be rounded
//! up to the system clock granularity (usually 1ms), and kernel scheduling
//! delays mean that the blocking interval may be overrun by a small amount.
//!
//! ### Interrupts while polling
//!
//! Interrupts (`EINTR` in C and `io::ErrorKind::Interrupted` in Rust) are
//! **not** handled, they are returned as errors. In most cases however these
//! can simply be ignored, but it's up to the user how to deal with the "error".
//!
//! # Implementation notes
//!
//! `OsQueue` is backed by a readiness event queue provided by the operating
//! system. On all platforms a call to [`Source::poll`] is mostly just a direct
//! system call. The following system implementations back `OsQueue`:
//!
//! | OS      | Selector |
//! |---------|----------|
//! | FreeBSD | [kqueue](https://www.freebsd.org/cgi/man.cgi?query=kqueue) |
//! | Linux   | [epoll](http://man7.org/linux/man-pages/man7/epoll.7.html) |
//! | macOS   | [kqueue](https://developer.apple.com/legacy/library/documentation/Darwin/Reference/ManPages/man2/kqueue.2.html) |
//! | NetBSD  | [kqueue](http://netbsd.gw.com/cgi-bin/man-cgi?kqueue) |
//! | OpenBSD | [kqueue](https://man.openbsd.org/kqueue) |
//!
//! On all supported platforms socket operations are handled by using the system
//! queue. Platform specific extensions (e.g. [`EventedFd`]) allow accessing
//! other features provided by individual system selectors.
//!
//! [`Eventedfd`]: crate::sys::unix::EventedFd
//! [`signalfd`]: http://man7.org/linux/man-pages/man2/signalfd.2.html

use std::io;
use std::time::Duration;

use log::trace;

use crate::{event, sys};

mod awakener;
mod evented;
mod interests;
mod option;

pub mod signals;

pub use self::awakener::Awakener;
pub use self::evented::Evented;
pub use self::interests::Interests;
pub use self::option::RegisterOption;
pub use self::signals::{Signal, SignalSet, Signals};

/// Readiness event queue backed by the OS.
///
/// This queue allows a program to monitor a large number of [`Evented`]
/// handles, waiting until one or more become "ready" for some class of
/// operations; e.g. [reading] or [writing]. An [`Evented`] type is considered
/// ready if it is possible to immediately perform a corresponding operation;
/// e.g. read or write.
///
/// To use this queue an [`Evented`] handle must first be registered using the
/// [`register`] method, supplying an associated id, readiness interests and
/// polling option. The [associated id] is used to associate a readiness event
/// with an `Evented` handle. The readiness [interests] defines which specific
/// operations on the handle to monitor for readiness. And the final argument,
/// [`RegisterOption`], defines how to deliver the readiness events, see
/// [`RegisterOption`] for more information.
///
/// See to [module documentation] for information.
///
/// [reading]: crate::event::Ready::READABLE
/// [writing]: crate::event::Ready::WRITABLE
/// [`register`]: OsQueue::register
/// [associated id]: event::Id
/// [interests]: Interests
/// [module documentation]: crate::os
#[derive(Debug)]
pub struct OsQueue {
    selector: sys::Selector,
}

impl OsQueue {
    /// Create a new OS backed readiness event queue.
    ///
    /// This function will make a syscall to the operating system to create the
    /// system selector. If this syscall fails it will return the error.
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use std::io;
    /// use std::time::Duration;
    ///
    /// use gaea::os::OsQueue;
    /// use gaea::poll;
    ///
    /// // Create a new OS backed readiness event queue.
    /// let mut os_queue = OsQueue::new()?;
    ///
    /// // Create an event sink.
    /// let mut events = Vec::new();
    ///
    /// // Poll the queue for new readiness events.
    /// // But since no `Evented` handles have been registered we'll receive no
    /// // events.
    /// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, Some(Duration::from_millis(500)))?;
    /// #     Ok(())
    /// # }
    /// ```
    pub fn new() -> io::Result<OsQueue> {
        sys::Selector::new().map(|selector| OsQueue { selector })
    }

    /// Register an [`Evented`] handle with the `OsQueue`.
    ///
    /// Once registered, the [`Evented`] handle will be monitored for readiness
    /// state changes. When it notices a state change, it will return a
    /// readiness event for the handle the next time the queue is [`polled`].
    ///
    /// [`polled`]: crate::poll
    ///
    /// # Arguments
    ///
    /// `handle`: This is the handle that the `OsQueue` should monitor for
    /// readiness state changes.
    ///
    /// `id`: The caller picks a id to associate with the handle. When [`poll`]
    /// returns an [event] for the handle, this id is [included]. This allows
    /// the caller to map the event to its handle. The id associated with the
    /// `Evented` handle can be changed at any time by calling [`reregister`].
    ///
    /// `interests`: Specifies which operations `OsQueue` should monitor for
    /// readiness. `OsQueue` will only return readiness events for operations
    /// specified by this argument. If a socket is registered with [readable]
    /// interests and the socket becomes writable, no event will be returned
    /// from [`poll`]. The readiness interests for an `Evented` handle can be
    /// changed at any time by calling [`reregister`]. Most types that
    /// implemented [`Evented`] have a associated constant named `INTERESTS`
    /// which provide a sane interest for that type, e.g. [`TcpStream`
    /// interests] are readable and writable.
    ///
    /// `opt`: Specifies the registration option. Just like the interests and
    /// id, the option can be changed for an `Evented` handle at any time by
    /// calling [`reregister`].
    ///
    /// [`poll`]: crate::poll
    /// [event]: crate::event::Event
    /// [included]: crate::event::Event::id
    /// [`reregister`]: OsQueue::reregister
    /// [readable]: Interests::READABLE
    /// [`TcpStream` interests]: crate::net::TcpStream::INTERESTS
    ///
    /// # Notes
    ///
    /// Unless otherwise specified, the caller should assume that once an
    /// `Evented` handle is registered with a `OsQueue` instance, it is bound to
    /// that `OsQueue` for the lifetime of the `Evented` handle. This remains
    /// true even if the `Evented` handle is [deregistered].
    ///
    /// [deregistered]: OsQueue::deregister
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use std::io;
    ///
    /// use gaea::net::TcpStream;
    /// use gaea::os::{OsQueue, RegisterOption};
    /// use gaea::{event, poll};
    ///
    /// // Create a new `OsQueue` as well a containers for the events.
    /// let mut os_queue = OsQueue::new()?;
    /// let mut events = Vec::new();
    ///
    /// // Create a TCP connection. `TcpStream` implements the `Evented` trait.
    /// let address = "216.58.193.100:80".parse()?;
    /// let mut stream = TcpStream::connect(address)?;
    ///
    /// // Register the connection with queue.
    /// os_queue.register(&mut stream, event::Id(0), TcpStream::INTERESTS, RegisterOption::EDGE)?;
    ///
    /// // Run the event loop.
    /// loop {
    ///     poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
    ///
    ///     for event in events.drain(..) {
    ///         if event.id() == event::Id(0) {
    ///             // The TCP connection is (likely) ready for use.
    ///             # return Ok(());
    ///         }
    ///     }
    /// }
    /// # }
    /// ```
    pub fn register<E>(&mut self, handle: &mut E, id: event::Id, interests: Interests, opt: RegisterOption) -> io::Result<()>
        where E: Evented + ?Sized,
    {
        trace!("registering handle: id={}, interests={:?}, opt={:?}", id, interests, opt);
        handle.register(self, id, interests, opt)
    }

    /// Re-register an `Evented` handle with `OsQueue`.
    ///
    /// Re-registering an `Evented` handle allows changing the details of the
    /// registration. Specifically, it allows updating the associated `id`,
    /// `interests`, and `opt` specified in previous `register` and `reregister`
    /// calls.
    ///
    /// The `reregister` arguments **fully override** the previous values. In
    /// other words, if a socket is registered with [readable] interest and the
    /// call to `reregister` specifies only [writable], then read interest is no
    /// longer monitored for the handle.
    ///
    /// The `Evented` handle must have previously been registered with this
    /// `OsQueue` otherwise the call to `reregister` may return an error.
    ///
    /// See the [`register`] documentation for details about the function
    /// arguments.
    ///
    /// [readable]: Interests::READABLE
    /// [writable]: Interests::WRITABLE
    /// [`register`]: OsQueue::register
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use std::io;
    ///
    /// use gaea::{event, poll};
    /// use gaea::net::TcpStream;
    /// use gaea::os::{Interests, RegisterOption, OsQueue};
    ///
    /// let mut os_queue = OsQueue::new()?;
    /// let mut events = Vec::new();
    ///
    /// // Create a TCP connection. `TcpStream` implements the `Evented` trait.
    /// let address = "216.58.193.100:80".parse()?;
    /// let mut stream = TcpStream::connect(address)?;
    ///
    /// // Register the connection with `OsQueue`, only with readable interest.
    /// os_queue.register(&mut stream, event::Id(0), Interests::READABLE, RegisterOption::EDGE)?;
    ///
    /// // Reregister the connection specifying a different id and write interest
    /// // instead. `RegisterOption::EDGE` must be specified even though that value
    /// // is not being changed.
    /// os_queue.reregister(&mut stream, event::Id(2), Interests::WRITABLE, RegisterOption::EDGE)?;
    ///
    /// // Run the event loop.
    /// loop {
    ///     poll::<_, io::Error>(&mut [&mut os_queue], &mut events, None)?;
    ///
    ///     for event in events.drain(..) {
    ///         if event.id() == event::Id(2) {
    ///             // The TCP connection is (likely) ready for use.
    ///             # return Ok(());
    ///         } else if event.id() == event::Id(0) {
    ///             // We won't receive events with the old id anymore.
    ///             unreachable!();
    ///         }
    ///     }
    /// }
    /// # }
    /// ```
    pub fn reregister<E>(&mut self, handle: &mut E, id: event::Id, interests: Interests, opt: RegisterOption) -> io::Result<()>
        where E: Evented + ?Sized,
    {
        trace!("reregistering handle: id={}, interests={:?}, opt={:?}", id, interests, opt);
        handle.reregister(self, id, interests, opt)
    }

    /// Deregister an `Evented` handle from `OsQueue`.
    ///
    /// When an `Evented` handle is deregistered, the handle will no longer be
    /// monitored for readiness state changes. Unlike disabling handles with
    /// [`oneshot`], deregistering clears up any internal resources needed to
    /// track the handle.
    ///
    /// A handle can be registered again using [`register`] after it has been
    /// deregistered; however, it must be passed back to the **same** `OsQueue`.
    ///
    /// # Notes
    ///
    /// Calling [`reregister`] after `deregister` may be work on some platforms
    /// but not all. To properly re-register a handle after deregistering use
    /// `register`, this works on all platforms.
    ///
    /// [`oneshot`]: RegisterOption::ONESHOT
    /// [`register`]: OsQueue::register
    /// [`reregister`]: OsQueue::reregister
    ///
    /// # Examples
    ///
    /// ```
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// use std::io;
    /// use std::time::Duration;
    ///
    /// use gaea::{event, poll};
    /// use gaea::net::TcpStream;
    /// use gaea::os::{OsQueue, RegisterOption};
    ///
    /// let mut os_queue = OsQueue::new()?;
    /// let mut events = Vec::new();
    ///
    /// // Create a TCP connection. `TcpStream` implements the `Evented` trait.
    /// let address = "216.58.193.100:80".parse()?;
    /// let mut stream = TcpStream::connect(address)?;
    ///
    /// // Register the connection with `OsQueue`.
    /// os_queue.register(&mut stream, event::Id(0), TcpStream::INTERESTS, RegisterOption::EDGE)?;
    ///
    /// // Do stuff with the connection etc.
    ///
    /// // Deregister it so the resources can be cleaned up.
    /// os_queue.deregister(&mut stream)?;
    ///
    /// // Set a timeout because we shouldn't receive any events anymore.
    /// poll::<_, io::Error>(&mut [&mut os_queue], &mut events, Some(Duration::from_millis(100)))?;
    /// assert!(events.is_empty());
    /// #     Ok(())
    /// # }
    /// ```
    pub fn deregister<E>(&mut self, handle: &mut E) -> io::Result<()>
        where E: Evented + ?Sized,
    {
        trace!("deregistering handle");
        handle.deregister(self)
    }

    /// Get access to the system selector. Used by platform specific code, e.g.
    /// `EventedFd`.
    pub(crate) fn selector(&self) -> &sys::Selector {
        &self.selector
    }
}

impl<ES, E> event::Source<ES, E> for OsQueue
    where ES: event::Sink,
          E: From<io::Error>,
{
    fn max_timeout(&self) -> Option<Duration> {
        // Can't tell if an event is available.
        None
    }

    fn poll(&mut self, event_sink: &mut ES) -> Result<(), E> {
        self.blocking_poll(event_sink, Some(Duration::from_millis(0)))
    }

    fn blocking_poll(&mut self, event_sink: &mut ES, timeout: Option<Duration>) -> Result<(), E> {
        trace!("polling OS queue: timeout={:?}", timeout);
        self.selector.select(event_sink, timeout)
            .map_err(Into::into)
    }
}